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

saluki_io/net/client/http/
conn.rs
