diff --git a/Dockerfile b/Dockerfile index 5a5f811..cbd3966 100644 --- a/Dockerfile +++ b/Dockerfile @@ -33,5 +33,5 @@ COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certifi ENV HERMES_CONFIG_PATH=/etc/hermes/hermes.yaml WORKDIR /etc/hermes USER 65532:65532 -EXPOSE 8080 +EXPOSE 8080 1080 ENTRYPOINT ["/usr/local/bin/hermes"] diff --git a/README.md b/README.md index 7174f7b..2feecd6 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,8 @@ Hermes is a minimal HTTP/1.1 forward proxy written in Rust. It supports streamin - Absolute-form HTTP forwarding (`GET http://host/path`) - HTTPS and arbitrary TCP tunneling with `CONNECT host:port` +- Optional SOCKS5 CONNECT listener with no-auth and username/password authentication +- SOCKS5 IPv4, IPv6, and domain destinations with shared ACL enforcement - Streaming request and response bodies - Upstream connection pooling and configurable timeouts - Ordered allow/deny ACLs using Basic authentication, client IP networks, and destination domains @@ -67,7 +69,9 @@ The Gitea workflow in `.gitea/workflows/build.yaml` builds and publishes `linux/ ```yaml listen_addr: "127.0.0.1:8080" +socks5_listen_addr: "127.0.0.1:1080" connect_timeout_ms: 10000 +socks5_handshake_timeout_ms: 10000 upstream_timeout_ms: 30000 telemetry: @@ -110,7 +114,9 @@ acl: | Field | Default | Description | | --- | --- | --- | | `listen_addr` | `127.0.0.1:8080` | Downstream listen address | +| `socks5_listen_addr` | `null` | Optional SOCKS5 listen address; disabled when omitted | | `connect_timeout_ms` | `10000` | TCP connection timeout for `CONNECT` and the HTTP connector | +| `socks5_handshake_timeout_ms` | `10000` | Timeout for SOCKS5 negotiation and CONNECT request parsing | | `upstream_timeout_ms` | `30000` | Time limit for receiving upstream HTTP response headers | | `telemetry` | see OpenTelemetry below | Logging, resource, and OTLP configuration | | `acl.default_action` | `deny` | Action when no rule matches | @@ -200,6 +206,35 @@ Password verification is synchronous and inexpensive, and digest comparisons are **Basic authentication is not encrypted in transit.** A client sending Basic credentials over cleartext HTTP exposes them to the network. Use Basic authentication only on localhost or a trusted private network unless downstream TLS is added. Do not expose Hermes publicly without appropriate network controls. +## SOCKS5 + +Set `socks5_listen_addr` to enable a separate SOCKS5 listener alongside the HTTP proxy: + +```yaml +listen_addr: "127.0.0.1:8080" +socks5_listen_addr: "127.0.0.1:1080" +socks5_handshake_timeout_ms: 10000 +``` + +The listener supports SOCKS5 CONNECT with IPv4, IPv6, and domain destinations. Domain destinations are resolved by Hermes; use `--socks5-hostname` with curl to request proxy-side DNS resolution: + +```sh +# Client-side DNS with an IP destination +curl --socks5 127.0.0.1:1080 http://198.51.100.10/ + +# Proxy-side DNS with a domain destination +curl --socks5-hostname 127.0.0.1:1080 http://example.com/ + +# Username/password authentication using the configured ACL user +curl --socks5-hostname 127.0.0.1:1080 \ + --proxy-user 'alice:your-password' \ + http://example.com/ +``` + +SOCKS5 reuses the ACL's salted password hashes and applies `users`, `networks`, and `domains` before upstream dialing. Username/password negotiation is selected when required by the ACL; otherwise no-auth is used. Authentication is cleartext on the network, so restrict the listener to localhost or trusted networks. ACL denial returns SOCKS reply `0x02` before any upstream connection. + +Only CONNECT is supported. SOCKS4, BIND, and UDP ASSOCIATE are not implemented. Docker users can publish the listener with `-p 127.0.0.1:1080:1080` when it is enabled. + ## OpenTelemetry Hermes exports traces, metrics, and logs independently to an OpenTelemetry Collector using OTLP/HTTP (protobuf). All settings come from `telemetry` in YAML; `OTEL_*` environment variables are not read. @@ -247,6 +282,7 @@ In another terminal, run Hermes and send a request through it. The collector wil - `proxy.upstream`: upstream HTTP client operation - `proxy.connect`: `CONNECT` handshake and tunnel lifetime - `proxy.upstream_connect`: upstream TCP connection establishment +- `socks5.connect`: SOCKS5 negotiation, authorization, upstream connection, and tunnel lifetime For plain HTTP, incoming W3C `traceparent` and `tracestate` headers establish the parent context, and a new context is injected upstream. Trace context cannot be propagated inside an encrypted `CONNECT` tunnel. Request spans include the ACL decision, reason, and matched rule name without credentials. @@ -264,7 +300,7 @@ Hermes uses standard HTTP instrument names where applicable and adds proxy-speci - `hermes.proxy.tunnel.bytes` - `hermes.proxy.acl.decisions` -Metric attributes are deliberately bounded. Raw URLs, destination hostnames, usernames, rule names, and client addresses are not metric attributes. +Metric attributes are deliberately bounded. Tunnel and ACL metrics include `proxy.protocol` to distinguish HTTP, HTTP CONNECT, and SOCKS5. Raw URLs, destination hostnames, usernames, rule names, and client addresses are not metric attributes. ### Logs diff --git a/examples/hermes.yaml b/examples/hermes.yaml index bccc9b6..f2170d0 100644 --- a/examples/hermes.yaml +++ b/examples/hermes.yaml @@ -1,5 +1,8 @@ listen_addr: "127.0.0.1:8080" +# To enable SOCKS5, use "127.0.0.1:1080" or another non-conflicting address. +socks5_listen_addr: null connect_timeout_ms: 10000 +socks5_handshake_timeout_ms: 10000 upstream_timeout_ms: 30000 telemetry: diff --git a/src/acl.rs b/src/acl.rs index 5447361..2ee3054 100644 --- a/src/acl.rs +++ b/src/acl.rs @@ -248,6 +248,38 @@ impl Acl { }) } + pub fn requires_user_password(&self) -> bool { + self.default_action == Action::Deny && self.has_user_rules + } + + pub fn authenticate(&self, username: &str, password: &[u8]) -> bool { + self.users + .get(username) + .is_some_and(|user| user.password_hash.verify(password)) + } + + pub fn authorize_identity( + &self, + peer: IpAddr, + destination: Option<&DestinationHost>, + authenticated_user: Option<&str>, + ) -> AclOutcome { + for rule in &self.rules { + if rule_matches(rule, peer, authenticated_user, destination) { + return AclOutcome::rule(rule); + } + } + + if authenticated_user.is_none() + && self.default_action == Action::Deny + && self.has_user_rules + { + return AclOutcome::challenge("authentication_required"); + } + + AclOutcome::default(self.default_action) + } + pub fn authorize( &self, peer: IpAddr, @@ -262,10 +294,7 @@ impl Acl { let authenticated_user = match credentials { Some(credentials) => { - let Some(user) = self.users.get(&credentials.username) else { - return AclOutcome::challenge("invalid_credentials"); - }; - if !user.password_hash.verify(credentials.password.as_slice()) { + if !self.authenticate(&credentials.username, credentials.password.as_slice()) { return AclOutcome::challenge("invalid_credentials"); } Some(credentials.username) @@ -273,20 +302,7 @@ impl Acl { None => None, }; - for rule in &self.rules { - if rule_matches(rule, peer, authenticated_user.as_deref(), destination) { - return AclOutcome::rule(rule); - } - } - - if authenticated_user.is_none() - && self.default_action == Action::Deny - && self.has_user_rules - { - return AclOutcome::challenge("authentication_required"); - } - - AclOutcome::default(self.default_action) + self.authorize_identity(peer, destination, authenticated_user.as_deref()) } } diff --git a/src/config.rs b/src/config.rs index 05e4b45..174341c 100644 --- a/src/config.rs +++ b/src/config.rs @@ -9,11 +9,14 @@ use crate::acl::{Acl, AclError}; const DEFAULT_CONNECT_TIMEOUT_MS: u64 = 10_000; const DEFAULT_UPSTREAM_TIMEOUT_MS: u64 = 30_000; +const DEFAULT_SOCKS5_HANDSHAKE_TIMEOUT_MS: u64 = 10_000; #[derive(Clone)] pub struct Config { pub listen_addr: SocketAddr, + pub socks5_listen_addr: Option, pub connect_timeout: Duration, + pub socks5_handshake_timeout: Duration, pub upstream_timeout: Duration, pub telemetry: TelemetryConfig, pub(crate) acl: Arc, @@ -37,11 +40,19 @@ impl Config { if file.upstream_timeout_ms == 0 { return Err(ConfigError::ZeroDuration("upstream_timeout_ms")); } + if file.socks5_handshake_timeout_ms == 0 { + return Err(ConfigError::ZeroDuration("socks5_handshake_timeout_ms")); + } + if file.socks5_listen_addr == Some(file.listen_addr) { + return Err(ConfigError::ListenerConflict); + } file.telemetry.validate()?; Ok(Self { listen_addr: file.listen_addr, + socks5_listen_addr: file.socks5_listen_addr, connect_timeout: Duration::from_millis(file.connect_timeout_ms), + socks5_handshake_timeout: Duration::from_millis(file.socks5_handshake_timeout_ms), upstream_timeout: Duration::from_millis(file.upstream_timeout_ms), telemetry: file.telemetry, acl: Arc::new(Acl::from_config(file.acl)?), @@ -54,10 +65,14 @@ impl Config { struct FileConfig { #[serde(default = "default_listen_addr")] listen_addr: SocketAddr, + #[serde(default)] + socks5_listen_addr: Option, #[serde(default = "default_connect_timeout_ms")] connect_timeout_ms: u64, #[serde(default = "default_upstream_timeout_ms")] upstream_timeout_ms: u64, + #[serde(default = "default_socks5_handshake_timeout_ms")] + socks5_handshake_timeout_ms: u64, #[serde(default)] telemetry: TelemetryConfig, acl: AclConfig, @@ -75,6 +90,10 @@ const fn default_upstream_timeout_ms() -> u64 { DEFAULT_UPSTREAM_TIMEOUT_MS } +const fn default_socks5_handshake_timeout_ms() -> u64 { + DEFAULT_SOCKS5_HANDSHAKE_TIMEOUT_MS +} + #[derive(Clone, Deserialize)] #[serde(default, deny_unknown_fields)] pub struct TelemetryConfig { @@ -290,6 +309,8 @@ pub enum ConfigError { InvalidHeader(String), #[error("telemetry.service_name must not be empty")] EmptyServiceName, + #[error("listen_addr and socks5_listen_addr must differ")] + ListenerConflict, #[error(transparent)] Acl(#[from] AclError), } @@ -314,6 +335,8 @@ acl: assert_eq!(config.listen_addr, default_listen_addr()); assert_eq!(config.connect_timeout, Duration::from_secs(10)); assert_eq!(config.upstream_timeout, Duration::from_secs(30)); + assert_eq!(config.socks5_listen_addr, None); + assert_eq!(config.socks5_handshake_timeout, Duration::from_secs(10)); assert_eq!(config.telemetry.service_name, "hermes"); assert_eq!(config.telemetry.log_filter, "info"); assert_eq!(config.telemetry.otlp.endpoint, "http://127.0.0.1:4318"); @@ -453,5 +476,11 @@ acl: let zero = Config::from_yaml("connect_timeout_ms: 0\nacl: {}\n"); assert!(matches!(zero, Err(ConfigError::ZeroDuration(_)))); + + let zero_socks = Config::from_yaml("socks5_handshake_timeout_ms: 0\nacl: {}\n"); + assert!(matches!(zero_socks, Err(ConfigError::ZeroDuration(_)))); + + let conflict = Config::from_yaml("socks5_listen_addr: \"127.0.0.1:8080\"\nacl: {}\n"); + assert!(matches!(conflict, Err(ConfigError::ListenerConflict))); } } diff --git a/src/lib.rs b/src/lib.rs index 12bccc0..ff8db96 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,6 +4,8 @@ pub mod telemetry; mod metrics; mod proxy; +mod socks5; +mod tunnel; use std::error::Error; diff --git a/src/metrics.rs b/src/metrics.rs index 0995694..8dd27b4 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -128,42 +128,53 @@ impl Metrics { ); } - pub fn acl_decision(&self, decision: &str, reason: &str) { + pub fn acl_decision(&self, protocol: &str, decision: &str, reason: &str) { self.acl_decisions.add( 1, &[ + KeyValue::new("proxy.protocol", protocol.to_owned()), KeyValue::new("acl.decision", decision.to_owned()), KeyValue::new("acl.reason", reason.to_owned()), ], ); } - pub fn tunnel_started(&self) -> ActiveTunnel { - self.active_tunnels.add(1, &[]); + pub fn tunnel_started(&self, protocol: &'static str) -> ActiveTunnel { + self.active_tunnels + .add(1, &[KeyValue::new("proxy.protocol", protocol)]); ActiveTunnel { metrics: self.clone(), + protocol, } } pub fn tunnel_finished( &self, + protocol: &str, duration: Duration, client_to_upstream: u64, upstream_to_client: u64, error_type: Option<&str>, ) { - let attributes = error_type - .map(|value| vec![KeyValue::new("error.type", value.to_owned())]) - .unwrap_or_default(); + let mut attributes = vec![KeyValue::new("proxy.protocol", protocol.to_owned())]; + if let Some(error_type) = error_type { + attributes.push(KeyValue::new("error.type", error_type.to_owned())); + } self.tunnel_duration .record(duration.as_secs_f64(), &attributes); self.tunnel_bytes.add( client_to_upstream, - &[KeyValue::new("direction", "client_to_upstream")], + &[ + KeyValue::new("proxy.protocol", protocol.to_owned()), + KeyValue::new("direction", "client_to_upstream"), + ], ); self.tunnel_bytes.add( upstream_to_client, - &[KeyValue::new("direction", "upstream_to_client")], + &[ + KeyValue::new("proxy.protocol", protocol.to_owned()), + KeyValue::new("direction", "upstream_to_client"), + ], ); } } @@ -180,10 +191,13 @@ impl Drop for ActiveRequest { pub(crate) struct ActiveTunnel { metrics: Metrics, + protocol: &'static str, } impl Drop for ActiveTunnel { fn drop(&mut self) { - self.metrics.active_tunnels.add(-1, &[]); + self.metrics + .active_tunnels + .add(-1, &[KeyValue::new("proxy.protocol", self.protocol)]); } } diff --git a/src/proxy.rs b/src/proxy.rs index 5c0bf85..adb9b77 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -20,7 +20,7 @@ use hyper_util::{ use opentelemetry::global; use tokio::{ net::{TcpListener, TcpStream}, - task::JoinSet, + task::{JoinHandle, JoinSet}, time::timeout, }; use tracing::{Instrument, Span, debug, error, field, info, info_span, warn}; @@ -31,7 +31,9 @@ use crate::{ acl::{Acl, Decision, DestinationHost}, config::Config, metrics::Metrics, + socks5::{self, SocksState}, telemetry::{HeaderExtractor, HeaderInjector, remove_hop_by_hop_headers}, + tunnel::{self, TunnelProtocol}, }; type ProxyBody = UnsyncBoxBody; @@ -66,6 +68,14 @@ impl ProxyState { pub async fn run(config: Config) -> Result<(), BoxError> { let listener = TcpListener::bind(config.listen_addr).await?; let state = ProxyState::new(&config); + let mut socks_task = match config.socks5_listen_addr { + Some(address) => { + let listener = TcpListener::bind(address).await?; + let state = SocksState::new(&config); + Some(tokio::spawn(socks5::run(listener, state))) + } + None => None, + }; let mut connections = JoinSet::new(); let shutdown = tokio::signal::ctrl_c(); tokio::pin!(shutdown); @@ -96,13 +106,33 @@ pub async fn run(config: Config) -> Result<(), BoxError> { error!(error.message = %error, "connection task failed"); } } + completed = wait_for_socks(&mut socks_task) => { + match completed { + Ok(Ok(())) => error!("SOCKS5 listener stopped unexpectedly"), + Ok(Err(error)) => error!(error.message = %error, "SOCKS5 listener failed"), + Err(error) => error!(error.message = %error, "SOCKS5 listener task failed"), + } + break; + } } } + if let Some(task) = socks_task { + task.abort(); + } info!("proxy listener stopped"); Ok(()) } +async fn wait_for_socks( + task: &mut Option>>, +) -> Result, tokio::task::JoinError> { + match task { + Some(task) => task.await, + None => std::future::pending().await, + } +} + async fn serve_connection(stream: TcpStream, peer: SocketAddr, state: ProxyState) { let peer_ip = peer.ip(); let service = service_fn(move |request| { @@ -188,9 +218,14 @@ fn authorize_request( .acl .authorize(peer, request.headers(), destination.as_ref()); + let protocol = if request.method() == Method::CONNECT { + "http_connect" + } else { + "http" + }; state .metrics - .acl_decision(outcome.decision.as_str(), outcome.reason); + .acl_decision(protocol, outcome.decision.as_str(), outcome.reason); Span::current().record("acl.decision", outcome.decision.as_str()); Span::current().record("acl.reason", outcome.reason); if let Some(rule) = &outcome.rule { @@ -424,41 +459,14 @@ async fn run_tunnel(request: Request, mut upstream: TcpStream, metrics } }; - let started = Instant::now(); - let _active = metrics.tunnel_started(); let mut downstream = TokioIo::new(upgraded); - - match tokio::io::copy_bidirectional(&mut downstream, &mut upstream).await { - Ok((client_to_upstream, upstream_to_client)) => { - Span::current().record( - "tunnel.bytes.client_to_upstream", - i64::try_from(client_to_upstream).unwrap_or(i64::MAX), - ); - Span::current().record( - "tunnel.bytes.upstream_to_client", - i64::try_from(upstream_to_client).unwrap_or(i64::MAX), - ); - metrics.tunnel_finished( - started.elapsed(), - client_to_upstream, - upstream_to_client, - None, - ); - info!( - tunnel.bytes.client_to_upstream = client_to_upstream, - tunnel.bytes.upstream_to_client = upstream_to_client, - tunnel.duration_ms = started.elapsed().as_millis(), - "CONNECT tunnel closed" - ); - } - Err(error) => { - Span::current().record("error.type", "tunnel_io"); - Span::current().record("otel.status_code", "ERROR"); - metrics.error("tunnel", "tunnel_io"); - metrics.tunnel_finished(started.elapsed(), 0, 0, Some("tunnel_io")); - error!(error.message = %error, "CONNECT tunnel failed"); - } - } + tunnel::relay( + &mut downstream, + &mut upstream, + &metrics, + TunnelProtocol::HttpConnect, + ) + .await; } fn finish_request( @@ -869,7 +877,9 @@ mod tests { let address = listener.local_addr().unwrap(); let state = ProxyState::new(&Config { listen_addr: address, + socks5_listen_addr: None, connect_timeout: Duration::from_secs(2), + socks5_handshake_timeout: Duration::from_secs(2), upstream_timeout: Duration::from_secs(2), telemetry: TelemetryConfig::default(), acl: Arc::new(acl), diff --git a/src/socks5.rs b/src/socks5.rs new file mode 100644 index 0000000..8dccff9 --- /dev/null +++ b/src/socks5.rs @@ -0,0 +1,892 @@ +use std::{ + io, + net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::Arc, + time::Duration, +}; + +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + task::JoinSet, + time::timeout, +}; +use tracing::{Instrument, Span, debug, error, field, info, info_span, warn}; +use zeroize::Zeroizing; + +use crate::{ + BoxError, + acl::{Acl, Decision, DestinationHost}, + config::Config, + metrics::Metrics, + tunnel::{self, TunnelProtocol}, +}; + +const SOCKS_VERSION: u8 = 5; +const AUTH_NONE: u8 = 0; +const AUTH_USER_PASSWORD: u8 = 2; +const AUTH_NO_ACCEPTABLE: u8 = 0xff; +const AUTH_VERSION: u8 = 1; +const COMMAND_CONNECT: u8 = 1; +const ADDRESS_IPV4: u8 = 1; +const ADDRESS_DOMAIN: u8 = 3; +const ADDRESS_IPV6: u8 = 4; +const REPLY_SUCCESS: u8 = 0; +const REPLY_GENERAL_FAILURE: u8 = 1; +const REPLY_NOT_ALLOWED: u8 = 2; +const REPLY_NETWORK_UNREACHABLE: u8 = 3; +const REPLY_HOST_UNREACHABLE: u8 = 4; +const REPLY_CONNECTION_REFUSED: u8 = 5; +const REPLY_TIMEOUT: u8 = 6; +const REPLY_COMMAND_UNSUPPORTED: u8 = 7; +const REPLY_ADDRESS_UNSUPPORTED: u8 = 8; + +#[derive(Clone)] +pub(crate) struct SocksState { + acl: Arc, + metrics: Metrics, + connect_timeout: Duration, + handshake_timeout: Duration, +} + +impl SocksState { + pub fn new(config: &Config) -> Self { + Self { + acl: config.acl.clone(), + metrics: Metrics::new(opentelemetry::global::meter("hermes")), + connect_timeout: config.connect_timeout, + handshake_timeout: config.socks5_handshake_timeout, + } + } +} + +pub async fn run(listener: TcpListener, state: SocksState) -> Result<(), BoxError> { + let address = listener.local_addr()?; + info!(listen.address = %address, "SOCKS5 proxy listening"); + let mut connections = JoinSet::new(); + + loop { + tokio::select! { + accepted = listener.accept() => { + match accepted { + Ok((stream, peer)) => { + let state = state.clone(); + let span = connection_span(peer); + connections.spawn(handle_connection(stream, peer, state).instrument(span)); + } + Err(error) => { + error!(error.message = %error, "failed to accept SOCKS5 connection"); + } + } + } + completed = connections.join_next(), if !connections.is_empty() => { + if let Some(Err(error)) = completed { + error!(error.message = %error, "SOCKS5 connection task failed"); + } + } + } + } +} + +fn connection_span(peer: SocketAddr) -> tracing::Span { + info_span!( + "socks5.connect", + otel.name = "SOCKS5 CONNECT", + otel.kind = "server", + client.address = %peer.ip(), + client.port = peer.port(), + server.address = field::Empty, + server.port = field::Empty, + socks5.auth_method = field::Empty, + socks5.reply_code = field::Empty, + error.type = field::Empty, + acl.decision = field::Empty, + acl.reason = field::Empty, + acl.rule = field::Empty, + tunnel.bytes.client_to_upstream = field::Empty, + tunnel.bytes.upstream_to_client = field::Empty, + otel.status_code = field::Empty, + ) +} + +async fn handle_connection(mut stream: TcpStream, peer: SocketAddr, state: SocksState) { + let negotiated = match timeout(state.handshake_timeout, negotiate(&mut stream, &state)).await { + Ok(Ok(negotiated)) => negotiated, + Ok(Err(error)) => { + fail(&mut stream, error, &state.metrics).await; + return; + } + Err(_) => { + fail( + &mut stream, + SocksError::new(None, "handshake_timeout", "SOCKS5 negotiation timed out"), + &state.metrics, + ) + .await; + return; + } + }; + + let request = match timeout(state.handshake_timeout, read_request(&mut stream)).await { + Ok(Ok(request)) => request, + Ok(Err(error)) => { + fail(&mut stream, error, &state.metrics).await; + return; + } + Err(_) => { + fail( + &mut stream, + SocksError::new( + Some(REPLY_TIMEOUT), + "handshake_timeout", + "SOCKS5 request timed out", + ), + &state.metrics, + ) + .await; + return; + } + }; + + Span::current().record("server.address", &request.host); + Span::current().record("server.port", request.port); + + let outcome = state.acl.authorize_identity( + peer.ip(), + Some(&request.destination), + negotiated.username.as_deref(), + ); + state + .metrics + .acl_decision("socks5", outcome.decision.as_str(), outcome.reason); + Span::current().record("acl.decision", outcome.decision.as_str()); + Span::current().record("acl.reason", outcome.reason); + if let Some(rule) = &outcome.rule { + Span::current().record("acl.rule", rule); + } + + if outcome.decision != Decision::Allow { + Span::current().record("error.type", "acl_denied"); + Span::current().record("otel.status_code", "ERROR"); + state.metrics.error("acl", "acl_denied"); + warn!( + acl.reason = outcome.reason, + acl.rule = outcome.rule.as_deref().unwrap_or_default(), + "ACL denied SOCKS5 connection" + ); + if let Err(error) = write_reply( + &mut stream, + REPLY_NOT_ALLOWED, + SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0)), + ) + .await + { + debug!(error.message = %error, "failed to write SOCKS5 ACL denial"); + } + return; + } + + debug!( + acl.reason = outcome.reason, + acl.rule = outcome.rule.as_deref().unwrap_or_default(), + "ACL allowed SOCKS5 connection" + ); + + let connect_span = info_span!( + "proxy.upstream_connect", + otel.name = "TCP CONNECT", + otel.kind = "client", + server.address = %request.host, + server.port = request.port, + error.type = field::Empty, + otel.status_code = field::Empty, + ); + let connected = async { + match request.target { + ConnectTarget::Socket(address) => { + timeout(state.connect_timeout, TcpStream::connect(address)).await + } + ConnectTarget::Domain(domain) => { + timeout( + state.connect_timeout, + TcpStream::connect((domain.as_str(), request.port)), + ) + .await + } + } + } + .instrument(connect_span) + .await; + + let mut upstream = match connected { + Ok(Ok(upstream)) => upstream, + Ok(Err(error)) => { + let reply = reply_for_io_error(&error); + fail( + &mut stream, + SocksError::new( + Some(reply), + "upstream_connect", + format!("upstream connection failed: {error}"), + ), + &state.metrics, + ) + .await; + return; + } + Err(_) => { + fail( + &mut stream, + SocksError::new( + Some(REPLY_TIMEOUT), + "upstream_timeout", + "upstream connection timed out", + ), + &state.metrics, + ) + .await; + return; + } + }; + + let bind = upstream + .local_addr() + .unwrap_or_else(|_| SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))); + if let Err(error) = write_reply(&mut stream, REPLY_SUCCESS, bind).await { + debug!(error.message = %error, "failed to write SOCKS5 success reply"); + return; + } + + tunnel::relay( + &mut stream, + &mut upstream, + &state.metrics, + TunnelProtocol::Socks5, + ) + .await; +} + +struct Negotiated { + username: Option, +} + +async fn negotiate(stream: &mut S, state: &SocksState) -> Result +where + S: AsyncRead + AsyncWrite + Unpin, +{ + if read_u8(stream).await? != SOCKS_VERSION { + return Err(SocksError::new( + None, + "invalid_version", + "unsupported SOCKS version", + )); + } + let method_count = read_u8(stream).await? as usize; + if method_count == 0 { + return Err(SocksError::new( + None, + "invalid_methods", + "SOCKS5 greeting contains no authentication methods", + )); + } + let mut methods = vec![0_u8; method_count]; + read_exact(stream, &mut methods).await?; + + let required = state.acl.requires_user_password(); + let method = if required { + AUTH_USER_PASSWORD + } else { + AUTH_NONE + }; + if !methods.contains(&method) { + stream + .write_all(&[SOCKS_VERSION, AUTH_NO_ACCEPTABLE]) + .await?; + return Err(SocksError::new( + None, + "no_acceptable_method", + "client offered no acceptable SOCKS5 authentication method", + )); + } + stream.write_all(&[SOCKS_VERSION, method]).await?; + Span::current().record( + "socks5.auth_method", + if method == AUTH_USER_PASSWORD { + "username_password" + } else { + "none" + }, + ); + + if method == AUTH_NONE { + return Ok(Negotiated { username: None }); + } + + if read_u8(stream).await? != AUTH_VERSION { + stream.write_all(&[AUTH_VERSION, 1]).await?; + return Err(SocksError::new( + None, + "invalid_auth_version", + "unsupported username/password authentication version", + )); + } + + let username_len = read_u8(stream).await? as usize; + if username_len == 0 { + return Err(SocksError::new( + None, + "invalid_credentials", + "SOCKS5 username must not be empty", + )); + } + let mut username = vec![0_u8; username_len]; + read_exact(stream, &mut username).await?; + let username = String::from_utf8(username).map_err(|_| { + SocksError::new( + None, + "invalid_credentials", + "SOCKS5 username is not valid UTF-8", + ) + })?; + + let password_len = read_u8(stream).await? as usize; + if password_len == 0 { + return Err(SocksError::new( + None, + "invalid_credentials", + "SOCKS5 password must not be empty", + )); + } + let mut password = Zeroizing::new(vec![0_u8; password_len]); + read_exact(stream, password.as_mut_slice()).await?; + + if !state.acl.authenticate(&username, password.as_slice()) { + stream.write_all(&[AUTH_VERSION, 1]).await?; + return Err(SocksError::new( + None, + "invalid_credentials", + "SOCKS5 username or password is invalid", + )); + } + stream.write_all(&[AUTH_VERSION, 0]).await?; + Ok(Negotiated { + username: Some(username), + }) +} + +struct SocksRequest { + destination: DestinationHost, + target: ConnectTarget, + host: String, + port: u16, +} + +enum ConnectTarget { + Socket(SocketAddr), + Domain(String), +} + +async fn read_request(stream: &mut S) -> Result +where + S: AsyncRead + Unpin, +{ + if read_u8(stream).await? != SOCKS_VERSION { + return Err(SocksError::new( + Some(REPLY_GENERAL_FAILURE), + "invalid_version", + "unsupported SOCKS version in CONNECT request", + )); + } + if read_u8(stream).await? != COMMAND_CONNECT { + return Err(SocksError::new( + Some(REPLY_COMMAND_UNSUPPORTED), + "unsupported_command", + "only SOCKS5 CONNECT is supported", + )); + } + if read_u8(stream).await? != 0 { + return Err(SocksError::new( + Some(REPLY_GENERAL_FAILURE), + "invalid_request", + "SOCKS5 reserved byte must be zero", + )); + } + + let address_type = read_u8(stream).await?; + let (destination, target, host) = match address_type { + ADDRESS_IPV4 => { + let mut octets = [0_u8; 4]; + read_exact(stream, &mut octets).await?; + let ip = IpAddr::from(octets); + let address = SocketAddr::new(ip, 0); + ( + DestinationHost::Ip(ip), + ConnectTarget::Socket(address), + ip.to_string(), + ) + } + ADDRESS_IPV6 => { + let mut octets = [0_u8; 16]; + read_exact(stream, &mut octets).await?; + let ip = IpAddr::from(octets); + let address = SocketAddr::new(ip, 0); + ( + DestinationHost::Ip(ip), + ConnectTarget::Socket(address), + ip.to_string(), + ) + } + ADDRESS_DOMAIN => { + let length = read_u8(stream).await? as usize; + if length == 0 { + return Err(SocksError::new( + Some(REPLY_GENERAL_FAILURE), + "invalid_destination", + "SOCKS5 domain must not be empty", + )); + } + let mut bytes = vec![0_u8; length]; + read_exact(stream, &mut bytes).await?; + let domain = String::from_utf8(bytes).map_err(|_| { + SocksError::new( + Some(REPLY_GENERAL_FAILURE), + "invalid_destination", + "SOCKS5 domain is not valid UTF-8", + ) + })?; + let destination = DestinationHost::parse(&domain).ok_or_else(|| { + SocksError::new( + Some(REPLY_GENERAL_FAILURE), + "invalid_destination", + "SOCKS5 domain is invalid", + ) + })?; + (destination, ConnectTarget::Domain(domain.clone()), domain) + } + _ => { + return Err(SocksError::new( + Some(REPLY_ADDRESS_UNSUPPORTED), + "unsupported_address_type", + "SOCKS5 address type is not supported", + )); + } + }; + + let port = read_u16(stream).await?; + if port == 0 { + return Err(SocksError::new( + Some(REPLY_GENERAL_FAILURE), + "invalid_destination", + "SOCKS5 destination port must not be zero", + )); + } + let target = match target { + ConnectTarget::Socket(mut address) => { + address.set_port(port); + ConnectTarget::Socket(address) + } + domain => domain, + }; + + Ok(SocksRequest { + destination, + target, + host, + port, + }) +} + +async fn write_reply(stream: &mut S, reply: u8, bind: SocketAddr) -> io::Result<()> +where + S: AsyncWrite + Unpin, +{ + Span::current().record("socks5.reply_code", reply); + let mut response = vec![SOCKS_VERSION, reply, 0]; + match bind.ip() { + IpAddr::V4(address) => { + response.push(ADDRESS_IPV4); + response.extend_from_slice(&address.octets()); + } + IpAddr::V6(address) => { + response.push(ADDRESS_IPV6); + response.extend_from_slice(&address.octets()); + } + } + response.extend_from_slice(&bind.port().to_be_bytes()); + stream.write_all(&response).await +} + +async fn fail(stream: &mut S, error: SocksError, metrics: &Metrics) +where + S: AsyncWrite + Unpin, +{ + Span::current().record("error.type", error.error_type); + Span::current().record("otel.status_code", "ERROR"); + metrics.error("handshake", error.error_type); + debug!(error.message = %error.message, "SOCKS5 connection failed"); + if let Some(reply) = error.reply { + if let Err(error) = + write_reply(stream, reply, SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))).await + { + debug!(error.message = %error, "failed to write SOCKS5 error reply"); + } + } +} + +fn reply_for_io_error(error: &io::Error) -> u8 { + match error.kind() { + io::ErrorKind::ConnectionRefused => REPLY_CONNECTION_REFUSED, + io::ErrorKind::NetworkUnreachable => REPLY_NETWORK_UNREACHABLE, + io::ErrorKind::HostUnreachable + | io::ErrorKind::AddrNotAvailable + | io::ErrorKind::NotFound => REPLY_HOST_UNREACHABLE, + _ => REPLY_GENERAL_FAILURE, + } +} + +#[derive(Debug)] +struct SocksError { + reply: Option, + error_type: &'static str, + message: String, +} + +impl SocksError { + fn new(reply: Option, error_type: &'static str, message: impl Into) -> Self { + Self { + reply, + error_type, + message: message.into(), + } + } +} + +impl From for SocksError { + fn from(error: io::Error) -> Self { + Self::new(None, "handshake_io", error.to_string()) + } +} + +async fn read_u8(stream: &mut S) -> io::Result +where + S: AsyncRead + Unpin, +{ + stream.read_u8().await +} + +async fn read_u16(stream: &mut S) -> io::Result +where + S: AsyncRead + Unpin, +{ + stream.read_u16().await +} + +async fn read_exact(stream: &mut S, buffer: &mut [u8]) -> io::Result<()> +where + S: AsyncRead + Unpin, +{ + stream.read_exact(buffer).await.map(|_| ()) +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use tokio::task::JoinHandle; + + use super::*; + use crate::config::{AclConfig, Action, RuleConfig, UserConfig}; + + const PASSWORD: &[u8] = b"correct horse battery staple"; + + fn test_state(acl: Acl) -> SocksState { + SocksState { + acl: Arc::new(acl), + metrics: Metrics::new(opentelemetry::global::meter("hermes-test")), + connect_timeout: Duration::from_secs(2), + handshake_timeout: Duration::from_secs(2), + } + } + + fn allow_all_acl() -> Acl { + Acl::from_config(AclConfig { + default_action: Action::Allow, + users: BTreeMap::new(), + rules: vec![], + }) + .unwrap() + } + + fn basic_acl() -> Acl { + Acl::from_config(AclConfig { + default_action: Action::Deny, + users: BTreeMap::from([( + "alice".to_owned(), + UserConfig { + password_hash: crate::acl::hash_password(PASSWORD).unwrap(), + }, + )]), + rules: vec![RuleConfig { + name: "alice".to_owned(), + action: Action::Allow, + users: vec!["alice".to_owned()], + networks: vec![], + domains: vec![], + }], + }) + .unwrap() + } + + fn localhost_domain_acl() -> Acl { + Acl::from_config(AclConfig { + default_action: Action::Deny, + users: BTreeMap::new(), + rules: vec![RuleConfig { + name: "localhost-domain".to_owned(), + action: Action::Allow, + users: vec![], + networks: vec!["127.0.0.0/8".to_owned()], + domains: vec!["localhost".to_owned()], + }], + }) + .unwrap() + } + + async fn spawn_socks(acl: Acl) -> (SocketAddr, JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let task = tokio::spawn(run(listener, test_state(acl))); + (address, task) + } + + async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let task = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = [0_u8; 4]; + stream.read_exact(&mut request).await.unwrap(); + assert_eq!(&request, b"ping"); + stream.write_all(b"pong").await.unwrap(); + stream.shutdown().await.unwrap(); + }); + (address, task) + } + + fn ipv4_request(address: SocketAddr) -> Vec { + let mut request = vec![SOCKS_VERSION, COMMAND_CONNECT, 0, ADDRESS_IPV4]; + let IpAddr::V4(ip) = address.ip() else { + panic!("expected IPv4 address"); + }; + request.extend_from_slice(&ip.octets()); + request.extend_from_slice(&address.port().to_be_bytes()); + request + } + + fn domain_request(host: &str, port: u16) -> Vec { + let mut request = vec![ + SOCKS_VERSION, + COMMAND_CONNECT, + 0, + ADDRESS_DOMAIN, + host.len() as u8, + ]; + request.extend_from_slice(host.as_bytes()); + request.extend_from_slice(&port.to_be_bytes()); + request + } + + async fn negotiate_without_auth(stream: &mut TcpStream) { + stream + .write_all(&[SOCKS_VERSION, 1, AUTH_NONE]) + .await + .unwrap(); + let mut response = [0_u8; 2]; + stream.read_exact(&mut response).await.unwrap(); + assert_eq!(response, [SOCKS_VERSION, AUTH_NONE]); + } + + async fn read_reply(stream: &mut TcpStream) -> u8 { + let mut header = [0_u8; 4]; + stream.read_exact(&mut header).await.unwrap(); + assert_eq!(header[0], SOCKS_VERSION); + assert_eq!(header[2], 0); + match header[3] { + ADDRESS_IPV4 => { + let mut bytes = [0_u8; 4]; + stream.read_exact(&mut bytes).await.unwrap(); + } + ADDRESS_IPV6 => { + let mut bytes = [0_u8; 16]; + stream.read_exact(&mut bytes).await.unwrap(); + } + other => panic!("unexpected reply address type {other}"), + } + let mut port = [0_u8; 2]; + stream.read_exact(&mut port).await.unwrap(); + header[1] + } + + async fn assert_ping_pong(stream: &mut TcpStream) { + stream.write_all(b"ping").await.unwrap(); + let mut response = [0_u8; 4]; + stream.read_exact(&mut response).await.unwrap(); + assert_eq!(&response, b"pong"); + } + + #[tokio::test] + async fn connects_ipv4_without_authentication() { + let (echo_addr, echo_task) = spawn_echo().await; + let (socks_addr, socks_task) = spawn_socks(allow_all_acl()).await; + let mut stream = TcpStream::connect(socks_addr).await.unwrap(); + negotiate_without_auth(&mut stream).await; + stream.write_all(&ipv4_request(echo_addr)).await.unwrap(); + assert_eq!(read_reply(&mut stream).await, REPLY_SUCCESS); + assert_ping_pong(&mut stream).await; + echo_task.await.unwrap(); + socks_task.abort(); + } + + #[tokio::test] + async fn connects_domain_and_applies_acl() { + let (echo_addr, echo_task) = spawn_echo().await; + let (socks_addr, socks_task) = spawn_socks(localhost_domain_acl()).await; + let mut stream = TcpStream::connect(socks_addr).await.unwrap(); + negotiate_without_auth(&mut stream).await; + stream + .write_all(&domain_request("localhost", echo_addr.port())) + .await + .unwrap(); + assert_eq!(read_reply(&mut stream).await, REPLY_SUCCESS); + assert_ping_pong(&mut stream).await; + echo_task.await.unwrap(); + socks_task.abort(); + } + + #[tokio::test] + async fn authenticates_username_password() { + let (echo_addr, echo_task) = spawn_echo().await; + let (socks_addr, socks_task) = spawn_socks(basic_acl()).await; + let mut stream = TcpStream::connect(socks_addr).await.unwrap(); + stream + .write_all(&[SOCKS_VERSION, 2, AUTH_NONE, AUTH_USER_PASSWORD]) + .await + .unwrap(); + let mut method = [0_u8; 2]; + stream.read_exact(&mut method).await.unwrap(); + assert_eq!(method, [SOCKS_VERSION, AUTH_USER_PASSWORD]); + + let mut authentication = vec![AUTH_VERSION, 5]; + authentication.extend_from_slice(b"alice"); + authentication.push(PASSWORD.len() as u8); + authentication.extend_from_slice(PASSWORD); + stream.write_all(&authentication).await.unwrap(); + let mut response = [0_u8; 2]; + stream.read_exact(&mut response).await.unwrap(); + assert_eq!(response, [AUTH_VERSION, 0]); + + stream.write_all(&ipv4_request(echo_addr)).await.unwrap(); + assert_eq!(read_reply(&mut stream).await, REPLY_SUCCESS); + assert_ping_pong(&mut stream).await; + echo_task.await.unwrap(); + socks_task.abort(); + } + + #[tokio::test] + async fn rejects_invalid_credentials_and_missing_auth_method() { + let (socks_addr, socks_task) = spawn_socks(basic_acl()).await; + + let mut no_auth = TcpStream::connect(socks_addr).await.unwrap(); + no_auth + .write_all(&[SOCKS_VERSION, 1, AUTH_NONE]) + .await + .unwrap(); + let mut method = [0_u8; 2]; + no_auth.read_exact(&mut method).await.unwrap(); + assert_eq!(method, [SOCKS_VERSION, AUTH_NO_ACCEPTABLE]); + + let mut wrong_password = TcpStream::connect(socks_addr).await.unwrap(); + wrong_password + .write_all(&[SOCKS_VERSION, 1, AUTH_USER_PASSWORD]) + .await + .unwrap(); + wrong_password.read_exact(&mut method).await.unwrap(); + assert_eq!(method, [SOCKS_VERSION, AUTH_USER_PASSWORD]); + let mut authentication = vec![AUTH_VERSION, 5]; + authentication.extend_from_slice(b"alice"); + authentication.push(5); + authentication.extend_from_slice(b"wrong"); + wrong_password.write_all(&authentication).await.unwrap(); + let mut response = [0_u8; 2]; + wrong_password.read_exact(&mut response).await.unwrap(); + assert_eq!(response, [AUTH_VERSION, 1]); + + socks_task.abort(); + } + + #[tokio::test] + async fn denies_acl_with_not_allowed_reply() { + let acl = Acl::from_config(AclConfig { + default_action: Action::Deny, + users: BTreeMap::new(), + rules: vec![], + }) + .unwrap(); + let (socks_addr, socks_task) = spawn_socks(acl).await; + let mut stream = TcpStream::connect(socks_addr).await.unwrap(); + negotiate_without_auth(&mut stream).await; + stream + .write_all(&ipv4_request("192.0.2.1:80".parse().unwrap())) + .await + .unwrap(); + assert_eq!(read_reply(&mut stream).await, REPLY_NOT_ALLOWED); + socks_task.abort(); + } + + #[tokio::test] + async fn reports_unsupported_command_and_address_type() { + let (socks_addr, socks_task) = spawn_socks(allow_all_acl()).await; + + let mut bind = TcpStream::connect(socks_addr).await.unwrap(); + negotiate_without_auth(&mut bind).await; + bind.write_all(&[SOCKS_VERSION, 2, 0, ADDRESS_IPV4, 127, 0, 0, 1, 0, 80]) + .await + .unwrap(); + assert_eq!(read_reply(&mut bind).await, REPLY_COMMAND_UNSUPPORTED); + + let mut invalid_address = TcpStream::connect(socks_addr).await.unwrap(); + negotiate_without_auth(&mut invalid_address).await; + invalid_address + .write_all(&[SOCKS_VERSION, COMMAND_CONNECT, 0, 9]) + .await + .unwrap(); + assert_eq!( + read_reply(&mut invalid_address).await, + REPLY_ADDRESS_UNSUPPORTED + ); + + socks_task.abort(); + } + + #[tokio::test] + async fn parses_fragmented_ipv6_request() { + let (mut client, mut server) = tokio::io::duplex(64); + let writer = tokio::spawn(async move { + let mut request = vec![SOCKS_VERSION, COMMAND_CONNECT, 0, ADDRESS_IPV6]; + request.extend_from_slice(&std::net::Ipv6Addr::LOCALHOST.octets()); + request.extend_from_slice(&8080_u16.to_be_bytes()); + for byte in request { + client.write_all(&[byte]).await.unwrap(); + } + }); + + let request = read_request(&mut server).await.unwrap(); + writer.await.unwrap(); + assert_eq!(request.host, "::1"); + assert_eq!(request.port, 8080); + assert!(matches!( + request.destination, + DestinationHost::Ip(IpAddr::V6(_)) + )); + } +} diff --git a/src/tunnel.rs b/src/tunnel.rs new file mode 100644 index 0000000..28fb05a --- /dev/null +++ b/src/tunnel.rs @@ -0,0 +1,78 @@ +use std::time::Instant; + +use tokio::io::{AsyncRead, AsyncWrite}; +use tracing::{Span, error, info}; + +use crate::metrics::Metrics; + +#[derive(Clone, Copy)] +pub(crate) enum TunnelProtocol { + HttpConnect, + Socks5, +} + +impl TunnelProtocol { + pub const fn as_str(self) -> &'static str { + match self { + Self::HttpConnect => "http_connect", + Self::Socks5 => "socks5", + } + } +} + +pub async fn relay( + downstream: &mut D, + upstream: &mut U, + metrics: &Metrics, + protocol: TunnelProtocol, +) where + D: AsyncRead + AsyncWrite + Unpin, + U: AsyncRead + AsyncWrite + Unpin, +{ + let started = Instant::now(); + let _active = metrics.tunnel_started(protocol.as_str()); + + match tokio::io::copy_bidirectional(downstream, upstream).await { + Ok((client_to_upstream, upstream_to_client)) => { + Span::current().record( + "tunnel.bytes.client_to_upstream", + i64::try_from(client_to_upstream).unwrap_or(i64::MAX), + ); + Span::current().record( + "tunnel.bytes.upstream_to_client", + i64::try_from(upstream_to_client).unwrap_or(i64::MAX), + ); + metrics.tunnel_finished( + protocol.as_str(), + started.elapsed(), + client_to_upstream, + upstream_to_client, + None, + ); + info!( + proxy.protocol = protocol.as_str(), + tunnel.bytes.client_to_upstream = client_to_upstream, + tunnel.bytes.upstream_to_client = upstream_to_client, + tunnel.duration_ms = started.elapsed().as_millis(), + "TCP tunnel closed" + ); + } + Err(error) => { + Span::current().record("error.type", "tunnel_io"); + Span::current().record("otel.status_code", "ERROR"); + metrics.error("tunnel", "tunnel_io"); + metrics.tunnel_finished( + protocol.as_str(), + started.elapsed(), + 0, + 0, + Some("tunnel_io"), + ); + error!( + proxy.protocol = protocol.as_str(), + error.message = %error, + "TCP tunnel failed" + ); + } + } +}