diff --git a/Cargo.lock b/Cargo.lock index 3dd3b403830..1fc2a06672e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6429,8 +6429,11 @@ dependencies = [ "sha2", "thiserror 2.0.18", "tokio", + "tokio-socks", + "tonic 0.14.6", "tonic-web-wasm-client", "tower 0.5.3", + "tower-service", "tracing", "wasm-bindgen-futures", ] @@ -7884,6 +7887,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..1c0ff1e9178 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 = [ @@ -73,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/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..e8f07b8ed35 --- /dev/null +++ b/packages/rs-dapi-client/src/transport/proxy.rs @@ -0,0 +1,556 @@ +//! 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 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, + started: tokio::time::Instant, + deadline: tokio::time::Instant, + ) -> Result, BoxError> { + let target = target(&uri)?; + let negotiated_by = deadline.min(started + 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]). +/// +/// 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, + 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 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) + }) + } +} + +#[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 779bdb33f52..7d311adb33c 100644 --- a/packages/rs-dapi-client/src/transport/tonic_channel.rs +++ b/packages/rs-dapi-client/src/transport/tonic_channel.rs @@ -1,8 +1,11 @@ +use super::proxy::{Socks5Connector, PROXY_CONNECT_TIMEOUT}; use super::TransportError; 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. @@ -20,11 +23,14 @@ 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>, ) -> 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 +38,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( @@ -45,15 +54,10 @@ pub fn create_channel( tls_config = tls_config.with_native_roots(); } - if let Some(settings) = settings { - if let Some(timeout) = settings.connect_timeout { - builder = builder.connect_timeout(timeout); - } - - 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); - }; + let proxy = settings.and_then(|settings| settings.proxy.clone()); + let connect_timeout = settings.and_then(AppliedRequestSettings::effective_connect_timeout); + 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 @@ -65,9 +69,104 @@ 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(invalid_tls_config)?; + + Ok(match proxy { + 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) +} - 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::transport::{ProxyEndpoint, Socks5Auth, Socks5Proxy}; + 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.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"); + } + + #[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/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/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, } } 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..5f943e67d85 --- /dev/null +++ b/packages/rs-dapi-client/tests/socks5_proxy.rs @@ -0,0 +1,711 @@ +//! 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)); +} + +#[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 { + 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); +} 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();