From 2810a11a47d42b49e0c884e4c3a7497c376ab49f Mon Sep 17 00:00:00 2001 From: pasta Date: Sat, 26 Sep 2026 19:53:44 -0500 Subject: [PATCH 1/5] fix(sdk): pass an unbracketed host as the TLS server name so IPv6 endpoints do not panic create_channel let tonic take the TLS server name from Uri::host, which keeps the brackets of an IPv6 literal ([2001:db8::1]). rustls rejects that as a server name, so tls_config(..).expect(..) panicked for every IPv6 evonode endpoint, and the explicit-CA path did the same. The host is now always passed without brackets through domain_name, and a TLS configuration error is returned as a transport error instead of panicking. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/transport/tonic_channel.rs | 51 ++++++++++++++++--- 1 file changed, 45 insertions(+), 6 deletions(-) diff --git a/packages/rs-dapi-client/src/transport/tonic_channel.rs b/packages/rs-dapi-client/src/transport/tonic_channel.rs index 779bdb33f52..2f774b043f0 100644 --- a/packages/rs-dapi-client/src/transport/tonic_channel.rs +++ b/packages/rs-dapi-client/src/transport/tonic_channel.rs @@ -24,7 +24,7 @@ pub fn create_channel( uri: Uri, settings: Option<&AppliedRequestSettings>, ) -> Result { - let host = uri.host().expect("Failed to get host from URI").to_string(); + let host = unbracketed(uri.host().expect("Failed to get host from URI")).to_string(); let mut builder = Channel::builder(uri); @@ -32,7 +32,10 @@ pub fn create_channel( // Try to add native roots only on platforms where they're available (not iOS) let mut tls_config = ClientTlsConfig::new() .with_webpki_roots() - .assume_http2(true); + .assume_http2(true) + // Without an explicit name tonic takes `Uri::host`, which keeps the + // brackets of an IPv6 literal and is not a valid TLS server name. + .domain_name(host); // Try to add native roots - this may fail on iOS/Android, which is fine since we have webpki roots #[cfg(not(any( @@ -52,7 +55,7 @@ pub fn create_channel( if let Some(pem) = settings.ca_certificate.as_ref() { let cert = Certificate::from_pem(pem); - tls_config = tls_config.ca_certificate(cert).domain_name(host); + tls_config = tls_config.ca_certificate(cert); }; } @@ -65,9 +68,45 @@ pub fn create_channel( .keep_alive_timeout(HTTP2_KEEP_ALIVE_TIMEOUT) .keep_alive_while_idle(false); - builder = builder - .tls_config(tls_config) - .expect("Failed to set TLS config"); + builder = builder.tls_config(tls_config).map_err(|e| { + TransportError::Grpc(dapi_grpc::tonic::Status::invalid_argument(format!( + "invalid TLS configuration: {e}" + ))) + })?; Ok(builder.connect_lazy()) } + +/// The host of a URI without the brackets an IPv6 literal carries in it +/// (`[2001:db8::1]` becomes `2001:db8::1`); any other host is unchanged. +pub(crate) fn unbracketed(host: &str) -> &str { + host.strip_prefix('[') + .and_then(|host| host.strip_suffix(']')) + .unwrap_or(host) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::RequestSettings; + + #[test] + fn should_strip_only_ipv6_brackets() { + assert_eq!(unbracketed("[2001:db8::1]"), "2001:db8::1"); + assert_eq!(unbracketed("127.0.0.1"), "127.0.0.1"); + assert_eq!(unbracketed("evo.example"), "evo.example"); + } + + /// `create_channel` used to panic for every IPv6 endpoint: the bracketed + /// host is not a valid TLS server name. + #[tokio::test] + async fn should_create_a_channel_for_an_ipv6_endpoint() { + let uri: Uri = "https://[2001:db8::1]:443".parse().expect("uri"); + create_channel(uri.clone(), None).expect("default roots"); + + let settings = RequestSettings::default() + .finalize() + .with_ca_certificate(Some(Certificate::from_pem("fake-pem-data"))); + create_channel(uri, Some(&settings)).expect("explicit CA"); + } +} From 435e92ba12c50765b363166b5e954531e4c06bec Mon Sep 17 00:00:00 2001 From: pasta Date: Sat, 26 Sep 2026 20:11:58 -0500 Subject: [PATCH 2/5] feat(sdk): tunnel DAPI connections through a SOCKS5 proxy DapiClient::with_proxy(Socks5Proxy) makes every connection of the client go through a SOCKS5 proxy (RFC 1928), over TCP or a Unix socket, with no fallback to a direct connection. The evonode's host goes to the proxy unresolved, so no endpoint is ever resolved locally, and TLS still runs end to end with the evonode and is validated against the endpoint host. Native only; wasm is unchanged. Socks5Auth::RandomPerConnection sends fresh random credentials on every new connection, so Tor (IsolateSOCKSAuth) builds a separate circuit for each. With credentials the client offers both no-auth and username/password, so a proxy that only allows no-auth still works. A failure on the proxy's side (unreachable, not accepting the connection or not answering the SOCKS5 greeting within 5 s, SOCKS protocol or auth error, general failure, unsupported command) carries a ProxyError in the tonic status source chain. CanRetry returns false for it, so it is neither retried nor held against the node, and TransportError::is_proxy_failure reports it. Tor also sends a general failure when one circuit fails; the request then fails at once and the next one gets a fresh circuit. Replies about the destination (host or network unreachable, refused, TTL expired, not allowed by ruleset, which is how Tor reports exit policy refusals, and address type not supported) and a connect timeout after the proxy answered the greeting (a CONNECT reply held while Tor builds a circuit, or a slow TLS handshake) stay ordinary retryable failures that ban the node as before. TransportError::clone now keeps the status source so the marker survives. With a proxy and no connect timeout, channels get a 20 s connect timeout (tonic applies it around the SOCKS5 handshake and TLS). The proxy is part of the connection pool key (endpoint, auth kind and username, never the password). Adds tokio-socks 0.5.3 (MIT), whose dependencies are all already in the tree. Co-Authored-By: Claude Opus 5.5 (1M context) --- Cargo.lock | 14 + packages/rs-dapi-client/Cargo.toml | 10 +- packages/rs-dapi-client/src/dapi_client.rs | 43 +- .../rs-dapi-client/src/request_settings.rs | 112 +++- packages/rs-dapi-client/src/transport.rs | 34 +- packages/rs-dapi-client/src/transport/grpc.rs | 17 + .../rs-dapi-client/src/transport/proxy.rs | 544 ++++++++++++++++++ .../src/transport/tonic_channel.rs | 34 +- .../rs-dapi-client/tests/rate_limit_ban.rs | 2 + 9 files changed, 788 insertions(+), 22 deletions(-) create mode 100644 packages/rs-dapi-client/src/transport/proxy.rs diff --git a/Cargo.lock b/Cargo.lock index 3dd3b403830..39e1c76bc9a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6429,8 +6429,10 @@ dependencies = [ "sha2", "thiserror 2.0.18", "tokio", + "tokio-socks", "tonic-web-wasm-client", "tower 0.5.3", + "tower-service", "tracing", "wasm-bindgen-futures", ] @@ -7884,6 +7886,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-socks" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7e2948f60dbe26b35f2c7fb74ac2854c1fddded0fe9d7548fcc674a246f7615" +dependencies = [ + "either", + "futures-util", + "thiserror 1.0.69", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" diff --git a/packages/rs-dapi-client/Cargo.toml b/packages/rs-dapi-client/Cargo.toml index 36043a61304..85c5174c404 100644 --- a/packages/rs-dapi-client/Cargo.toml +++ b/packages/rs-dapi-client/Cargo.toml @@ -25,7 +25,15 @@ offline-testing = [] backon = { version = "1.3", default-features = false, features = [ "tokio-sleep", ] } -tokio = { version = "1.40", features = ["time"] } +tokio = { version = "1.40", features = ["time", "net"] } +# SOCKS5 proxy support (transport/proxy.rs). +tokio-socks = { version = "0.5.3", default-features = false, features = [ + "tokio", +] } +hyper-util = { version = "0.1.11", default-features = false, features = [ + "tokio", +] } +tower-service = { version = "0.3" } # The native transport (tonic channel + TLS) comes from dapi-grpc's transport # feature; wasm builds use tonic-web-wasm-client instead and must not pull it. dapi-grpc = { path = "../dapi-grpc", features = [ diff --git a/packages/rs-dapi-client/src/dapi_client.rs b/packages/rs-dapi-client/src/dapi_client.rs index 5557af91361..bcbddf61113 100644 --- a/packages/rs-dapi-client/src/dapi_client.rs +++ b/packages/rs-dapi-client/src/dapi_client.rs @@ -13,6 +13,8 @@ use tracing::Instrument; use crate::address_list::AddressListError; use crate::connection_pool::{ConnectionPool, DEFAULT_POOL_CAPACITY}; use crate::request_settings::AppliedRequestSettings; +#[cfg(not(target_arch = "wasm32"))] +use crate::transport::Socks5Proxy; use crate::transport::{self, TransportError}; use crate::{ transport::{TransportClient, TransportRequest}, @@ -113,6 +115,9 @@ pub struct DapiClient { #[cfg(not(target_arch = "wasm32"))] /// Certificate Authority certificate to use for verifying the server's certificate. pub ca_certificate: Option, + #[cfg(not(target_arch = "wasm32"))] + /// SOCKS5 proxy every connection is tunnelled through. + pub proxy: Option, #[cfg(feature = "dump")] pub(crate) dump_dir: Option, } @@ -135,6 +140,8 @@ impl DapiClient { dump_dir: None, #[cfg(not(target_arch = "wasm32"))] ca_certificate: None, + #[cfg(not(target_arch = "wasm32"))] + proxy: None, } } @@ -153,6 +160,21 @@ impl DapiClient { self } + /// Tunnel every connection through a SOCKS5 proxy. + /// + /// There is no fallback to a direct connection. A request that fails + /// because the proxy failed is not retried and does not ban the node + /// (see [TransportError::is_proxy_failure]). + /// + /// # Returns + /// [DapiClient] with the proxy set. + #[cfg(not(target_arch = "wasm32"))] + pub fn with_proxy(mut self, proxy: Socks5Proxy) -> Self { + self.proxy = Some(proxy); + + self + } + /// Return the [DapiClient] address list. pub fn address_list(&self) -> &AddressList { &self.address_list @@ -284,6 +306,8 @@ mod tests { max_decoding_message_size: None, #[cfg(not(target_arch = "wasm32"))] ca_certificate: None, + #[cfg(not(target_arch = "wasm32"))] + proxy: None, } } @@ -690,6 +714,21 @@ mod tests { assert!(client.ca_certificate.is_some()); } + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn test_dapi_client_with_proxy() { + use crate::transport::{ProxyEndpoint, Socks5Auth}; + + let address_list: AddressList = "http://127.0.0.1:3000".parse().unwrap(); + let proxy = Socks5Proxy { + endpoint: ProxyEndpoint::Tcp("127.0.0.1:9050".parse().unwrap()), + auth: Socks5Auth::RandomPerConnection, + }; + let client = + DapiClient::new(address_list, RequestSettings::default()).with_proxy(proxy.clone()); + assert_eq!(client.proxy, Some(proxy)); + } + #[cfg(feature = "mocks")] #[test] fn test_dapi_client_error_mock_serialize_deserialize() { @@ -1075,7 +1114,9 @@ impl DapiRequestExecutor for DapiClient { .override_by(settings) .finalize(); #[cfg(not(target_arch = "wasm32"))] - let applied_settings = applied_settings.with_ca_certificate(self.ca_certificate.clone()); + let applied_settings = applied_settings + .with_ca_certificate(self.ca_certificate.clone()) + .with_proxy(self.proxy.clone()); // Save dump dir for later use #[cfg(feature = "dump")] diff --git a/packages/rs-dapi-client/src/request_settings.rs b/packages/rs-dapi-client/src/request_settings.rs index b503d21e9e5..49b6be33a6d 100644 --- a/packages/rs-dapi-client/src/request_settings.rs +++ b/packages/rs-dapi-client/src/request_settings.rs @@ -1,5 +1,7 @@ //! DAPI client request settings processing. +#[cfg(not(target_arch = "wasm32"))] +use crate::transport::Socks5Proxy; #[cfg(not(target_arch = "wasm32"))] use dapi_grpc::tonic::transport::Certificate; use std::time::Duration; @@ -26,7 +28,9 @@ pub struct RequestSettings { /// It is sent to the server as the `grpc-timeout` header. On native targets /// it also bounds the whole attempt on the client: an attempt still running /// after `timeout + connect_timeout` fails with `DeadlineExceeded` and is - /// retried like any other retryable error. For unary RPCs the attempt runs + /// retried like any other retryable error (through a proxy without a + /// `connect_timeout`, the proxy's 20-second connect default counts as the + /// `connect_timeout` here). For unary RPCs the attempt runs /// from dispatch through the response body and trailers; for streaming /// RPCs it ends when the response headers arrive, so consuming the /// returned stream is not bounded by it. Zero disables both limits. @@ -83,6 +87,8 @@ impl RequestSettings { max_decoding_message_size: self.max_decoding_message_size, #[cfg(not(target_arch = "wasm32"))] ca_certificate: None, + #[cfg(not(target_arch = "wasm32"))] + proxy: None, } } } @@ -107,6 +113,9 @@ pub struct AppliedRequestSettings { /// Certificate Authority certificate to use for verifying the server's certificate. #[cfg(not(target_arch = "wasm32"))] pub ca_certificate: Option, + /// SOCKS5 proxy every connection is tunnelled through. + #[cfg(not(target_arch = "wasm32"))] + pub proxy: Option, } impl AppliedRequestSettings { /// Use provided CA certificate for verifying the server's certificate. @@ -130,13 +139,33 @@ impl AppliedRequestSettings { } Some( self.timeout - .saturating_add(self.connect_timeout.unwrap_or_default()), + .saturating_add(self.effective_connect_timeout().unwrap_or_default()), ) } + /// The connect budget the transport applies: `connect_timeout`, or, + /// through a proxy without one, the proxy default (a proxy that accepts + /// the connection and never answers would otherwise hang the request). + pub(crate) fn effective_connect_timeout(&self) -> Option { + #[cfg(not(target_arch = "wasm32"))] + if self.connect_timeout.is_none() && self.proxy.is_some() { + return Some(crate::transport::PROXY_CONNECT_TIMEOUT); + } + self.connect_timeout + } + + /// Tunnel every connection through the given SOCKS5 proxy. + /// + /// If set to None, connections are made directly. + #[cfg(not(target_arch = "wasm32"))] + pub fn with_proxy(mut self, proxy: Option) -> Self { + self.proxy = proxy; + self + } + /// Cache key fragment for the [ConnectionPool](crate::ConnectionPool), /// covering only the fields that affect the constructed transport client: - /// connect timeout, response decoding limit and CA certificate. + /// connect timeout, response decoding limit, CA certificate and proxy. /// Per-request knobs (request timeout, retries, address banning) are /// deliberately excluded so requests that differ only in those reuse the /// same pooled connection. @@ -154,6 +183,8 @@ impl AppliedRequestSettings { max_decoding_message_size, #[cfg(not(target_arch = "wasm32"))] ca_certificate, + #[cfg(not(target_arch = "wasm32"))] + proxy, } = self; // The wasm channel builder ignores `connect_timeout`, so it does not @@ -178,9 +209,15 @@ impl AppliedRequestSettings { #[cfg(target_arch = "wasm32")] let ca_certificate: Option = None; + // A proxied and a direct channel to one node never share a key. + #[cfg(not(target_arch = "wasm32"))] + let proxy = proxy.as_ref().map(Socks5Proxy::connection_key); + #[cfg(target_arch = "wasm32")] + let proxy: Option = None; + format!( - "connect_timeout={:?},max_decoding_message_size={:?},ca_certificate={:?}", - connect_timeout, max_decoding_message_size, ca_certificate + "connect_timeout={:?},max_decoding_message_size={:?},ca_certificate={:?},proxy={:?}", + connect_timeout, max_decoding_message_size, ca_certificate, proxy ) } } @@ -292,6 +329,41 @@ mod tests { assert_eq!(default.attempt_deadline(), Some(Duration::from_secs(10))); } + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn should_count_the_proxy_connect_default_in_the_attempt_deadline() { + use crate::transport::{ProxyEndpoint, Socks5Auth}; + + let proxied = RequestSettings::default() + .finalize() + .with_proxy(Some(Socks5Proxy { + endpoint: ProxyEndpoint::Tcp("127.0.0.1:9050".parse().expect("address")), + auth: Socks5Auth::None, + })); + assert_eq!( + proxied.effective_connect_timeout(), + Some(crate::transport::PROXY_CONNECT_TIMEOUT) + ); + assert_eq!( + proxied.attempt_deadline(), + Some(DEFAULT_TIMEOUT + crate::transport::PROXY_CONNECT_TIMEOUT) + ); + + let explicit = RequestSettings { + connect_timeout: Some(Duration::from_secs(3)), + ..RequestSettings::default() + } + .finalize() + .with_proxy(proxied.proxy.clone()); + assert_eq!( + explicit.effective_connect_timeout(), + Some(Duration::from_secs(3)) + ); + + let direct = RequestSettings::default().finalize(); + assert_eq!(direct.effective_connect_timeout(), None); + } + #[test] fn should_not_bound_attempt_when_timeout_is_zero() { let applied = RequestSettings { @@ -366,4 +438,34 @@ mod tests { .with_ca_certificate(Some(Certificate::from_pem("other-pem-data"))); assert_ne!(with_ca.connection_key(), with_other_ca.connection_key()); } + + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn test_connection_key_differs_on_proxy() { + use crate::transport::{ProxyEndpoint, Socks5Auth}; + + let with_proxy = |auth| { + RequestSettings::default() + .finalize() + .with_proxy(Some(Socks5Proxy { + endpoint: ProxyEndpoint::Tcp("127.0.0.1:9050".parse().expect("address")), + auth, + })) + .connection_key() + }; + let password = |password: &str| Socks5Auth::Password { + username: "alice".to_string(), + password: password.to_string(), + }; + let direct = RequestSettings::default().finalize().connection_key(); + + assert_ne!(direct, with_proxy(Socks5Auth::None)); + assert_ne!( + with_proxy(Socks5Auth::None), + with_proxy(Socks5Auth::RandomPerConnection) + ); + assert_eq!(with_proxy(password("one")), with_proxy(password("one"))); + assert_ne!(with_proxy(password("one")), with_proxy(password("two"))); + assert!(!with_proxy(password("secret")).contains("secret")); + } } diff --git a/packages/rs-dapi-client/src/transport.rs b/packages/rs-dapi-client/src/transport.rs index bad291c6b26..78e9a11110d 100644 --- a/packages/rs-dapi-client/src/transport.rs +++ b/packages/rs-dapi-client/src/transport.rs @@ -2,6 +2,8 @@ pub(crate) mod grpc; #[cfg(not(target_arch = "wasm32"))] +mod proxy; +#[cfg(not(target_arch = "wasm32"))] pub(crate) mod tonic_channel; #[cfg(target_arch = "wasm32")] pub(crate) mod wasm_channel; @@ -15,6 +17,10 @@ use std::any; use std::fmt::Debug; use std::time::Duration; +#[cfg(not(target_arch = "wasm32"))] +pub(crate) use proxy::PROXY_CONNECT_TIMEOUT; +#[cfg(not(target_arch = "wasm32"))] +pub use proxy::{ProxyEndpoint, ProxyError, Socks5Auth, Socks5Proxy}; #[cfg(not(target_arch = "wasm32"))] pub use tonic_channel::{ create_channel, CoreGrpcClient, PlatformGrpcClient, TokioBackonSleeper as BackonSleeper, @@ -85,17 +91,23 @@ pub enum TransportError { impl Clone for TransportError { fn clone(&self) -> Self { match self { - TransportError::Grpc(status) => { - // tonic::Status doesn't implement Clone, so we reconstruct it - // from its components. Note: this loses the original error source. - let cloned_status = dapi_grpc::tonic::Status::with_details_and_metadata( - status.code(), - status.message(), - status.details().to_vec().into(), - status.metadata().clone(), - ); - TransportError::Grpc(cloned_status) - } + // Keeps the error source, which carries a [ProxyError]. + TransportError::Grpc(status) => TransportError::Grpc(status.clone()), + } + } +} + +impl TransportError { + /// Whether the request failed because the SOCKS5 proxy failed rather + /// than the node (a [ProxyError] is in the error's source chain). Such + /// a failure says nothing about the node, so it is neither retried nor + /// held against it. + pub fn is_proxy_failure(&self) -> bool { + match self { + #[cfg(not(target_arch = "wasm32"))] + TransportError::Grpc(status) => proxy::is_proxy_failure(status), + #[cfg(target_arch = "wasm32")] + TransportError::Grpc(_) => false, } } } diff --git a/packages/rs-dapi-client/src/transport/grpc.rs b/packages/rs-dapi-client/src/transport/grpc.rs index ce02519e845..f985deb1438 100644 --- a/packages/rs-dapi-client/src/transport/grpc.rs +++ b/packages/rs-dapi-client/src/transport/grpc.rs @@ -3,6 +3,8 @@ use std::time::Duration; use super::create_channel; +#[cfg(not(target_arch = "wasm32"))] +use super::proxy::is_proxy_failure; use super::{CanRetry, TransportClient, TransportError, TransportRequest}; use super::{CoreGrpcClient, PlatformGrpcClient}; use crate::connection_pool::{ConnectionPool, PoolPrefix}; @@ -13,6 +15,10 @@ use dapi_grpc::tonic::{IntoRequest, Streaming}; use futures::{future::BoxFuture, FutureExt, TryFutureExt}; impl TransportClient for PlatformGrpcClient { + /// Ignores any proxy the [DapiClient](crate::DapiClient) has: without + /// settings the channel connects directly. Use + /// [with_uri_and_settings](TransportClient::with_uri_and_settings) with + /// the client's settings to connect through its proxy. fn with_uri(uri: Uri, pool: &ConnectionPool) -> Result { Ok(pool .get_or_create(PoolPrefix::Platform, &uri, None, || { @@ -61,6 +67,10 @@ impl TransportClient for PlatformGrpcClient { } impl TransportClient for CoreGrpcClient { + /// Ignores any proxy the [DapiClient](crate::DapiClient) has: without + /// settings the channel connects directly. Use + /// [with_uri_and_settings](TransportClient::with_uri_and_settings) with + /// the client's settings to connect through its proxy. fn with_uri(uri: Uri, pool: &ConnectionPool) -> Result { Ok(pool .get_or_create(PoolPrefix::Core, &uri, None, || { @@ -110,6 +120,13 @@ impl TransportClient for CoreGrpcClient { impl CanRetry for dapi_grpc::tonic::Status { fn can_retry(&self) -> bool { + // A failing SOCKS5 proxy fails every node alike: retrying elsewhere + // cannot help, and the node must not be banned for it. + #[cfg(not(target_arch = "wasm32"))] + if is_proxy_failure(self) { + return false; + } + let code = self.code(); use dapi_grpc::tonic::Code::*; diff --git a/packages/rs-dapi-client/src/transport/proxy.rs b/packages/rs-dapi-client/src/transport/proxy.rs new file mode 100644 index 00000000000..b8875eb7b6b --- /dev/null +++ b/packages/rs-dapi-client/src/transport/proxy.rs @@ -0,0 +1,544 @@ +//! SOCKS5 proxy (RFC 1928) for DAPI connections. +//! +//! A [DapiClient](crate::DapiClient) configured with a [Socks5Proxy] opens +//! every connection through it: the evonode's host goes to the proxy +//! unresolved (an IP literal as such, anything else as a domain name the +//! proxy resolves), so there is never a local DNS lookup for an endpoint, +//! and TLS still runs end to end with the evonode, validated against the +//! endpoint host. There is no fallback to a direct connection. + +use std::error::Error as StdError; +use std::fmt; +use std::future::Future; +use std::hash::BuildHasher; +use std::net::{IpAddr, SocketAddr}; +#[cfg(unix)] +use std::path::PathBuf; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::task::{Context, Poll}; +use std::time::Duration; + +use dapi_grpc::tonic::Status; +use futures::future::{select, Either}; +use hyper_util::rt::TokioIo; +use rand::RngCore; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use tokio::net::TcpStream; +#[cfg(unix)] +use tokio::net::UnixStream; +use tokio_socks::tcp::Socks5Stream; +use tokio_socks::{Error as SocksError, TargetAddr}; + +use super::tonic_channel::unbracketed; +use crate::Uri; + +type BoxError = Box; + +/// How long the proxy itself may take to accept the connection and to +/// complete the SOCKS5 negotiation, method selection and any +/// authentication (Dash Core's default `-timeout`). Running out of it is a +/// [ProxyError]; the rest of the connect budget covers the CONNECT reply. +const PROXY_ANSWER_TIMEOUT: Duration = Duration::from_secs(5); + +/// Connect budget through a proxy when the settings give none: a request's +/// own deadline only starts once the connection is up, so a proxy that +/// accepts the TCP connection and never answers would otherwise hang the +/// request. Dash Core waits as long for each SOCKS5 reply. +pub(crate) const PROXY_CONNECT_TIMEOUT: Duration = Duration::from_secs(20); + +/// Stand-in for a connect budget too large to add to the current instant. +const UNLIMITED: Duration = Duration::from_secs(100 * 365 * 24 * 60 * 60); + +/// A SOCKS5 proxy every DAPI connection is tunnelled through. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Socks5Proxy { + /// Where the proxy listens. + pub endpoint: ProxyEndpoint, + /// How the client authenticates to the proxy. + pub auth: Socks5Auth, +} + +/// Where a [Socks5Proxy] listens. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ProxyEndpoint { + /// A TCP address. Numeric: resolving the proxy's own name is up to the + /// caller. + Tcp(SocketAddr), + /// A Unix domain socket. + #[cfg(unix)] + Unix(PathBuf), +} + +/// How a client authenticates to a [Socks5Proxy]. +/// +/// With credentials, the client offers both "no authentication" and +/// username/password (RFC 1929) and accepts whichever the proxy picks. +#[derive(Clone, PartialEq, Eq)] +pub enum Socks5Auth { + /// No credentials. + None, + /// Fixed username and password, 1 to 255 bytes each. + Password { + /// Username. + username: String, + /// Password. + password: String, + }, + /// Fresh random credentials for every new connection. Tor isolates + /// streams by SOCKS credentials by default (`IsolateSOCKSAuth`), so each + /// connection gets its own circuit. + RandomPerConnection, +} + +/// A request failed because the proxy failed, not the evonode: the proxy +/// is unreachable, broke the SOCKS5 protocol, refused the credentials, or +/// answered with a general failure or an unsupported command. It is carried +/// in the source chain of the transport error; see +/// [TransportError::is_proxy_failure](super::TransportError::is_proxy_failure). +/// +/// Such a failure is not retried and does not ban the evonode: every +/// endpoint shares the proxy. That includes a general failure, which Tor +/// also sends when a single circuit fails; the request fails at once rather +/// than being retried on another node without a ban, and the next request +/// (with [Socks5Auth::RandomPerConnection], on a fresh circuit) tries again. +/// +/// A reply about the destination (unreachable, refused, TTL expired, not +/// allowed by the proxy's rules, address type not supported) produces an +/// ordinary, retryable connection error instead, as does the connect +/// timeout once the proxy has completed the SOCKS5 negotiation (the CONNECT +/// reply is pending): a proxy that holds its reply while it builds the route (Tor does while bootstrapping) cannot +/// be told apart from a slow node, and Dash Core counts it against the peer +/// too. A proxy that does not accept the TCP connection, or does not finish +/// the SOCKS5 negotiation (method selection and any authentication), within +/// five seconds or the connect budget, whichever is shorter, is the proxy's +/// failure: that is a proxy that is stuck (a suspended Tor), not a route +/// being built. +#[derive(Debug, thiserror::Error)] +#[error("SOCKS5 proxy {proxy}: {source}")] +pub struct ProxyError { + proxy: String, + #[source] + source: BoxError, +} + +/// A process-local keyed hash of a proxy password, so that two fixed +/// passwords never share a pooled connection (Tor isolates streams by +/// username and password) while the pool key never holds the password or a +/// digest of it that could be checked offline. +fn password_identity(password: &str) -> u64 { + static KEY: OnceLock = OnceLock::new(); + KEY.get_or_init(Default::default).hash_one(password) +} + +impl Socks5Proxy { + /// The part of the connection pool key this proxy contributes. Never + /// includes the password, only a process-local keyed hash of it. + pub(crate) fn connection_key(&self) -> String { + let auth = match &self.auth { + Socks5Auth::None => "none".to_string(), + Socks5Auth::Password { username, password } => { + format!("password {username} {:016x}", password_identity(password)) + } + Socks5Auth::RandomPerConnection => "random".to_string(), + }; + format!("{} {auth}", self.endpoint) + } + + /// Opens a tunnel to `uri`'s host through the proxy within + /// `connect_timeout`: the proxy gets [PROXY_ANSWER_TIMEOUT] (or less, if + /// the budget is shorter) to accept and negotiate, and running out of it + /// is the proxy's failure; the CONNECT reply gets what is left, and + /// running out of that is the destination's. + async fn connect( + self, + uri: Uri, + connect_timeout: Duration, + ) -> Result, BoxError> { + let target = target(&uri)?; + let now = tokio::time::Instant::now(); + // A budget too large to add (an "unlimited" Duration::MAX) is + // capped rather than overflowing. + let deadline = now + .checked_add(connect_timeout) + .unwrap_or_else(|| now + UNLIMITED); + let negotiated_by = deadline.min(now + PROXY_ANSWER_TIMEOUT); + let proxy_error = |source: BoxError| ProxyError { + proxy: self.endpoint.to_string(), + source, + }; + let not_accepted = || proxy_error("the proxy did not accept the connection in time".into()); + match &self.endpoint { + ProxyEndpoint::Tcp(address) => { + let socket = tokio::time::timeout_at(negotiated_by, TcpStream::connect(address)) + .await + .map_err(|_| not_accepted())? + .map_err(|e| proxy_error(e.into()))?; + socket + .set_nodelay(true) + .map_err(|e| proxy_error(e.into()))?; + let stream = self + .handshake_timed(socket, target, negotiated_by, deadline, proxy_error) + .await?; + Ok(Box::new(stream)) + } + #[cfg(unix)] + ProxyEndpoint::Unix(path) => { + let socket = tokio::time::timeout_at(negotiated_by, UnixStream::connect(path)) + .await + .map_err(|_| not_accepted())? + .map_err(|e| proxy_error(e.into()))?; + let stream = self + .handshake_timed(socket, target, negotiated_by, deadline, proxy_error) + .await?; + Ok(Box::new(stream)) + } + } + } + + /// [handshake](Self::handshake), failing as the proxy's when the + /// negotiation (method selection and any authentication) is not complete + /// by `negotiated_by`, and as the destination's when the CONNECT reply, + /// which only then is asked for, is not in by `deadline`. + async fn handshake_timed( + &self, + socket: S, + target: TargetAddr<'static>, + negotiated_by: tokio::time::Instant, + deadline: tokio::time::Instant, + proxy_error: impl Fn(BoxError) -> ProxyError, + ) -> Result { + let negotiated = Arc::new(AtomicBool::new(false)); + let socket = Negotiation::new(socket, negotiated.clone()); + let handshake = Box::pin(self.handshake(socket, target, &proxy_error)); + let stalled = || -> BoxError { + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "no SOCKS5 CONNECT reply within the connect timeout", + ) + .into() + }; + match select(handshake, Box::pin(tokio::time::sleep_until(negotiated_by))).await { + Either::Left((result, _)) => result, + Either::Right(((), handshake)) if negotiated.load(Ordering::Relaxed) => { + tokio::time::timeout_at(deadline, handshake) + .await + .map_err(|_| stalled())? + } + Either::Right(_) => { + Err(proxy_error("no complete SOCKS5 negotiation in time".into()).into()) + } + } + .map(|stream| stream.inner) + } + + async fn handshake( + &self, + socket: S, + target: TargetAddr<'static>, + proxy_error: impl Fn(BoxError) -> ProxyError, + ) -> Result { + let result = match &self.auth { + Socks5Auth::None => Socks5Stream::connect_with_socket(socket, target.clone()).await, + Socks5Auth::Password { username, password } => { + Socks5Stream::connect_with_password_and_socket( + socket, + target.clone(), + username, + password, + ) + .await + } + Socks5Auth::RandomPerConnection => { + let (username, password) = (random_credential(), random_credential()); + Socks5Stream::connect_with_password_and_socket( + socket, + target.clone(), + &username, + &password, + ) + .await + } + }; + match result { + Ok(stream) => Ok(stream.into_inner()), + // Replies about the destination: the evonode, or the route to it, + // failed. Tor reports its exit and entry policy refusals as "not + // allowed by ruleset". + Err( + e @ (SocksError::HostUnreachable + | SocksError::ConnectionRefused + | SocksError::NetworkUnreachable + | SocksError::TtlExpired + | SocksError::ConnectionNotAllowedByRuleset + | SocksError::AddressTypeNotSupported), + ) => Err(e.into()), + // Everything else is the proxy's, including a general failure, + // which is how Tor answers while it cannot build circuits, and an + // invalid address, which can only come from the proxy's reply + // since the target was checked before connecting. + Err(e) => Err(proxy_error(e.into()).into()), + } + } +} + +/// The CONNECT target for `uri`: an IP literal as an address, any other +/// host as a domain name for the proxy to resolve. An unusable host is the +/// endpoint's fault, not the proxy's. +fn target(uri: &Uri) -> Result, BoxError> { + let host = unbracketed(uri.host().ok_or("the endpoint URI has no host")?); + let default_port = match uri.scheme_str() { + Some("http") => 80, + _ => 443, + }; + let port = uri.port_u16().unwrap_or(default_port); + if let Ok(ip) = host.parse::() { + return Ok(TargetAddr::Ip(SocketAddr::new(ip, port))); + } + if host.len() > 255 { + return Err(format!("the endpoint host is longer than 255 bytes: {host}").into()); + } + Ok(TargetAddr::Domain(host.to_string().into(), port)) +} + +/// 16 random bytes, hex encoded. +fn random_credential() -> String { + let mut bytes = [0u8; 16]; + rand::rngs::OsRng.fill_bytes(&mut bytes); + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +/// Whether `status` failed because of the proxy (a [ProxyError] is in its +/// source chain). +pub(crate) fn is_proxy_failure(status: &Status) -> bool { + let mut error: Option<&(dyn StdError + 'static)> = Some(status); + while let Some(current) = error { + if current.is::() { + return true; + } + error = current.source(); + } + false +} + +impl fmt::Display for ProxyEndpoint { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ProxyEndpoint::Tcp(address) => write!(f, "{address}"), + #[cfg(unix)] + ProxyEndpoint::Unix(path) => write!(f, "unix:{}", path.display()), + } + } +} + +// Request settings, and with them the proxy, are logged at debug level. +impl fmt::Debug for Socks5Auth { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Socks5Auth::None => write!(f, "None"), + Socks5Auth::Password { username, .. } => f + .debug_struct("Password") + .field("username", username) + .field("password", &"") + .finish(), + Socks5Auth::RandomPerConnection => write!(f, "RandomPerConnection"), + } + } +} + +/// A stream to the proxy that notes when the SOCKS5 negotiation is +/// complete: the proxy's method selection (RFC 1928 §3, two bytes) and, when +/// it chose username/password, its authentication status (RFC 1929, two +/// more). Only then does the client send CONNECT, so a stall before that is +/// the proxy's, not the destination's. +struct Negotiation { + inner: S, + negotiated: Arc, + /// Bytes the proxy has sent so far, up to the end of the negotiation. + received: usize, + /// Bytes the negotiation takes: 2, or 4 once username/password is chosen. + expected: usize, +} + +impl Negotiation { + fn new(inner: S, negotiated: Arc) -> Self { + Self { + inner, + negotiated, + received: 0, + expected: 2, + } + } + + fn note(&mut self, bytes: &[u8]) { + for &byte in bytes { + if self.received >= self.expected { + break; + } + // The second byte of the method selection is the method chosen. + if self.received == 1 && byte == 0x02 { + self.expected = 4; + } + self.received += 1; + } + if self.received >= self.expected { + self.negotiated.store(true, Ordering::Relaxed); + } + } +} + +impl AsyncRead for Negotiation { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let before = buf.filled().len(); + let poll = Pin::new(&mut self.inner).poll_read(cx, buf); + if buf.filled().len() > before && self.received < self.expected { + self.note(&buf.filled()[before..]); + } + poll + } +} + +impl AsyncWrite for Negotiation { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.inner).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } + + fn poll_write_vectored( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &[std::io::IoSlice<'_>], + ) -> Poll> { + Pin::new(&mut self.inner).poll_write_vectored(cx, bufs) + } + + fn is_write_vectored(&self) -> bool { + self.inner.is_write_vectored() + } +} + +/// A connected tunnel, TCP or Unix underneath. +pub(crate) trait ProxiedStream: AsyncRead + AsyncWrite + Send + Unpin {} +impl ProxiedStream for S {} + +/// The tonic connector that tunnels every connection through the proxy +/// within the connect budget (see [Socks5Proxy::connect]). +#[derive(Clone)] +pub(crate) struct Socks5Connector { + pub(crate) proxy: Socks5Proxy, + pub(crate) connect_timeout: Duration, +} + +impl tower_service::Service for Socks5Connector { + type Response = TokioIo>; + type Error = BoxError; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, uri: Uri) -> Self::Future { + let proxy = self.proxy.clone(); + let connect_timeout = self.connect_timeout; + Box::pin(async move { proxy.connect(uri, connect_timeout).await.map(TokioIo::new) }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn proxy(auth: Socks5Auth) -> Socks5Proxy { + Socks5Proxy { + endpoint: ProxyEndpoint::Tcp("127.0.0.1:9050".parse().expect("address")), + auth, + } + } + + #[test] + fn should_redact_the_password_in_debug_output() { + let debug = format!( + "{:?}", + proxy(Socks5Auth::Password { + username: "alice".to_string(), + password: "hunter2".to_string(), + }) + ); + assert!(debug.contains("alice"), "{debug}"); + assert!(!debug.contains("hunter2"), "{debug}"); + } + + #[test] + fn should_leave_the_password_out_of_the_connection_key() { + let with = |password: &str| { + proxy(Socks5Auth::Password { + username: "alice".to_string(), + password: password.to_string(), + }) + .connection_key() + }; + assert_eq!(with("one"), with("one")); + assert_ne!(with("one"), with("two")); + assert!(!with("hunter2").contains("hunter2")); + assert_ne!( + proxy(Socks5Auth::None).connection_key(), + proxy(Socks5Auth::RandomPerConnection).connection_key() + ); + } + + #[test] + fn should_draw_fresh_hex_credentials() { + let (first, second) = (random_credential(), random_credential()); + assert_eq!(first.len(), 32); + assert!(first.chars().all(|c| c.is_ascii_hexdigit())); + assert_ne!(first, second); + } + + #[test] + fn should_send_ip_literals_as_addresses_and_names_as_domains() { + let target = |uri: &str| target(&uri.parse().expect("uri")); + assert_eq!( + target("https://[2001:db8::1]:1443").expect("ipv6"), + TargetAddr::Ip("[2001:db8::1]:1443".parse().expect("address")) + ); + assert_eq!( + target("https://10.0.0.1").expect("default port"), + TargetAddr::Ip("10.0.0.1:443".parse().expect("address")) + ); + assert_eq!( + target("https://evo.example:1443").expect("name"), + TargetAddr::Domain("evo.example".into(), 1443) + ); + let overlong = format!("https://{}.example", "a".repeat(250)); + assert!(target(&overlong).is_err()); + } + + #[test] + fn should_find_a_proxy_failure_in_the_source_chain() { + let error = ProxyError { + proxy: "127.0.0.1:9050".to_string(), + source: "refused".into(), + }; + let status = Status::from_error(Box::new(error)); + assert!(is_proxy_failure(&status)); + assert!(is_proxy_failure(&status.clone()), "clones keep the source"); + assert!(!is_proxy_failure(&Status::unavailable("node down"))); + } +} diff --git a/packages/rs-dapi-client/src/transport/tonic_channel.rs b/packages/rs-dapi-client/src/transport/tonic_channel.rs index 2f774b043f0..20afa2778f0 100644 --- a/packages/rs-dapi-client/src/transport/tonic_channel.rs +++ b/packages/rs-dapi-client/src/transport/tonic_channel.rs @@ -1,3 +1,4 @@ +use super::proxy::{Socks5Connector, PROXY_CONNECT_TIMEOUT}; use super::TransportError; use crate::{request_settings::AppliedRequestSettings, Uri}; use dapi_grpc::core::v0::core_client::CoreClient; @@ -20,6 +21,9 @@ const HTTP2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15); const HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(10); /// Create channel (connection) for gRPC transport. +/// +/// With a proxy in `settings`, every connection of the channel is tunnelled +/// through it. pub fn create_channel( uri: Uri, settings: Option<&AppliedRequestSettings>, @@ -48,9 +52,18 @@ pub fn create_channel( tls_config = tls_config.with_native_roots(); } + let proxy = settings.and_then(|settings| settings.proxy.clone()); + let connect_timeout = settings.and_then(AppliedRequestSettings::effective_connect_timeout); if let Some(settings) = settings { - if let Some(timeout) = settings.connect_timeout { - builder = builder.connect_timeout(timeout); + match (connect_timeout, &proxy) { + // Through a proxy the connector enforces the budget itself, so + // it can tell a stuck proxy from a slow destination; TLS, which + // runs after it, gets the same budget of its own (a connection + // can so take up to twice the budget; the executor's attempt + // deadline still bounds the whole attempt). + (Some(timeout), Some(_)) => tls_config = tls_config.timeout(timeout), + (Some(timeout), None) => builder = builder.connect_timeout(timeout), + (None, _) => {} } if let Some(pem) = settings.ca_certificate.as_ref() { @@ -74,7 +87,13 @@ pub fn create_channel( ))) })?; - Ok(builder.connect_lazy()) + Ok(match proxy { + None => builder.connect_lazy(), + Some(proxy) => builder.connect_with_connector_lazy(Socks5Connector { + proxy, + connect_timeout: connect_timeout.unwrap_or(PROXY_CONNECT_TIMEOUT), + }), + }) } /// The host of a URI without the brackets an IPv6 literal carries in it @@ -88,6 +107,7 @@ pub(crate) fn unbracketed(host: &str) -> &str { #[cfg(test)] mod tests { use super::*; + use crate::transport::{ProxyEndpoint, Socks5Auth, Socks5Proxy}; use crate::RequestSettings; #[test] @@ -107,6 +127,12 @@ mod tests { let settings = RequestSettings::default() .finalize() .with_ca_certificate(Some(Certificate::from_pem("fake-pem-data"))); - create_channel(uri, Some(&settings)).expect("explicit CA"); + create_channel(uri.clone(), Some(&settings)).expect("explicit CA"); + + let settings = settings.with_proxy(Some(Socks5Proxy { + endpoint: ProxyEndpoint::Tcp("127.0.0.1:9050".parse().expect("address")), + auth: Socks5Auth::None, + })); + create_channel(uri, Some(&settings)).expect("through a proxy"); } } diff --git a/packages/rs-dapi-client/tests/rate_limit_ban.rs b/packages/rs-dapi-client/tests/rate_limit_ban.rs index 6f63722a1c7..a5dfc6b7ff2 100644 --- a/packages/rs-dapi-client/tests/rate_limit_ban.rs +++ b/packages/rs-dapi-client/tests/rate_limit_ban.rs @@ -26,6 +26,8 @@ fn applied_settings(ban: bool) -> AppliedRequestSettings { max_decoding_message_size: None, #[cfg(not(target_arch = "wasm32"))] ca_certificate: None, + #[cfg(not(target_arch = "wasm32"))] + proxy: None, } } From e7b7a2d9ed2cbfceb93c8f173986a280b6d808a5 Mon Sep 17 00:00:00 2001 From: pasta Date: Sat, 26 Sep 2026 20:14:40 -0500 Subject: [PATCH 3/5] feat(sdk): add SdkBuilder::with_proxy Passes a SOCKS5 proxy to the DapiClient the builder creates, the way with_ca_certificate passes the CA. Native only. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/rs-sdk/src/sdk.rs | 44 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/packages/rs-sdk/src/sdk.rs b/packages/rs-sdk/src/sdk.rs index cd8769b136a..4ba4ffd125d 100644 --- a/packages/rs-sdk/src/sdk.rs +++ b/packages/rs-sdk/src/sdk.rs @@ -27,6 +27,8 @@ use drive_proof_verifier::FromProof; pub use http::Uri; #[cfg(feature = "mocks")] use rs_dapi_client::mock::MockDapiClient; +#[cfg(not(target_arch = "wasm32"))] +use rs_dapi_client::transport::Socks5Proxy; pub use rs_dapi_client::Address; pub use rs_dapi_client::AddressBanInfo; pub use rs_dapi_client::AddressList; @@ -873,6 +875,10 @@ pub struct SdkBuilder { /// CA certificate to use for TLS connections. #[cfg(not(target_arch = "wasm32"))] ca_certificate: Option, + + /// SOCKS5 proxy every DAPI connection is tunnelled through. + #[cfg(not(target_arch = "wasm32"))] + proxy: Option, } impl Default for SdkBuilder { @@ -916,6 +922,8 @@ impl Default for SdkBuilder { protocol_version_observer: None, #[cfg(not(target_arch = "wasm32"))] ca_certificate: None, + #[cfg(not(target_arch = "wasm32"))] + proxy: None, #[cfg(feature = "mocks")] dump_dir: None, @@ -1019,6 +1027,17 @@ impl SdkBuilder { Ok(self.with_ca_certificate(cert)) } + /// Tunnel every DAPI connection through a SOCKS5 proxy. + /// + /// There is no fallback to a direct connection. See + /// [DapiClient::with_proxy] for how proxy failures are handled. Core RPC + /// connections made for [SdkBuilder::with_core] are not proxied. + #[cfg(not(target_arch = "wasm32"))] + pub fn with_proxy(mut self, proxy: Socks5Proxy) -> Self { + self.proxy = Some(proxy); + self + } + /// Configure request settings. /// /// Tune request settings used to connect to the Dash Platform. @@ -1229,6 +1248,10 @@ impl SdkBuilder { if let Some(pem) = self.ca_certificate { dapi = dapi.with_ca_certificate(pem); } + #[cfg(not(target_arch = "wasm32"))] + if let Some(proxy) = self.proxy { + dapi = dapi.with_proxy(proxy); + } #[cfg(feature = "mocks")] let dapi = dapi.dump_dir(self.dump_dir.clone()); @@ -1387,6 +1410,27 @@ mod test { /// Testnet Evo masternodes expose the Platform HTTP endpoint on 1443. const TESTNET_PLATFORM_HTTP_PORT: u16 = 1443; + #[cfg(all(feature = "mocks", not(target_arch = "wasm32")))] + #[test] + fn with_proxy_reaches_the_dapi_client() { + use rs_dapi_client::transport::{ProxyEndpoint, Socks5Auth, Socks5Proxy}; + + let proxy = Socks5Proxy { + endpoint: ProxyEndpoint::Tcp("127.0.0.1:9050".parse().expect("address")), + auth: Socks5Auth::RandomPerConnection, + }; + let sdk = SdkBuilder::new("https://127.0.0.1:1443".parse().expect("address list")) + .with_context_provider(dash_context_provider::MockContextProvider::new()) + .with_proxy(proxy.clone()) + .build() + .expect("sdk"); + + match &sdk.inner { + super::SdkInstance::Dapi { dapi } => assert_eq!(dapi.proxy, Some(proxy)), + super::SdkInstance::Mock { .. } => panic!("expected a network SDK"), + } + } + #[test] fn new_testnet_sources_bootstrap_from_seeds() { let builder = SdkBuilder::new_testnet(); From f6be27b98a30b38a6f576700dd1e53fc255dcac1 Mon Sep 17 00:00:00 2001 From: pasta Date: Sat, 26 Sep 2026 20:14:50 -0500 Subject: [PATCH 4/5] test(sdk): SOCKS5 proxy end-to-end suite on loopback A real DapiClient sends GetStatus through a minimal in-test SOCKS5 server to a TLS gRPC server that serves no methods, so Unimplemented proves TCP, SOCKS5, TLS, HTTP/2 and gRPC all ran through the tunnel. Covers TLS validation through the proxy, fresh random credentials per connection, a no-auth-only proxy, IPv6 literals (ATYP 4), names resolved by the proxy (ATYP 3), a proxy that is down (not retried, nothing banned), a proxy that never answers the greeting (the proxy's failure, within 5 s), a proxy that answers but holds the CONNECT reply (the node's, after the connect timeout), a proxy's general failure versus a node refusal (only the refusal bans), and a Unix-socket proxy. The tests bind loopback ports. The test certificates are a throwaway CA and a server certificate for 127.0.0.1, ::1 and localhost; tests/data/proxy/README.md has the commands that made them. Co-Authored-By: Claude Opus 5.5 (1M context) --- Cargo.lock | 1 + packages/rs-dapi-client/Cargo.toml | 4 + .../rs-dapi-client/tests/data/proxy/README.md | 19 + .../rs-dapi-client/tests/data/proxy/ca.pem | 11 + .../tests/data/proxy/server.key | 5 + .../tests/data/proxy/server.pem | 12 + packages/rs-dapi-client/tests/socks5_proxy.rs | 661 ++++++++++++++++++ 7 files changed, 713 insertions(+) create mode 100644 packages/rs-dapi-client/tests/data/proxy/README.md create mode 100644 packages/rs-dapi-client/tests/data/proxy/ca.pem create mode 100644 packages/rs-dapi-client/tests/data/proxy/server.key create mode 100644 packages/rs-dapi-client/tests/data/proxy/server.pem create mode 100644 packages/rs-dapi-client/tests/socks5_proxy.rs diff --git a/Cargo.lock b/Cargo.lock index 39e1c76bc9a..1fc2a06672e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6430,6 +6430,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tokio-socks", + "tonic 0.14.6", "tonic-web-wasm-client", "tower 0.5.3", "tower-service", diff --git a/packages/rs-dapi-client/Cargo.toml b/packages/rs-dapi-client/Cargo.toml index 85c5174c404..1c0ff1e9178 100644 --- a/packages/rs-dapi-client/Cargo.toml +++ b/packages/rs-dapi-client/Cargo.toml @@ -81,6 +81,10 @@ tokio = { version = "1.40", features = ["macros", "rt", "test-util", "io-util"] h2 = "0.4" hyper-util = { version = "0.1", features = ["tokio"] } tower = { version = "0.5", features = ["util"] } +# tests/socks5_proxy.rs runs a TLS gRPC server on loopback. +tonic = { version = "0.14.2", default-features = false, features = [ + "router", +] } [package.metadata.cargo-machete] diff --git a/packages/rs-dapi-client/tests/data/proxy/README.md b/packages/rs-dapi-client/tests/data/proxy/README.md new file mode 100644 index 00000000000..d0a137c5814 --- /dev/null +++ b/packages/rs-dapi-client/tests/data/proxy/README.md @@ -0,0 +1,19 @@ +# SOCKS5 proxy test certificates + +Used by `tests/socks5_proxy.rs`. A throwaway CA (its key was discarded) and a +server certificate for `IP:127.0.0.1`, `IP:::1` and `DNS:localhost`, both +valid for 100 years. `server.key` is the server's test-only private key. + +Regenerate with: + +```sh +openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:P-256 -nodes -days 36500 \ + -subj "/CN=rs-dapi-client test CA" -keyout ca.key -out ca.pem +openssl req -newkey ec -pkeyopt ec_paramgen_curve:P-256 -nodes \ + -subj "/CN=localhost" -keyout server.key.sec1 -out server.csr +printf 'subjectAltName=IP:127.0.0.1,IP:::1,DNS:localhost\nbasicConstraints=CA:FALSE\nextendedKeyUsage=serverAuth\n' > ext.cnf +openssl x509 -req -in server.csr -CA ca.pem -CAkey ca.key -CAcreateserial \ + -days 36500 -extfile ext.cnf -out server.pem +openssl pkcs8 -topk8 -nocrypt -in server.key.sec1 -out server.key +rm ca.key ca.srl server.csr server.key.sec1 ext.cnf +``` diff --git a/packages/rs-dapi-client/tests/data/proxy/ca.pem b/packages/rs-dapi-client/tests/data/proxy/ca.pem new file mode 100644 index 00000000000..835f685c393 --- /dev/null +++ b/packages/rs-dapi-client/tests/data/proxy/ca.pem @@ -0,0 +1,11 @@ +-----BEGIN CERTIFICATE----- +MIIBmjCCAT+gAwIBAgIUGKaEwr4hJuE6kzmdWlQsMPPOlyEwCgYIKoZIzj0EAwIw +ITEfMB0GA1UEAwwWcnMtZGFwaS1jbGllbnQgdGVzdCBDQTAgFw0yNjA5MjcwMTA0 +NTdaGA8yMTI2MDkwMzAxMDQ1N1owITEfMB0GA1UEAwwWcnMtZGFwaS1jbGllbnQg +dGVzdCBDQTBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABOBnFHX6vlIZGphCY60F +9oTP6iMalDGk8x9Nsw4qMBjVcvAHTOIR582MRoOLxinU15zVq+PI8YUOEyaNeFoe +tA+jUzBRMB0GA1UdDgQWBBQixb/FybX4YltZC50OuUfx8EfzKTAfBgNVHSMEGDAW +gBQixb/FybX4YltZC50OuUfx8EfzKTAPBgNVHRMBAf8EBTADAQH/MAoGCCqGSM49 +BAMCA0kAMEYCIQDNRVK074i2Dn0uBQSStYmZlhoAtKOnkqn8SVQspTDUnQIhANLc +KT7EyJLGLG4gR5y2ViW7SHZbuf+ZJm26HgKG1Ipp +-----END CERTIFICATE----- diff --git a/packages/rs-dapi-client/tests/data/proxy/server.key b/packages/rs-dapi-client/tests/data/proxy/server.key new file mode 100644 index 00000000000..71d8ac3fea5 --- /dev/null +++ b/packages/rs-dapi-client/tests/data/proxy/server.key @@ -0,0 +1,5 @@ +-----BEGIN PRIVATE KEY----- +MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQg47vbuAz42f51KN10 +n9lL4jfsfXJGvQSEKR6x6dYiVwWhRANCAATntdIOlHW3ssiM34HJHOZIE50vG9RX +BxXPaC34aWiMErifSc700BbDSEUVOPU/FKFLQo/roiVuTFOsMBE6YTBU +-----END PRIVATE KEY----- diff --git a/packages/rs-dapi-client/tests/data/proxy/server.pem b/packages/rs-dapi-client/tests/data/proxy/server.pem new file mode 100644 index 00000000000..9b459821c9d --- /dev/null +++ b/packages/rs-dapi-client/tests/data/proxy/server.pem @@ -0,0 +1,12 @@ +-----BEGIN CERTIFICATE----- +MIIBzDCCAXGgAwIBAgIUeTOKPIq3htiJAfBw5gZIZz+uM/UwCgYIKoZIzj0EAwIw +ITEfMB0GA1UEAwwWcnMtZGFwaS1jbGllbnQgdGVzdCBDQTAgFw0yNjA5MjcwMTA0 +NTdaGA8yMTI2MDkwMzAxMDQ1N1owFDESMBAGA1UEAwwJbG9jYWxob3N0MFkwEwYH +KoZIzj0CAQYIKoZIzj0DAQcDQgAE57XSDpR1t7LIjN+ByRzmSBOdLxvUVwcVz2gt ++GlojBK4n0nO9NAWw0hFFTj1PxShS0KP66IlbkxTrDAROmEwVKOBkTCBjjAsBgNV +HREEJTAjhwR/AAABhxAAAAAAAAAAAAAAAAAAAAABgglsb2NhbGhvc3QwCQYDVR0T +BAIwADATBgNVHSUEDDAKBggrBgEFBQcDATAdBgNVHQ4EFgQUuEZoB7QlRh5D52PT +HTVW8O1bn8EwHwYDVR0jBBgwFoAUIsW/xcm1+GJbWQudDrlH8fBH8ykwCgYIKoZI +zj0EAwIDSQAwRgIhAMGPM9qe6i2E1pArC6k68ITCcxD+zGxMhds4YRYqYMPvAiEA +gSX/Q1XMfeQqv3S7U6IRY0CumdRKB9Itsa1GOh2rwlw= +-----END CERTIFICATE----- diff --git a/packages/rs-dapi-client/tests/socks5_proxy.rs b/packages/rs-dapi-client/tests/socks5_proxy.rs new file mode 100644 index 00000000000..7ba80c8366e --- /dev/null +++ b/packages/rs-dapi-client/tests/socks5_proxy.rs @@ -0,0 +1,661 @@ +//! End-to-end SOCKS5 proxy suite on loopback: a real `DapiClient` sends a +//! gRPC request through a minimal SOCKS5 server to a TLS gRPC server. The +//! gRPC server serves no methods, so a request that made it all the way +//! (TCP, SOCKS5, TLS, HTTP/2, gRPC) comes back `Unimplemented`. +//! +//! The certificates in `tests/data/proxy/` are described in its README. +#![cfg(not(target_arch = "wasm32"))] + +use std::io; +use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use dapi_grpc::platform::v0::{get_status_request, GetStatusRequest}; +use dapi_grpc::tonic::transport::{Certificate, Identity, Server, ServerTlsConfig}; +use dapi_grpc::tonic::{Code, Status}; +use rs_dapi_client::transport::{ProxyEndpoint, Socks5Auth, Socks5Proxy, TransportError}; +use rs_dapi_client::{ + Address, AddressList, CanRetry, DapiClient, DapiClientError, DapiRequestExecutor, + ExecutionError, RequestSettings, +}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tonic::service::Routes; +use tonic::transport::server::TcpIncoming; + +const CA: &str = include_str!("data/proxy/ca.pem"); +const SERVER_CERT: &str = include_str!("data/proxy/server.pem"); +const SERVER_KEY: &str = include_str!("data/proxy/server.key"); + +/// SOCKS5 reply codes (RFC 1928 §6). +const SUCCEEDED: u8 = 0x00; +const GENERAL_FAILURE: u8 = 0x01; +const NOT_ALLOWED_BY_RULESET: u8 = 0x02; +const CONNECTION_REFUSED: u8 = 0x05; + +/// What the proxy saw for one connection. +#[derive(Debug, Clone)] +struct Connect { + /// Authentication methods the client offered. + methods: Vec, + /// RFC 1929 username, if the proxy chose username/password. + username: Option, + /// Address type of the CONNECT target. + atyp: u8, + target: String, +} + +type Log = Arc>>; + +#[derive(Clone, Copy)] +struct ProxyBehaviour { + /// Accept "no authentication" only, like a proxy without credentials. + no_auth_only: bool, + /// Reply code for the CONNECT; anything but `SUCCEEDED` closes. + reply: u8, +} + +impl Default for ProxyBehaviour { + fn default() -> Self { + Self { + no_auth_only: false, + reply: SUCCEEDED, + } + } +} + +/// Starts the TLS gRPC server ("the evonode"). +async fn start_node() -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind node"); + let address = listener.local_addr().expect("node address"); + let tls = ServerTlsConfig::new().identity(Identity::from_pem(SERVER_CERT, SERVER_KEY)); + let server = Server::builder() + .tls_config(tls) + .expect("server TLS") + .add_routes(Routes::default()); + tokio::spawn(server.serve_with_incoming(TcpIncoming::from(listener))); + address +} + +/// Starts a SOCKS5 proxy on loopback TCP that splices every accepted +/// CONNECT to `upstream`, whatever its target. +async fn start_proxy(upstream: SocketAddr, behaviour: ProxyBehaviour) -> (SocketAddr, Log) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind proxy"); + let address = listener.local_addr().expect("proxy address"); + let log = Log::default(); + let connections = log.clone(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + tokio::spawn(serve_socks( + stream, + upstream, + behaviour, + connections.clone(), + )); + } + }); + (address, log) +} + +async fn serve_socks( + mut client: S, + upstream: SocketAddr, + behaviour: ProxyBehaviour, + log: Log, +) -> io::Result<()> { + let mut greeting = [0u8; 2]; + client.read_exact(&mut greeting).await?; + let mut methods = vec![0u8; greeting[1] as usize]; + client.read_exact(&mut methods).await?; + + let mut username = None; + if methods.contains(&0x02) && !behaviour.no_auth_only { + client.write_all(&[0x05, 0x02]).await?; + let mut header = [0u8; 2]; + client.read_exact(&mut header).await?; + let mut name = vec![0u8; header[1] as usize]; + client.read_exact(&mut name).await?; + let mut password_len = [0u8; 1]; + client.read_exact(&mut password_len).await?; + let mut password = vec![0u8; password_len[0] as usize]; + client.read_exact(&mut password).await?; + username = Some(String::from_utf8_lossy(&name).into_owned()); + client.write_all(&[0x01, 0x00]).await?; + } else if methods.contains(&0x00) { + client.write_all(&[0x05, 0x00]).await?; + } else { + return client.write_all(&[0x05, 0xff]).await; + } + + let mut request = [0u8; 4]; + client.read_exact(&mut request).await?; + let atyp = request[3]; + let host = match atyp { + 0x01 => { + let mut ip = [0u8; 4]; + client.read_exact(&mut ip).await?; + Ipv4Addr::from(ip).to_string() + } + 0x03 => { + let mut len = [0u8; 1]; + client.read_exact(&mut len).await?; + let mut name = vec![0u8; len[0] as usize]; + client.read_exact(&mut name).await?; + String::from_utf8_lossy(&name).into_owned() + } + 0x04 => { + let mut ip = [0u8; 16]; + client.read_exact(&mut ip).await?; + format!("[{}]", Ipv6Addr::from(ip)) + } + other => return Err(io::Error::other(format!("address type {other}"))), + }; + let port = client.read_u16().await?; + log.lock().expect("log").push(Connect { + methods, + username, + atyp, + target: format!("{host}:{port}"), + }); + + client + .write_all(&[0x05, behaviour.reply, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await?; + if behaviour.reply == SUCCEEDED { + let mut node = TcpStream::connect(upstream).await?; + tokio::io::copy_bidirectional(&mut client, &mut node).await?; + } + Ok(()) +} + +fn tcp_proxy(address: SocketAddr, auth: Socks5Auth) -> Socks5Proxy { + Socks5Proxy { + endpoint: ProxyEndpoint::Tcp(address), + auth, + } +} + +fn get_status_request() -> GetStatusRequest { + GetStatusRequest { + version: Some(get_status_request::Version::V0( + get_status_request::GetStatusRequestV0 {}, + )), + } +} + +/// A client over `uris` (comma-separated) through `proxy`, trusting the +/// test CA when `trust_ca`. +fn client(uris: &str, proxy: Socks5Proxy, retries: usize, trust_ca: bool) -> DapiClient { + let addresses: AddressList = uris.parse().expect("address list"); + let settings = RequestSettings { + connect_timeout: Some(Duration::from_secs(5)), + timeout: Some(Duration::from_secs(5)), + retries: Some(retries), + ..RequestSettings::default() + }; + let client = DapiClient::new(addresses, settings).with_proxy(proxy); + if trust_ca { + client.with_ca_certificate(Certificate::from_pem(CA)) + } else { + client + } +} + +/// A client over one `uri` through `proxy` that waits `connect_timeout` +/// for a connection and never retries. +fn client_with_connect_timeout( + uri: &str, + proxy: Socks5Proxy, + connect_timeout: Duration, +) -> DapiClient { + let settings = RequestSettings { + connect_timeout: Some(connect_timeout), + timeout: Some(Duration::from_secs(5)), + retries: Some(0), + ..RequestSettings::default() + }; + DapiClient::new(uri.parse().expect("address list"), settings).with_proxy(proxy) +} + +/// Sends one GetStatus; the node serves nothing, so the request always +/// fails and the failure is returned. +async fn get_status(client: &DapiClient) -> ExecutionError { + client + .execute(get_status_request(), RequestSettings::default()) + .await + .expect_err("the test node serves no methods") +} + +fn transport_error(error: &ExecutionError) -> &TransportError { + match &error.inner { + DapiClientError::Transport(transport) => transport, + other => panic!("expected a transport error, got {other:?}"), + } +} + +fn status(error: &ExecutionError) -> &Status { + let TransportError::Grpc(status) = transport_error(error); + status +} + +fn connects(log: &Log) -> Vec { + log.lock().expect("log").clone() +} + +fn is_banned(client: &DapiClient, uri: &str) -> bool { + let address: Address = uri.parse().expect("address"); + client.address_list().is_banned(&address) +} + +#[tokio::test] +async fn should_send_the_request_through_the_proxy_over_tls() { + let node = start_node().await; + let (proxy, log) = start_proxy(node, ProxyBehaviour::default()).await; + let client = client( + &format!("https://{node}"), + tcp_proxy(proxy, Socks5Auth::None), + 0, + true, + ); + + let error = get_status(&client).await; + + assert_eq!(status(&error).code(), Code::Unimplemented, "{error:?}"); + let seen = connects(&log); + assert_eq!(seen.len(), 1, "{seen:?}"); + assert_eq!(seen[0].methods, vec![0x00]); + assert_eq!(seen[0].atyp, 0x01); + assert_eq!(seen[0].target, node.to_string()); +} + +#[tokio::test] +async fn should_validate_the_node_certificate_through_the_proxy() { + let node = start_node().await; + let (proxy, _) = start_proxy(node, ProxyBehaviour::default()).await; + let client = client( + &format!("https://{node}"), + tcp_proxy(proxy, Socks5Auth::None), + 0, + false, + ); + + let error = get_status(&client).await; + + let status = status(&error); + assert_eq!(status.code(), Code::Unavailable, "{status:?}"); + assert!( + status.message().contains("invalid peer certificate"), + "{status:?}" + ); + assert!(!transport_error(&error).is_proxy_failure()); +} + +#[tokio::test] +async fn should_isolate_every_connection_with_fresh_credentials() { + let node = start_node().await; + let (proxy, log) = start_proxy(node, ProxyBehaviour::default()).await; + let port = node.port(); + // Two endpoints, two connections: the IPv4 and the IPv6 loopback. + for uri in [ + format!("https://127.0.0.1:{port}"), + format!("https://[::1]:{port}"), + ] { + let client = client( + &uri, + tcp_proxy(proxy, Socks5Auth::RandomPerConnection), + 0, + true, + ); + let error = get_status(&client).await; + assert_eq!(status(&error).code(), Code::Unimplemented, "{error:?}"); + } + + let seen = connects(&log); + assert_eq!(seen.len(), 2, "{seen:?}"); + let usernames: Vec<&str> = seen + .iter() + .map(|connect| connect.username.as_deref().expect("credentials")) + .collect(); + for (connect, username) in seen.iter().zip(&usernames) { + assert_eq!(connect.methods, vec![0x00, 0x02]); + assert_eq!(username.len(), 32); + assert!(username.chars().all(|c| c.is_ascii_hexdigit())); + } + assert_ne!(usernames[0], usernames[1]); +} + +#[tokio::test] +async fn should_accept_a_proxy_that_only_allows_no_authentication() { + let node = start_node().await; + let behaviour = ProxyBehaviour { + no_auth_only: true, + ..ProxyBehaviour::default() + }; + let (proxy, log) = start_proxy(node, behaviour).await; + let client = client( + &format!("https://{node}"), + tcp_proxy(proxy, Socks5Auth::RandomPerConnection), + 0, + true, + ); + + let error = get_status(&client).await; + + assert_eq!(status(&error).code(), Code::Unimplemented, "{error:?}"); + let seen = connects(&log); + assert_eq!(seen.len(), 1); + assert_eq!(seen[0].username, None); +} + +#[tokio::test] +async fn should_send_an_ipv6_literal_as_an_ipv6_address() { + let node = start_node().await; + let (proxy, log) = start_proxy(node, ProxyBehaviour::default()).await; + let port = node.port(); + let client = client( + &format!("https://[::1]:{port}"), + tcp_proxy(proxy, Socks5Auth::None), + 0, + true, + ); + + let error = get_status(&client).await; + + assert_eq!(status(&error).code(), Code::Unimplemented, "{error:?}"); + let seen = connects(&log); + assert_eq!(seen[0].atyp, 0x04); + assert_eq!(seen[0].target, format!("[::1]:{port}")); +} + +#[tokio::test] +async fn should_leave_name_resolution_to_the_proxy() { + let node = start_node().await; + let (proxy, log) = start_proxy(node, ProxyBehaviour::default()).await; + let port = node.port(); + let client = client( + &format!("https://localhost:{port}"), + tcp_proxy(proxy, Socks5Auth::None), + 0, + true, + ); + + let error = get_status(&client).await; + + assert_eq!(status(&error).code(), Code::Unimplemented, "{error:?}"); + let seen = connects(&log); + assert_eq!(seen[0].atyp, 0x03); + assert_eq!(seen[0].target, format!("localhost:{port}")); +} + +#[tokio::test] +async fn should_neither_ban_nor_retry_when_the_proxy_fails() { + // A proxy that hangs up on every connection before answering. + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let closed = listener.local_addr().expect("address"); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + drop(stream); + } + }); + let uris = ["https://127.0.0.1:1", "https://127.0.0.1:2"]; + let client = client( + &uris.join(","), + tcp_proxy(closed, Socks5Auth::None), + 2, + true, + ); + + let error = get_status(&client).await; + + let transport = transport_error(&error); + assert!(transport.is_proxy_failure(), "{error:?}"); + assert_eq!(status(&error).code(), Code::Unavailable); + assert!(!error.can_retry()); + assert_eq!(error.retries, 0, "no other node is tried"); + for uri in uris { + assert!(!is_banned(&client, uri), "{uri} must not be banned"); + } + assert!( + transport.clone().is_proxy_failure(), + "a clone keeps the source" + ); +} + +#[tokio::test] +async fn should_blame_the_proxy_for_its_own_failure_and_the_node_for_a_destination_reply() { + let node = start_node().await; + let uri = format!("https://{node}"); + for (reply, proxy_failure) in [ + (GENERAL_FAILURE, true), + (NOT_ALLOWED_BY_RULESET, false), + (CONNECTION_REFUSED, false), + ] { + let behaviour = ProxyBehaviour { + reply, + ..ProxyBehaviour::default() + }; + let (proxy, _) = start_proxy(node, behaviour).await; + let client = client(&uri, tcp_proxy(proxy, Socks5Auth::None), 0, true); + + let error = get_status(&client).await; + + assert_eq!(status(&error).code(), Code::Unavailable, "{error:?}"); + assert_eq!( + transport_error(&error).is_proxy_failure(), + proxy_failure, + "reply {reply:#04x}: {error:?}" + ); + assert_eq!(error.can_retry(), !proxy_failure, "reply {reply:#04x}"); + assert_eq!( + is_banned(&client, &uri), + !proxy_failure, + "reply {reply:#04x}" + ); + } +} + +#[tokio::test] +async fn should_blame_a_proxy_that_never_answers_the_greeting() { + // Accepts every connection and never says anything, like a suspended + // proxy. + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let silent = listener.local_addr().expect("address"); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((stream, _)) = listener.accept().await { + held.push(stream); + } + }); + let uri = "https://127.0.0.1:1"; + let client = client_with_connect_timeout( + uri, + tcp_proxy(silent, Socks5Auth::None), + Duration::from_secs(30), + ); + + let started = std::time::Instant::now(); + let error = get_status(&client).await; + + assert!(started.elapsed() < Duration::from_secs(20), "{error:?}"); + assert!(transport_error(&error).is_proxy_failure(), "{error:?}"); + assert!(!is_banned(&client, uri)); +} + +#[tokio::test] +async fn should_hold_a_stalled_route_against_the_node() { + // Answers the greeting, then holds the CONNECT reply as Tor does while + // it builds a circuit: only the connect timeout ends it. + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let stalling = listener.local_addr().expect("address"); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((mut stream, _)) = listener.accept().await { + let mut greeting = [0u8; 3]; + if stream.read_exact(&mut greeting).await.is_ok() + && stream.write_all(&[0x05, 0x00]).await.is_ok() + { + held.push(stream); + } + } + }); + let uri = "https://127.0.0.1:1"; + let client = client_with_connect_timeout( + uri, + tcp_proxy(stalling, Socks5Auth::None), + Duration::from_secs(8), + ); + + let started = std::time::Instant::now(); + let error = get_status(&client).await; + + assert!(started.elapsed() >= Duration::from_secs(8), "{error:?}"); + assert!(!transport_error(&error).is_proxy_failure(), "{error:?}"); + assert!(is_banned(&client, uri)); +} + +/// Starts a proxy that reads the greeting (and, if `auth`, the RFC 1929 +/// request), writes `reply`, and then holds the connection open. +async fn start_stalling_proxy(reply: &'static [u8], auth: bool) -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let address = listener.local_addr().expect("address"); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((mut stream, _)) = listener.accept().await { + let mut greeting = [0u8; 2]; + if stream.read_exact(&mut greeting).await.is_err() { + continue; + } + let mut methods = vec![0u8; greeting[1] as usize]; + if stream.read_exact(&mut methods).await.is_err() { + continue; + } + if auth { + // Method selection first, then read the credentials. + if stream.write_all(&[0x05, 0x02]).await.is_err() { + continue; + } + let mut header = [0u8; 2]; + if stream.read_exact(&mut header).await.is_err() { + continue; + } + let mut rest = vec![0u8; header[1] as usize + 1]; + if stream.read_exact(&mut rest).await.is_err() { + continue; + } + let mut password = vec![0u8; rest[header[1] as usize] as usize]; + if stream.read_exact(&mut password).await.is_err() { + continue; + } + } + if stream.write_all(reply).await.is_ok() { + held.push(stream); + } + } + }); + address +} + +#[tokio::test] +async fn should_blame_a_proxy_that_stops_mid_negotiation() { + // Half a method selection; a username/password selection with no + // authentication status; a silent proxy under a budget below five + // seconds; and one under a budget too large to add to the current + // instant. None of them has been asked for a CONNECT yet. + let cases: [(&'static [u8], bool, Socks5Auth, Duration); 4] = [ + (&[0x05], false, Socks5Auth::None, Duration::from_secs(8)), + ( + &[], + true, + Socks5Auth::RandomPerConnection, + Duration::from_secs(8), + ), + (&[], false, Socks5Auth::None, Duration::from_millis(300)), + (&[], false, Socks5Auth::None, Duration::MAX), + ]; + for (reply, auth, credentials, connect_timeout) in cases { + let proxy = start_stalling_proxy(reply, auth).await; + let uri = "https://127.0.0.1:1"; + let client = + client_with_connect_timeout(uri, tcp_proxy(proxy, credentials), connect_timeout); + + let error = get_status(&client).await; + + assert!( + transport_error(&error).is_proxy_failure(), + "reply {reply:?}, budget {connect_timeout:?}: {error:?}" + ); + assert!(!is_banned(&client, uri), "reply {reply:?}"); + } +} + +#[tokio::test] +async fn should_give_a_proxied_connection_the_default_connect_budget() { + // Negotiates, then never answers the CONNECT: only the twenty-second + // proxy connect default ends it, which the executor's attempt deadline + // (ten-second request timeout plus that default) now leaves room for. + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let proxy = listener.local_addr().expect("address"); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((mut stream, _)) = listener.accept().await { + let mut greeting = [0u8; 3]; + if stream.read_exact(&mut greeting).await.is_ok() + && stream.write_all(&[0x05, 0x00]).await.is_ok() + { + held.push(stream); + } + } + }); + let uri = "https://127.0.0.1:1"; + let settings = RequestSettings { + retries: Some(0), + ..RequestSettings::default() + }; + let client = DapiClient::new(uri.parse().expect("address list"), settings) + .with_proxy(tcp_proxy(proxy, Socks5Auth::None)); + + let started = std::time::Instant::now(); + let error = get_status(&client).await; + + // The connect budget ran out (about twenty seconds), not the ten-second + // attempt deadline that ignored it. + assert!(started.elapsed() >= Duration::from_secs(19), "{error:?}"); + assert_ne!(status(&error).code(), Code::DeadlineExceeded, "{error:?}"); + assert!(!transport_error(&error).is_proxy_failure(), "{error:?}"); + assert!(is_banned(&client, uri)); +} + +#[cfg(unix)] +#[tokio::test] +async fn should_reach_the_proxy_over_a_unix_socket() { + use tokio::net::UnixListener; + + let node = start_node().await; + let path = std::env::temp_dir().join(format!("rs-dapi-client-socks-{}.sock", node.port())); + let _ = std::fs::remove_file(&path); + let listener = UnixListener::bind(&path).expect("bind unix proxy"); + let log = Log::default(); + let connections = log.clone(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + tokio::spawn(serve_socks( + stream, + node, + ProxyBehaviour::default(), + connections.clone(), + )); + } + }); + let proxy = Socks5Proxy { + endpoint: ProxyEndpoint::Unix(path.clone()), + auth: Socks5Auth::None, + }; + let client = client(&format!("https://{node}"), proxy, 0, true); + + let error = get_status(&client).await; + let _ = std::fs::remove_file(&path); + + assert_eq!(status(&error).code(), Code::Unimplemented, "{error:?}"); + assert_eq!(connects(&log).len(), 1); +} From 6109d6eab8c864b388189355171ab7062c789deb Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 5 Oct 2026 00:13:45 -0500 Subject: [PATCH 5/5] fix(sdk): bound a proxied connection by one deadline and keep the TLS config error's cause Through a SOCKS5 proxy the connector got the whole connect budget and TLS then started a second one of the same length, so a delayed CONNECT followed by a stalled handshake could take twice `connect_timeout`. The channel's own connect timeout now covers the tunnel and TLS together; the connector fixes its phase deadlines when it is called, before tonic's clock starts, so a stuck proxy is still reported as a ProxyError. An invalid TLS configuration now keeps tonic's cause (for example the invalid DNS name) in the status message and source chain instead of only "transport error". Co-Authored-By: Claude Opus 5.5 (1M context) --- .../rs-dapi-client/src/transport/proxy.rs | 42 +++++---- .../src/transport/tonic_channel.rs | 86 +++++++++++++------ packages/rs-dapi-client/tests/socks5_proxy.rs | 50 +++++++++++ 3 files changed, 137 insertions(+), 41 deletions(-) diff --git a/packages/rs-dapi-client/src/transport/proxy.rs b/packages/rs-dapi-client/src/transport/proxy.rs index b8875eb7b6b..e8f07b8ed35 100644 --- a/packages/rs-dapi-client/src/transport/proxy.rs +++ b/packages/rs-dapi-client/src/transport/proxy.rs @@ -146,24 +146,19 @@ impl Socks5Proxy { format!("{} {auth}", self.endpoint) } - /// Opens a tunnel to `uri`'s host through the proxy within - /// `connect_timeout`: the proxy gets [PROXY_ANSWER_TIMEOUT] (or less, if - /// the budget is shorter) to accept and negotiate, and running out of it - /// is the proxy's failure; the CONNECT reply gets what is left, and - /// running out of that is the destination's. + /// Opens a tunnel to `uri`'s host through the proxy by `deadline`: the + /// proxy gets [PROXY_ANSWER_TIMEOUT] (or less, if the deadline is + /// closer) to accept and negotiate, and running out of it is the proxy's + /// failure; the CONNECT reply gets what is left, and running out of that + /// is the destination's. async fn connect( self, uri: Uri, - connect_timeout: Duration, + started: tokio::time::Instant, + deadline: tokio::time::Instant, ) -> Result, BoxError> { let target = target(&uri)?; - let now = tokio::time::Instant::now(); - // A budget too large to add (an "unlimited" Duration::MAX) is - // capped rather than overflowing. - let deadline = now - .checked_add(connect_timeout) - .unwrap_or_else(|| now + UNLIMITED); - let negotiated_by = deadline.min(now + PROXY_ANSWER_TIMEOUT); + let negotiated_by = deadline.min(started + PROXY_ANSWER_TIMEOUT); let proxy_error = |source: BoxError| ProxyError { proxy: self.endpoint.to_string(), source, @@ -439,6 +434,13 @@ impl ProxiedStream for S {} /// The tonic connector that tunnels every connection through the proxy /// within the connect budget (see [Socks5Proxy::connect]). +/// +/// The channel's own connect timeout, set to the same budget, bounds the +/// tunnel and the TLS handshake after it together. Its clock starts when +/// the connection future is first polled, after [call](tower_service::Service::call) +/// returns, and it polls the connection before checking its own deadline, so +/// the connector's deadlines, fixed in `call`, always fire first: a proxy +/// that stalls the negotiation is still reported as a [ProxyError]. #[derive(Clone)] pub(crate) struct Socks5Connector { pub(crate) proxy: Socks5Proxy, @@ -456,8 +458,18 @@ impl tower_service::Service for Socks5Connector { fn call(&mut self, uri: Uri) -> Self::Future { let proxy = self.proxy.clone(); - let connect_timeout = self.connect_timeout; - Box::pin(async move { proxy.connect(uri, connect_timeout).await.map(TokioIo::new) }) + let started = tokio::time::Instant::now(); + // A budget too large to add (an "unlimited" Duration::MAX) is + // capped rather than overflowing. + let deadline = started + .checked_add(self.connect_timeout) + .unwrap_or_else(|| started + UNLIMITED); + Box::pin(async move { + proxy + .connect(uri, started, deadline) + .await + .map(TokioIo::new) + }) } } diff --git a/packages/rs-dapi-client/src/transport/tonic_channel.rs b/packages/rs-dapi-client/src/transport/tonic_channel.rs index 20afa2778f0..7d311adb33c 100644 --- a/packages/rs-dapi-client/src/transport/tonic_channel.rs +++ b/packages/rs-dapi-client/src/transport/tonic_channel.rs @@ -4,6 +4,8 @@ use crate::{request_settings::AppliedRequestSettings, Uri}; use dapi_grpc::core::v0::core_client::CoreClient; use dapi_grpc::platform::v0::platform_client::PlatformClient; use dapi_grpc::tonic::transport::{Certificate, Channel, ClientTlsConfig}; +use std::error::Error as _; +use std::sync::Arc; use std::time::Duration; /// Platform Client using gRPC transport. @@ -54,22 +56,8 @@ pub fn create_channel( let proxy = settings.and_then(|settings| settings.proxy.clone()); let connect_timeout = settings.and_then(AppliedRequestSettings::effective_connect_timeout); - if let Some(settings) = settings { - match (connect_timeout, &proxy) { - // Through a proxy the connector enforces the budget itself, so - // it can tell a stuck proxy from a slow destination; TLS, which - // runs after it, gets the same budget of its own (a connection - // can so take up to twice the budget; the executor's attempt - // deadline still bounds the whole attempt). - (Some(timeout), Some(_)) => tls_config = tls_config.timeout(timeout), - (Some(timeout), None) => builder = builder.connect_timeout(timeout), - (None, _) => {} - } - - if let Some(pem) = settings.ca_certificate.as_ref() { - let cert = Certificate::from_pem(pem); - tls_config = tls_config.ca_certificate(cert); - }; + if let Some(pem) = settings.and_then(|settings| settings.ca_certificate.as_ref()) { + tls_config = tls_config.ca_certificate(Certificate::from_pem(pem)); } // Ping only while a request is in flight: a connection whose network path @@ -81,21 +69,48 @@ pub fn create_channel( .keep_alive_timeout(HTTP2_KEEP_ALIVE_TIMEOUT) .keep_alive_while_idle(false); - builder = builder.tls_config(tls_config).map_err(|e| { - TransportError::Grpc(dapi_grpc::tonic::Status::invalid_argument(format!( - "invalid TLS configuration: {e}" - ))) - })?; + builder = builder.tls_config(tls_config).map_err(invalid_tls_config)?; Ok(match proxy { - None => builder.connect_lazy(), - Some(proxy) => builder.connect_with_connector_lazy(Socks5Connector { - proxy, - connect_timeout: connect_timeout.unwrap_or(PROXY_CONNECT_TIMEOUT), - }), + None => { + if let Some(timeout) = connect_timeout { + builder = builder.connect_timeout(timeout); + } + builder.connect_lazy() + } + Some(proxy) => { + // One absolute deadline for the whole connection: tonic applies + // it around the connector, so it covers the SOCKS5 tunnel and the + // TLS handshake after it. The connector times its own phases + // against the same budget from a start that is never later than + // tonic's, so it still tells a stuck proxy from a slow + // destination before this deadline fires. + let connect_timeout = connect_timeout.unwrap_or(PROXY_CONNECT_TIMEOUT); + builder + .connect_timeout(connect_timeout) + .connect_with_connector_lazy(Socks5Connector { + proxy, + connect_timeout, + }) + } }) } +/// A TLS configuration tonic rejects, as an `InvalidArgument` status that +/// keeps the cause: tonic's own message names only the error kind, the cause +/// (an invalid server name, a malformed certificate) is in its source chain. +fn invalid_tls_config(error: dapi_grpc::tonic::transport::Error) -> TransportError { + let mut message = format!("invalid TLS configuration: {error}"); + let mut cause = error.source(); + while let Some(current) = cause { + message.push_str(&format!(": {current}")); + cause = current.source(); + } + let mut status = dapi_grpc::tonic::Status::invalid_argument(message); + status.set_source(Arc::new(error)); + TransportError::Grpc(status) +} + /// The host of a URI without the brackets an IPv6 literal carries in it /// (`[2001:db8::1]` becomes `2001:db8::1`); any other host is unchanged. pub(crate) fn unbracketed(host: &str) -> &str { @@ -135,4 +150,23 @@ mod tests { })); create_channel(uri, Some(&settings)).expect("through a proxy"); } + + #[test] + fn should_keep_the_cause_of_an_invalid_tls_configuration() { + // Not a valid TLS server name. + let uri: Uri = "https://foo..bar:1443".parse().expect("uri"); + let Err(TransportError::Grpc(status)) = create_channel(uri, None) else { + panic!("an invalid server name must be rejected"); + }; + assert_eq!(status.code(), dapi_grpc::tonic::Code::InvalidArgument); + assert!( + status.message().contains("invalid dns name"), + "{}", + status.message() + ); + assert!( + status.source().is_some(), + "the tonic error stays in the source chain" + ); + } } diff --git a/packages/rs-dapi-client/tests/socks5_proxy.rs b/packages/rs-dapi-client/tests/socks5_proxy.rs index 7ba80c8366e..5f943e67d85 100644 --- a/packages/rs-dapi-client/tests/socks5_proxy.rs +++ b/packages/rs-dapi-client/tests/socks5_proxy.rs @@ -514,6 +514,56 @@ async fn should_hold_a_stalled_route_against_the_node() { assert!(is_banned(&client, uri)); } +#[tokio::test] +async fn should_bound_the_tunnel_and_tls_by_one_connect_deadline() { + // Answers the greeting, holds the CONNECT reply for most of the budget, + // grants it, and then never answers the TLS ClientHello. One budget + // covers the tunnel and TLS together, so the connection fails when it + // runs out instead of TLS starting a full budget of its own. + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let slow = listener.local_addr().expect("address"); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((mut stream, _)) = listener.accept().await { + let mut greeting = [0u8; 3]; + if stream.read_exact(&mut greeting).await.is_err() + || stream.write_all(&[0x05, 0x00]).await.is_err() + { + continue; + } + // CONNECT to an IPv4 address: header, address and port. + let mut request = [0u8; 10]; + if stream.read_exact(&mut request).await.is_err() { + continue; + } + tokio::time::sleep(Duration::from_millis(1500)).await; + if stream + .write_all(&[0x05, SUCCEEDED, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .is_ok() + { + held.push(stream); + } + } + }); + let uri = "https://127.0.0.1:1"; + let client = client_with_connect_timeout( + uri, + tcp_proxy(slow, Socks5Auth::None), + Duration::from_secs(2), + ); + + let started = std::time::Instant::now(); + let error = get_status(&client).await; + let elapsed = started.elapsed(); + + // With a TLS timeout of its own this took 1.5 s + 2 s. + assert!(elapsed >= Duration::from_secs(2), "{elapsed:?}: {error:?}"); + assert!(elapsed < Duration::from_secs(3), "{elapsed:?}: {error:?}"); + assert!(!transport_error(&error).is_proxy_failure(), "{error:?}"); + assert!(is_banned(&client, uri)); +} + /// Starts a proxy that reads the greeting (and, if `auth`, the RFC 1929 /// request), writes `reply`, and then holds the connection open. async fn start_stalling_proxy(reply: &'static [u8], auth: bool) -> SocketAddr {