client: Extract proxy handshakes into a new proxy_handshake crate (#61427)
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.
This commit is contained in:
Eric Holk authored and GitHub committed
1 parent
1db5b3ae06
commit
414e6fe31e
13 files changed
+1571
-487
No files matched your search
Generated
+16
-4
@@ -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",
|
||||
|
||||
+3
-4
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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?),
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
+80
-54
@@ -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<Box<dyn AsyncReadWrite>> {
|
||||
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<T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static> 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<Box<dyn AsyncReadWrite>> {
|
||||
// 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<dyn AsyncReadWrite> = 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<tokio_native_tls::TlsStream<TcpStream>> {
|
||||
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<tokio_rustls::client::TlsStream<TcpStream>> {
|
||||
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?)
|
||||
}
|
||||
@@ -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<HttpProxyAuthorization<'t>>),
|
||||
HTTPS(Option<HttpProxyAuthorization<'t>>),
|
||||
}
|
||||
|
||||
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<Box<dyn AsyncReadWrite>> {
|
||||
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<T>(
|
||||
stream: T,
|
||||
target: (&str, u16),
|
||||
auth: Option<HttpProxyAuthorization<'_>>,
|
||||
) -> Result<Box<dyn AsyncReadWrite>>
|
||||
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<T>(
|
||||
stream: T,
|
||||
target: (&str, u16),
|
||||
auth: Option<HttpProxyAuthorization<'_>>,
|
||||
proxy_domain: &str,
|
||||
) -> Result<Box<dyn AsyncReadWrite>>
|
||||
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<T>(
|
||||
stream: T,
|
||||
target: (&str, u16),
|
||||
auth: Option<HttpProxyAuthorization<'_>>,
|
||||
proxy_domain: &str,
|
||||
) -> Result<Box<dyn AsyncReadWrite>>
|
||||
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<HttpProxyAuthorization<'_>>) -> 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<T>(stream: &mut BufStream<T>) -> 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<T>(stream: &mut BufStream<T>) -> Result<String>
|
||||
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"
|
||||
}))
|
||||
))
|
||||
}
|
||||
}
|
||||
@@ -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<Socks4Identification<'a>>,
|
||||
},
|
||||
V5 {
|
||||
local_dns: bool,
|
||||
authorization: Option<Socks5Authorization<'a>>,
|
||||
},
|
||||
}
|
||||
|
||||
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<Box<dyn AsyncReadWrite>> {
|
||||
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
|
||||
}
|
||||
))
|
||||
}
|
||||
}
|
||||
@@ -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"] }
|
||||
+1
@@ -0,0 +1 @@
|
||||
../../LICENSE-APACHE
|
||||
@@ -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<S>(
|
||||
mut stream: S,
|
||||
spec: &ProxySpec,
|
||||
target: &Target,
|
||||
) -> Result<Tunneled<S>, 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<S> {
|
||||
leftover: Vec<u8>,
|
||||
offset: usize,
|
||||
stream: S,
|
||||
}
|
||||
|
||||
impl<S: AsyncRead + Unpin> AsyncRead for Tunneled<S> {
|
||||
fn poll_read(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut [u8],
|
||||
) -> Poll<io::Result<usize>> {
|
||||
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<S: AsyncWrite + Unpin> AsyncWrite for Tunneled<S> {
|
||||
fn poll_write(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &[u8],
|
||||
) -> Poll<io::Result<usize>> {
|
||||
Pin::new(&mut self.get_mut().stream).poll_write(cx, buf)
|
||||
}
|
||||
|
||||
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.get_mut().stream).poll_flush(cx)
|
||||
}
|
||||
|
||||
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
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<io::Result<usize>> {
|
||||
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<io::Result<usize>> {
|
||||
Pin::new(&mut self.get_mut().writer).poll_write(cx, buf)
|
||||
}
|
||||
|
||||
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.get_mut().writer).poll_flush(cx)
|
||||
}
|
||||
|
||||
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.get_mut().writer).poll_close(cx)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<Exchange>,
|
||||
input: Vec<u8>,
|
||||
}
|
||||
|
||||
/// 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<u8>),
|
||||
/// 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<u8> },
|
||||
}
|
||||
|
||||
enum Exchange {
|
||||
Send(Vec<u8>),
|
||||
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<Self, ProxyError> {
|
||||
// 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<Step, ProxyError> {
|
||||
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<Option<usize>, 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<Option<usize>, 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<u8> {
|
||||
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<Vec<u8>, 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<Vec<u8>, 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<Vec<u8>, 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<u8>)> {
|
||||
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<u8>>, Vec<u8>), 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));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<Credentials>,
|
||||
}
|
||||
|
||||
/// 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", &"<redacted>")
|
||||
.field("password", &"<redacted>")
|
||||
.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<Self, ProxyError> {
|
||||
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<String, ProxyError> {
|
||||
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::<u16>().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("<redacted>"));
|
||||
}
|
||||
|
||||
#[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()
|
||||
}
|
||||
}
|
||||
@@ -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<S>(
|
||||
mut stream: S,
|
||||
spec: &ProxySpec,
|
||||
target: &Target,
|
||||
) -> Result<Tunneled<S>, 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<S> {
|
||||
leftover: Vec<u8>,
|
||||
offset: usize,
|
||||
stream: S,
|
||||
}
|
||||
|
||||
impl<S: AsyncRead + Unpin> AsyncRead for Tunneled<S> {
|
||||
fn poll_read(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut ReadBuf<'_>,
|
||||
) -> Poll<io::Result<()>> {
|
||||
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<S: AsyncWrite + Unpin> AsyncWrite for Tunneled<S> {
|
||||
fn poll_write(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &[u8],
|
||||
) -> Poll<io::Result<usize>> {
|
||||
Pin::new(&mut self.get_mut().stream).poll_write(cx, buf)
|
||||
}
|
||||
|
||||
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.get_mut().stream).poll_flush(cx)
|
||||
}
|
||||
|
||||
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
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");
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user