saluki_io/net/client/http/
conn.rs

1#[cfg(unix)]
2use std::path::PathBuf;
3use std::{
4    future::Future,
5    io,
6    pin::Pin,
7    sync::Arc,
8    task::{Context, Poll},
9    time::{Duration, Instant},
10};
11
12use http::{Extensions, Uri};
13use hyper_rustls::MaybeHttpsStream;
14use hyper_util::{
15    client::legacy::connect::{CaptureConnection, Connected, Connection, HttpConnector},
16    rt::TokioIo,
17};
18use metrics::Counter;
19use pin_project_lite::pin_project;
20use rustls::{pki_types::ServerName, ClientConfig};
21use saluki_error::GenericError;
22use tokio::net::TcpStream;
23use tokio_rustls::TlsConnector;
24#[cfg(target_os = "linux")]
25use tokio_vsock::{VsockAddr, VsockStream};
26use tower::{BoxError, Service};
27use tracing::debug;
28
29use super::telemetry::HttpTransactionErrorTelemetry;
30use crate::net::dns::{DnsError, SystemHttpConnector, SystemResolver};
31
32/// Imposes a limit on the age of a connection.
33///
34/// In many cases, it's undesirable to hold onto a connection indefinitely, even if it can be theoretically reused.
35/// Doing so can make it more difficult to perform maintenance on infrastructure, as the expectation of old connections
36/// being eventually closed and replaced isn't upheld.
37///
38/// This extension allows tracking the age of a connection (based on when the connector creates the connection) and
39/// checking if it's expired, or past the configured limit. Callers can then decide how to handle the expiration, such
40/// as by closing the connection.
41#[derive(Clone)]
42struct ConnectionAgeLimit {
43    limit: Duration,
44    created: Instant,
45}
46
47impl ConnectionAgeLimit {
48    fn new(limit: Duration) -> Self {
49        ConnectionAgeLimit {
50            limit,
51            created: Instant::now(),
52        }
53    }
54
55    fn is_expired(&self) -> bool {
56        self.created.elapsed() >= self.limit
57    }
58}
59
60/// An inner transport that abstracts over TCP, Unix domain socket, and vsock connections.
61///
62/// This allows using a single monomorphization of the HTTP/2 and TLS stacks regardless of the
63/// underlying transport, avoiding duplicate code generation for each transport type.
64enum Transport {
65    Tcp(TokioIo<TcpStream>),
66    #[cfg(unix)]
67    Unix(TokioIo<tokio::net::UnixStream>),
68    #[cfg(target_os = "linux")]
69    Vsock(TokioIo<VsockStream>),
70}
71
72impl Connection for Transport {
73    fn connected(&self) -> Connected {
74        match self {
75            Self::Tcp(s) => s.connected(),
76            #[cfg(unix)]
77            Self::Unix(_) => Connected::new(),
78            #[cfg(target_os = "linux")]
79            Self::Vsock(_) => Connected::new(),
80        }
81    }
82}
83
84impl hyper::rt::Read for Transport {
85    fn poll_read(
86        self: Pin<&mut Self>, cx: &mut Context<'_>, buf: hyper::rt::ReadBufCursor<'_>,
87    ) -> Poll<io::Result<()>> {
88        match Pin::get_mut(self) {
89            Self::Tcp(s) => Pin::new(s).poll_read(cx, buf),
90            #[cfg(unix)]
91            Self::Unix(s) => Pin::new(s).poll_read(cx, buf),
92            #[cfg(target_os = "linux")]
93            Self::Vsock(s) => Pin::new(s).poll_read(cx, buf),
94        }
95    }
96}
97
98impl hyper::rt::Write for Transport {
99    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
100        match Pin::get_mut(self) {
101            Self::Tcp(s) => Pin::new(s).poll_write(cx, buf),
102            #[cfg(unix)]
103            Self::Unix(s) => Pin::new(s).poll_write(cx, buf),
104            #[cfg(target_os = "linux")]
105            Self::Vsock(s) => Pin::new(s).poll_write(cx, buf),
106        }
107    }
108
109    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
110        match Pin::get_mut(self) {
111            Self::Tcp(s) => Pin::new(s).poll_flush(cx),
112            #[cfg(unix)]
113            Self::Unix(s) => Pin::new(s).poll_flush(cx),
114            #[cfg(target_os = "linux")]
115            Self::Vsock(s) => Pin::new(s).poll_flush(cx),
116        }
117    }
118
119    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
120        match Pin::get_mut(self) {
121            Self::Tcp(s) => Pin::new(s).poll_shutdown(cx),
122            #[cfg(unix)]
123            Self::Unix(s) => Pin::new(s).poll_shutdown(cx),
124            #[cfg(target_os = "linux")]
125            Self::Vsock(s) => Pin::new(s).poll_shutdown(cx),
126        }
127    }
128
129    fn is_write_vectored(&self) -> bool {
130        match self {
131            Self::Tcp(s) => s.is_write_vectored(),
132            #[cfg(unix)]
133            Self::Unix(s) => s.is_write_vectored(),
134            #[cfg(target_os = "linux")]
135            Self::Vsock(s) => s.is_write_vectored(),
136        }
137    }
138
139    fn poll_write_vectored(
140        self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[io::IoSlice<'_>],
141    ) -> Poll<io::Result<usize>> {
142        match Pin::get_mut(self) {
143            Self::Tcp(s) => Pin::new(s).poll_write_vectored(cx, bufs),
144            #[cfg(unix)]
145            Self::Unix(s) => Pin::new(s).poll_write_vectored(cx, bufs),
146            #[cfg(target_os = "linux")]
147            Self::Vsock(s) => Pin::new(s).poll_write_vectored(cx, bufs),
148        }
149    }
150}
151
152pin_project! {
153    /// A connection that supports both HTTP and HTTPS.
154    pub struct HttpsCapableConnection {
155        #[pin]
156        inner: MaybeHttpsStream<Transport>,
157        bytes_sent: Option<Counter>,
158        error_telemetry: Option<HttpTransactionErrorTelemetry>,
159        conn_age_limit: Option<Duration>,
160    }
161}
162
163impl Connection for HttpsCapableConnection {
164    fn connected(&self) -> Connected {
165        let connected = self.inner.connected();
166
167        if let Some(conn_age_limit) = self.conn_age_limit {
168            debug!("setting connection age limit to {:?}", conn_age_limit);
169            connected.extra(ConnectionAgeLimit::new(conn_age_limit))
170        } else {
171            connected
172        }
173    }
174}
175
176impl hyper::rt::Read for HttpsCapableConnection {
177    fn poll_read(
178        self: Pin<&mut Self>, cx: &mut Context<'_>, buf: hyper::rt::ReadBufCursor<'_>,
179    ) -> Poll<io::Result<()>> {
180        let this = self.project();
181        this.inner.poll_read(cx, buf)
182    }
183}
184
185impl hyper::rt::Write for HttpsCapableConnection {
186    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
187        let this = self.project();
188        match this.inner.poll_write(cx, buf) {
189            Poll::Ready(Ok(n)) => {
190                if let Some(bytes_sent) = this.bytes_sent {
191                    bytes_sent.increment(n as u64);
192                }
193                Poll::Ready(Ok(n))
194            }
195            Poll::Ready(Err(error)) => {
196                if let Some(error_telemetry) = this.error_telemetry.as_ref() {
197                    error_telemetry.increment_wrote_request_error();
198                }
199                Poll::Ready(Err(error))
200            }
201            other => other,
202        }
203    }
204
205    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
206        let this = self.project();
207        match this.inner.poll_flush(cx) {
208            Poll::Ready(Err(error)) => {
209                if let Some(error_telemetry) = this.error_telemetry.as_ref() {
210                    error_telemetry.increment_wrote_request_error();
211                }
212                Poll::Ready(Err(error))
213            }
214            other => other,
215        }
216    }
217
218    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
219        let this = self.project();
220        this.inner.poll_shutdown(cx)
221    }
222
223    fn is_write_vectored(&self) -> bool {
224        self.inner.is_write_vectored()
225    }
226
227    fn poll_write_vectored(
228        self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[io::IoSlice<'_>],
229    ) -> Poll<io::Result<usize>> {
230        let this = self.project();
231        match this.inner.poll_write_vectored(cx, bufs) {
232            Poll::Ready(Ok(n)) => {
233                if let Some(bytes_sent) = this.bytes_sent {
234                    bytes_sent.increment(n as u64);
235                }
236                Poll::Ready(Ok(n))
237            }
238            Poll::Ready(Err(error)) => {
239                if let Some(error_telemetry) = this.error_telemetry.as_ref() {
240                    error_telemetry.increment_wrote_request_error();
241                }
242                Poll::Ready(Err(error))
243            }
244            other => other,
245        }
246    }
247}
248
249/// An inner connector that routes to TCP (via DNS), a Unix domain socket, or a vsock socket.
250///
251/// When a Unix socket path is configured, all connections are routed through that socket regardless
252/// of the URI host. When a vsock CID is configured, all connections are routed through that vsock
253/// socket using the port from the destination URI. Otherwise, connections use the standard DNS +
254/// TCP path.
255#[derive(Clone)]
256struct InnerConnector {
257    http: SystemHttpConnector,
258    #[cfg(unix)]
259    connect_timeout: Duration,
260    error_telemetry: Option<HttpTransactionErrorTelemetry>,
261    #[cfg(unix)]
262    unix_socket_path: Option<Arc<std::path::Path>>,
263    #[cfg(target_os = "linux")]
264    vsock_addr: Option<VsockAddr>,
265}
266
267impl Service<Uri> for InnerConnector {
268    type Response = Transport;
269    type Error = BoxError;
270    type Future = Pin<Box<dyn Future<Output = Result<Transport, BoxError>> + Send>>;
271
272    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
273        // When routing via vsock or a Unix domain socket, the TCP/DNS connector is not used, so we
274        // consider the service immediately ready. vsock takes priority over Unix (matching Agent
275        // behavior) when both are configured.
276        #[cfg(target_os = "linux")]
277        if self.vsock_addr.is_some() {
278            return Poll::Ready(Ok(()));
279        }
280
281        #[cfg(unix)]
282        if self.unix_socket_path.is_some() {
283            return Poll::Ready(Ok(()));
284        }
285
286        self.http.poll_ready(cx).map_err(Into::into)
287    }
288
289    fn call(&mut self, dst: Uri) -> Self::Future {
290        #[cfg(target_os = "linux")]
291        if let Some(addr) = self.vsock_addr {
292            let connect_timeout = self.connect_timeout;
293            let error_telemetry = self.error_telemetry.clone();
294            return Box::pin(async move {
295                let stream = tokio::time::timeout(connect_timeout, VsockStream::connect(addr))
296                    .await
297                    .map_err(|_| -> BoxError {
298                        if let Some(error_telemetry) = &error_telemetry {
299                            error_telemetry.increment_connection_error();
300                        }
301                        Box::new(io::Error::new(io::ErrorKind::TimedOut, "vsock connect timed out"))
302                    })?
303                    .map_err(|e| -> BoxError {
304                        if let Some(error_telemetry) = &error_telemetry {
305                            error_telemetry.increment_connection_error();
306                        }
307                        Box::new(e)
308                    })?;
309                Ok(Transport::Vsock(TokioIo::new(stream)))
310            });
311        }
312
313        #[cfg(unix)]
314        if let Some(path) = self.unix_socket_path.clone() {
315            let connect_timeout = self.connect_timeout;
316            let error_telemetry = self.error_telemetry.clone();
317            return Box::pin(async move {
318                let stream = tokio::time::timeout(connect_timeout, tokio::net::UnixStream::connect(&*path))
319                    .await
320                    .map_err(|_| -> BoxError {
321                        if let Some(error_telemetry) = &error_telemetry {
322                            error_telemetry.increment_connection_error();
323                        }
324                        Box::new(io::Error::new(io::ErrorKind::TimedOut, "unix socket connect timed out"))
325                    })?
326                    .map_err(|e| -> BoxError {
327                        if let Some(error_telemetry) = &error_telemetry {
328                            error_telemetry.increment_connection_error();
329                        }
330                        Box::new(e)
331                    })?;
332                Ok(Transport::Unix(TokioIo::new(stream)))
333            });
334        }
335
336        let fut = self.http.call(dst);
337        let error_telemetry = self.error_telemetry.clone();
338        Box::pin(async move {
339            let tcp = fut.await.map_err(|error| {
340                if !is_dns_error(&error) {
341                    if let Some(error_telemetry) = &error_telemetry {
342                        error_telemetry.increment_connection_error();
343                    }
344                }
345                BoxError::from(error)
346            })?;
347            Ok(Transport::Tcp(tcp))
348        })
349    }
350}
351
352/// HTTP protocol selection for client connections.
353#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
354pub enum HttpProtocol {
355    /// Automatically negotiate HTTP/2 with HTTP/1.1 fallback.
356    #[default]
357    Auto,
358
359    /// Use HTTP/1.1 only.
360    Http1,
361}
362
363/// A connector that supports HTTP or HTTPS.
364///
365/// Unlike [`hyper_rustls::HttpsConnector`], which fuses the transport connect and TLS handshake into a single
366/// opaque future, this connector performs them as two distinct steps. That split allows a timeout to be scoped to
367/// just the handshake, rather than the combined connect-and-handshake duration.
368#[derive(Clone)]
369pub struct HttpsCapableConnector {
370    inner: InnerConnector,
371    tls_config: Arc<ClientConfig>,
372    tls_handshake_timeout: Duration,
373    bytes_sent: Option<Counter>,
374    error_telemetry: Option<HttpTransactionErrorTelemetry>,
375    conn_age_limit: Option<Duration>,
376}
377
378impl HttpsCapableConnector {
379    /// Returns the timeout applied to the TLS handshake for HTTPS connections.
380    pub(crate) fn tls_handshake_timeout(&self) -> Duration {
381        self.tls_handshake_timeout
382    }
383}
384
385impl Service<Uri> for HttpsCapableConnector {
386    type Response = HttpsCapableConnection;
387    type Error = BoxError;
388    type Future = Pin<Box<dyn Future<Output = Result<HttpsCapableConnection, BoxError>> + Send>>;
389
390    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
391        self.inner.poll_ready(cx)
392    }
393
394    fn call(&mut self, dst: Uri) -> Self::Future {
395        let is_https = match dst.scheme_str() {
396            Some("https") => true,
397            Some("http") => false,
398            scheme => {
399                let scheme = scheme.map(str::to_owned);
400                return Box::pin(async move {
401                    Err(Box::new(io::Error::new(
402                        io::ErrorKind::InvalidInput,
403                        format!("unsupported URI scheme: {scheme:?}"),
404                    )) as BoxError)
405                });
406            }
407        };
408        let transport_fut = self.inner.call(dst.clone());
409        let tls_config = Arc::clone(&self.tls_config);
410        let tls_handshake_timeout = self.tls_handshake_timeout;
411        let bytes_sent = self.bytes_sent.clone();
412        let error_telemetry = self.error_telemetry.clone();
413        let conn_age_limit = self.conn_age_limit;
414
415        Box::pin(async move {
416            let transport = transport_fut.await?;
417
418            let inner = if is_https {
419                let host = dst.host().ok_or_else(|| -> BoxError {
420                    Box::new(io::Error::new(io::ErrorKind::InvalidInput, "URI has no host"))
421                })?;
422                let host = strip_ipv6_brackets(host);
423                let server_name = ServerName::try_from(host)
424                    .map_err(|error| -> BoxError { Box::new(error) })?
425                    .to_owned();
426
427                let handshake = TlsConnector::from(tls_config).connect(server_name, TokioIo::new(transport));
428
429                match await_handshake_with_deadline(tls_handshake_timeout, handshake).await {
430                    Ok(stream) => MaybeHttpsStream::from(stream),
431                    Err(error) => {
432                        if let Some(error_telemetry) = &error_telemetry {
433                            error_telemetry.increment_tls_error();
434                        }
435                        return Err(error);
436                    }
437                }
438            } else {
439                MaybeHttpsStream::from(transport)
440            };
441
442            Ok(HttpsCapableConnection {
443                inner,
444                bytes_sent,
445                error_telemetry,
446                conn_age_limit,
447            })
448        })
449    }
450}
451
452/// Strips the surrounding brackets from a bracketed IPv6 host, as found in a URI authority.
453///
454/// [`rustls::pki_types::ServerName`] accepts unbracketed IPv6 addresses but rejects the bracketed form that
455/// [`http::Uri::host`] returns (for example, `[::1]`), so this normalizes the host before constructing the server name.
456fn strip_ipv6_brackets(host: &str) -> &str {
457    host.strip_prefix('[').and_then(|h| h.strip_suffix(']')).unwrap_or(host)
458}
459
460/// Awaits a TLS handshake future, bounding it by `timeout` unless `timeout` is zero.
461///
462/// A zero duration means the handshake deadline is disabled.
463/// `tokio::time::timeout` with a zero duration fires immediately rather than never, so that case is handled by
464/// awaiting the handshake directly instead of wrapping it in a timeout.
465async fn await_handshake_with_deadline<F, T, E>(timeout: Duration, handshake: F) -> Result<T, BoxError>
466where
467    F: Future<Output = Result<T, E>>,
468    E: std::error::Error + Send + Sync + 'static,
469{
470    if timeout.is_zero() {
471        return handshake.await.map_err(|error| Box::new(error) as BoxError);
472    }
473
474    match tokio::time::timeout(timeout, handshake).await {
475        Ok(result) => result.map_err(|error| Box::new(error) as BoxError),
476        Err(_) => Err(Box::new(io::Error::new(io::ErrorKind::TimedOut, "TLS handshake timed out")) as BoxError),
477    }
478}
479
480fn build_dns_resolver(error_telemetry: &Option<HttpTransactionErrorTelemetry>) -> SystemResolver {
481    let mut r = SystemResolver::new();
482    if let Some(et) = error_telemetry {
483        r = r.with_lookup_errors_counter(et.dns_errors());
484    }
485    r
486}
487
488/// A builder for `HttpsCapableConnector`.
489#[derive(Default)]
490pub struct HttpsCapableConnectorBuilder {
491    connect_timeout: Option<Duration>,
492    tls_handshake_timeout: Option<Duration>,
493    bytes_sent: Option<Counter>,
494    error_telemetry: Option<HttpTransactionErrorTelemetry>,
495    conn_age_limit: Option<Duration>,
496    http_protocol: HttpProtocol,
497    #[cfg(unix)]
498    unix_socket_path: Option<PathBuf>,
499    #[cfg(target_os = "linux")]
500    vsock_addr: Option<VsockAddr>,
501}
502
503impl HttpsCapableConnectorBuilder {
504    /// Sets the timeout when connecting to the remote host.
505    ///
506    /// Defaults to 30 seconds.
507    pub fn with_connect_timeout(mut self, timeout: Duration) -> Self {
508        self.connect_timeout = Some(timeout);
509        self
510    }
511
512    /// Sets the timeout for completing the TLS handshake after a connection is established.
513    ///
514    /// Defaults to 10 seconds.
515    pub fn with_tls_handshake_timeout(mut self, timeout: Duration) -> Self {
516        self.tls_handshake_timeout = Some(timeout);
517        self
518    }
519
520    /// Sets the HTTP protocol selection for client connections.
521    ///
522    /// Defaults to [`HttpProtocol::Auto`].
523    pub fn with_http_protocol(mut self, protocol: HttpProtocol) -> Self {
524        self.http_protocol = protocol;
525        self
526    }
527
528    /// Sets the maximum age of a connection before it's closed.
529    ///
530    /// This is distinct from the maximum idle time: if any connection's age exceeds `limit`, it will be closed rather
531    /// than being reused and added to the idle connection pool.
532    ///
533    /// Defaults to no limit.
534    pub fn with_connection_age_limit<L>(mut self, limit: L) -> Self
535    where
536        L: Into<Option<Duration>>,
537    {
538        self.conn_age_limit = limit.into();
539        self
540    }
541
542    /// Sets a counter that gets incremented with the number of bytes sent over the connection.
543    ///
544    /// This tracks bytes sent at the HTTP client level, which includes headers and body but doesn't include underlying
545    /// transport overhead, such as TLS handshaking, and so on.
546    ///
547    /// Defaults to unset.
548    pub fn with_bytes_sent_counter(mut self, counter: Counter) -> Self {
549        self.bytes_sent = Some(counter);
550        self
551    }
552
553    /// Sets the telemetry counters used to track HTTP request lifecycle failures.
554    pub(super) fn with_error_telemetry(mut self, error_telemetry: HttpTransactionErrorTelemetry) -> Self {
555        self.error_telemetry = Some(error_telemetry);
556        self
557    }
558
559    /// Sets a Unix domain socket path to route all connections through.
560    ///
561    /// When set, the connector will connect to this Unix socket instead of performing DNS resolution
562    /// and TCP connection. The URI host is ignored in this case—all requests are sent through the
563    /// configured socket.
564    ///
565    /// Defaults to unset (TCP connections via DNS).
566    #[cfg(unix)]
567    pub fn with_unix_socket_path<P: Into<PathBuf>>(mut self, path: P) -> Self {
568        self.unix_socket_path = Some(path.into());
569        self
570    }
571
572    /// Sets a vsock address to route all connections through.
573    ///
574    /// When set, the connector will connect via AF_VSOCK using the given address, bypassing
575    /// DNS and TCP. This allows connecting to a server process running in a host or hypervisor
576    /// context from within a guest VM (for example, Nitro Enclaves).
577    ///
578    /// Defaults to unset (TCP connections via DNS).
579    #[cfg(target_os = "linux")]
580    pub fn with_vsock_addr(mut self, addr: VsockAddr) -> Self {
581        self.vsock_addr = Some(addr);
582        self
583    }
584
585    /// Builds the `HttpsCapableConnector` from the given TLS configuration.
586    pub fn build(self, mut tls_config: ClientConfig) -> Result<HttpsCapableConnector, GenericError> {
587        let connect_timeout = self.connect_timeout.unwrap_or(Duration::from_secs(30));
588        let tls_handshake_timeout = self.tls_handshake_timeout.unwrap_or(Duration::from_secs(10));
589
590        // Create the HTTP connector, and ensure that we don't enforce _only_ HTTP, since that will break being able to
591        // wrap this in an HTTPS connector.
592        let mut http_connector = HttpConnector::new_with_resolver(build_dns_resolver(&self.error_telemetry));
593        http_connector.set_connect_timeout(Some(connect_timeout));
594        http_connector.enforce_http(false);
595
596        let inner_connector = InnerConnector {
597            http: http_connector,
598            #[cfg(unix)]
599            connect_timeout,
600            error_telemetry: self.error_telemetry.clone(),
601            #[cfg(unix)]
602            unix_socket_path: self.unix_socket_path.map(PathBuf::into_boxed_path).map(Arc::from),
603            #[cfg(target_os = "linux")]
604            vsock_addr: self.vsock_addr,
605        };
606
607        tls_config.alpn_protocols = http_protocol_alpns(self.http_protocol);
608
609        Ok(HttpsCapableConnector {
610            inner: inner_connector,
611            tls_config: Arc::new(tls_config),
612            tls_handshake_timeout,
613            bytes_sent: self.bytes_sent,
614            error_telemetry: self.error_telemetry,
615            conn_age_limit: self.conn_age_limit,
616        })
617    }
618}
619
620/// Selects the ALPN protocols to advertise for the given HTTP protocol.
621fn http_protocol_alpns(protocol: HttpProtocol) -> Vec<Vec<u8>> {
622    match protocol {
623        HttpProtocol::Auto => vec![b"h2".to_vec(), b"http/1.1".to_vec()],
624        HttpProtocol::Http1 => Vec::new(),
625    }
626}
627
628fn is_dns_error(error: &(dyn std::error::Error + 'static)) -> bool {
629    let mut current = Some(error);
630    while let Some(error) = current {
631        if error.downcast_ref::<DnsError>().is_some() {
632            return true;
633        }
634        current = error.source();
635    }
636    false
637}
638
639pub(super) fn check_connection_state(captured_conn: CaptureConnection) {
640    let maybe_conn_metadata = captured_conn.connection_metadata();
641    if let Some(conn_metadata) = maybe_conn_metadata.as_ref() {
642        let mut extensions = Extensions::new();
643        conn_metadata.get_extras(&mut extensions);
644
645        // If the connection has an age limit, check to see if the connection is expired (i.e. too old) and "poison"
646        // it if so. Poisoning indicates to `hyper` that the connection should be closed/dropped instead of
647        // returning it back to the idle connection pool.
648        if let Some(conn_age_limit) = extensions.get::<ConnectionAgeLimit>() {
649            if conn_age_limit.is_expired() {
650                debug!("connection is expired; poisoning it");
651                conn_metadata.poison();
652            }
653        }
654    }
655}
656
657#[cfg(test)]
658mod tests {
659    use std::{io, time::Duration};
660
661    use super::{await_handshake_with_deadline, http_protocol_alpns, HttpProtocol};
662
663    #[tokio::test(start_paused = true)]
664    async fn handshake_deadline_of_zero_disables_the_timeout() {
665        let handshake = async {
666            tokio::time::sleep(Duration::from_secs(3600)).await;
667            Ok::<_, io::Error>(())
668        };
669
670        let result = await_handshake_with_deadline(Duration::ZERO, handshake).await;
671        assert!(result.is_ok());
672    }
673
674    #[tokio::test(start_paused = true)]
675    async fn handshake_deadline_times_out_when_exceeded() {
676        let handshake = async {
677            tokio::time::sleep(Duration::from_secs(3600)).await;
678            Ok::<_, io::Error>(())
679        };
680
681        let result = await_handshake_with_deadline(Duration::from_secs(10), handshake).await;
682        let error = result.expect_err("expected handshake to time out");
683        assert!(error.to_string().contains("TLS handshake timed out"));
684    }
685
686    #[tokio::test]
687    async fn handshake_deadline_propagates_success() {
688        let handshake = async { Ok::<_, io::Error>(42) };
689
690        let result = await_handshake_with_deadline(Duration::from_secs(10), handshake).await;
691        assert_eq!(result.unwrap(), 42);
692    }
693
694    #[test]
695    fn strip_ipv6_brackets_unwraps_bracketed_addresses() {
696        use super::strip_ipv6_brackets;
697
698        assert_eq!(strip_ipv6_brackets("[::1]"), "::1");
699        assert_eq!(strip_ipv6_brackets("[2001:db8::1]"), "2001:db8::1");
700    }
701
702    #[test]
703    fn strip_ipv6_brackets_leaves_unbracketed_hosts_alone() {
704        use super::strip_ipv6_brackets;
705
706        assert_eq!(strip_ipv6_brackets("example.com"), "example.com");
707        assert_eq!(strip_ipv6_brackets("::1"), "::1");
708    }
709
710    #[cfg(unix)]
711    #[tokio::test]
712    async fn call_rejects_unsupported_uri_scheme() {
713        use std::sync::Arc;
714
715        use rustls::{ClientConfig, RootCertStore};
716        use tower::Service as _;
717
718        use super::{HttpsCapableConnector, InnerConnector};
719        use crate::net::dns::SystemResolver;
720
721        let inner = InnerConnector {
722            http: SystemResolver::new().into_http_connector(),
723            connect_timeout: Duration::from_secs(1),
724            error_telemetry: None,
725            unix_socket_path: None,
726            #[cfg(target_os = "linux")]
727            vsock_addr: None,
728        };
729
730        let tls_config = Arc::new(
731            ClientConfig::builder()
732                .with_root_certificates(RootCertStore::empty())
733                .with_no_client_auth(),
734        );
735
736        let mut connector = HttpsCapableConnector {
737            inner,
738            tls_config,
739            tls_handshake_timeout: Duration::from_secs(1),
740            bytes_sent: None,
741            error_telemetry: None,
742            conn_age_limit: None,
743        };
744
745        let uri: http::Uri = "ftp://example.com/".parse().unwrap();
746        let error = connector.call(uri).await.err().expect("expected scheme to be rejected");
747        assert!(error.to_string().contains("unsupported URI scheme"));
748    }
749
750    #[test]
751    fn auto_protocol_advertises_h2_and_http1_alpn() {
752        let alpn_protocols = http_protocol_alpns(HttpProtocol::Auto);
753
754        assert_eq!(alpn_protocols, vec![b"h2".to_vec(), b"http/1.1".to_vec()]);
755    }
756
757    #[test]
758    fn http1_protocol_leaves_alpn_empty() {
759        let alpn_protocols = http_protocol_alpns(HttpProtocol::Http1);
760
761        assert!(alpn_protocols.is_empty());
762    }
763
764    // vsock takes priority over unix when both are configured, matching Agent behavior.
765    // We verify by checking the error does not mention "unix" — if unix had priority it would
766    // fail with a socket-path error; vsock produces a connection or device error instead.
767    #[cfg(target_os = "linux")]
768    #[tokio::test]
769    async fn vsock_takes_priority_over_unix_when_both_set() {
770        use std::sync::Arc;
771
772        use tower::Service as _;
773
774        use super::{InnerConnector, VsockAddr};
775        use crate::net::dns::SystemResolver;
776
777        let mut connector = InnerConnector {
778            http: SystemResolver::new().into_http_connector(),
779            connect_timeout: std::time::Duration::from_secs(1),
780            error_telemetry: None,
781            unix_socket_path: Some(Arc::from(std::path::Path::new("/tmp/test.sock"))),
782            vsock_addr: Some(VsockAddr::new(2, 5001)),
783        };
784
785        // Verify vsock path was taken: if unix had priority the error would mention the socket
786        // path or "unix"; a vsock attempt produces a connection or device error instead.
787        let uri: http::Uri = "https://127.0.0.1:5001/".parse().unwrap();
788        let err = connector.call(uri).await.err().expect("expected a connection error");
789        assert!(
790            !err.to_string().contains("unix"),
791            "expected vsock error (not unix socket error), got: {err}"
792        );
793    }
794}
saluki_io/net/client/http/conn.rs

saluki_io/net/client/http/
conn.rs
