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) <noreply@anthropic.com>

Author: PastaPastaPasta

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<Box<dyn ProxiedStream>, 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<S: AsyncRead + AsyncWrite + Send + Unpin> 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<Uri> 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)
+        })
     }
 }
 

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"
+        );
+    }
 }

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 {