From 414e6fe31e0df5d0f84a4b24360b8fe22d2bcd89 Mon Sep 17 00:00:00 2001 From: Eric Holk Date: Thu, 23 Jul 2026 15:46:06 -0700 Subject: [PATCH] client: Extract proxy handshakes into a new proxy_handshake crate (#61427) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This consolidates the proxy code in `client/src/proxy` into a new `proxy_handshake` crate that implements the client side of HTTP CONNECT and SOCKS4/4a/5/5h handshakes sans-IO: the protocol logic is a pure state machine that never touches a socket, with thin drivers behind `futures-io` and `tokio` feature flags. `client` now drives the websocket's proxy tunnel through the tokio driver, and the `tokio-socks` dependency goes away. The main win is testability: the handshakes are now covered by synchronous byte-level tests on both sides of each exchange, plus a property test that the outcome is invariant under how the proxy's bytes are chunked across reads — the class of partial-read bug that's hard to hit with hand-written tests but easy to hit in production behind a buffering proxy. The crate is written as if it were standalone (no zed-internal dependencies, structured errors instead of `anyhow`), which is also why it grows both driver flavors rather than just the one `client` needs today. Two deliberate behavior changes, both converging on spec/curl semantics: - Proxy credentials embedded in URLs are now percent-decoded before use, per RFC 3986 §3.2.1. Previously they were sent in their encoded form, which only worked for credentials with no reserved characters. A password that literally contains something like `%40` and relied on the old behavior now needs the RFC-correct `%2540` spelling. - `NO_PROXY` is now honored for the collaboration connection, matching the exclusions reqwest already applies to the rest of Zed's HTTP traffic. TLS to the proxy itself (for `https://` proxy URLs) and DNS policy stay in `client`, since those are I/O concerns the sans-IO crate deliberately doesn't own. Release Notes: - Improved proxy support: the collaboration connection now respects `NO_PROXY` and percent-encoded proxy credentials. --- Cargo.lock | 20 +- Cargo.toml | 7 +- crates/client/Cargo.toml | 4 +- crates/client/src/client.rs | 8 +- crates/client/src/proxy.rs | 134 ++-- crates/client/src/proxy/http_proxy.rs | 193 ----- crates/client/src/proxy/socks_proxy.rs | 226 ------ crates/proxy_handshake/Cargo.toml | 34 + crates/proxy_handshake/LICENSE-APACHE | 1 + crates/proxy_handshake/src/futures_io.rs | 191 +++++ crates/proxy_handshake/src/handshake.rs | 708 ++++++++++++++++++ crates/proxy_handshake/src/proxy_handshake.rs | 393 ++++++++++ crates/proxy_handshake/src/tokio.rs | 139 ++++ 13 files changed, 1571 insertions(+), 487 deletions(-) delete mode 100644 crates/client/src/proxy/http_proxy.rs delete mode 100644 crates/client/src/proxy/socks_proxy.rs create mode 100644 crates/proxy_handshake/Cargo.toml create mode 120000 crates/proxy_handshake/LICENSE-APACHE create mode 100644 crates/proxy_handshake/src/futures_io.rs create mode 100644 crates/proxy_handshake/src/handshake.rs create mode 100644 crates/proxy_handshake/src/proxy_handshake.rs create mode 100644 crates/proxy_handshake/src/tokio.rs diff --git a/Cargo.lock b/Cargo.lock index 6b4c6950b3..300c9230fd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3188,7 +3188,6 @@ dependencies = [ "anyhow", "async-channel 2.5.0", "async-tungstenite", - "base64 0.22.1", "chrono", "clock", "cloud_api_client", @@ -3204,12 +3203,12 @@ dependencies = [ "gpui_tokio", "http_client", "http_client_tls", - "httparse", "log", "objc2-foundation 0.3.2", "parking_lot", "paths", "postage", + "proxy_handshake", "rand 0.9.4", "regex", "release_channel", @@ -3231,7 +3230,6 @@ dependencies = [ "tokio", "tokio-native-tls", "tokio-rustls 0.26.4", - "tokio-socks", "url", "util", "windows 0.61.3", @@ -14493,6 +14491,21 @@ dependencies = [ "thiserror 1.0.69", ] +[[package]] +name = "proxy_handshake" +version = "0.1.0" +dependencies = [ + "base64 0.22.1", + "futures 0.3.32", + "httparse", + "percent-encoding", + "piper", + "proptest", + "thiserror 2.0.17", + "tokio", + "url", +] + [[package]] name = "proxyvars" version = "0.2.0" @@ -19015,7 +19028,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d4770b8024672c1101b3f6733eab95b18007dbe0847a8afe341fcf79e06043f" dependencies = [ "either", - "futures-io", "futures-util", "thiserror 1.0.69", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 5741492485..44e6df8376 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -164,6 +164,7 @@ members = [ "crates/project_symbols", "crates/prompt_store", "crates/proto", + "crates/proxy_handshake", "crates/recent_projects", "crates/refineable", "crates/refineable/derive_refineable", @@ -428,6 +429,7 @@ project_panel = { path = "crates/project_panel" } project_symbols = { path = "crates/project_symbols" } prompt_store = { path = "crates/prompt_store" } proto = { path = "crates/proto" } +proxy_handshake = { path = "crates/proxy_handshake" } recent_projects = { path = "crates/recent_projects" } refineable = { path = "crates/refineable" } release_channel = { path = "crates/release_channel" } @@ -734,6 +736,7 @@ pet-fs = { git = "https://github.com/microsoft/python-environment-tools.git", re pet-poetry = { git = "https://github.com/microsoft/python-environment-tools.git", rev = "9e61a22af989fe54937bf07c9f9cff1bc53d9056" } pet-reporter = { git = "https://github.com/microsoft/python-environment-tools.git", rev = "9e61a22af989fe54937bf07c9f9cff1bc53d9056" } pet-virtualenv = { git = "https://github.com/microsoft/python-environment-tools.git", rev = "9e61a22af989fe54937bf07c9f9cff1bc53d9056" } +piper = "0.2" portable-pty = "0.9.0" postage = { version = "0.5", features = ["futures-traits"] } pretty_assertions = { version = "1.3.0", features = ["unstable"] } @@ -822,10 +825,6 @@ time = { version = "0.3", features = [ ] } tiny_http = "0.12" tokio = { version = "1" } -tokio-socks = { version = "0.5.2", default-features = false, features = [ - "futures-io", - "tokio", -] } toml = "0.8" toml_edit = { version = "0.22", default-features = false, features = [ "display", diff --git a/crates/client/Cargo.toml b/crates/client/Cargo.toml index 91c35a15bb..f2892ba611 100644 --- a/crates/client/Cargo.toml +++ b/crates/client/Cargo.toml @@ -19,7 +19,6 @@ test-support = ["clock/test-support", "collections/test-support", "gpui/test-sup anyhow.workspace = true async-channel.workspace = true async-tungstenite = { workspace = true, features = ["tokio", "tokio-rustls-manual-roots"] } -base64.workspace = true chrono = { workspace = true, features = ["serde"] } clock.workspace = true cloud_api_client.workspace = true @@ -35,11 +34,11 @@ gpui.workspace = true gpui_tokio.workspace = true http_client.workspace = true http_client_tls.workspace = true -httparse = "1.10" log.workspace = true parking_lot.workspace = true paths.workspace = true postage.workspace = true +proxy_handshake = { workspace = true, features = ["tokio"] } rand.workspace = true regex.workspace = true release_channel.workspace = true @@ -56,7 +55,6 @@ text.workspace = true thiserror.workspace = true time.workspace = true tiny_http.workspace = true -tokio-socks.workspace = true tokio.workspace = true url.workspace = true util.workspace = true diff --git a/crates/client/src/client.rs b/crates/client/src/client.rs index 2c8dfa8ba4..7273da7222 100644 --- a/crates/client/src/client.rs +++ b/crates/client/src/client.rs @@ -30,7 +30,7 @@ use gpui::{App, AsyncApp, Entity, Global, Task, TaskExt, WeakEntity, actions}; use http_client::{HttpClient, HttpClientWithUrl, http, read_proxy_from_env}; use parking_lot::{Mutex, RwLock}; use postage::watch; -use proxy::connect_proxy_stream; +use proxy::{connect_proxy_stream, excluded_from_proxy}; use rand::prelude::*; use release_channel::{AppVersion, ReleaseChannel}; use rpc::proto::{AnyTypedEnvelope, EnvelopedMessage, PeerId, RequestMessage}; @@ -1362,8 +1362,10 @@ impl Client { .zip(rpc_url.port_or_known_default()) .context("missing host in rpc url")?; Ok(match proxy { - Some(proxy) => connect_proxy_stream(&proxy, rpc_host).await?, - None => Box::new(TcpStream::connect(rpc_host).await?), + Some(proxy) if !excluded_from_proxy(rpc_host.0) => { + connect_proxy_stream(&proxy, rpc_host).await? + } + _ => Box::new(TcpStream::connect(rpc_host).await?), }) } }) diff --git a/crates/client/src/proxy.rs b/crates/client/src/proxy.rs index eb3812ca07..53328b6e09 100644 --- a/crates/client/src/proxy.rs +++ b/crates/client/src/proxy.rs @@ -1,60 +1,14 @@ -//! client proxy - -mod http_proxy; -mod socks_proxy; +//! Proxied connections for the collaboration WebSocket. +//! +//! Proxy URLs come from settings or the environment (see +//! `ProxySettings::proxy_url`); this module dials the proxy, wraps the +//! connection in TLS when an `https://` proxy asks for it, and tunnels +//! through it with `proxy_handshake`. use anyhow::{Context as _, Result}; use http_client::Url; -use http_proxy::{HttpProxyType, connect_http_proxy_stream, parse_http_proxy}; -use socks_proxy::{SocksVersion, connect_socks_proxy_stream, parse_socks_proxy}; - -pub(crate) async fn connect_proxy_stream( - proxy: &Url, - rpc_host: (&str, u16), -) -> Result> { - let Some(((proxy_domain, proxy_port), proxy_type)) = parse_proxy_type(proxy) else { - // If parsing the proxy URL fails, we must avoid falling back to an insecure connection. - // SOCKS proxies are often used in contexts where security and privacy are critical, - // so any fallback could expose users to significant risks. - anyhow::bail!("Parsing proxy url failed"); - }; - - // Connect to proxy and wrap protocol later - let stream = tokio::net::TcpStream::connect((proxy_domain.as_str(), proxy_port)) - .await - .context("Failed to connect to proxy")?; - - let proxy_stream = match proxy_type { - ProxyType::SocksProxy(proxy) => connect_socks_proxy_stream(stream, proxy, rpc_host).await?, - ProxyType::HttpProxy(proxy) => { - connect_http_proxy_stream(stream, proxy, rpc_host, &proxy_domain).await? - } - }; - - Ok(proxy_stream) -} - -enum ProxyType<'t> { - SocksProxy(SocksVersion<'t>), - HttpProxy(HttpProxyType<'t>), -} - -fn parse_proxy_type(proxy: &Url) -> Option<((String, u16), ProxyType<'_>)> { - let scheme = proxy.scheme(); - let host = proxy.host()?.to_string(); - let port = proxy.port_or_known_default()?; - let proxy_type = match scheme { - scheme if scheme.starts_with("socks") => { - Some(ProxyType::SocksProxy(parse_socks_proxy(scheme, proxy))) - } - scheme if scheme.starts_with("http") => { - Some(ProxyType::HttpProxy(parse_http_proxy(scheme, proxy))) - } - _ => None, - }?; - - Some(((host, port), proxy_type)) -} +use proxy_handshake::{ProxyScheme, ProxySpec, Target}; +use tokio::net::TcpStream; pub(crate) trait AsyncReadWrite: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static @@ -64,3 +18,75 @@ impl A for T { } + +/// Whether `NO_PROXY` in the environment excludes `host` from proxying, +/// matching the exclusions the HTTP client already applies to its own +/// requests. +pub(crate) fn excluded_from_proxy(host: &str) -> bool { + http_client::read_no_proxy_from_env() + .is_some_and(|no_proxy| proxy_handshake::no_proxy_matches(&no_proxy, host)) +} + +pub(crate) async fn connect_proxy_stream( + proxy: &Url, + rpc_host: (&str, u16), +) -> Result> { + // If parsing the proxy URL fails, we must avoid falling back to an + // insecure connection. Proxies are often used in contexts where security + // and privacy are critical, so any fallback could expose users to + // significant risks. + let spec = ProxySpec::parse(proxy).context("parsing proxy URL")?; + + let target = if spec.remote_dns() { + Target::Domain(rpc_host.0.to_string(), rpc_host.1) + } else { + // SOCKS4 requests carry a raw IPv4 address, so the target must + // resolve to one. + let requires_ipv4 = matches!(spec.scheme, ProxyScheme::Socks4 { .. }); + let address = tokio::net::lookup_host(rpc_host) + .await + .with_context(|| format!("failed to lookup domain {}", rpc_host.0))? + .find(|address| !requires_ipv4 || address.is_ipv4()) + .with_context(|| format!("failed to lookup domain {}", rpc_host.0))?; + Target::Address(address) + }; + + let stream = TcpStream::connect((spec.host.as_str(), spec.port)) + .await + .context("Failed to connect to proxy")?; + + let stream: Box = if spec.tls() { + Box::new(connect_tls_to_proxy(stream, &spec.host).await?) + } else { + Box::new(stream) + }; + + let stream = proxy_handshake::tokio::establish(stream, &spec, &target) + .await + .context("error connecting through proxy")?; + Ok(Box::new(stream)) +} + +#[cfg(any(target_os = "windows", target_os = "macos"))] +async fn connect_tls_to_proxy( + stream: TcpStream, + proxy_domain: &str, +) -> Result> { + use tokio_native_tls::{TlsConnector, native_tls}; + + let tls_connector = TlsConnector::from(native_tls::TlsConnector::new()?); + Ok(tls_connector.connect(proxy_domain, stream).await?) +} + +#[cfg(not(any(target_os = "windows", target_os = "macos")))] +async fn connect_tls_to_proxy( + stream: TcpStream, + proxy_domain: &str, +) -> Result> { + let proxy_domain = rustls_pki_types::ServerName::try_from(proxy_domain) + .context("invalid DNS name for proxy TLS")? + .to_owned(); + let tls_connector = + tokio_rustls::TlsConnector::from(std::sync::Arc::new(http_client_tls::tls_config())); + Ok(tls_connector.connect(proxy_domain, stream).await?) +} diff --git a/crates/client/src/proxy/http_proxy.rs b/crates/client/src/proxy/http_proxy.rs deleted file mode 100644 index f64c56b16c..0000000000 --- a/crates/client/src/proxy/http_proxy.rs +++ /dev/null @@ -1,193 +0,0 @@ -use anyhow::{Context, Result}; -use base64::Engine; -use httparse::{EMPTY_HEADER, Response}; -use tokio::{ - io::{AsyncBufReadExt, AsyncWriteExt, BufStream}, - net::TcpStream, -}; -#[cfg(any(target_os = "windows", target_os = "macos"))] -use tokio_native_tls::{TlsConnector, native_tls}; -#[cfg(not(any(target_os = "windows", target_os = "macos")))] -use tokio_rustls::TlsConnector; -use url::Url; - -use super::AsyncReadWrite; - -pub(super) enum HttpProxyType<'t> { - HTTP(Option>), - HTTPS(Option>), -} - -pub(super) struct HttpProxyAuthorization<'t> { - username: &'t str, - password: &'t str, -} - -pub(super) fn parse_http_proxy<'t>(scheme: &str, proxy: &'t Url) -> HttpProxyType<'t> { - let auth = proxy.password().map(|password| HttpProxyAuthorization { - username: proxy.username(), - password, - }); - if scheme.starts_with("https") { - HttpProxyType::HTTPS(auth) - } else { - HttpProxyType::HTTP(auth) - } -} - -pub(crate) async fn connect_http_proxy_stream( - stream: TcpStream, - http_proxy: HttpProxyType<'_>, - rpc_host: (&str, u16), - proxy_domain: &str, -) -> Result> { - match http_proxy { - HttpProxyType::HTTP(auth) => http_connect(stream, rpc_host, auth).await, - HttpProxyType::HTTPS(auth) => https_connect(stream, rpc_host, auth, proxy_domain).await, - } - .context("error connecting to http/https proxy") -} - -async fn http_connect( - stream: T, - target: (&str, u16), - auth: Option>, -) -> Result> -where - T: AsyncReadWrite, -{ - let mut stream = BufStream::new(stream); - let request = make_request(target, auth); - stream.write_all(request.as_bytes()).await?; - stream.flush().await?; - check_response(&mut stream).await?; - Ok(Box::new(stream)) -} - -#[cfg(any(target_os = "windows", target_os = "macos"))] -async fn https_connect( - stream: T, - target: (&str, u16), - auth: Option>, - proxy_domain: &str, -) -> Result> -where - T: AsyncReadWrite, -{ - let tls_connector = TlsConnector::from(native_tls::TlsConnector::new()?); - let stream = tls_connector.connect(proxy_domain, stream).await?; - http_connect(stream, target, auth).await -} - -#[cfg(not(any(target_os = "windows", target_os = "macos")))] -async fn https_connect( - stream: T, - target: (&str, u16), - auth: Option>, - proxy_domain: &str, -) -> Result> -where - T: AsyncReadWrite, -{ - let proxy_domain = rustls_pki_types::ServerName::try_from(proxy_domain) - .context("Address resolution failed")? - .to_owned(); - let tls_connector = TlsConnector::from(std::sync::Arc::new(http_client_tls::tls_config())); - let stream = tls_connector.connect(proxy_domain, stream).await?; - http_connect(stream, target, auth).await -} - -fn make_request(target: (&str, u16), auth: Option>) -> String { - let (host, port) = target; - let mut request = format!( - "CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\nProxy-Connection: Keep-Alive\r\n" - ); - if let Some(HttpProxyAuthorization { username, password }) = auth { - let auth = - base64::prelude::BASE64_STANDARD.encode(format!("{username}:{password}").as_bytes()); - let auth = format!("Proxy-Authorization: Basic {auth}\r\n"); - request.push_str(&auth); - } - request.push_str("\r\n"); - request -} - -async fn check_response(stream: &mut BufStream) -> Result<()> -where - T: AsyncReadWrite, -{ - let response = recv_response(stream).await?; - let mut dummy_headers = [EMPTY_HEADER; MAX_RESPONSE_HEADERS]; - let mut parser = Response::new(&mut dummy_headers); - parser.parse(response.as_bytes())?; - - match parser.code { - Some(code) => { - if code == 200 { - Ok(()) - } else { - Err(anyhow::anyhow!( - "Proxy connection failed with HTTP code: {code}" - )) - } - } - None => Err(anyhow::anyhow!( - "Proxy connection failed with no HTTP code: {}", - parser.reason.unwrap_or("Unknown reason") - )), - } -} - -const MAX_RESPONSE_HEADER_LENGTH: usize = 4096; -const MAX_RESPONSE_HEADERS: usize = 16; - -async fn recv_response(stream: &mut BufStream) -> Result -where - T: AsyncReadWrite, -{ - let mut response = String::new(); - loop { - if stream.read_line(&mut response).await? == 0 { - return Err(anyhow::anyhow!("End of stream")); - } - - if MAX_RESPONSE_HEADER_LENGTH < response.len() { - return Err(anyhow::anyhow!("Maximum response header length exceeded")); - } - - if response.ends_with("\r\n\r\n") { - return Ok(response); - } - } -} - -#[cfg(test)] -mod tests { - use url::Url; - - use super::{HttpProxyAuthorization, HttpProxyType, parse_http_proxy}; - - #[test] - fn test_parse_http_proxy() { - let proxy = Url::parse("http://proxy.example.com:1080").unwrap(); - let scheme = proxy.scheme(); - - let version = parse_http_proxy(scheme, &proxy); - assert!(matches!(version, HttpProxyType::HTTP(None))) - } - - #[test] - fn test_parse_http_proxy_with_auth() { - let proxy = Url::parse("http://username:password@proxy.example.com:1080").unwrap(); - let scheme = proxy.scheme(); - - let version = parse_http_proxy(scheme, &proxy); - assert!(matches!( - version, - HttpProxyType::HTTP(Some(HttpProxyAuthorization { - username: "username", - password: "password" - })) - )) - } -} diff --git a/crates/client/src/proxy/socks_proxy.rs b/crates/client/src/proxy/socks_proxy.rs deleted file mode 100644 index bf2a5eab62..0000000000 --- a/crates/client/src/proxy/socks_proxy.rs +++ /dev/null @@ -1,226 +0,0 @@ -//! socks proxy - -use anyhow::{Context as _, Result}; -use http_client::Url; -use tokio::net::TcpStream; -use tokio_socks::{ - IntoTargetAddr, TargetAddr, - tcp::{Socks4Stream, Socks5Stream}, -}; - -use super::AsyncReadWrite; - -/// Identification to a Socks V4 Proxy -pub(super) struct Socks4Identification<'a> { - user_id: &'a str, -} - -/// Authorization to a Socks V5 Proxy -pub(super) struct Socks5Authorization<'a> { - username: &'a str, - password: &'a str, -} - -/// Socks Proxy Protocol Version -/// -/// V4 allows identification using a user_id -/// V5 allows authorization using a username and password -pub(super) enum SocksVersion<'a> { - V4 { - local_dns: bool, - identification: Option>, - }, - V5 { - local_dns: bool, - authorization: Option>, - }, -} - -pub(super) fn parse_socks_proxy<'t>(scheme: &str, proxy: &'t Url) -> SocksVersion<'t> { - if scheme.starts_with("socks4") { - let identification = match proxy.username() { - "" => None, - username => Some(Socks4Identification { user_id: username }), - }; - SocksVersion::V4 { - local_dns: scheme != "socks4a", - identification, - } - } else { - let authorization = proxy.password().map(|password| Socks5Authorization { - username: proxy.username(), - password, - }); - SocksVersion::V5 { - local_dns: scheme != "socks5h", - authorization, - } - } -} - -pub(super) async fn connect_socks_proxy_stream( - stream: TcpStream, - socks_version: SocksVersion<'_>, - rpc_host: (&str, u16), -) -> Result> { - let rpc_host = rpc_host - .into_target_addr() - .context("Failed to parse target addr")?; - - let local_dns = match &socks_version { - SocksVersion::V4 { local_dns, .. } => local_dns, - SocksVersion::V5 { local_dns, .. } => local_dns, - }; - let rpc_host = match (rpc_host, local_dns) { - (TargetAddr::Domain(domain, port), true) => { - let ip_addr = tokio::net::lookup_host((domain.as_ref(), port)) - .await - .with_context(|| format!("Failed to lookup domain {}", domain))? - .next() - .ok_or_else(|| anyhow::anyhow!("Failed to lookup domain {}", domain))?; - TargetAddr::Ip(ip_addr) - } - (rpc_host, _) => rpc_host, - }; - - match socks_version { - SocksVersion::V4 { - identification: None, - .. - } => { - let socks = Socks4Stream::connect_with_socket(stream, rpc_host) - .await - .context("error connecting to socks")?; - Ok(Box::new(socks)) - } - SocksVersion::V4 { - identification: Some(Socks4Identification { user_id }), - .. - } => { - let socks = Socks4Stream::connect_with_userid_and_socket(stream, rpc_host, user_id) - .await - .context("error connecting to socks")?; - Ok(Box::new(socks)) - } - SocksVersion::V5 { - authorization: None, - .. - } => { - let socks = Socks5Stream::connect_with_socket(stream, rpc_host) - .await - .context("error connecting to socks")?; - Ok(Box::new(socks)) - } - SocksVersion::V5 { - authorization: Some(Socks5Authorization { username, password }), - .. - } => { - let socks = Socks5Stream::connect_with_password_and_socket( - stream, rpc_host, username, password, - ) - .await - .context("error connecting to socks")?; - Ok(Box::new(socks)) - } - } -} - -#[cfg(test)] -mod tests { - use url::Url; - - use super::*; - - #[test] - fn parse_socks4() { - let proxy = Url::parse("socks4://proxy.example.com:1080").unwrap(); - let scheme = proxy.scheme(); - - let version = parse_socks_proxy(scheme, &proxy); - assert!(matches!( - version, - SocksVersion::V4 { - local_dns: true, - identification: None - } - )) - } - - #[test] - fn parse_socks4_with_identification() { - let proxy = Url::parse("socks4://userid@proxy.example.com:1080").unwrap(); - let scheme = proxy.scheme(); - - let version = parse_socks_proxy(scheme, &proxy); - assert!(matches!( - version, - SocksVersion::V4 { - local_dns: true, - identification: Some(Socks4Identification { user_id: "userid" }) - } - )) - } - - #[test] - fn parse_socks4_with_remote_dns() { - let proxy = Url::parse("socks4a://proxy.example.com:1080").unwrap(); - let scheme = proxy.scheme(); - - let version = parse_socks_proxy(scheme, &proxy); - assert!(matches!( - version, - SocksVersion::V4 { - local_dns: false, - identification: None - } - )) - } - - #[test] - fn parse_socks5() { - let proxy = Url::parse("socks5://proxy.example.com:1080").unwrap(); - let scheme = proxy.scheme(); - - let version = parse_socks_proxy(scheme, &proxy); - assert!(matches!( - version, - SocksVersion::V5 { - local_dns: true, - authorization: None - } - )) - } - - #[test] - fn parse_socks5_with_authorization() { - let proxy = Url::parse("socks5://username:password@proxy.example.com:1080").unwrap(); - let scheme = proxy.scheme(); - - let version = parse_socks_proxy(scheme, &proxy); - assert!(matches!( - version, - SocksVersion::V5 { - local_dns: true, - authorization: Some(Socks5Authorization { - username: "username", - password: "password" - }) - } - )) - } - - #[test] - fn parse_socks5_with_remote_dns() { - let proxy = Url::parse("socks5h://proxy.example.com:1080").unwrap(); - let scheme = proxy.scheme(); - - let version = parse_socks_proxy(scheme, &proxy); - assert!(matches!( - version, - SocksVersion::V5 { - local_dns: false, - authorization: None - } - )) - } -} diff --git a/crates/proxy_handshake/Cargo.toml b/crates/proxy_handshake/Cargo.toml new file mode 100644 index 0000000000..6ea582ab58 --- /dev/null +++ b/crates/proxy_handshake/Cargo.toml @@ -0,0 +1,34 @@ +[package] +name = "proxy_handshake" +version = "0.1.0" +edition.workspace = true +publish.workspace = true +license = "Apache-2.0" +description = "Sans-IO client handshakes for HTTP CONNECT and SOCKS proxies" + +[lints] +workspace = true + +[lib] +path = "src/proxy_handshake.rs" +doctest = true + +[features] +default = [] +futures-io = ["dep:futures"] +tokio = ["dep:tokio"] + +[dependencies] +base64.workspace = true +futures = { workspace = true, optional = true } +httparse.workspace = true +percent-encoding.workspace = true +thiserror.workspace = true +tokio = { workspace = true, features = ["io-util"], optional = true } +url.workspace = true + +[dev-dependencies] +futures.workspace = true +piper.workspace = true +proptest.workspace = true +tokio = { workspace = true, features = ["io-util", "macros", "rt"] } diff --git a/crates/proxy_handshake/LICENSE-APACHE b/crates/proxy_handshake/LICENSE-APACHE new file mode 120000 index 0000000000..1cd601d0a3 --- /dev/null +++ b/crates/proxy_handshake/LICENSE-APACHE @@ -0,0 +1 @@ +../../LICENSE-APACHE \ No newline at end of file diff --git a/crates/proxy_handshake/src/futures_io.rs b/crates/proxy_handshake/src/futures_io.rs new file mode 100644 index 0000000000..dc1107ab77 --- /dev/null +++ b/crates/proxy_handshake/src/futures_io.rs @@ -0,0 +1,191 @@ +//! A driver that runs a [`Handshake`] over [`futures`] I/O traits. + +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _}; + +use crate::{EstablishError, Handshake, ProxySpec, Step, Target}; + +/// Runs the proxy handshake over `stream` and returns the tunneled stream. +/// +/// The stream must already be connected to the proxy (and wrapped in TLS if +/// [`ProxySpec::tls`] asks for it). See [`Handshake::new`] for how `target` +/// interacts with DNS resolution. +pub async fn establish( + mut stream: S, + spec: &ProxySpec, + target: &Target, +) -> Result, EstablishError> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let mut handshake = Handshake::new(spec, target)?; + let mut buffer = [0u8; 4096]; + let mut received_length = 0; + loop { + let step = handshake.advance(&buffer[..received_length])?; + received_length = 0; + match step { + Step::Send(bytes) => { + stream.write_all(&bytes).await?; + stream.flush().await?; + } + Step::NeedMoreInput => { + received_length = stream.read(&mut buffer).await?; + if received_length == 0 { + return Err(EstablishError::Io(io::ErrorKind::UnexpectedEof.into())); + } + } + Step::Done { leftover } => { + return Ok(Tunneled { + leftover, + offset: 0, + stream, + }); + } + } + } +} + +/// A tunneled stream returned by [`establish`]: replays the handshake's +/// leftover bytes before reading from the underlying transport. Writes pass +/// straight through. +pub struct Tunneled { + leftover: Vec, + offset: usize, + stream: S, +} + +impl AsyncRead for Tunneled { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + let this = self.get_mut(); + if this.offset < this.leftover.len() { + let remaining = &this.leftover[this.offset..]; + let length = remaining.len().min(buf.len()); + buf[..length].copy_from_slice(&remaining[..length]); + this.offset += length; + if this.offset == this.leftover.len() { + this.leftover = Vec::new(); + this.offset = 0; + } + return Poll::Ready(Ok(length)); + } + Pin::new(&mut this.stream).poll_read(cx, buf) + } +} + +impl AsyncWrite for Tunneled { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_flush(cx) + } + + fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_close(cx) + } +} + +#[cfg(test)] +mod tests { + use futures::executor::block_on; + use futures::join; + + use super::*; + + #[test] + fn establishes_http_tunnel_and_preserves_leftover() { + let (client, mut server) = duplex(); + let spec = ProxySpec::parse(&"http://proxy:8080".parse().unwrap()).unwrap(); + let target = Target::Domain("cloud.example.com".to_string(), 443); + block_on(async { + join!( + async { + let mut stream = establish(client, &spec, &target).await.unwrap(); + let mut buffer = [0u8; 12]; + stream.read_exact(&mut buffer).await.unwrap(); + assert_eq!(&buffer, b"early-tunnel"); + stream.write_all(b"hello").await.unwrap(); + stream.flush().await.unwrap(); + }, + async { + let mut head = Vec::new(); + let mut byte = [0u8; 1]; + while !head.ends_with(b"\r\n\r\n") { + server.read_exact(&mut byte).await.unwrap(); + head.push(byte[0]); + } + // Send the response and some tunnel bytes in one write so + // the driver has to hand them back through `leftover`. + server + .write_all(b"HTTP/1.1 200 OK\r\n\r\nearly-tunnel") + .await + .unwrap(); + let mut buffer = [0u8; 5]; + server.read_exact(&mut buffer).await.unwrap(); + assert_eq!(&buffer, b"hello"); + }, + ); + }); + } + + fn duplex() -> (PipeStream, PipeStream) { + let (client_reader, server_writer) = piper::pipe(1024); + let (server_reader, client_writer) = piper::pipe(1024); + ( + PipeStream { + reader: client_reader, + writer: client_writer, + }, + PipeStream { + reader: server_reader, + writer: server_writer, + }, + ) + } + + struct PipeStream { + reader: piper::Reader, + writer: piper::Writer, + } + + impl AsyncRead for PipeStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().reader).poll_read(cx, buf) + } + } + + impl AsyncWrite for PipeStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().writer).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().writer).poll_flush(cx) + } + + fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().writer).poll_close(cx) + } + } +} diff --git a/crates/proxy_handshake/src/handshake.rs b/crates/proxy_handshake/src/handshake.rs new file mode 100644 index 0000000000..233e420d9c --- /dev/null +++ b/crates/proxy_handshake/src/handshake.rs @@ -0,0 +1,708 @@ +//! The sans-IO proxy handshake state machine. + +use std::collections::VecDeque; +use std::net::{IpAddr, SocketAddr}; + +use base64::Engine as _; + +use crate::{Credentials, MAX_HTTP_RESPONSE_LENGTH, ProxyError, ProxyScheme, ProxySpec, Target}; + +const MAX_HTTP_RESPONSE_HEADERS: usize = 64; + +const SOCKS4_VERSION: u8 = 0x04; +const SOCKS5_VERSION: u8 = 0x05; +const SOCKS_COMMAND_CONNECT: u8 = 0x01; +const SOCKS4_REPLY_GRANTED: u8 = 0x5A; +const SOCKS5_METHOD_NO_AUTHENTICATION: u8 = 0x00; +const SOCKS5_METHOD_USERNAME_PASSWORD: u8 = 0x02; +const SOCKS5_ADDRESS_IPV4: u8 = 0x01; +const SOCKS5_ADDRESS_DOMAIN: u8 = 0x03; +const SOCKS5_ADDRESS_IPV6: u8 = 0x04; + +/// A proxy handshake in progress. +/// +/// Sans-IO: the handshake never performs I/O itself. Drive it by calling +/// [`advance`](Self::advance) with bytes received from the proxy and acting +/// on the returned [`Step`]s until it reports [`Step::Done`]. +pub struct Handshake { + exchanges: VecDeque, + input: Vec, +} + +/// What the caller must do next to make progress on the handshake. +#[derive(Debug, PartialEq, Eq)] +pub enum Step { + /// Write these bytes to the proxy. + Send(Vec), + /// Read more bytes from the proxy and pass them to the next `advance`. + NeedMoreInput, + /// The tunnel is established. `leftover` holds any bytes the proxy sent + /// past the end of the handshake; they belong to the tunneled stream and + /// must be consumed before reading from the transport again. + Done { leftover: Vec }, +} + +enum Exchange { + Send(Vec), + Expect(Expectation), +} + +enum Expectation { + HttpResponseHead, + Socks5MethodSelection { offered: u8 }, + Socks5AuthReply, + Socks5ConnectReply, + Socks4ConnectReply, +} + +impl Handshake { + /// Plans the handshake for tunneling to `target` through the proxy + /// described by `spec`. + /// + /// Passing a [`Target::Domain`] delegates DNS resolution to the proxy + /// regardless of scheme (SOCKS4 proxies are sent the 4a extension); + /// resolve the host locally and pass a [`Target::Address`] when + /// [`ProxySpec::remote_dns`] is false, to follow the proxy URL's intent. + pub fn new(spec: &ProxySpec, target: &Target) -> Result { + // The domain is spliced into protocol messages (the CONNECT request + // line, NUL-terminated SOCKS4a fields), so control bytes would let a + // hostile target string forge protocol structure. No valid host name + // contains them; fail closed rather than trusting the caller to have + // validated. + if let Target::Domain(domain, _) = target + && domain + .bytes() + .any(|byte| byte.is_ascii_control() || byte == 0x7F) + { + return Err(ProxyError::DomainContainsControlBytes); + } + + let mut exchanges = VecDeque::new(); + match spec.scheme { + ProxyScheme::Http { .. } => { + exchanges.push_back(Exchange::Send(http_connect_request( + target, + spec.credentials.as_ref(), + ))); + exchanges.push_back(Exchange::Expect(Expectation::HttpResponseHead)); + } + ProxyScheme::Socks4 { .. } => { + exchanges.push_back(Exchange::Send(socks4_connect_request( + target, + spec.credentials.as_ref(), + )?)); + exchanges.push_back(Exchange::Expect(Expectation::Socks4ConnectReply)); + } + ProxyScheme::Socks5 { .. } => { + let method = if spec.credentials.is_some() { + SOCKS5_METHOD_USERNAME_PASSWORD + } else { + SOCKS5_METHOD_NO_AUTHENTICATION + }; + exchanges.push_back(Exchange::Send(vec![SOCKS5_VERSION, 1, method])); + exchanges.push_back(Exchange::Expect(Expectation::Socks5MethodSelection { + offered: method, + })); + if let Some(credentials) = &spec.credentials { + exchanges.push_back(Exchange::Send(socks5_auth_request(credentials)?)); + exchanges.push_back(Exchange::Expect(Expectation::Socks5AuthReply)); + } + exchanges.push_back(Exchange::Send(socks5_connect_request(target)?)); + exchanges.push_back(Exchange::Expect(Expectation::Socks5ConnectReply)); + } + } + Ok(Self { + exchanges, + input: Vec::new(), + }) + } + + /// Advances the state machine with bytes newly received from the proxy + /// (empty on the first call, and whenever there is nothing new to feed). + /// + /// Callers must not feed the same bytes twice: partial input is buffered + /// internally until it can be parsed. + pub fn advance(&mut self, received: &[u8]) -> Result { + self.input.extend_from_slice(received); + loop { + match self.exchanges.pop_front() { + None => { + return Ok(Step::Done { + leftover: std::mem::take(&mut self.input), + }); + } + Some(Exchange::Send(bytes)) => return Ok(Step::Send(bytes)), + Some(Exchange::Expect(expectation)) => { + match try_parse(&expectation, &self.input)? { + Some(consumed) => { + self.input.drain(..consumed); + } + None => { + self.exchanges.push_front(Exchange::Expect(expectation)); + return Ok(Step::NeedMoreInput); + } + } + } + } + } + } +} + +/// Attempts to parse one expected proxy message from the buffered input. +/// Returns the number of bytes consumed, or `None` when more input is needed. +fn try_parse(expectation: &Expectation, input: &[u8]) -> Result, ProxyError> { + match expectation { + Expectation::HttpResponseHead => parse_http_response_head(input), + Expectation::Socks5MethodSelection { offered } => { + let Some(reply) = input.get(..2) else { + return Ok(None); + }; + if reply[0] != SOCKS5_VERSION { + return Err(ProxyError::UnexpectedSocksVersion(reply[0])); + } + if reply[1] != *offered { + return Err(ProxyError::AuthMethodRejected { selected: reply[1] }); + } + Ok(Some(2)) + } + Expectation::Socks5AuthReply => { + let Some(reply) = input.get(..2) else { + return Ok(None); + }; + // RFC 1929: any non-zero status is failure. + if reply[1] != 0x00 { + return Err(ProxyError::AuthenticationFailed); + } + Ok(Some(2)) + } + Expectation::Socks5ConnectReply => { + let Some(header) = input.get(..4) else { + return Ok(None); + }; + if header[0] != SOCKS5_VERSION { + return Err(ProxyError::UnexpectedSocksVersion(header[0])); + } + if header[1] != 0x00 { + return Err(ProxyError::Socks5ConnectRefused(header[1])); + } + let address_length = match header[3] { + SOCKS5_ADDRESS_IPV4 => 4, + SOCKS5_ADDRESS_IPV6 => 16, + SOCKS5_ADDRESS_DOMAIN => match input.get(4) { + Some(length) => 1 + *length as usize, + None => return Ok(None), + }, + other => return Err(ProxyError::UnknownAddressType(other)), + }; + // The reply's bound address and port only matter for BIND, so + // they are consumed and discarded. + let total = 4 + address_length + 2; + if input.len() < total { + return Ok(None); + } + Ok(Some(total)) + } + Expectation::Socks4ConnectReply => { + if input.len() < 8 { + return Ok(None); + } + if input[1] != SOCKS4_REPLY_GRANTED { + return Err(ProxyError::Socks4ConnectRefused(input[1])); + } + Ok(Some(8)) + } + } +} + +fn parse_http_response_head(input: &[u8]) -> Result, ProxyError> { + let mut headers = [httparse::EMPTY_HEADER; MAX_HTTP_RESPONSE_HEADERS]; + let mut response = httparse::Response::new(&mut headers); + match response.parse(input) { + Ok(httparse::Status::Complete(consumed)) => { + // The cap applies to complete heads too: a driver that reads in + // large chunks could otherwise hand a well-formed multi-megabyte + // head to a single `advance` and have it accepted. + if consumed > MAX_HTTP_RESPONSE_LENGTH { + return Err(ProxyError::HttpResponseTooLarge); + } + let code = response.code.ok_or_else(|| { + ProxyError::MalformedHttpResponse("missing status code".to_string()) + })?; + if !(200..300).contains(&code) { + return Err(ProxyError::HttpConnectRefused(code)); + } + Ok(Some(consumed)) + } + Ok(httparse::Status::Partial) => { + if input.len() > MAX_HTTP_RESPONSE_LENGTH { + Err(ProxyError::HttpResponseTooLarge) + } else { + Ok(None) + } + } + Err(error) => Err(ProxyError::MalformedHttpResponse(error.to_string())), + } +} + +fn http_connect_request(target: &Target, credentials: Option<&Credentials>) -> Vec { + let host = match target { + Target::Domain(domain, port) => format!("{domain}:{port}"), + // `SocketAddr`'s `Display` brackets IPv6 addresses, matching the + // `host:port` form CONNECT requires. + Target::Address(address) => address.to_string(), + }; + let mut request = + format!("CONNECT {host} HTTP/1.1\r\nHost: {host}\r\nProxy-Connection: Keep-Alive\r\n"); + if let Some(Credentials { username, password }) = credentials { + let encoded = base64::prelude::BASE64_STANDARD.encode(format!("{username}:{password}")); + request.push_str(&format!("Proxy-Authorization: Basic {encoded}\r\n")); + } + request.push_str("\r\n"); + request.into_bytes() +} + +fn socks5_auth_request(credentials: &Credentials) -> Result, ProxyError> { + let username = credentials.username.as_bytes(); + let password = credentials.password.as_bytes(); + if username.len() > 255 || password.len() > 255 { + return Err(ProxyError::CredentialsTooLong); + } + let mut request = Vec::with_capacity(3 + username.len() + password.len()); + // RFC 1929 subnegotiation version. + request.push(0x01); + request.push(username.len() as u8); + request.extend_from_slice(username); + request.push(password.len() as u8); + request.extend_from_slice(password); + Ok(request) +} + +fn socks5_connect_request(target: &Target) -> Result, ProxyError> { + let mut request = vec![SOCKS5_VERSION, SOCKS_COMMAND_CONNECT, 0x00]; + let port = match target { + Target::Domain(domain, port) => { + let domain = domain.as_bytes(); + if domain.len() > 255 { + return Err(ProxyError::DomainTooLong); + } + request.push(SOCKS5_ADDRESS_DOMAIN); + request.push(domain.len() as u8); + request.extend_from_slice(domain); + *port + } + Target::Address(address) => { + match address.ip() { + IpAddr::V4(ip) => { + request.push(SOCKS5_ADDRESS_IPV4); + request.extend_from_slice(&ip.octets()); + } + IpAddr::V6(ip) => { + request.push(SOCKS5_ADDRESS_IPV6); + request.extend_from_slice(&ip.octets()); + } + } + address.port() + } + }; + request.extend_from_slice(&port.to_be_bytes()); + Ok(request) +} + +fn socks4_connect_request( + target: &Target, + credentials: Option<&Credentials>, +) -> Result, ProxyError> { + let user_id = credentials + .map(|credentials| credentials.username.as_str()) + .unwrap_or(""); + // The user id field is NUL-terminated on the wire, so an embedded NUL + // would truncate it and shift the fields that follow. + if user_id.as_bytes().contains(&0x00) { + return Err(ProxyError::Socks4UserIdContainsNul); + } + let mut request = vec![SOCKS4_VERSION, SOCKS_COMMAND_CONNECT]; + match target { + Target::Address(SocketAddr::V4(address)) => { + request.extend_from_slice(&address.port().to_be_bytes()); + request.extend_from_slice(&address.ip().octets()); + request.extend_from_slice(user_id.as_bytes()); + request.push(0x00); + } + Target::Address(SocketAddr::V6(_)) => return Err(ProxyError::Socks4Ipv4Only), + Target::Domain(domain, port) => { + // SOCKS4a: the invalid destination address 0.0.0.1 signals that + // a host name follows the user id. + request.extend_from_slice(&port.to_be_bytes()); + request.extend_from_slice(&[0, 0, 0, 1]); + request.extend_from_slice(user_id.as_bytes()); + request.push(0x00); + request.extend_from_slice(domain.as_bytes()); + request.push(0x00); + } + } + Ok(request) +} + +#[cfg(test)] +mod tests { + use proptest::prelude::*; + + use super::*; + + #[test] + fn http_connect_completes_on_200_and_preserves_leftover() { + let server_bytes = b"HTTP/1.1 200 Connection Established\r\n\r\ntunnel-bytes"; + let (sends, leftover) = run( + "http://proxy:8080", + &domain_target(), + server_bytes, + &[server_bytes.len()], + ) + .unwrap(); + + assert_eq!(sends.len(), 1); + let request = String::from_utf8(sends[0].clone()).unwrap(); + assert!( + request.starts_with("CONNECT cloud.example.com:443 HTTP/1.1\r\n"), + "unexpected request: {request:?}" + ); + assert!(request.contains("Host: cloud.example.com:443\r\n")); + assert!(request.ends_with("\r\n\r\n")); + assert_eq!(leftover, b"tunnel-bytes"); + } + + #[test] + fn http_connect_sends_basic_authorization() { + let (sends, _) = run( + "http://user:pass@proxy:8080", + &domain_target(), + b"HTTP/1.1 200 OK\r\n\r\n", + &[64], + ) + .unwrap(); + + let request = String::from_utf8(sends[0].clone()).unwrap(); + // "dXNlcjpwYXNz" is base64("user:pass"). + assert!( + request.contains("Proxy-Authorization: Basic dXNlcjpwYXNz\r\n"), + "unexpected request: {request:?}" + ); + } + + #[test] + fn http_connect_fails_on_error_status() { + let error = run( + "http://proxy:8080", + &domain_target(), + b"HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic\r\n\r\n", + &[64], + ) + .unwrap_err(); + assert!(matches!(error, ProxyError::HttpConnectRefused(407))); + } + + #[test] + fn http_connect_rejects_oversized_response() { + // A single never-ending header keeps the response incomplete (and + // under httparse's header-count limit) until the size cap trips. + let mut server_bytes = b"HTTP/1.1 200 OK\r\nX-Padding: ".to_vec(); + server_bytes.resize(MAX_HTTP_RESPONSE_LENGTH + 1024, b'a'); + let error = run("http://proxy:8080", &domain_target(), &server_bytes, &[64]).unwrap_err(); + assert!(matches!(error, ProxyError::HttpResponseTooLarge)); + } + + #[test] + fn http_connect_rejects_oversized_complete_response_in_one_chunk() { + // A *well-formed* head over the cap, delivered whole in a single + // `advance`, must be rejected too: the cap can't depend on the + // driver's read chunking. + let mut server_bytes = b"HTTP/1.1 200 OK\r\nX-Padding: ".to_vec(); + server_bytes.resize(MAX_HTTP_RESPONSE_LENGTH + 1024, b'a'); + server_bytes.extend_from_slice(b"\r\n\r\n"); + let error = run( + "http://proxy:8080", + &domain_target(), + &server_bytes, + &[server_bytes.len()], + ) + .unwrap_err(); + assert!(matches!(error, ProxyError::HttpResponseTooLarge)); + } + + #[test] + fn rejects_domains_containing_control_bytes() { + for scheme in ["http", "socks4a", "socks5h"] { + for domain in ["evil.com\r\nX-Injected: x", "evil.com\0.example.com"] { + let spec = spec(&format!("{scheme}://proxy:1080")); + let target = Target::Domain(domain.to_string(), 443); + let Err(error) = Handshake::new(&spec, &target) else { + panic!("expected {scheme} handshake construction to fail for {domain:?}"); + }; + assert!(matches!(error, ProxyError::DomainContainsControlBytes)); + } + } + } + + #[test] + fn socks4_rejects_user_id_containing_nul() { + let spec = spec("socks4://user%00id@proxy:1080"); + let target = Target::Address("192.0.2.10:443".parse().unwrap()); + let Err(error) = Handshake::new(&spec, &target) else { + panic!("expected handshake construction to fail"); + }; + assert!(matches!(error, ProxyError::Socks4UserIdContainsNul)); + } + + #[test] + fn socks5_connect_with_domain_target() { + let mut server_bytes = vec![0x05, 0x00]; + server_bytes.extend_from_slice(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0]); + server_bytes.extend_from_slice(b"tunnel"); + let (sends, leftover) = run( + "socks5h://proxy:1080", + &domain_target(), + &server_bytes, + &[server_bytes.len()], + ) + .unwrap(); + + assert_eq!(sends[0], vec![0x05, 0x01, 0x00]); + let mut expected_connect = vec![0x05, 0x01, 0x00, 0x03, 17]; + expected_connect.extend_from_slice(b"cloud.example.com"); + expected_connect.extend_from_slice(&443u16.to_be_bytes()); + assert_eq!(sends[1], expected_connect); + assert_eq!(leftover, b"tunnel"); + } + + #[test] + fn socks5_connect_with_username_password() { + let mut server_bytes = vec![0x05, 0x02]; + server_bytes.extend_from_slice(&[0x01, 0x00]); + server_bytes.extend_from_slice(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0]); + let (sends, _) = run( + "socks5h://user:pass@proxy:1080", + &domain_target(), + &server_bytes, + &[server_bytes.len()], + ) + .unwrap(); + + assert_eq!(sends[0], vec![0x05, 0x01, 0x02]); + assert_eq!(sends[1], b"\x01\x04user\x04pass".to_vec()); + assert_eq!(sends.len(), 3); + } + + #[test] + fn socks5_reports_connection_refusal() { + let mut server_bytes = vec![0x05, 0x00]; + server_bytes.extend_from_slice(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0]); + let error = run( + "socks5h://proxy:1080", + &domain_target(), + &server_bytes, + &[server_bytes.len()], + ) + .unwrap_err(); + assert!(matches!(error, ProxyError::Socks5ConnectRefused(0x02))); + } + + #[test] + fn socks5_reports_rejected_authentication_method() { + let error = run( + "socks5h://proxy:1080", + &domain_target(), + &[0x05, 0xFF], + &[2], + ) + .unwrap_err(); + assert!(matches!( + error, + ProxyError::AuthMethodRejected { selected: 0xFF } + )); + } + + #[test] + fn socks5_encodes_ip_targets() { + let mut server_bytes = vec![0x05, 0x00]; + server_bytes.extend_from_slice(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0]); + + let target = Target::Address("192.0.2.10:443".parse().unwrap()); + let (sends, _) = run("socks5://proxy:1080", &target, &server_bytes, &[64]).unwrap(); + assert_eq!( + sends[1], + vec![0x05, 0x01, 0x00, 0x01, 192, 0, 2, 10, 0x01, 0xBB] + ); + + let target = Target::Address("[2001:db8::1]:443".parse().unwrap()); + let (sends, _) = run("socks5://proxy:1080", &target, &server_bytes, &[64]).unwrap(); + assert_eq!(sends[1][3], SOCKS5_ADDRESS_IPV6); + assert_eq!(sends[1].len(), 4 + 16 + 2); + } + + #[test] + fn socks5_reply_with_domain_bound_address_is_consumed() { + let mut server_bytes = vec![0x05, 0x00]; + server_bytes.extend_from_slice(&[0x05, 0x00, 0x00, 0x03, 4]); + server_bytes.extend_from_slice(b"host"); + server_bytes.extend_from_slice(&[0, 80]); + server_bytes.extend_from_slice(b"rest"); + let (_, leftover) = run( + "socks5h://proxy:1080", + &domain_target(), + &server_bytes, + &[server_bytes.len()], + ) + .unwrap(); + assert_eq!(leftover, b"rest"); + } + + #[test] + fn socks4_connect_with_ipv4_target() { + let target = Target::Address("192.0.2.10:443".parse().unwrap()); + let (sends, leftover) = run( + "socks4://proxy:1080", + &target, + &[0x00, 0x5A, 0, 0, 0, 0, 0, 0], + &[8], + ) + .unwrap(); + assert_eq!(sends[0], vec![0x04, 0x01, 0x01, 0xBB, 192, 0, 2, 10, 0x00]); + assert_eq!(leftover, b""); + } + + #[test] + fn socks4a_connect_with_domain_target_and_user_id() { + let (sends, _) = run( + "socks4a://userid@proxy:1080", + &domain_target(), + &[0x00, 0x5A, 0, 0, 0, 0, 0, 0], + &[8], + ) + .unwrap(); + let mut expected = vec![0x04, 0x01, 0x01, 0xBB, 0, 0, 0, 1]; + expected.extend_from_slice(b"userid\x00"); + expected.extend_from_slice(b"cloud.example.com\x00"); + assert_eq!(sends[0], expected); + } + + #[test] + fn socks4_rejects_ipv6_targets() { + let spec = spec("socks4://proxy:1080"); + let target = Target::Address("[2001:db8::1]:443".parse().unwrap()); + let Err(error) = Handshake::new(&spec, &target) else { + panic!("expected handshake construction to fail"); + }; + assert!(matches!(error, ProxyError::Socks4Ipv4Only)); + } + + #[test] + fn socks4_reports_rejection() { + let target = Target::Address("192.0.2.10:443".parse().unwrap()); + let error = run( + "socks4://proxy:1080", + &target, + &[0x00, 0x5B, 0, 0, 0, 0, 0, 0], + &[8], + ) + .unwrap_err(); + assert!(matches!(error, ProxyError::Socks4ConnectRefused(0x5B))); + } + + proptest! { + /// The outcome of a handshake must not depend on how the proxy's + /// bytes are chunked across reads. + #[test] + fn chunking_does_not_change_the_outcome( + chunk_sizes in proptest::collection::vec(1usize..16, 1..32), + ) { + for (url, server_bytes) in transcripts() { + let baseline = run(url, &domain_target(), &server_bytes, &[server_bytes.len()]); + let chunked = run(url, &domain_target(), &server_bytes, &chunk_sizes); + match (baseline, chunked) { + (Ok(expected), Ok(actual)) => prop_assert_eq!(expected, actual), + (Err(_), Err(_)) => {} + (baseline, chunked) => { + return Err(TestCaseError::fail(format!( + "outcomes diverged: baseline {baseline:?} vs chunked {chunked:?}" + ))); + } + } + } + } + } + + /// Scripted proxy conversations used by the chunking property test, each + /// with trailing bytes to pin leftover handling. + fn transcripts() -> Vec<(&'static str, Vec)> { + let mut socks5 = vec![0x05, 0x02]; + socks5.extend_from_slice(&[0x01, 0x00]); + socks5.extend_from_slice(&[0x05, 0x00, 0x00, 0x03, 4]); + socks5.extend_from_slice(b"host"); + socks5.extend_from_slice(&[0, 80]); + socks5.extend_from_slice(b"leftover-bytes"); + + vec![ + ( + "http://user:pass@proxy:8080", + b"HTTP/1.1 200 Connection Established\r\nVia: test\r\n\r\nleftover-bytes".to_vec(), + ), + ("socks5h://user:pass@proxy:1080", socks5), + ( + "socks4a://userid@proxy:1080", + b"\x00\x5A\x00\x00\x00\x00\x00\x00leftover-bytes".to_vec(), + ), + ] + } + + fn spec(url: &str) -> ProxySpec { + ProxySpec::parse(&url.parse().unwrap()).unwrap() + } + + fn domain_target() -> Target { + Target::Domain("cloud.example.com".to_string(), 443) + } + + /// Drives a handshake to completion against scripted proxy bytes, + /// feeding input in the given chunk sizes (cycled). Returns everything + /// the client sent and the leftover bytes reported at completion. + fn run( + url: &str, + target: &Target, + server_bytes: &[u8], + chunk_sizes: &[usize], + ) -> Result<(Vec>, Vec), ProxyError> { + let mut handshake = Handshake::new(&spec(url), target)?; + let mut sends = Vec::new(); + let mut remaining = server_bytes; + let mut sizes = chunk_sizes.iter().copied().cycle(); + let mut pending: &[u8] = &[]; + loop { + let step = handshake.advance(pending)?; + pending = &[]; + match step { + Step::Send(bytes) => sends.push(bytes), + Step::NeedMoreInput => { + assert!( + !remaining.is_empty(), + "handshake wants more input than the transcript provides" + ); + let size = sizes + .next() + .expect("chunk_sizes must not be empty") + .clamp(1, remaining.len()); + let (chunk, rest) = remaining.split_at(size); + pending = chunk; + remaining = rest; + } + Step::Done { leftover } => { + let mut leftover = leftover; + // Any transcript bytes not yet fed to the machine are + // also tunnel bytes; account for them so results don't + // depend on chunking. + leftover.extend_from_slice(remaining); + return Ok((sends, leftover)); + } + } + } + } +} diff --git a/crates/proxy_handshake/src/proxy_handshake.rs b/crates/proxy_handshake/src/proxy_handshake.rs new file mode 100644 index 0000000000..d098086232 --- /dev/null +++ b/crates/proxy_handshake/src/proxy_handshake.rs @@ -0,0 +1,393 @@ +//! Sans-IO client handshakes for HTTP CONNECT and SOCKS proxies. +//! +//! Speaks the client side of the protocols needed to tunnel a TCP connection +//! through a proxy — HTTP CONNECT (RFC 9110 §9.3.6), SOCKS4/4a, and SOCKS5 +//! (RFC 1928, with RFC 1929 username/password authentication) — as a pure +//! state machine that never touches a socket. Callers own all I/O: resolving +//! and connecting to the proxy, wrapping the connection in TLS when the proxy +//! requires it, and pumping bytes between the [`Handshake`] and their +//! transport. The `futures-io` and `tokio` features add ready-made drivers +//! for the corresponding I/O traits. +//! +//! ``` +//! use proxy_handshake::{Handshake, ProxySpec, Step, Target, Url}; +//! +//! let url: Url = "http://localhost:8888".parse().unwrap(); +//! let spec = ProxySpec::parse(&url).unwrap(); +//! let target = Target::Domain("example.com".into(), 443); +//! let mut handshake = Handshake::new(&spec, &target).unwrap(); +//! +//! // The sans-IO loop: write `Send` bytes to the proxy, feed bytes the proxy +//! // sent back through `advance`, and start using the stream for the +//! // tunneled protocol once `Done` is returned. +//! let Ok(Step::Send(connect_request)) = handshake.advance(&[]) else { +//! unreachable!() +//! }; +//! assert!(connect_request.starts_with(b"CONNECT example.com:443 HTTP/1.1\r\n")); +//! ``` + +mod handshake; + +#[cfg(feature = "futures-io")] +pub mod futures_io; +#[cfg(feature = "tokio")] +pub mod tokio; + +use std::net::SocketAddr; + +pub use handshake::{Handshake, Step}; +pub use url::Url; + +/// How to reach and authenticate with a proxy: the validated interpretation +/// of a proxy [`Url`]. +/// +/// A raw [`Url`] can hold any scheme, leaves default ports implicit, and +/// carries credentials percent-encoded. Parsing into a `ProxySpec` settles +/// all of that once — unsupported schemes are rejected, scheme-specific +/// default ports are applied, and credentials are decoded — so the handshake +/// and its callers never re-derive them. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProxySpec { + pub scheme: ProxyScheme, + /// The proxy's host: a domain name or an IP address literal (without + /// brackets for IPv6). + pub host: String, + pub port: u16, + pub credentials: Option, +} + +/// The protocol the proxy speaks, taken from the proxy URL's scheme. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ProxyScheme { + /// An HTTP CONNECT proxy (`http://` or `https://`). `tls` is whether the + /// connection *to the proxy itself* is wrapped in TLS (an `https://` + /// proxy URL); it is independent of any TLS the tunneled protocol layers + /// inside the tunnel. + Http { tls: bool }, + /// A SOCKS4 proxy (`socks4://` or `socks4a://`). + Socks4 { remote_dns: bool }, + /// A SOCKS5 proxy (`socks5://` or `socks5h://`). + Socks5 { remote_dns: bool }, +} + +/// Proxy credentials, percent-decoded from the URL's userinfo component. +#[derive(Clone, PartialEq, Eq)] +pub struct Credentials { + pub username: String, + pub password: String, +} + +/// Redacts both fields so credentials can't leak through debug formatting of +/// any containing type. Implemented manually (rather than leaving `Debug` +/// underived) so a future `#[derive(Debug)]` on this struct is a conscious +/// choice instead of an accident. +impl std::fmt::Debug for Credentials { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Credentials") + .field("username", &"") + .field("password", &"") + .finish() + } +} + +/// The destination to tunnel to. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Target { + /// An unresolved host name and port: DNS resolution is delegated to the + /// proxy. With SOCKS4 proxies this is encoded using the SOCKS4a + /// extension, which not every SOCKS4 proxy supports. + /// + /// The host must be a name, not arbitrary text: strings containing ASCII + /// control bytes are rejected by [`Handshake::new`], since they could + /// otherwise forge protocol structure in the messages the host is + /// embedded in. + Domain(String, u16), + /// An already-resolved socket address. Use this when + /// [`ProxySpec::remote_dns`] is false, resolving the host locally first. + Address(SocketAddr), +} + +/// The size cap on an HTTP CONNECT response head; a proxy that sends more +/// than this before the blank line is misbehaving. +pub const MAX_HTTP_RESPONSE_LENGTH: usize = 8192; + +#[derive(Debug, thiserror::Error)] +pub enum ProxyError { + #[error( + "unsupported proxy scheme `{0}` (expected http, https, socks4, socks4a, socks5, or socks5h)" + )] + UnsupportedScheme(String), + #[error("proxy URL is missing a host")] + MissingHost, + #[error("proxy credentials are not valid UTF-8 after percent-decoding")] + InvalidCredentials, + #[error("SOCKS5 usernames and passwords are limited to 255 bytes")] + CredentialsTooLong, + #[error("SOCKS target domains are limited to 255 bytes")] + DomainTooLong, + #[error("target domain contains ASCII control bytes")] + DomainContainsControlBytes, + #[error("SOCKS4 proxies only support IPv4 targets")] + Socks4Ipv4Only, + #[error("SOCKS4 user ids cannot contain NUL bytes")] + Socks4UserIdContainsNul, + #[error("proxy sent a malformed HTTP response: {0}")] + MalformedHttpResponse(String), + #[error("proxy refused CONNECT with HTTP status {0}")] + HttpConnectRefused(u16), + #[error("HTTP response from proxy exceeded {MAX_HTTP_RESPONSE_LENGTH} bytes")] + HttpResponseTooLarge, + #[error("proxy replied with SOCKS version {0:#04x}")] + UnexpectedSocksVersion(u8), + #[error("SOCKS5 proxy rejected the offered authentication method (selected {selected:#04x})")] + AuthMethodRejected { selected: u8 }, + #[error("SOCKS5 proxy rejected the username/password")] + AuthenticationFailed, + #[error("SOCKS5 proxy refused the connection: {}", socks5_refusal_message(*.0))] + Socks5ConnectRefused(u8), + #[error("SOCKS5 proxy sent unknown address type {0:#04x}")] + UnknownAddressType(u8), + #[error("SOCKS4 proxy rejected the connection (code {0:#04x})")] + Socks4ConnectRefused(u8), +} + +/// Errors from a driver establishing a tunnel: either the proxy conversation +/// failed or the transport did. +#[cfg(any(feature = "futures-io", feature = "tokio"))] +#[derive(Debug, thiserror::Error)] +pub enum EstablishError { + #[error(transparent)] + Proxy(#[from] ProxyError), + #[error(transparent)] + Io(#[from] std::io::Error), +} + +impl ProxySpec { + /// Parses a proxy URL with an `http`, `https`, `socks4`, `socks4a`, + /// `socks5`, or `socks5h` scheme. Missing ports default by scheme (80, + /// 443, or 1080). Credentials in the userinfo component are + /// percent-decoded, per RFC 3986 §3.2.1. + pub fn parse(url: &Url) -> Result { + let scheme = match url.scheme() { + "http" => ProxyScheme::Http { tls: false }, + "https" => ProxyScheme::Http { tls: true }, + "socks4" => ProxyScheme::Socks4 { remote_dns: false }, + "socks4a" => ProxyScheme::Socks4 { remote_dns: true }, + "socks5" => ProxyScheme::Socks5 { remote_dns: false }, + "socks5h" => ProxyScheme::Socks5 { remote_dns: true }, + other => return Err(ProxyError::UnsupportedScheme(other.to_string())), + }; + + let host = match url.host().ok_or(ProxyError::MissingHost)? { + url::Host::Domain(domain) => domain.to_string(), + url::Host::Ipv4(ip) => ip.to_string(), + url::Host::Ipv6(ip) => ip.to_string(), + }; + let default_port = match scheme { + ProxyScheme::Http { tls: false } => 80, + ProxyScheme::Http { tls: true } => 443, + ProxyScheme::Socks4 { .. } | ProxyScheme::Socks5 { .. } => 1080, + }; + let port = url.port().unwrap_or(default_port); + + let credentials = if url.username().is_empty() && url.password().is_none() { + None + } else { + Some(Credentials { + username: decode_userinfo(url.username())?, + password: decode_userinfo(url.password().unwrap_or_default())?, + }) + }; + + Ok(Self { + scheme, + host, + port, + credentials, + }) + } + + /// Whether the target should be handed to the proxy as a host name + /// ([`Target::Domain`]) instead of being resolved locally. True for HTTP + /// CONNECT and the `socks4a`/`socks5h` schemes, matching curl's scheme + /// semantics. + pub fn remote_dns(&self) -> bool { + match self.scheme { + ProxyScheme::Http { .. } => true, + ProxyScheme::Socks4 { remote_dns } | ProxyScheme::Socks5 { remote_dns } => remote_dns, + } + } + + /// Whether the connection to the proxy itself must be wrapped in TLS + /// before the handshake starts (an `https://` proxy URL). + pub fn tls(&self) -> bool { + matches!(self.scheme, ProxyScheme::Http { tls: true }) + } +} + +/// Decodes one component (username or password) of a URL's userinfo, which +/// RFC 3986 §3.2.1 defines as percent-encoded: the proxy must be sent the +/// decoded form. +fn decode_userinfo(component: &str) -> Result { + percent_encoding::percent_decode_str(component) + .decode_utf8() + .map(|decoded| decoded.into_owned()) + .map_err(|_| ProxyError::InvalidCredentials) +} + +/// Matches `host` against a `NO_PROXY`-style exclusion list: comma-separated +/// entries that are `*`, an exact host, or a domain suffix (bare, with a +/// leading dot, or with a leading `*.`). Entries may carry a `:port`, which +/// is ignored — exclusion is by host. CIDR ranges are not supported. +pub fn no_proxy_matches(no_proxy: &str, host: &str) -> bool { + no_proxy + .split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .any(|entry| { + if entry == "*" { + return true; + } + // Bare IPv6 literals also contain colons, so only strip a port + // when there is exactly one colon and the suffix parses as one. + let entry = match entry.rsplit_once(':') { + Some((prefix, suffix)) + if !prefix.contains(':') && suffix.parse::().is_ok() => + { + prefix + } + _ => entry, + }; + let entry = entry.strip_prefix("*.").unwrap_or(entry); + let entry = entry.strip_prefix('.').unwrap_or(entry); + host.eq_ignore_ascii_case(entry) + || (host.len() > entry.len() + && host.as_bytes()[host.len() - entry.len() - 1] == b'.' + && host[host.len() - entry.len()..].eq_ignore_ascii_case(entry)) + }) +} + +/// Human-readable descriptions of the SOCKS5 reply field's error codes, +/// verbatim from RFC 1928 §6. +fn socks5_refusal_message(code: u8) -> String { + match code { + 0x01 => "general SOCKS server failure".to_string(), + 0x02 => "connection not allowed by ruleset".to_string(), + 0x03 => "network unreachable".to_string(), + 0x04 => "host unreachable".to_string(), + 0x05 => "connection refused".to_string(), + 0x06 => "TTL expired".to_string(), + 0x07 => "command not supported".to_string(), + 0x08 => "address type not supported".to_string(), + other => format!("unknown error code {other:#04x}"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_applies_default_ports_by_scheme() { + assert_eq!(parse("http://proxy.example.com").port, 80); + assert_eq!(parse("https://proxy.example.com").port, 443); + assert_eq!(parse("socks4://proxy.example.com").port, 1080); + assert_eq!(parse("socks5://proxy.example.com").port, 1080); + assert_eq!(parse("http://proxy.example.com:3128").port, 3128); + } + + #[test] + fn parse_maps_schemes() { + assert_eq!(parse("http://p").scheme, ProxyScheme::Http { tls: false }); + assert_eq!(parse("https://p").scheme, ProxyScheme::Http { tls: true }); + assert_eq!( + parse("socks4://p").scheme, + ProxyScheme::Socks4 { remote_dns: false } + ); + assert_eq!( + parse("socks4a://p").scheme, + ProxyScheme::Socks4 { remote_dns: true } + ); + assert_eq!( + parse("socks5://p").scheme, + ProxyScheme::Socks5 { remote_dns: false } + ); + assert_eq!( + parse("socks5h://p").scheme, + ProxyScheme::Socks5 { remote_dns: true } + ); + } + + #[test] + fn parse_rejects_unknown_scheme() { + let url: Url = "ftp://proxy.example.com".parse().unwrap(); + let Err(error) = ProxySpec::parse(&url) else { + panic!("expected parsing to fail"); + }; + assert!(matches!(error, ProxyError::UnsupportedScheme(scheme) if scheme == "ftp")); + } + + #[test] + fn parse_percent_decodes_credentials() { + let spec = parse("http://user:p%40ss@proxy.example.com:8080"); + let credentials = spec.credentials.unwrap(); + assert_eq!(credentials.username, "user"); + assert_eq!(credentials.password, "p@ss"); + } + + #[test] + fn parse_supports_username_only_credentials() { + let spec = parse("socks4://userid@proxy.example.com"); + let credentials = spec.credentials.unwrap(); + assert_eq!(credentials.username, "userid"); + assert_eq!(credentials.password, ""); + } + + #[test] + fn parse_unbrackets_ipv6_hosts() { + assert_eq!(parse("http://[::1]:8080").host, "::1"); + } + + #[test] + fn remote_dns_follows_scheme() { + assert!(parse("http://p").remote_dns()); + assert!(parse("https://p").remote_dns()); + assert!(!parse("socks4://p").remote_dns()); + assert!(parse("socks4a://p").remote_dns()); + assert!(!parse("socks5://p").remote_dns()); + assert!(parse("socks5h://p").remote_dns()); + } + + #[test] + fn credentials_debug_output_is_redacted() { + let credentials = Credentials { + username: "user".to_string(), + password: "hunter2".to_string(), + }; + let debug_output = format!("{credentials:?}"); + assert!(!debug_output.contains("hunter2")); + assert!(debug_output.contains("")); + } + + #[test] + fn no_proxy_matching() { + assert!(no_proxy_matches("example.com", "example.com")); + assert!(no_proxy_matches("example.com", "sub.example.com")); + assert!(no_proxy_matches(".example.com", "sub.example.com")); + assert!(no_proxy_matches("*.example.com", "sub.example.com")); + assert!(no_proxy_matches("other.org, example.com", "example.com")); + assert!(no_proxy_matches("example.com:443", "example.com")); + assert!(no_proxy_matches("*", "anything.at.all")); + assert!(no_proxy_matches("EXAMPLE.com", "example.COM")); + assert!(no_proxy_matches("::1", "::1")); + + assert!(!no_proxy_matches("example.com", "notexample.com")); + assert!(!no_proxy_matches("example.com", "example.org")); + assert!(!no_proxy_matches("", "example.com")); + assert!(!no_proxy_matches("sub.example.com", "example.com")); + } + + fn parse(url: &str) -> ProxySpec { + ProxySpec::parse(&url.parse().unwrap()).unwrap() + } +} diff --git a/crates/proxy_handshake/src/tokio.rs b/crates/proxy_handshake/src/tokio.rs new file mode 100644 index 0000000000..eedc4c58bb --- /dev/null +++ b/crates/proxy_handshake/src/tokio.rs @@ -0,0 +1,139 @@ +//! A driver that runs a [`Handshake`] over [`tokio`](::tokio) I/O traits. + +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use ::tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _, ReadBuf}; + +use crate::{EstablishError, Handshake, ProxySpec, Step, Target}; + +/// Runs the proxy handshake over `stream` and returns the tunneled stream. +/// +/// The stream must already be connected to the proxy (and wrapped in TLS if +/// [`ProxySpec::tls`] asks for it). See [`Handshake::new`] for how `target` +/// interacts with DNS resolution. +pub async fn establish( + mut stream: S, + spec: &ProxySpec, + target: &Target, +) -> Result, EstablishError> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let mut handshake = Handshake::new(spec, target)?; + let mut buffer = [0u8; 4096]; + let mut received_length = 0; + loop { + let step = handshake.advance(&buffer[..received_length])?; + received_length = 0; + match step { + Step::Send(bytes) => { + stream.write_all(&bytes).await?; + stream.flush().await?; + } + Step::NeedMoreInput => { + received_length = stream.read(&mut buffer).await?; + if received_length == 0 { + return Err(EstablishError::Io(io::ErrorKind::UnexpectedEof.into())); + } + } + Step::Done { leftover } => { + return Ok(Tunneled { + leftover, + offset: 0, + stream, + }); + } + } + } +} + +/// A tunneled stream returned by [`establish`]: replays the handshake's +/// leftover bytes before reading from the underlying transport. Writes pass +/// straight through. +pub struct Tunneled { + leftover: Vec, + offset: usize, + stream: S, +} + +impl AsyncRead for Tunneled { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + if this.offset < this.leftover.len() { + let remaining = &this.leftover[this.offset..]; + let length = remaining.len().min(buf.remaining()); + buf.put_slice(&remaining[..length]); + this.offset += length; + if this.offset == this.leftover.len() { + this.leftover = Vec::new(); + this.offset = 0; + } + return Poll::Ready(Ok(())); + } + Pin::new(&mut this.stream).poll_read(cx, buf) + } +} + +impl AsyncWrite for Tunneled { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_shutdown(cx) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn establishes_socks5_tunnel_and_preserves_leftover() { + let (client, mut server) = ::tokio::io::duplex(1024); + let spec = ProxySpec::parse(&"socks5h://proxy:1080".parse().unwrap()).unwrap(); + let target = Target::Domain("cloud.example.com".to_string(), 443); + ::tokio::join!( + async { + let mut stream = establish(client, &spec, &target).await.unwrap(); + let mut buffer = [0u8; 12]; + stream.read_exact(&mut buffer).await.unwrap(); + assert_eq!(&buffer, b"early-tunnel"); + stream.write_all(b"hello").await.unwrap(); + stream.flush().await.unwrap(); + }, + async { + let mut greeting = [0u8; 3]; + server.read_exact(&mut greeting).await.unwrap(); + assert_eq!(greeting, [0x05, 0x01, 0x00]); + server.write_all(&[0x05, 0x00]).await.unwrap(); + + let mut connect_request = vec![0u8; 4 + 1 + 17 + 2]; + server.read_exact(&mut connect_request).await.unwrap(); + // Send the reply and some tunnel bytes in one write so the + // driver has to hand them back through `leftover`. + let mut reply = vec![0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0]; + reply.extend_from_slice(b"early-tunnel"); + server.write_all(&reply).await.unwrap(); + + let mut buffer = [0u8; 5]; + server.read_exact(&mut buffer).await.unwrap(); + assert_eq!(&buffer, b"hello"); + }, + ); + } +}