feat(common)!: add HTTPS_PROXY support for hyper_backend

.github/workflows/test.yml

@@ -153,14 +153,14 @@ jobs:
             cargo test --workspace $CRASHTRACKER_FEATURE --exclude builder --doc --verbose
             # shellcheck disable=SC2086
             cargo nextest run --workspace $CRASHTRACKER_FEATURE --profile ci --verbose --no-tests=pass -E '!test(tracing_integration_tests::)'
-            cargo nextest run -p libdd-http-client --no-default-features --features hyper-backend,https --profile ci --verbose
+            cargo nextest run -p libdd-http-client --no-default-features --features hyper-backend,hyper-proxy,https --profile ci --verbose
           else
             # shellcheck disable=SC2086
             cargo test $PACKAGES $CRASHTRACKER_FEATURE --doc --verbose
             # shellcheck disable=SC2086
             cargo nextest run $PACKAGES $CRASHTRACKER_FEATURE --profile ci --verbose --no-tests=pass -E '!test(tracing_integration_tests::)'
             if echo "$PACKAGES" | grep -q "libdd-http-client"; then
-              cargo nextest run -p libdd-http-client --no-default-features --features hyper-backend,https --profile ci --verbose
+              cargo nextest run -p libdd-http-client --no-default-features --features hyper-backend,hyper-proxy,https --profile ci --verbose
             fi
           fi
       - name: "[${{ steps.rust-version.outputs.version}}] cargo nextest run -p libdd-common --no-default-features --features fips"

libdd-common/Cargo.toml

@@ -107,6 +107,8 @@ regex-lite = ["dep:regex-lite"]
 require-regex-full = []
 # FIPS mode uses the FIPS-compliant cryptographic provider (Unix only)
 fips = ["tls-core", "hyper-rustls/fips"]
+# Honor HTTP(S)_PROXY/NO_PROXY environment variables for the hyper connector,
+hyper-proxy = ["hyper-util/client-proxy"]
 # Enable reqwest client builder support with file dump debugging
 reqwest = ["dep:reqwest", "test-utils"]
 # Enable test utilities for use in other crates

libdd-common/src/connector/conn_stream.rs

@@ -12,16 +12,16 @@ use pin_project::pin_project;
 
 #[pin_project(project=ConnStreamProj)]
 #[derive(Debug)]
+// `BoxedConn` is internal, and the field anyway not pub.
+#[allow(private_interfaces)]
 pub enum ConnStream {
     Tcp {
         #[pin]
         transport: TokioIo<tokio::net::TcpStream>,
     },
-    #[cfg(feature = "tls-core")]
     Tls {
         #[pin]
-        transport:
-            Box<TokioIo<tokio_rustls::client::TlsStream<TokioIo<TokioIo<tokio::net::TcpStream>>>>>,
+        transport: wrapped::BoxedConn,
     },
     #[cfg(unix)]
     Udp {
@@ -35,6 +35,73 @@ pub enum ConnStream {
     },
 }
 
+/// Wrapped connection, usable from hyper directly, or e.g. a hyper proxy wrapper
+mod wrapped {
+    use super::*;
+
+    pub trait EstablishedConn:
+        hyper::rt::Read + hyper::rt::Write + connect::Connection + Send
+    {
+    }
+
+    impl<T: hyper::rt::Read + hyper::rt::Write + connect::Connection + Send> EstablishedConn for T {}
+
+    #[derive(Debug)]
+    pub struct BoxedConn(Pin<Box<dyn EstablishedConn>>);
+
+    impl core::fmt::Debug for dyn EstablishedConn {
+        fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+            f.write_str("EstablishedConn")
+        }
+    }
+
+    impl BoxedConn {
+        pub fn new<T: EstablishedConn + 'static>(inner: T) -> Self {
+            Self(Box::pin(inner))
+        }
+    }
+
+    impl hyper::rt::Read for BoxedConn {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            buf: ReadBufCursor<'_>,
+        ) -> Poll<std::io::Result<()>> {
+            self.get_mut().0.as_mut().poll_read(cx, buf)
+        }
+    }
+
+    impl hyper::rt::Write for BoxedConn {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            buf: &[u8],
+        ) -> Poll<Result<usize, std::io::Error>> {
+            self.get_mut().0.as_mut().poll_write(cx, buf)
+        }
+
+        fn poll_flush(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<(), std::io::Error>> {
+            self.get_mut().0.as_mut().poll_flush(cx)
+        }
+
+        fn poll_shutdown(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<(), std::io::Error>> {
+            self.get_mut().0.as_mut().poll_shutdown(cx)
+        }
+    }
+
+    impl connect::Connection for BoxedConn {
+        fn connected(&self) -> connect::Connected {
+            self.0.connected()
+        }
+    }
+}
+
 pub type ConnStreamError = Box<dyn core::error::Error + Send + Sync>;
 
 use hyper_util::client::legacy::connect::{self, HttpConnector};
@@ -87,28 +154,48 @@ impl ConnStream {
     }
 
     #[cfg(feature = "tls-core")]
-    pub fn from_https_connector_with_uri(
-        c: &mut hyper_rustls::HttpsConnector<connect::HttpConnector>,
+    pub fn from_https_connector_with_uri<C>(
+        c: &mut hyper_rustls::HttpsConnector<C>,
         uri: hyper::Uri,
         require_tls: bool,
-    ) -> impl Future<Output = Result<ConnStream, ConnStreamError>> + 'static {
+    ) -> impl Future<Output = Result<ConnStream, ConnStreamError>> + 'static
+    where
+        C: Service<hyper::Uri> + 'static,
+        C::Response:
+            hyper::rt::Read + hyper::rt::Write + connect::Connection + Send + Unpin + 'static,
+        C::Future: Send + 'static,
+        C::Error: Into<Box<dyn core::error::Error + Send + Sync>>,
+    {
         #[allow(clippy::unwrap_used)]
         let stream_fut = c.call(uri.to_string().parse().unwrap());
         async move {
             let stream = stream_fut.await?;
-            match stream {
-                // move only require_tls
-                hyper_rustls::MaybeHttpsStream::Http(t) => {
-                    if require_tls {
-                        Err(super::errors::Error::CannotEstablishTlsConnection.into())
-                    } else {
-                        Ok(ConnStream::Tcp { transport: t })
-                    }
-                }
-                hyper_rustls::MaybeHttpsStream::Https(t) => Ok(ConnStream::Tls {
-                    transport: Box::from(t),
-                }),
+            if require_tls && matches!(stream, hyper_rustls::MaybeHttpsStream::Http(_)) {
+                return Err(super::errors::Error::CannotEstablishTlsConnection.into());
             }
+            Ok(ConnStream::Tls {
+                transport: wrapped::BoxedConn::new(stream),
+            })
+        }
+    }
+
+    pub fn from_tunnel_with_uri<C>(
+        c: &mut C,
+        uri: hyper::Uri,
+    ) -> impl Future<Output = Result<ConnStream, ConnStreamError>> + 'static
+    where
+        C: Service<hyper::Uri> + 'static,
+        C::Response:
+            hyper::rt::Read + hyper::rt::Write + connect::Connection + Send + Unpin + 'static,
+        C::Future: Send + 'static,
+        C::Error: Into<Box<dyn core::error::Error + Send + Sync>>,
+    {
+        let stream_fut = c.call(uri);
+        async move {
+            let stream = stream_fut.await.map_err(Into::into)?;
+            Ok(ConnStream::Tls {
+                transport: wrapped::BoxedConn::new(stream),
+            })
         }
     }
 }
@@ -121,7 +208,6 @@ impl hyper::rt::Read for ConnStream {
     ) -> Poll<std::io::Result<()>> {
         match self.project() {
             ConnStreamProj::Tcp { transport } => transport.poll_read(cx, buf),
-            #[cfg(feature = "tls-core")]
             ConnStreamProj::Tls { transport } => transport.poll_read(cx, buf),
             #[cfg(unix)]
             ConnStreamProj::Udp { transport } => transport.poll_read(cx, buf),
@@ -135,11 +221,7 @@ impl connect::Connection for ConnStream {
     fn connected(&self) -> connect::Connected {
         match self {
             Self::Tcp { transport } => transport.connected(),
-            #[cfg(feature = "tls-core")]
-            Self::Tls { transport } => {
-                let (tcp, _) = transport.inner().get_ref();
-                tcp.inner().inner().connected()
-            }
+            Self::Tls { transport } => transport.connected(),
             #[cfg(unix)]
             Self::Udp { transport: _ } => connect::Connected::new(),
             #[cfg(windows)]
@@ -156,7 +238,6 @@ impl hyper::rt::Write for ConnStream {
     ) -> Poll<Result<usize, std::io::Error>> {
         match self.project() {
             ConnStreamProj::Tcp { transport } => transport.poll_write(cx, buf),
-            #[cfg(feature = "tls-core")]
             ConnStreamProj::Tls { transport } => transport.poll_write(cx, buf),
             #[cfg(unix)]
             ConnStreamProj::Udp { transport } => transport.poll_write(cx, buf),
@@ -171,7 +252,6 @@ impl hyper::rt::Write for ConnStream {
     ) -> Poll<Result<(), std::io::Error>> {
         match self.project() {
             ConnStreamProj::Tcp { transport } => transport.poll_shutdown(cx),
-            #[cfg(feature = "tls-core")]
             ConnStreamProj::Tls { transport } => transport.poll_shutdown(cx),
             #[cfg(unix)]
             ConnStreamProj::Udp { transport } => transport.poll_shutdown(cx),
@@ -183,7 +263,6 @@ impl hyper::rt::Write for ConnStream {
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
         match self.project() {
             ConnStreamProj::Tcp { transport } => transport.poll_flush(cx),
-            #[cfg(feature = "tls-core")]
             ConnStreamProj::Tls { transport } => transport.poll_flush(cx),
             #[cfg(unix)]
             ConnStreamProj::Udp { transport } => transport.poll_flush(cx),

libdd-common/src/connector/mod.rs

@@ -20,11 +20,18 @@ pub mod errors;
 mod conn_stream;
 use conn_stream::{ConnStream, ConnStreamError};
 
+#[cfg(feature = "hyper-proxy")]
+mod proxy;
+
 #[derive(Clone)]
+// `proxy::HttpProxyConnector` is crate internal, and the field anyway not pub.
+#[allow(private_interfaces)]
 pub enum Connector {
     Http(connect::HttpConnector),
     #[cfg(feature = "tls-core")]
     Https(hyper_rustls::HttpsConnector<connect::HttpConnector>),
+    #[cfg(feature = "hyper-proxy")]
+    Proxy(Box<proxy::HttpProxyConnector>),
 }
 
 static DEFAULT_CONNECTOR: LazyLock<Connector> = LazyLock::new(Connector::new);
@@ -39,14 +46,22 @@ impl Connector {
     /// Make sure this function is not called frequently. Fetching the root certificates is an
     /// expensive operation. Access the globally cached connector via Connector::default().
     fn new() -> Self {
-        #[cfg(feature = "tls-core")]
+        #[cfg(feature = "hyper-proxy")]
+        {
+            Connector::Proxy(Box::new(proxy::HttpProxyConnector::new(
+                Self::new_no_proxy(),
+            )))
+        }
+        #[cfg(not(feature = "hyper-proxy"))]
         {
-            #[cfg(feature = "use_webpki_roots")]
-            let https_connector_fn = https::build_https_connector_with_webpki_roots;
-            #[cfg(not(feature = "use_webpki_roots"))]
-            let https_connector_fn = https::build_https_connector;
+            Self::new_no_proxy()
+        }
+    }
 
-            match https_connector_fn() {
+    pub(super) fn new_no_proxy() -> Self {
+        #[cfg(feature = "tls-core")]
+        {
+            match https::build_https_connector() {
                 Ok(connector) => Connector::Https(connector),
                 Err(_) => Connector::Http(connect::HttpConnector::new()),
             }
@@ -77,6 +92,8 @@ impl Connector {
             Self::Https(c) => {
                 ConnStream::from_https_connector_with_uri(c, uri, require_tls).boxed()
             }
+            #[cfg(feature = "hyper-proxy")]
+            Self::Proxy(p) => p.build_conn_stream(uri, require_tls),
         }
     }
 }
@@ -106,39 +123,49 @@ mod https {
     #[cfg(not(feature = "https"))]
     fn ensure_crypto_provider_initialized() {}
 
+    #[cfg(any(feature = "https", feature = "fips"))]
+    fn require_crypto_provider() -> anyhow::Result<()> {
+        Ok(())
+    }
+
+    #[cfg(not(any(feature = "https", feature = "fips")))]
+    fn require_crypto_provider() -> anyhow::Result<()> {
+        if rustls::crypto::CryptoProvider::get_default().is_none() {
+            anyhow::bail!("no rustls CryptoProvider installed");
+        }
+        Ok(())
+    }
+
     #[cfg(feature = "use_webpki_roots")]
-    pub(super) fn build_https_connector_with_webpki_roots() -> anyhow::Result<
-        hyper_rustls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>,
-    > {
+    pub(super) fn build_tls_config() -> anyhow::Result<ClientConfig> {
         ensure_crypto_provider_initialized(); // One-time initialization of a crypto provider if needed
+        require_crypto_provider()?;
 
-        let client_config = ClientConfig::builder()
+        Ok(ClientConfig::builder()
             .with_webpki_roots()
-            .with_no_client_auth();
-        Ok(hyper_rustls::HttpsConnectorBuilder::new()
-            .with_tls_config(client_config)
-            .https_or_http()
-            .enable_http1()
-            .build())
+            .with_no_client_auth())
     }
 
     #[cfg(not(feature = "use_webpki_roots"))]
-    /// Returns a default connector that uses the system trust roots.
+    /// Builds the client TLS config using the system trust roots.
     /// `SSL_CERT_FILE` and `SSL_CERT_DIR` variable are only supported on linux, see
     /// `rustls_platform_verifier` doc for details.
-    pub(super) fn build_https_connector() -> anyhow::Result<
-        hyper_rustls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>,
-    > {
+    pub(super) fn build_tls_config() -> anyhow::Result<ClientConfig> {
         use rustls_platform_verifier::BuilderVerifierExt;
 
         ensure_crypto_provider_initialized(); // One-time initialization of a crypto provider if needed
+        require_crypto_provider()?;
 
-        let client_config = ClientConfig::builder()
+        Ok(ClientConfig::builder()
             .with_platform_verifier()?
-            .with_no_client_auth();
+            .with_no_client_auth())
+    }
 
+    pub(super) fn build_https_connector() -> anyhow::Result<
+        hyper_rustls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>,
+    > {
         Ok(hyper_rustls::HttpsConnectorBuilder::new()
-            .with_tls_config(client_config)
+            .with_tls_config(build_tls_config()?)
             .https_or_http()
             .enable_http1()
             .build())
@@ -168,6 +195,8 @@ impl tower_service::Service<hyper::Uri> for Connector {
             Connector::Http(c) => c.poll_ready(cx).map_err(|e| e.into()),
             #[cfg(feature = "tls-core")]
             Connector::Https(c) => c.poll_ready(cx),
+            #[cfg(feature = "hyper-proxy")]
+            Connector::Proxy(p) => p.poll_ready(cx),
         }
     }
 }
@@ -212,7 +241,7 @@ mod tests {
 
         env::set_var(ENV_SSL_CERT_FILE, "this/folder/does/not/exist");
         env::set_var(ENV_SSL_CERT_DIR, "this/folder/does/not/exist");
-        let connector = Connector::new();
+        let connector = Connector::new_no_proxy();
 
         assert!(matches!(connector, Connector::Http(_)));
 
@@ -231,7 +260,7 @@ mod tests {
         let old_value = env::var(ENV_SSL_CERT_FILE).unwrap_or_default();
 
         env::set_var(ENV_SSL_CERT_FILE, "this/folder/does/not/exist");
-        let connector = Connector::new();
+        let connector = Connector::new_no_proxy();
         assert!(matches!(connector, Connector::Https(_)));
 
         env::set_var(ENV_SSL_CERT_FILE, old_value);