CONNECT proxy support

src/agent.rs

@@ -141,7 +141,10 @@ impl Agent {
 
         let mut unit = Unit::new(self.config.clone(), current_time(), request, body)?;
 
+        // For CONNECT proxy, this is the address of the proxy server, for
+        // all other cases it's the address of the URL being requested.
         let mut addr = None;
+
         let mut connection: Option<Connection> = None;
         let mut response;
         let mut no_buffers = NoBuffers;
@@ -235,11 +238,16 @@ impl Agent {
                 }
 
                 Event::Resolve { uri, timeout } => {
+                    // If we're using a CONNECT proxy, we need to resolve that hostname.
+                    let maybe_connect_uri = self.config.connect_proxy_uri();
+
+                    let effective_uri = maybe_connect_uri.unwrap_or(uri);
+
                     // Before resolving the URI we need to ensure it is a full URI. We
                     // cannot make requests with partial uri like "/path".
-                    uri.ensure_valid_url()?;
+                    effective_uri.ensure_valid_url()?;
 
-                    addr = Some(self.resolver.resolve(uri, timeout)?);
+                    addr = Some(self.resolver.resolve(effective_uri, timeout)?);
                     unit.handle_input(current_time(), Input::Resolved, &mut [])?;
                 }
 

src/config.rs

@@ -2,6 +2,7 @@ use std::fmt;
 use std::time::Duration;
 
 use hoot::client::flow::RedirectAuthHeaders;
+use http::Uri;
 
 use crate::Proxy;
 
@@ -184,6 +185,18 @@ mod private {
     pub struct Private;
 }
 
+impl AgentConfig {
+    pub(crate) fn connect_proxy_uri(&self) -> Option<&Uri> {
+        let proxy = self.proxy.as_ref()?;
+
+        if !proxy.proto().is_connect() {
+            return None;
+        }
+
+        Some(proxy.uri())
+    }
+}
+
 impl Default for AgentConfig {
     fn default() -> Self {
         Self {

src/error.rs

@@ -153,6 +153,10 @@ pub enum Error {
     #[error("json: {0}")]
     Json(#[from] serde_json::Error),
 
+    /// Attempt to connect to a CONNECT proxy failed.
+    #[error("CONNECT proxy failed: {0}")]
+    ConnectProxyFailed(String),
+
     /// hoot made no progress and there is no more input to read.
     ///
     /// We should never see this value.

src/lib.rs

@@ -595,5 +595,3 @@ pub(crate) mod test {
         is_sync(owned_reader);
     }
 }
-
-// TODO(martin): CONNECT proxy

src/proxy.rs

@@ -1,9 +1,14 @@
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use hoot::parser::try_parse_response;
 use std::convert::{TryFrom, TryInto};
 use std::fmt;
+use std::io::Write;
 
-use http::Uri;
+use http::{StatusCode, Uri};
 
-use crate::util::{AuthorityExt, DebugUri};
+use crate::transport::{ConnectionDetails, Connector, Transport, TransportAdapter};
+use crate::util::{AuthorityExt, DebugUri, SchemeExt, UriExt};
 use crate::Error;
 
 /// Proxy protocol
@@ -29,6 +34,10 @@ impl Proto {
     pub fn is_socks(&self) -> bool {
         matches!(self, Self::Socks4 | Self::Socks4A | Self::Socks5)
     }
+
+    pub(crate) fn is_connect(&self) -> bool {
+        matches!(self, Self::Http | Self::Https)
+    }
 }
 
 /// Proxy server settings
@@ -157,46 +166,90 @@ impl Proxy {
     pub fn is_from_env(&self) -> bool {
         self.from_env
     }
+}
+
+/// Connector for CONNECT proxy settings.
+///
+/// This operates on the previous chained transport typically a TcpConnector optionally
+/// wrapped in TLS.
+pub struct ConnectProxyConnector;
+
+impl Connector for ConnectProxyConnector {
+    fn connect(
+        &self,
+        details: &ConnectionDetails,
+        chained: Option<Box<dyn Transport>>,
+    ) -> Result<Option<Box<dyn Transport>>, Error> {
+        let Some(transport) = chained else {
+            return Ok(None);
+        };
+
+        let is_connect_proxy = details.config.connect_proxy_uri().is_some();
+
+        if is_connect_proxy {
+            // unwrap is ok because connect_proxy_uri() above checks it.
+            let proxy = details.config.proxy.as_ref().unwrap();
+
+            let mut w = TransportAdapter::new(transport);
+
+            let uri = &details.uri;
+            uri.ensure_valid_url()?;
+
+            // All these unwrap() are ok because ensure_valid_uri() above checks them.
+            let host = uri.host().unwrap();
+            let port = uri
+                .port_u16()
+                .unwrap_or(uri.scheme().unwrap().default_port().unwrap());
+
+            write!(w, "CONNECT {}:{} HTTP/1.1\r\n", host, port)?;
+            write!(w, "Host: {}:{}\r\n", host, port)?;
+            write!(w, "User-Agent: {}\r\n", details.config.user_agent)?;
+            write!(w, "Proxy-Connection: Keep-Alive\r\n")?;
+
+            let use_creds = proxy.username().is_some() || proxy.password().is_some();
+
+            if use_creds {
+                let user = proxy.username().unwrap_or_default();
+                let pass = proxy.password().unwrap_or_default();
+                let creds = BASE64_STANDARD.encode(format!("{}:{}", user, pass));
+                write!(w, "Proxy-Authorization: basic {}\r\n", creds)?;
+            }
+
+            write!(w, "\r\n")?;
+            w.flush()?;
+
+            let mut transport = w.into_inner();
+
+            let response = loop {
+                let made_progress = transport.await_input(details.timeout)?;
+                let buffers = transport.buffers();
+                let input = buffers.input();
+                let Some((used_input, response)) = try_parse_response::<20>(input)? else {
+                    if !made_progress {
+                        let reason = "proxy server did not respond".to_string();
+                        return Err(Error::ConnectProxyFailed(reason));
+                    }
+                    continue;
+                };
+                buffers.consume(used_input);
+                break response;
+            };
 
-    //     pub(crate) fn connect<S: AsRef<str>>(&self, host: S, port: u16, user_agent: &str) -> String {
-    //         let authorization = if self.use_authorization() {
-    //             let creds = BASE64_STANDARD.encode(format!(
-    //                 "{}:{}",
-    //                 self.username.clone().unwrap_or_default(),
-    //                 self.password.clone().unwrap_or_default()
-    //             ));
-
-    //             match self.proto {
-    //                 Proto::HTTP => format!("Proxy-Authorization: basic {}\r\n", creds),
-    //                 _ => String::new(),
-    //             }
-    //         } else {
-    //             String::new()
-    //         };
-
-    //         format!(
-    //             "CONNECT {}:{} HTTP/1.1\r\n\
-    // Host: {}:{}\r\n\
-    // User-Agent: {}\r\n\
-    // Proxy-Connection: Keep-Alive\r\n\
-    // {}\
-    // \r\n",
-    //             host.as_ref(),
-    //             port,
-    //             host.as_ref(),
-    //             port,
-    //             user_agent,
-    //             authorization
-    //         )
-    //     }
-
-    // pub(crate) fn verify_response(response: &Response) -> Result<(), Error> {
-    //     match response.status() {
-    //         200 => Ok(()),
-    //         401 | 407 => Err(ErrorKind::ProxyUnauthorized.new()),
-    //         _ => Err(ErrorKind::ProxyConnect.new()),
-    //     }
-    // }
+            match response.status() {
+                StatusCode::OK => {
+                    trace!("CONNECT proxy connected");
+                }
+                x => {
+                    let reason = format!("proxy server responded {}/{}", x.as_u16(), x.as_str());
+                    return Err(Error::ConnectProxyFailed(reason));
+                }
+            }
+
+            Ok(Some(transport))
+        } else {
+            Ok(Some(transport))
+        }
+    }
 }
 
 impl TryFrom<&str> for Proto {
@@ -225,6 +278,12 @@ impl fmt::Debug for Proxy {
     }
 }
 
+impl fmt::Debug for ConnectProxyConnector {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ProxyConnector").finish()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/tls/native_tls.rs

@@ -8,7 +8,6 @@ use crate::transport::time::NextTimeout;
 use crate::{transport::*, Error};
 use der::pem::LineEnding;
 use der::Document;
-use http::uri::Scheme;
 use native_tls::{Certificate, HandshakeError, Identity, TlsConnector};
 use native_tls::{TlsConnectorBuilder, TlsStream};
 use once_cell::sync::OnceCell;
@@ -35,7 +34,7 @@ impl Connector for NativeTlsConnector {
 
         // Only add TLS if we are connecting via HTTPS and the transport isn't TLS
         // already, otherwise use chained transport as is.
-        if details.uri.scheme() != Some(&Scheme::HTTPS) || transport.is_tls() {
+        if !details.needs_tls() || transport.is_tls() {
             trace!("Skip");
             return Ok(Some(transport));
         }

src/tls/rustls.rs

@@ -3,7 +3,6 @@ use std::fmt;
 use std::io::{Read, Write};
 use std::sync::Arc;
 
-use http::uri::Scheme;
 use once_cell::sync::OnceCell;
 use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
 use rustls::{ClientConfig, ClientConnection, RootCertStore, StreamOwned, ALL_VERSIONS};
@@ -39,7 +38,7 @@ impl Connector for RustlsConnector {
 
         // Only add TLS if we are connecting via HTTPS and the transport isn't TLS
         // already, otherwise use chained transport as is.
-        if details.uri.scheme() != Some(&Scheme::HTTPS) || transport.is_tls() {
+        if !details.needs_tls() || transport.is_tls() {
             trace!("Skip");
             return Ok(Some(transport));
         }

src/transport/io.rs

@@ -36,6 +36,11 @@ impl TransportAdapter {
     pub fn get_mut(&mut self) -> &mut dyn Transport {
         &mut *self.transport
     }
+
+    /// Turn the adapter back into the wrapped transport
+    pub fn into_inner(self) -> Box<dyn Transport> {
+        self.transport
+    }
 }
 
 impl io::Read for TransportAdapter {

src/transport/mod.rs

@@ -24,8 +24,10 @@
 use std::fmt::Debug;
 use std::net::SocketAddr;
 
+use http::uri::Scheme;
 use http::Uri;
 
+use crate::proxy::Proto;
 use crate::resolver::Resolver;
 use crate::{AgentConfig, Error};
 
@@ -54,6 +56,8 @@ mod socks;
 #[cfg(feature = "socks-proxy")]
 pub use self::socks::SocksConnector;
 
+pub use crate::proxy::ConnectProxyConnector;
+
 pub mod time;
 
 /// Trait for components providing some aspect of connecting.
@@ -103,6 +107,8 @@ pub struct ConnectionDetails<'a> {
     pub uri: &'a Uri,
 
     /// A resolved IP address for the uri being requested. See [`Resolver`].
+    ///
+    /// For CONNECT proxy, this is the address of the proxy server.
     pub addr: SocketAddr,
 
     /// The Agent configuration.
@@ -112,7 +118,7 @@ pub struct ConnectionDetails<'a> {
     ///
     /// Typically the IP address of the host in the uri is already resolved to the `addr`
     /// property. However there might be cases where additional DNS lookups need to be
-    /// made in the connector itself, such as resolving a proxy server.
+    /// made in the connector itself, such as resolving a SOCKS proxy server.
     pub resolver: &'a dyn Resolver,
 
     /// Current time.
@@ -123,6 +129,22 @@ pub struct ConnectionDetails<'a> {
     pub timeout: NextTimeout,
 }
 
+impl<'a> ConnectionDetails<'a> {
+    /// Tell if the requested socket need TLS wrapping.
+    ///
+    /// This is (obviously) true for URLs starting `https`, but
+    /// also in the case of using a CONNECT proxy over https.
+    pub fn needs_tls(&self) -> bool {
+        if let Some(p) = &self.config.proxy {
+            if p.proto() == Proto::Https {
+                return true;
+            }
+        }
+
+        self.uri.scheme() == Some(&Scheme::HTTPS)
+    }
+}
+
 /// Transport of HTTP/1.1 as created by a [`Connector`].
 ///
 /// In ureq, [`Transport`] and [`Buffers`] go hand in hand. The rest of ureq tries to minimize
@@ -241,6 +263,9 @@ impl Default for DefaultConnector {
             // TLS provider is not enabled by feature flags.
             #[cfg(feature = "_tls")]
             no_tls::WarnOnMissingTlsProvider(crate::tls::TlsProvider::NativeTls).boxed(),
+            //
+            // Do the final CONNECT proxy on top of the connection if indicated by config.
+            ConnectProxyConnector.boxed(),
         ]);
 
         DefaultConnector { chain }
@@ -296,8 +321,6 @@ mod no_proxy {
 
 #[cfg(feature = "_tls")]
 mod no_tls {
-    use http::uri::Scheme;
-
     use crate::tls::TlsProvider;
 
     use super::{ConnectionDetails, Connector, Debug, Error, Transport};
@@ -319,9 +342,7 @@ mod no_tls {
 
             let tls_config = &details.config.tls_config;
 
-            if details.uri.scheme() != Some(&Scheme::HTTPS)
-                && tls_config.provider == self.0
-                && !self.0.is_feature_enabled()
+            if details.needs_tls() && tls_config.provider == self.0 && !self.0.is_feature_enabled()
             {
                 panic!(
                     "uri scheme is https, provider is {:?} but feature is not enabled: {}",