feat(sdk): route DAPI connections through a SOCKS5 proxy

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]
 

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<Certificate>,
+    #[cfg(not(target_arch = "wasm32"))]
+    /// SOCKS5 proxy every connection is tunnelled through.
+    pub proxy: Option<Socks5Proxy>,
     #[cfg(feature = "dump")]
     pub(crate) dump_dir: Option<std::path::PathBuf>,
 }
@@ -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")]

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<Certificate>,
+    /// SOCKS5 proxy every connection is tunnelled through.
+    #[cfg(not(target_arch = "wasm32"))]
+    pub proxy: Option<Socks5Proxy>,
 }
 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<Duration> {
+        #[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<Socks5Proxy>) -> 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<String> = 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<String> = 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"));
+    }
 }

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,
         }
     }
 }