hyper-util - cargo Package Compare versions

Comparing version 0.1.11 to 0.1.12

src/client/legacy/connect/proxy/mod.rs

+//! Proxy helpers
+mod socks;
+mod tunnel;
+
+pub use self::socks::{SocksV4, SocksV5};
+pub use self::tunnel::Tunnel;

src/client/legacy/connect/proxy/socks/mod.rs

+mod v5;
+pub use v5::{SocksV5, SocksV5Error};
+
+mod v4;
+pub use v4::{SocksV4, SocksV4Error};
+
+use bytes::BytesMut;
+
+use hyper::rt::Read;
+
+#[derive(Debug)]
+pub enum SocksError<C> {
+    Inner(C),
+    Io(std::io::Error),
+
+    DnsFailure,
+    MissingHost,
+    MissingPort,
+
+    V4(SocksV4Error),
+    V5(SocksV5Error),
+
+    Parsing(ParsingError),
+    Serialize(SerializeError),
+}
+
+#[derive(Debug)]
+pub enum ParsingError {
+    Incomplete,
+    WouldOverflow,
+    Other,
+}
+
+#[derive(Debug)]
+pub enum SerializeError {
+    WouldOverflow,
+}
+
+async fn read_message<T, M, C>(mut conn: &mut T, buf: &mut BytesMut) -> Result<M, SocksError<C>>
+where
+    T: Read + Unpin,
+    M: for<'a> TryFrom<&'a mut BytesMut, Error = ParsingError>,
+{
+    let mut tmp = [0; 513];
+
+    loop {
+        let n = crate::rt::read(&mut conn, &mut tmp).await?;
+        buf.extend_from_slice(&tmp[..n]);
+
+        match M::try_from(buf) {
+            Err(ParsingError::Incomplete) => {
+                if n == 0 {
+                    if buf.spare_capacity_mut().len() == 0 {
+                        return Err(SocksError::Parsing(ParsingError::WouldOverflow));
+                    } else {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::UnexpectedEof,
+                            "unexpected eof",
+                        )
+                        .into());
+                    }
+                }
+            }
+            Err(err) => return Err(err.into()),
+            Ok(res) => return Ok(res),
+        }
+    }
+}
+
+impl<C> std::fmt::Display for SocksError<C> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("SOCKS error: ")?;
+
+        match self {
+            Self::Inner(_) => f.write_str("failed to create underlying connection"),
+            Self::Io(_) => f.write_str("io error during SOCKS handshake"),
+
+            Self::DnsFailure => f.write_str("could not resolve to acceptable address type"),
+            Self::MissingHost => f.write_str("missing destination host"),
+            Self::MissingPort => f.write_str("missing destination port"),
+
+            Self::Parsing(_) => f.write_str("failed parsing server response"),
+            Self::Serialize(_) => f.write_str("failed serialize request"),
+
+            Self::V4(e) => e.fmt(f),
+            Self::V5(e) => e.fmt(f),
+        }
+    }
+}
+
+impl<C: std::fmt::Debug + std::fmt::Display> std::error::Error for SocksError<C> {}
+
+impl<C> From<std::io::Error> for SocksError<C> {
+    fn from(err: std::io::Error) -> Self {
+        Self::Io(err)
+    }
+}
+
+impl<C> From<ParsingError> for SocksError<C> {
+    fn from(err: ParsingError) -> Self {
+        Self::Parsing(err)
+    }
+}
+
+impl<C> From<SerializeError> for SocksError<C> {
+    fn from(err: SerializeError) -> Self {
+        Self::Serialize(err)
+    }
+}
+
+impl<C> From<SocksV4Error> for SocksError<C> {
+    fn from(err: SocksV4Error) -> Self {
+        Self::V4(err)
+    }
+}
+
+impl<C> From<SocksV5Error> for SocksError<C> {
+    fn from(err: SocksV5Error) -> Self {
+        Self::V5(err)
+    }
+}

src/client/legacy/connect/proxy/socks/v4/errors.rs

+use super::Status;
+
+#[derive(Debug)]
+pub enum SocksV4Error {
+    IpV6,
+    Command(Status),
+}
+
+impl From<Status> for SocksV4Error {
+    fn from(err: Status) -> Self {
+        Self::Command(err)
+    }
+}
+
+impl std::fmt::Display for SocksV4Error {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::IpV6 => f.write_str("IPV6 is not supported"),
+            Self::Command(status) => status.fmt(f),
+        }
+    }
+}

src/client/legacy/connect/proxy/socks/v4/messages.rs

+use super::super::{ParsingError, SerializeError};
+
+use bytes::{Buf, BufMut, BytesMut};
+use std::net::SocketAddrV4;
+
+/// +-----+-----+----+----+----+----+----+----+-------------+------+------------+------+
+/// |  VN |  CD | DSTPORT |        DSTIP      |    USERID   | NULL |   DOMAIN   | NULL |
+/// +-----+-----+----+----+----+----+----+----+-------------+------+------------+------+
+/// |  1  |  1  |    2    |         4         |   Variable  |  1   |  Variable  |   1  |
+/// +-----+-----+----+----+----+----+----+----+-------------+------+------------+------+
+///                                                                ^^^^^^^^^^^^^^^^^^^^^
+///                                                      optional: only do IP is 0.0.0.X
+#[derive(Debug)]
+pub struct Request<'a>(pub &'a Address);
+
+/// +-----+-----+----+----+----+----+----+----+
+/// |  VN |  CD | DSTPORT |       DSTIP       |
+/// +-----+-----+----+----+----+----+----+----+
+/// |  1  |  1  |    2    |         4         |
+/// +-----+-----+----+----+----+----+----+----+
+///             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+///              ignore: only for SOCKSv4 BIND
+#[derive(Debug)]
+pub struct Response(pub Status);
+
+#[derive(Debug)]
+pub enum Address {
+    Socket(SocketAddrV4),
+    Domain(String, u16),
+}
+
+#[derive(Debug, PartialEq)]
+pub enum Status {
+    Success = 90,
+    Failed = 91,
+    IdentFailure = 92,
+    IdentMismatch = 93,
+}
+
+impl Request<'_> {
+    pub fn write_to_buf<B: BufMut>(&self, mut buf: B) -> Result<usize, SerializeError> {
+        match self.0 {
+            Address::Socket(socket) => {
+                if buf.remaining_mut() < 10 {
+                    return Err(SerializeError::WouldOverflow);
+                }
+
+                buf.put_u8(0x04); // Version
+                buf.put_u8(0x01); // CONNECT
+
+                buf.put_u16(socket.port()); // Port
+                buf.put_slice(&socket.ip().octets()); // IP
+
+                buf.put_u8(0x00); // USERID
+                buf.put_u8(0x00); // NULL
+
+                Ok(10)
+            }
+
+            Address::Domain(domain, port) => {
+                if buf.remaining_mut() < 10 + domain.len() + 1 {
+                    return Err(SerializeError::WouldOverflow);
+                }
+
+                buf.put_u8(0x04); // Version
+                buf.put_u8(0x01); // CONNECT
+
+                buf.put_u16(*port); // IP
+                buf.put_slice(&[0x00, 0x00, 0x00, 0xFF]); // Invalid IP
+
+                buf.put_u8(0x00); // USERID
+                buf.put_u8(0x00); // NULL
+
+                buf.put_slice(domain.as_bytes()); // Domain
+                buf.put_u8(0x00); // NULL
+
+                Ok(10 + domain.len() + 1)
+            }
+        }
+    }
+}
+
+impl TryFrom<&mut BytesMut> for Response {
+    type Error = ParsingError;
+
+    fn try_from(buf: &mut BytesMut) -> Result<Self, Self::Error> {
+        if buf.remaining() < 8 {
+            return Err(ParsingError::Incomplete);
+        }
+
+        if buf.get_u8() != 0x00 {
+            return Err(ParsingError::Other);
+        }
+
+        let status = buf.get_u8().try_into()?;
+        let _addr = {
+            let port = buf.get_u16();
+            let mut ip = [0; 4];
+            buf.copy_to_slice(&mut ip);
+
+            SocketAddrV4::new(ip.into(), port)
+        };
+
+        return Ok(Self(status));
+    }
+}
+
+impl TryFrom<u8> for Status {
+    type Error = ParsingError;
+
+    fn try_from(byte: u8) -> Result<Self, Self::Error> {
+        Ok(match byte {
+            90 => Self::Success,
+            91 => Self::Failed,
+            92 => Self::IdentFailure,
+            93 => Self::IdentMismatch,
+            _ => return Err(ParsingError::Other),
+        })
+    }
+}
+
+impl std::fmt::Display for Status {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Self::Success => "success",
+            Self::Failed => "server failed to execute command",
+            Self::IdentFailure => "server ident service failed",
+            Self::IdentMismatch => "server ident service did not recognise client identifier",
+        })
+    }
+}

src/client/legacy/connect/proxy/socks/v4/mod.rs

+mod errors;
+pub use errors::*;
+
+mod messages;
+use messages::*;
+
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use std::net::{IpAddr, SocketAddr, SocketAddrV4, ToSocketAddrs};
+
+use http::Uri;
+use hyper::rt::{Read, Write};
+use tower_service::Service;
+
+use bytes::BytesMut;
+
+use pin_project_lite::pin_project;
+
+/// Tunnel Proxy via SOCKSv4
+///
+/// This is a connector that can be used by the `legacy::Client`. It wraps
+/// another connector, and after getting an underlying connection, it established
+/// a TCP tunnel over it using SOCKSv4.
+#[derive(Debug, Clone)]
+pub struct SocksV4<C> {
+    inner: C,
+    config: SocksConfig,
+}
+
+#[derive(Debug, Clone)]
+struct SocksConfig {
+    proxy: Uri,
+    local_dns: bool,
+}
+
+pin_project! {
+    // Not publicly exported (so missing_docs doesn't trigger).
+    //
+    // We return this `Future` instead of the `Pin<Box<dyn Future>>` directly
+    // so that users don't rely on it fitting in a `Pin<Box<dyn Future>>` slot
+    // (and thus we can change the type in the future).
+    #[must_use = "futures do nothing unless polled"]
+    #[allow(missing_debug_implementations)]
+    pub struct Handshaking<F, T, E> {
+        #[pin]
+        fut: BoxHandshaking<T, E>,
+        _marker: std::marker::PhantomData<F>
+    }
+}
+
+type BoxHandshaking<T, E> = Pin<Box<dyn Future<Output = Result<T, super::SocksError<E>>> + Send>>;
+
+impl<C> SocksV4<C> {
+    /// Create a new SOCKSv4 handshake service
+    ///
+    /// Wraps an underlying connector and stores the address of a tunneling
+    /// proxying server.
+    ///
+    /// A `SocksV4` can then be called with any destination. The `dst` passed to
+    /// `call` will not be used to create the underlying connection, but will
+    /// be used in a SOCKS handshake with the proxy destination.
+    pub fn new(proxy_dst: Uri, connector: C) -> Self {
+        Self {
+            inner: connector,
+            config: SocksConfig::new(proxy_dst),
+        }
+    }
+
+    /// Resolve domain names locally on the client, rather than on the proxy server.
+    ///
+    /// Disabled by default as local resolution of domain names can be detected as a
+    /// DNS leak.
+    pub fn local_dns(mut self, local_dns: bool) -> Self {
+        self.config.local_dns = local_dns;
+        self
+    }
+}
+
+impl SocksConfig {
+    pub fn new(proxy: Uri) -> Self {
+        Self {
+            proxy,
+            local_dns: false,
+        }
+    }
+
+    async fn execute<T, E>(
+        self,
+        mut conn: T,
+        host: String,
+        port: u16,
+    ) -> Result<T, super::SocksError<E>>
+    where
+        T: Read + Write + Unpin,
+    {
+        let address = match host.parse::<IpAddr>() {
+            Ok(IpAddr::V6(_)) => return Err(SocksV4Error::IpV6.into()),
+            Ok(IpAddr::V4(ip)) => Address::Socket(SocketAddrV4::new(ip.into(), port)),
+            Err(_) => {
+                if self.local_dns {
+                    (host, port)
+                        .to_socket_addrs()?
+                        .find_map(|s| {
+                            if let SocketAddr::V4(v4) = s {
+                                Some(Address::Socket(v4))
+                            } else {
+                                None
+                            }
+                        })
+                        .ok_or(super::SocksError::DnsFailure)?
+                } else {
+                    Address::Domain(host, port)
+                }
+            }
+        };
+
+        let mut send_buf = BytesMut::with_capacity(1024);
+        let mut recv_buf = BytesMut::with_capacity(1024);
+
+        // Send Request
+        let req = Request(&address);
+        let n = req.write_to_buf(&mut send_buf)?;
+        crate::rt::write_all(&mut conn, &send_buf[..n]).await?;
+
+        // Read Response
+        let res: Response = super::read_message(&mut conn, &mut recv_buf).await?;
+        if res.0 == Status::Success {
+            Ok(conn)
+        } else {
+            Err(SocksV4Error::Command(res.0).into())
+        }
+    }
+}
+
+impl<C> Service<Uri> for SocksV4<C>
+where
+    C: Service<Uri>,
+    C::Future: Send + 'static,
+    C::Response: Read + Write + Unpin + Send + 'static,
+    C::Error: Send + 'static,
+{
+    type Response = C::Response;
+    type Error = super::SocksError<C::Error>;
+    type Future = Handshaking<C::Future, C::Response, C::Error>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx).map_err(super::SocksError::Inner)
+    }
+
+    fn call(&mut self, dst: Uri) -> Self::Future {
+        let config = self.config.clone();
+        let connecting = self.inner.call(config.proxy.clone());
+
+        let fut = async move {
+            let port = dst.port().map(|p| p.as_u16()).unwrap_or(443);
+            let host = dst
+                .host()
+                .ok_or(super::SocksError::MissingHost)?
+                .to_string();
+
+            let conn = connecting.await.map_err(super::SocksError::Inner)?;
+            config.execute(conn, host, port).await
+        };
+
+        Handshaking {
+            fut: Box::pin(fut),
+            _marker: Default::default(),
+        }
+    }
+}
+
+impl<F, T, E> Future for Handshaking<F, T, E>
+where
+    F: Future<Output = Result<T, E>>,
+{
+    type Output = Result<T, super::SocksError<E>>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().fut.poll(cx)
+    }
+}

src/client/legacy/connect/proxy/socks/v5/errors.rs

+use super::Status;
+
+#[derive(Debug)]
+pub enum SocksV5Error {
+    HostTooLong,
+    Auth(AuthError),
+    Command(Status),
+}
+
+#[derive(Debug)]
+pub enum AuthError {
+    Unsupported,
+    MethodMismatch,
+    Failed,
+}
+
+impl From<Status> for SocksV5Error {
+    fn from(err: Status) -> Self {
+        Self::Command(err)
+    }
+}
+
+impl From<AuthError> for SocksV5Error {
+    fn from(err: AuthError) -> Self {
+        Self::Auth(err)
+    }
+}
+
+impl std::fmt::Display for SocksV5Error {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::HostTooLong => f.write_str("host address is more than 255 characters"),
+            Self::Command(e) => e.fmt(f),
+            Self::Auth(e) => e.fmt(f),
+        }
+    }
+}
+
+impl std::fmt::Display for AuthError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Self::Unsupported => "server does not support user/pass authentication",
+            Self::MethodMismatch => "server implements authentication incorrectly",
+            Self::Failed => "credentials not accepted",
+        })
+    }
+}

src/client/legacy/connect/proxy/socks/v5/messages.rs

+use super::super::{ParsingError, SerializeError};
+
+use bytes::{Buf, BufMut, BytesMut};
+use std::net::SocketAddr;
+
+///  +----+----------+----------+
+/// |VER | NMETHODS | METHODS  |
+/// +----+----------+----------+
+/// | 1  |    1     | 1 to 255 |
+/// +----+----------+----------+
+#[derive(Debug)]
+pub struct NegotiationReq<'a>(pub &'a AuthMethod);
+
+/// +----+--------+
+/// |VER | METHOD |
+/// +----+--------+
+/// | 1  |   1    |
+/// +----+--------+
+#[derive(Debug)]
+pub struct NegotiationRes(pub AuthMethod);
+
+/// +----+------+----------+------+----------+
+/// |VER | ULEN |  UNAME   | PLEN |  PASSWD  |
+/// +----+------+----------+------+----------+
+/// | 1  |  1   | 1 to 255 |  1   | 1 to 255 |
+/// +----+------+----------+------+----------+
+#[derive(Debug)]
+pub struct AuthenticationReq<'a>(pub &'a str, pub &'a str);
+
+/// +----+--------+
+/// |VER | STATUS |
+/// +----+--------+
+/// | 1  |   1    |
+/// +----+--------+
+#[derive(Debug)]
+pub struct AuthenticationRes(pub bool);
+
+/// +----+-----+-------+------+----------+----------+
+/// |VER | CMD |  RSV  | ATYP | DST.ADDR | DST.PORT |
+/// +----+-----+-------+------+----------+----------+
+/// | 1  |  1  | X'00' |  1   | Variable |    2     |
+/// +----+-----+-------+------+----------+----------+
+#[derive(Debug)]
+pub struct ProxyReq<'a>(pub &'a Address);
+
+/// +----+-----+-------+------+----------+----------+
+/// |VER | REP |  RSV  | ATYP | BND.ADDR | BND.PORT |
+/// +----+-----+-------+------+----------+----------+
+/// | 1  |  1  | X'00' |  1   | Variable |    2     |
+/// +----+-----+-------+------+----------+----------+
+#[derive(Debug)]
+pub struct ProxyRes(pub Status);
+
+#[repr(u8)]
+#[derive(Debug, Copy, Clone, PartialEq)]
+pub enum AuthMethod {
+    NoAuth = 0x00,
+    UserPass = 0x02,
+    NoneAcceptable = 0xFF,
+}
+
+#[derive(Debug)]
+pub enum Address {
+    Socket(SocketAddr),
+    Domain(String, u16),
+}
+
+#[derive(Debug, Copy, Clone, PartialEq)]
+pub enum Status {
+    Success,
+    GeneralServerFailure,
+    ConnectionNotAllowed,
+    NetworkUnreachable,
+    HostUnreachable,
+    ConnectionRefused,
+    TtlExpired,
+    CommandNotSupported,
+    AddressTypeNotSupported,
+}
+
+impl NegotiationReq<'_> {
+    pub fn write_to_buf(&self, buf: &mut BytesMut) -> Result<usize, SerializeError> {
+        if buf.capacity() - buf.len() < 3 {
+            return Err(SerializeError::WouldOverflow);
+        }
+
+        buf.put_u8(0x05); // Version
+        buf.put_u8(0x01); // Number of authentication methods
+        buf.put_u8(*self.0 as u8); // Authentication method
+
+        Ok(3)
+    }
+}
+
+impl TryFrom<&mut BytesMut> for NegotiationRes {
+    type Error = ParsingError;
+
+    fn try_from(buf: &mut BytesMut) -> Result<Self, ParsingError> {
+        if buf.remaining() < 2 {
+            return Err(ParsingError::Incomplete);
+        }
+
+        if buf.get_u8() != 0x05 {
+            return Err(ParsingError::Other);
+        }
+
+        let method = buf.get_u8().try_into()?;
+        Ok(Self(method))
+    }
+}
+
+impl AuthenticationReq<'_> {
+    pub fn write_to_buf(&self, buf: &mut BytesMut) -> Result<usize, SerializeError> {
+        if buf.capacity() - buf.len() < 3 + self.0.len() + self.1.len() {
+            return Err(SerializeError::WouldOverflow);
+        }
+
+        buf.put_u8(0x01); // Version
+
+        buf.put_u8(self.0.len() as u8); // Username length (guarenteed to be 255 or less)
+        buf.put_slice(self.0.as_bytes()); // Username
+
+        buf.put_u8(self.1.len() as u8); // Password length (guarenteed to be 255 or less)
+        buf.put_slice(self.1.as_bytes()); // Password
+
+        Ok(3 + self.0.len() + self.1.len())
+    }
+}
+
+impl TryFrom<&mut BytesMut> for AuthenticationRes {
+    type Error = ParsingError;
+
+    fn try_from(buf: &mut BytesMut) -> Result<Self, ParsingError> {
+        if buf.remaining() < 2 {
+            return Err(ParsingError::Incomplete);
+        }
+
+        if buf.get_u8() != 0x01 {
+            return Err(ParsingError::Other);
+        }
+
+        if buf.get_u8() == 0 {
+            Ok(Self(true))
+        } else {
+            Ok(Self(false))
+        }
+    }
+}
+
+impl ProxyReq<'_> {
+    pub fn write_to_buf(&self, buf: &mut BytesMut) -> Result<usize, SerializeError> {
+        let addr_len = match self.0 {
+            Address::Socket(SocketAddr::V4(_)) => 1 + 4 + 2,
+            Address::Socket(SocketAddr::V6(_)) => 1 + 16 + 2,
+            Address::Domain(ref domain, _) => 1 + 1 + domain.len() + 2,
+        };
+
+        if buf.capacity() - buf.len() < 3 + addr_len {
+            return Err(SerializeError::WouldOverflow);
+        }
+
+        buf.put_u8(0x05); // Version
+        buf.put_u8(0x01); // TCP tunneling command
+        buf.put_u8(0x00); // Reserved
+        let _ = self.0.write_to_buf(buf); // Address
+
+        Ok(3 + addr_len)
+    }
+}
+
+impl TryFrom<&mut BytesMut> for ProxyRes {
+    type Error = ParsingError;
+
+    fn try_from(buf: &mut BytesMut) -> Result<Self, ParsingError> {
+        if buf.remaining() < 2 {
+            return Err(ParsingError::Incomplete);
+        }
+
+        // VER
+        if buf.get_u8() != 0x05 {
+            return Err(ParsingError::Other);
+        }
+
+        // REP
+        let status = buf.get_u8().try_into()?;
+
+        // RSV
+        if buf.get_u8() != 0x00 {
+            return Err(ParsingError::Other);
+        }
+
+        // ATYP + ADDR
+        Address::try_from(buf)?;
+
+        Ok(Self(status))
+    }
+}
+
+impl Address {
+    pub fn write_to_buf(&self, buf: &mut BytesMut) -> Result<usize, SerializeError> {
+        match self {
+            Self::Socket(SocketAddr::V4(v4)) => {
+                if buf.capacity() - buf.len() < 1 + 4 + 2 {
+                    return Err(SerializeError::WouldOverflow);
+                }
+
+                buf.put_u8(0x01);
+                buf.put_slice(&v4.ip().octets());
+                buf.put_u16(v4.port()); // Network Order/BigEndian for port
+
+                Ok(7)
+            }
+
+            Self::Socket(SocketAddr::V6(v6)) => {
+                if buf.capacity() - buf.len() < 1 + 16 + 2 {
+                    return Err(SerializeError::WouldOverflow);
+                }
+
+                buf.put_u8(0x04);
+                buf.put_slice(&v6.ip().octets());
+                buf.put_u16(v6.port()); // Network Order/BigEndian for port
+
+                Ok(19)
+            }
+
+            Self::Domain(domain, port) => {
+                if buf.capacity() - buf.len() < 1 + 1 + domain.len() + 2 {
+                    return Err(SerializeError::WouldOverflow);
+                }
+
+                buf.put_u8(0x03);
+                buf.put_u8(domain.len() as u8); // Guarenteed to be less than 255
+                buf.put_slice(domain.as_bytes());
+                buf.put_u16(*port);
+
+                Ok(4 + domain.len())
+            }
+        }
+    }
+}
+
+impl TryFrom<&mut BytesMut> for Address {
+    type Error = ParsingError;
+
+    fn try_from(buf: &mut BytesMut) -> Result<Self, Self::Error> {
+        if buf.remaining() < 2 {
+            return Err(ParsingError::Incomplete);
+        }
+
+        Ok(match buf.get_u8() {
+            0x01 => {
+                let mut ip = [0; 4];
+
+                if buf.remaining() < 6 {
+                    return Err(ParsingError::Incomplete);
+                }
+
+                buf.copy_to_slice(&mut ip);
+                let port = buf.get_u16();
+
+                Self::Socket(SocketAddr::new(ip.into(), port))
+            }
+
+            0x03 => {
+                let len = buf.get_u8();
+
+                if len == 0 {
+                    return Err(ParsingError::Other);
+                } else if buf.remaining() < (len as usize) + 2 {
+                    return Err(ParsingError::Incomplete);
+                }
+
+                let domain = std::str::from_utf8(&buf[..len as usize])
+                    .map_err(|_| ParsingError::Other)?
+                    .to_string();
+
+                let port = buf.get_u16();
+
+                Self::Domain(domain, port)
+            }
+
+            0x04 => {
+                let mut ip = [0; 16];
+
+                if buf.remaining() < 6 {
+                    return Err(ParsingError::Incomplete);
+                }
+                buf.copy_to_slice(&mut ip);
+                let port = buf.get_u16();
+
+                Self::Socket(SocketAddr::new(ip.into(), port))
+            }
+
+            _ => return Err(ParsingError::Other),
+        })
+    }
+}
+
+impl TryFrom<u8> for Status {
+    type Error = ParsingError;
+
+    fn try_from(byte: u8) -> Result<Self, Self::Error> {
+        Ok(match byte {
+            0x00 => Self::Success,
+
+            0x01 => Self::GeneralServerFailure,
+            0x02 => Self::ConnectionNotAllowed,
+            0x03 => Self::NetworkUnreachable,
+            0x04 => Self::HostUnreachable,
+            0x05 => Self::ConnectionRefused,
+            0x06 => Self::TtlExpired,
+            0x07 => Self::CommandNotSupported,
+            0x08 => Self::AddressTypeNotSupported,
+            _ => return Err(ParsingError::Other),
+        })
+    }
+}
+
+impl TryFrom<u8> for AuthMethod {
+    type Error = ParsingError;
+
+    fn try_from(byte: u8) -> Result<Self, Self::Error> {
+        Ok(match byte {
+            0x00 => Self::NoAuth,
+            0x02 => Self::UserPass,
+            0xFF => Self::NoneAcceptable,
+
+            _ => return Err(ParsingError::Other),
+        })
+    }
+}
+
+impl std::fmt::Display for Status {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Self::Success => "success",
+            Self::GeneralServerFailure => "general server failure",
+            Self::ConnectionNotAllowed => "connection not allowed",
+            Self::NetworkUnreachable => "network unreachable",
+            Self::HostUnreachable => "host unreachable",
+            Self::ConnectionRefused => "connection refused",
+            Self::TtlExpired => "ttl expired",
+            Self::CommandNotSupported => "command not supported",
+            Self::AddressTypeNotSupported => "address type not supported",
+        })
+    }
+}

src/client/legacy/connect/proxy/socks/v5/mod.rs

+mod errors;
+pub use errors::*;
+
+mod messages;
+use messages::*;
+
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
+
+use http::Uri;
+use hyper::rt::{Read, Write};
+use tower_service::Service;
+
+use bytes::BytesMut;
+
+use pin_project_lite::pin_project;
+
+/// Tunnel Proxy via SOCKSv5
+///
+/// This is a connector that can be used by the `legacy::Client`. It wraps
+/// another connector, and after getting an underlying connection, it established
+/// a TCP tunnel over it using SOCKSv5.
+#[derive(Debug, Clone)]
+pub struct SocksV5<C> {
+    inner: C,
+    config: SocksConfig,
+}
+
+#[derive(Debug, Clone)]
+pub struct SocksConfig {
+    proxy: Uri,
+    proxy_auth: Option<(String, String)>,
+
+    local_dns: bool,
+    optimistic: bool,
+}
+
+#[derive(Debug)]
+enum State {
+    SendingNegReq,
+    ReadingNegRes,
+    SendingAuthReq,
+    ReadingAuthRes,
+    SendingProxyReq,
+    ReadingProxyRes,
+}
+
+pin_project! {
+    // Not publicly exported (so missing_docs doesn't trigger).
+    //
+    // We return this `Future` instead of the `Pin<Box<dyn Future>>` directly
+    // so that users don't rely on it fitting in a `Pin<Box<dyn Future>>` slot
+    // (and thus we can change the type in the future).
+    #[must_use = "futures do nothing unless polled"]
+    #[allow(missing_debug_implementations)]
+    pub struct Handshaking<F, T, E> {
+        #[pin]
+        fut: BoxHandshaking<T, E>,
+        _marker: std::marker::PhantomData<F>
+    }
+}
+
+type BoxHandshaking<T, E> = Pin<Box<dyn Future<Output = Result<T, super::SocksError<E>>> + Send>>;
+
+impl<C> SocksV5<C> {
+    /// Create a new SOCKSv5 handshake service.
+    ///
+    /// Wraps an underlying connector and stores the address of a tunneling
+    /// proxying server.
+    ///
+    /// A `SocksV5` can then be called with any destination. The `dst` passed to
+    /// `call` will not be used to create the underlying connection, but will
+    /// be used in a SOCKS handshake with the proxy destination.
+    pub fn new(proxy_dst: Uri, connector: C) -> Self {
+        Self {
+            inner: connector,
+            config: SocksConfig::new(proxy_dst),
+        }
+    }
+
+    /// Use User/Pass authentication method during handshake.
+    ///
+    /// Username and Password must be maximum of 255 characters each.
+    /// 0 length strings are allowed despite RFC prohibiting it. This is done so that
+    /// for compatablity with server implementations that require it for IP authentication.
+    pub fn with_auth(mut self, user: String, pass: String) -> Self {
+        self.config.proxy_auth = Some((user, pass));
+        self
+    }
+
+    /// Resolve domain names locally on the client, rather than on the proxy server.
+    ///
+    /// Disabled by default as local resolution of domain names can be detected as a
+    /// DNS leak.
+    pub fn local_dns(mut self, local_dns: bool) -> Self {
+        self.config.local_dns = local_dns;
+        self
+    }
+
+    /// Send all messages of the handshake optmistically (without waiting for server response).
+    ///
+    /// Typical SOCKS handshake with auithentication takes 3 round trips. Optimistic sending
+    /// can reduce round trip times and dramatically increase speed of handshake at the cost of
+    /// reduced portability; many server implementations do not support optimistic sending as it
+    /// is not defined in the RFC (RFC 1928).
+    ///
+    /// Recommended to ensure connector works correctly without optimistic sending before trying
+    /// with optimistic sending.
+    pub fn send_optimistically(mut self, optimistic: bool) -> Self {
+        self.config.optimistic = optimistic;
+        self
+    }
+}
+
+impl SocksConfig {
+    fn new(proxy: Uri) -> Self {
+        Self {
+            proxy,
+            proxy_auth: None,
+
+            local_dns: false,
+            optimistic: false,
+        }
+    }
+
+    async fn execute<T, E>(
+        self,
+        mut conn: T,
+        host: String,
+        port: u16,
+    ) -> Result<T, super::SocksError<E>>
+    where
+        T: Read + Write + Unpin,
+    {
+        let address = match host.parse::<IpAddr>() {
+            Ok(ip) => Address::Socket(SocketAddr::new(ip, port)),
+            Err(_) if host.len() <= 255 => {
+                if self.local_dns {
+                    let socket = (host, port)
+                        .to_socket_addrs()?
+                        .next()
+                        .ok_or(super::SocksError::DnsFailure)?;
+
+                    Address::Socket(socket)
+                } else {
+                    Address::Domain(host, port)
+                }
+            }
+            Err(_) => return Err(SocksV5Error::HostTooLong.into()),
+        };
+
+        let method = if self.proxy_auth.is_some() {
+            AuthMethod::UserPass
+        } else {
+            AuthMethod::NoAuth
+        };
+
+        let mut recv_buf = BytesMut::with_capacity(513); // Max length of valid recievable message is 513 from Auth Request
+        let mut send_buf = BytesMut::with_capacity(262); // Max length of valid sendable message is 262 from Auth Response
+        let mut state = State::SendingNegReq;
+
+        loop {
+            match state {
+                State::SendingNegReq => {
+                    let req = NegotiationReq(&method);
+
+                    let start = send_buf.len();
+                    req.write_to_buf(&mut send_buf)?;
+                    crate::rt::write_all(&mut conn, &send_buf[start..]).await?;
+
+                    if self.optimistic {
+                        if method == AuthMethod::UserPass {
+                            state = State::SendingAuthReq;
+                        } else {
+                            state = State::SendingProxyReq;
+                        }
+                    } else {
+                        state = State::ReadingNegRes;
+                    }
+                }
+
+                State::ReadingNegRes => {
+                    let res: NegotiationRes = super::read_message(&mut conn, &mut recv_buf).await?;
+
+                    if res.0 == AuthMethod::NoneAcceptable {
+                        return Err(SocksV5Error::Auth(AuthError::Unsupported).into());
+                    }
+
+                    if res.0 != method {
+                        return Err(SocksV5Error::Auth(AuthError::MethodMismatch).into());
+                    }
+
+                    if self.optimistic {
+                        if res.0 == AuthMethod::UserPass {
+                            state = State::ReadingAuthRes;
+                        } else {
+                            state = State::ReadingProxyRes;
+                        }
+                    } else {
+                        if res.0 == AuthMethod::UserPass {
+                            state = State::SendingAuthReq;
+                        } else {
+                            state = State::SendingProxyReq;
+                        }
+                    }
+                }
+
+                State::SendingAuthReq => {
+                    let (user, pass) = self.proxy_auth.as_ref().unwrap();
+                    let req = AuthenticationReq(&user, &pass);
+
+                    let start = send_buf.len();
+                    req.write_to_buf(&mut send_buf)?;
+                    crate::rt::write_all(&mut conn, &send_buf[start..]).await?;
+
+                    if self.optimistic {
+                        state = State::SendingProxyReq;
+                    } else {
+                        state = State::ReadingAuthRes;
+                    }
+                }
+
+                State::ReadingAuthRes => {
+                    let res: AuthenticationRes =
+                        super::read_message(&mut conn, &mut recv_buf).await?;
+
+                    if !res.0 {
+                        return Err(SocksV5Error::Auth(AuthError::Failed).into());
+                    }
+
+                    if self.optimistic {
+                        state = State::ReadingProxyRes;
+                    } else {
+                        state = State::SendingProxyReq;
+                    }
+                }
+
+                State::SendingProxyReq => {
+                    let req = ProxyReq(&address);
+
+                    let start = send_buf.len();
+                    req.write_to_buf(&mut send_buf)?;
+                    crate::rt::write_all(&mut conn, &send_buf[start..]).await?;
+
+                    if self.optimistic {
+                        state = State::ReadingNegRes;
+                    } else {
+                        state = State::ReadingProxyRes;
+                    }
+                }
+
+                State::ReadingProxyRes => {
+                    let res: ProxyRes = super::read_message(&mut conn, &mut recv_buf).await?;
+
+                    if res.0 == Status::Success {
+                        return Ok(conn);
+                    } else {
+                        return Err(SocksV5Error::Command(res.0).into());
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl<C> Service<Uri> for SocksV5<C>
+where
+    C: Service<Uri>,
+    C::Future: Send + 'static,
+    C::Response: Read + Write + Unpin + Send + 'static,
+    C::Error: Send + 'static,
+{
+    type Response = C::Response;
+    type Error = super::SocksError<C::Error>;
+    type Future = Handshaking<C::Future, C::Response, C::Error>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx).map_err(super::SocksError::Inner)
+    }
+
+    fn call(&mut self, dst: Uri) -> Self::Future {
+        let config = self.config.clone();
+        let connecting = self.inner.call(config.proxy.clone());
+
+        let fut = async move {
+            let port = dst.port().map(|p| p.as_u16()).unwrap_or(443);
+            let host = dst
+                .host()
+                .ok_or(super::SocksError::MissingHost)?
+                .to_string();
+
+            let conn = connecting.await.map_err(super::SocksError::Inner)?;
+            config.execute(conn, host, port).await
+        };
+
+        Handshaking {
+            fut: Box::pin(fut),
+            _marker: Default::default(),
+        }
+    }
+}
+
+impl<F, T, E> Future for Handshaking<F, T, E>
+where
+    F: Future<Output = Result<T, E>>,
+{
+    type Output = Result<T, super::SocksError<E>>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().fut.poll(cx)
+    }
+}

src/client/legacy/connect/proxy/tunnel.rs

+use std::error::Error as StdError;
+use std::future::Future;
+use std::marker::{PhantomData, Unpin};
+use std::pin::Pin;
+use std::task::{self, Poll};
+
+use http::{HeaderMap, HeaderValue, Uri};
+use hyper::rt::{Read, Write};
+use pin_project_lite::pin_project;
+use tower_service::Service;
+
+/// Tunnel Proxy via HTTP CONNECT
+///
+/// This is a connector that can be used by the `legacy::Client`. It wraps
+/// another connector, and after getting an underlying connection, it creates
+/// an HTTP CONNECT tunnel over it.
+#[derive(Debug)]
+pub struct Tunnel<C> {
+    headers: Headers,
+    inner: C,
+    proxy_dst: Uri,
+}
+
+#[derive(Clone, Debug)]
+enum Headers {
+    Empty,
+    Auth(HeaderValue),
+    Extra(HeaderMap),
+}
+
+#[derive(Debug)]
+pub enum TunnelError {
+    ConnectFailed(Box<dyn StdError + Send + Sync>),
+    Io(std::io::Error),
+    MissingHost,
+    ProxyAuthRequired,
+    ProxyHeadersTooLong,
+    TunnelUnexpectedEof,
+    TunnelUnsuccessful,
+}
+
+pin_project! {
+    // Not publicly exported (so missing_docs doesn't trigger).
+    //
+    // We return this `Future` instead of the `Pin<Box<dyn Future>>` directly
+    // so that users don't rely on it fitting in a `Pin<Box<dyn Future>>` slot
+    // (and thus we can change the type in the future).
+    #[must_use = "futures do nothing unless polled"]
+    #[allow(missing_debug_implementations)]
+    pub struct Tunneling<F, T> {
+        #[pin]
+        fut: BoxTunneling<T>,
+        _marker: PhantomData<F>,
+    }
+}
+
+type BoxTunneling<T> = Pin<Box<dyn Future<Output = Result<T, TunnelError>> + Send>>;
+
+impl<C> Tunnel<C> {
+    /// Create a new Tunnel service.
+    ///
+    /// This wraps an underlying connector, and stores the address of a
+    /// tunneling proxy server.
+    ///
+    /// A `Tunnel` can then be called with any destination. The `dst` passed to
+    /// `call` will not be used to create the underlying connection, but will
+    /// be used in an HTTP CONNECT request sent to the proxy destination.
+    pub fn new(proxy_dst: Uri, connector: C) -> Self {
+        Self {
+            headers: Headers::Empty,
+            inner: connector,
+            proxy_dst,
+        }
+    }
+
+    /// Add `proxy-authorization` header value to the CONNECT request.
+    pub fn with_auth(mut self, mut auth: HeaderValue) -> Self {
+        // just in case the user forgot
+        auth.set_sensitive(true);
+        match self.headers {
+            Headers::Empty => {
+                self.headers = Headers::Auth(auth);
+            }
+            Headers::Auth(ref mut existing) => {
+                *existing = auth;
+            }
+            Headers::Extra(ref mut extra) => {
+                extra.insert(http::header::PROXY_AUTHORIZATION, auth);
+            }
+        }
+
+        self
+    }
+
+    /// Add extra headers to be sent with the CONNECT request.
+    ///
+    /// If existing headers have been set, these will be merged.
+    pub fn with_headers(mut self, mut headers: HeaderMap) -> Self {
+        match self.headers {
+            Headers::Empty => {
+                self.headers = Headers::Extra(headers);
+            }
+            Headers::Auth(auth) => {
+                headers
+                    .entry(http::header::PROXY_AUTHORIZATION)
+                    .or_insert(auth);
+                self.headers = Headers::Extra(headers);
+            }
+            Headers::Extra(ref mut extra) => {
+                extra.extend(headers);
+            }
+        }
+
+        self
+    }
+}
+
+impl<C> Service<Uri> for Tunnel<C>
+where
+    C: Service<Uri>,
+    C::Future: Send + 'static,
+    C::Response: Read + Write + Unpin + Send + 'static,
+    C::Error: Into<Box<dyn StdError + Send + Sync>>,
+{
+    type Response = C::Response;
+    type Error = TunnelError;
+    type Future = Tunneling<C::Future, C::Response>;
+
+    fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
+        futures_util::ready!(self.inner.poll_ready(cx))
+            .map_err(|e| TunnelError::ConnectFailed(e.into()))?;
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, dst: Uri) -> Self::Future {
+        let connecting = self.inner.call(self.proxy_dst.clone());
+        let headers = self.headers.clone();
+
+        Tunneling {
+            fut: Box::pin(async move {
+                let conn = connecting
+                    .await
+                    .map_err(|e| TunnelError::ConnectFailed(e.into()))?;
+                tunnel(
+                    conn,
+                    dst.host().ok_or(TunnelError::MissingHost)?,
+                    dst.port().map(|p| p.as_u16()).unwrap_or(443),
+                    &headers,
+                )
+                .await
+            }),
+            _marker: PhantomData,
+        }
+    }
+}
+
+impl<F, T, E> Future for Tunneling<F, T>
+where
+    F: Future<Output = Result<T, E>>,
+{
+    type Output = Result<T, TunnelError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
+        self.project().fut.poll(cx)
+    }
+}
+
+async fn tunnel<T>(mut conn: T, host: &str, port: u16, headers: &Headers) -> Result<T, TunnelError>
+where
+    T: Read + Write + Unpin,
+{
+    let mut buf = format!(
+        "\
+         CONNECT {host}:{port} HTTP/1.1\r\n\
+         Host: {host}:{port}\r\n\
+         "
+    )
+    .into_bytes();
+
+    match headers {
+        Headers::Auth(auth) => {
+            buf.extend_from_slice(b"Proxy-Authorization: ");
+            buf.extend_from_slice(auth.as_bytes());
+            buf.extend_from_slice(b"\r\n");
+        }
+        Headers::Extra(extra) => {
+            for (name, value) in extra {
+                buf.extend_from_slice(name.as_str().as_bytes());
+                buf.extend_from_slice(b": ");
+                buf.extend_from_slice(value.as_bytes());
+                buf.extend_from_slice(b"\r\n");
+            }
+        }
+        Headers::Empty => (),
+    }
+
+    // headers end
+    buf.extend_from_slice(b"\r\n");
+
+    crate::rt::write_all(&mut conn, &buf)
+        .await
+        .map_err(TunnelError::Io)?;
+
+    let mut buf = [0; 8192];
+    let mut pos = 0;
+
+    loop {
+        let n = crate::rt::read(&mut conn, &mut buf[pos..])
+            .await
+            .map_err(TunnelError::Io)?;
+
+        if n == 0 {
+            return Err(TunnelError::TunnelUnexpectedEof);
+        }
+        pos += n;
+
+        let recvd = &buf[..pos];
+        if recvd.starts_with(b"HTTP/1.1 200") || recvd.starts_with(b"HTTP/1.0 200") {
+            if recvd.ends_with(b"\r\n\r\n") {
+                return Ok(conn);
+            }
+            if pos == buf.len() {
+                return Err(TunnelError::ProxyHeadersTooLong);
+            }
+        // else read more
+        } else if recvd.starts_with(b"HTTP/1.1 407") {
+            return Err(TunnelError::ProxyAuthRequired);
+        } else {
+            return Err(TunnelError::TunnelUnsuccessful);
+        }
+    }
+}
+
+impl std::fmt::Display for TunnelError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("tunnel error: ")?;
+
+        f.write_str(match self {
+            TunnelError::MissingHost => "missing destination host",
+            TunnelError::ProxyAuthRequired => "proxy authorization required",
+            TunnelError::ProxyHeadersTooLong => "proxy response headers too long",
+            TunnelError::TunnelUnexpectedEof => "unexpected end of file",
+            TunnelError::TunnelUnsuccessful => "unsuccessful",
+            TunnelError::ConnectFailed(_) => "failed to create underlying connection",
+            TunnelError::Io(_) => "io error establishing tunnel",
+        })
+    }
+}
+
+impl std::error::Error for TunnelError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            TunnelError::Io(ref e) => Some(e),
+            TunnelError::ConnectFailed(ref e) => Some(&**e),
+            _ => None,
+        }
+    }
+}

src/client/proxy/matcher.rs

+//! Proxy matchers
+//!
+//! This module contains different matchers to configure rules for when a proxy
+//! should be used, and if so, with what arguments.
+//!
+//! A [`Matcher`] can be constructed either using environment variables, or
+//! a [`Matcher::builder()`].
+//!
+//! Once constructed, the `Matcher` can be asked if it intercepts a `Uri` by
+//! calling [`Matcher::intercept()`].
+//!
+//! An [`Intercept`] includes the destination for the proxy, and any parsed
+//! authentication to be used.
+
+use std::fmt;
+use std::net::IpAddr;
+
+use http::header::HeaderValue;
+use ipnet::IpNet;
+use percent_encoding::percent_decode_str;
+
+#[cfg(docsrs)]
+pub use self::builder::IntoValue;
+#[cfg(not(docsrs))]
+use self::builder::IntoValue;
+
+/// A proxy matcher, usually built from environment variables.
+pub struct Matcher {
+    http: Option<Intercept>,
+    https: Option<Intercept>,
+    no: NoProxy,
+}
+
+/// A matched proxy,
+///
+/// This is returned by a matcher if a proxy should be used.
+#[derive(Clone)]
+pub struct Intercept {
+    uri: http::Uri,
+    auth: Auth,
+}
+
+/// A builder to create a [`Matcher`].
+///
+/// Construct with [`Matcher::builder()`].
+#[derive(Default)]
+pub struct Builder {
+    is_cgi: bool,
+    all: String,
+    http: String,
+    https: String,
+    no: String,
+}
+
+#[derive(Clone)]
+enum Auth {
+    Empty,
+    Basic(http::header::HeaderValue),
+    Raw(String, String),
+}
+
+/// A filter for proxy matchers.
+///
+/// This type is based off the `NO_PROXY` rules used by curl.
+#[derive(Clone, Debug, Default)]
+struct NoProxy {
+    ips: IpMatcher,
+    domains: DomainMatcher,
+}
+
+#[derive(Clone, Debug, Default)]
+struct DomainMatcher(Vec<String>);
+
+#[derive(Clone, Debug, Default)]
+struct IpMatcher(Vec<Ip>);
+
+#[derive(Clone, Debug)]
+enum Ip {
+    Address(IpAddr),
+    Network(IpNet),
+}
+
+// ===== impl Matcher =====
+
+impl Matcher {
+    /// Create a matcher reading the current environment variables.
+    ///
+    /// This checks for values in the following variables, treating them the
+    /// same as curl does:
+    ///
+    /// - `ALL_PROXY`/`all_proxy`
+    /// - `HTTPS_PROXY`/`https_proxy`
+    /// - `HTTP_PROXY`/`http_proxy`
+    /// - `NO_PROXY`/`no_proxy`
+    pub fn from_env() -> Self {
+        Builder::from_env().build()
+    }
+
+    /// Create a matcher from the environment or system.
+    ///
+    /// This checks the same environment variables as `from_env()`, and if not
+    /// set, checks the system configuration for values for the OS.
+    ///
+    /// This constructor is always available, but if the `client-proxy-system`
+    /// feature is enabled, it will check more configuration. Use this
+    /// constructor if you want to allow users to optionally enable more, or
+    /// use `from_env` if you do not want the values to change based on an
+    /// enabled feature.
+    pub fn from_system() -> Self {
+        Builder::from_system().build()
+    }
+
+    /// Start a builder to configure a matcher.
+    pub fn builder() -> Builder {
+        Builder::default()
+    }
+
+    /// Check if the destination should be intercepted by a proxy.
+    ///
+    /// If the proxy rules match the destination, a new `Uri` will be returned
+    /// to connect to.
+    pub fn intercept(&self, dst: &http::Uri) -> Option<Intercept> {
+        // TODO(perf): don't need to check `no` if below doesn't match...
+        if self.no.contains(dst.host()?) {
+            return None;
+        }
+
+        match dst.scheme_str() {
+            Some("http") => self.http.clone(),
+            Some("https") => self.https.clone(),
+            _ => None,
+        }
+    }
+}
+
+impl fmt::Debug for Matcher {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut b = f.debug_struct("Matcher");
+
+        if let Some(ref http) = self.http {
+            b.field("http", http);
+        }
+
+        if let Some(ref https) = self.https {
+            b.field("https", https);
+        }
+
+        if !self.no.is_empty() {
+            b.field("no", &self.no);
+        }
+        b.finish()
+    }
+}
+
+// ===== impl Intercept =====
+
+impl Intercept {
+    /// Get the `http::Uri` for the target proxy.
+    pub fn uri(&self) -> &http::Uri {
+        &self.uri
+    }
+
+    /// Get any configured basic authorization.
+    ///
+    /// This should usually be used with a `Proxy-Authorization` header, to
+    /// send in Basic format.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// # use hyper_util::client::proxy::matcher::Matcher;
+    /// # let uri = http::Uri::from_static("https://hyper.rs");
+    /// let m = Matcher::builder()
+    ///     .all("https://Aladdin:opensesame@localhost:8887")
+    ///     .build();
+    ///
+    /// let proxy = m.intercept(&uri).expect("example");
+    /// let auth = proxy.basic_auth().expect("example");
+    /// assert_eq!(auth, "Basic QWxhZGRpbjpvcGVuc2VzYW1l");
+    /// ```
+    pub fn basic_auth(&self) -> Option<&HeaderValue> {
+        if let Auth::Basic(ref val) = self.auth {
+            Some(val)
+        } else {
+            None
+        }
+    }
+
+    /// Get any configured raw authorization.
+    ///
+    /// If not detected as another scheme, this is the username and password
+    /// that should be sent with whatever protocol the proxy handshake uses.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// # use hyper_util::client::proxy::matcher::Matcher;
+    /// # let uri = http::Uri::from_static("https://hyper.rs");
+    /// let m = Matcher::builder()
+    ///     .all("socks5h://Aladdin:opensesame@localhost:8887")
+    ///     .build();
+    ///
+    /// let proxy = m.intercept(&uri).expect("example");
+    /// let auth = proxy.raw_auth().expect("example");
+    /// assert_eq!(auth, ("Aladdin", "opensesame"));
+    /// ```
+    pub fn raw_auth(&self) -> Option<(&str, &str)> {
+        if let Auth::Raw(ref u, ref p) = self.auth {
+            Some((u.as_str(), p.as_str()))
+        } else {
+            None
+        }
+    }
+}
+
+impl fmt::Debug for Intercept {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Intercept")
+            .field("uri", &self.uri)
+            // dont output auth, its sensitive
+            .finish()
+    }
+}
+
+// ===== impl Builder =====
+
+impl Builder {
+    fn from_env() -> Self {
+        Builder {
+            is_cgi: std::env::var_os("REQUEST_METHOD").is_some(),
+            all: get_first_env(&["ALL_PROXY", "all_proxy"]),
+            http: get_first_env(&["HTTP_PROXY", "http_proxy"]),
+            https: get_first_env(&["HTTPS_PROXY", "https_proxy"]),
+            no: get_first_env(&["NO_PROXY", "no_proxy"]),
+        }
+    }
+
+    fn from_system() -> Self {
+        #[allow(unused_mut)]
+        let mut builder = Self::from_env();
+
+        #[cfg(all(feature = "client-proxy-system", target_os = "macos"))]
+        mac::with_system(&mut builder);
+
+        #[cfg(all(feature = "client-proxy-system", windows))]
+        win::with_system(&mut builder);
+
+        builder
+    }
+
+    /// Set the target proxy for all destinations.
+    pub fn all<S>(mut self, val: S) -> Self
+    where
+        S: IntoValue,
+    {
+        self.all = val.into_value();
+        self
+    }
+
+    /// Set the target proxy for HTTP destinations.
+    pub fn http<S>(mut self, val: S) -> Self
+    where
+        S: IntoValue,
+    {
+        self.http = val.into_value();
+        self
+    }
+
+    /// Set the target proxy for HTTPS destinations.
+    pub fn https<S>(mut self, val: S) -> Self
+    where
+        S: IntoValue,
+    {
+        self.https = val.into_value();
+        self
+    }
+
+    /// Set the "no" proxy filter.
+    ///
+    /// The rules are as follows:
+    /// * Entries are expected to be comma-separated (whitespace between entries is ignored)
+    /// * IP addresses (both IPv4 and IPv6) are allowed, as are optional subnet masks (by adding /size,
+    /// for example "`192.168.1.0/24`").
+    /// * An entry "`*`" matches all hostnames (this is the only wildcard allowed)
+    /// * Any other entry is considered a domain name (and may contain a leading dot, for example `google.com`
+    /// and `.google.com` are equivalent) and would match both that domain AND all subdomains.
+    ///
+    /// For example, if `"NO_PROXY=google.com, 192.168.1.0/24"` was set, all of the following would match
+    /// (and therefore would bypass the proxy):
+    /// * `http://google.com/`
+    /// * `http://www.google.com/`
+    /// * `http://192.168.1.42/`
+    ///
+    /// The URL `http://notgoogle.com/` would not match.
+    pub fn no<S>(mut self, val: S) -> Self
+    where
+        S: IntoValue,
+    {
+        self.no = val.into_value();
+        self
+    }
+
+    /// Construct a [`Matcher`] using the configured values.
+    pub fn build(self) -> Matcher {
+        if self.is_cgi {
+            return Matcher {
+                http: None,
+                https: None,
+                no: NoProxy::empty(),
+            };
+        }
+
+        let all = parse_env_uri(&self.all);
+
+        Matcher {
+            http: parse_env_uri(&self.http).or_else(|| all.clone()),
+            https: parse_env_uri(&self.https).or(all),
+            no: NoProxy::from_string(&self.no),
+        }
+    }
+}
+
+fn get_first_env(names: &[&str]) -> String {
+    for name in names {
+        if let Ok(val) = std::env::var(name) {
+            return val;
+        }
+    }
+
+    String::new()
+}
+
+fn parse_env_uri(val: &str) -> Option<Intercept> {
+    let uri = val.parse::<http::Uri>().ok()?;
+    let mut builder = http::Uri::builder();
+    let mut is_httpish = false;
+    let mut auth = Auth::Empty;
+
+    builder = builder.scheme(match uri.scheme() {
+        Some(s) => {
+            if s == &http::uri::Scheme::HTTP || s == &http::uri::Scheme::HTTPS {
+                is_httpish = true;
+                s.clone()
+            } else if s.as_str() == "socks5" || s.as_str() == "socks5h" {
+                s.clone()
+            } else {
+                // can't use this proxy scheme
+                return None;
+            }
+        }
+        // if no scheme provided, assume they meant 'http'
+        None => {
+            is_httpish = true;
+            http::uri::Scheme::HTTP
+        }
+    });
+
+    let authority = uri.authority()?;
+
+    if let Some((userinfo, host_port)) = authority.as_str().split_once('@') {
+        let (user, pass) = userinfo.split_once(':')?;
+        let user = percent_decode_str(user).decode_utf8_lossy();
+        let pass = percent_decode_str(pass).decode_utf8_lossy();
+        if is_httpish {
+            auth = Auth::Basic(encode_basic_auth(&user, Some(&pass)));
+        } else {
+            auth = Auth::Raw(user.into(), pass.into());
+        }
+        builder = builder.authority(host_port);
+    } else {
+        builder = builder.authority(authority.clone());
+    }
+
+    // removing any path, but we MUST specify one or the builder errors
+    builder = builder.path_and_query("/");
+
+    let dst = builder.build().ok()?;
+
+    Some(Intercept { uri: dst, auth })
+}
+
+fn encode_basic_auth(user: &str, pass: Option<&str>) -> HeaderValue {
+    use base64::prelude::BASE64_STANDARD;
+    use base64::write::EncoderWriter;
+    use std::io::Write;
+
+    let mut buf = b"Basic ".to_vec();
+    {
+        let mut encoder = EncoderWriter::new(&mut buf, &BASE64_STANDARD);
+        let _ = write!(encoder, "{user}:");
+        if let Some(password) = pass {
+            let _ = write!(encoder, "{password}");
+        }
+    }
+    let mut header = HeaderValue::from_bytes(&buf).expect("base64 is always valid HeaderValue");
+    header.set_sensitive(true);
+    header
+}
+
+impl NoProxy {
+    /*
+    fn from_env() -> NoProxy {
+        let raw = std::env::var("NO_PROXY")
+            .or_else(|_| std::env::var("no_proxy"))
+            .unwrap_or_default();
+
+        Self::from_string(&raw)
+    }
+    */
+
+    fn empty() -> NoProxy {
+        NoProxy {
+            ips: IpMatcher(Vec::new()),
+            domains: DomainMatcher(Vec::new()),
+        }
+    }
+
+    /// Returns a new no-proxy configuration based on a `no_proxy` string (or `None` if no variables
+    /// are set)
+    /// The rules are as follows:
+    /// * The environment variable `NO_PROXY` is checked, if it is not set, `no_proxy` is checked
+    /// * If neither environment variable is set, `None` is returned
+    /// * Entries are expected to be comma-separated (whitespace between entries is ignored)
+    /// * IP addresses (both IPv4 and IPv6) are allowed, as are optional subnet masks (by adding /size,
+    /// for example "`192.168.1.0/24`").
+    /// * An entry "`*`" matches all hostnames (this is the only wildcard allowed)
+    /// * Any other entry is considered a domain name (and may contain a leading dot, for example `google.com`
+    /// and `.google.com` are equivalent) and would match both that domain AND all subdomains.
+    ///
+    /// For example, if `"NO_PROXY=google.com, 192.168.1.0/24"` was set, all of the following would match
+    /// (and therefore would bypass the proxy):
+    /// * `http://google.com/`
+    /// * `http://www.google.com/`
+    /// * `http://192.168.1.42/`
+    ///
+    /// The URL `http://notgoogle.com/` would not match.
+    pub fn from_string(no_proxy_list: &str) -> Self {
+        let mut ips = Vec::new();
+        let mut domains = Vec::new();
+        let parts = no_proxy_list.split(',').map(str::trim);
+        for part in parts {
+            match part.parse::<IpNet>() {
+                // If we can parse an IP net or address, then use it, otherwise, assume it is a domain
+                Ok(ip) => ips.push(Ip::Network(ip)),
+                Err(_) => match part.parse::<IpAddr>() {
+                    Ok(addr) => ips.push(Ip::Address(addr)),
+                    Err(_) => {
+                        if !part.trim().is_empty() {
+                            domains.push(part.to_owned())
+                        }
+                    }
+                },
+            }
+        }
+        NoProxy {
+            ips: IpMatcher(ips),
+            domains: DomainMatcher(domains),
+        }
+    }
+
+    /// Return true if this matches the host (domain or IP).
+    pub fn contains(&self, host: &str) -> bool {
+        // According to RFC3986, raw IPv6 hosts will be wrapped in []. So we need to strip those off
+        // the end in order to parse correctly
+        let host = if host.starts_with('[') {
+            let x: &[_] = &['[', ']'];
+            host.trim_matches(x)
+        } else {
+            host
+        };
+        match host.parse::<IpAddr>() {
+            // If we can parse an IP addr, then use it, otherwise, assume it is a domain
+            Ok(ip) => self.ips.contains(ip),
+            Err(_) => self.domains.contains(host),
+        }
+    }
+
+    fn is_empty(&self) -> bool {
+        self.ips.0.is_empty() && self.domains.0.is_empty()
+    }
+}
+
+impl IpMatcher {
+    fn contains(&self, addr: IpAddr) -> bool {
+        for ip in &self.0 {
+            match ip {
+                Ip::Address(address) => {
+                    if &addr == address {
+                        return true;
+                    }
+                }
+                Ip::Network(net) => {
+                    if net.contains(&addr) {
+                        return true;
+                    }
+                }
+            }
+        }
+        false
+    }
+}
+
+impl DomainMatcher {
+    // The following links may be useful to understand the origin of these rules:
+    // * https://curl.se/libcurl/c/CURLOPT_NOPROXY.html
+    // * https://github.com/curl/curl/issues/1208
+    fn contains(&self, domain: &str) -> bool {
+        let domain_len = domain.len();
+        for d in &self.0 {
+            if d == domain || d.strip_prefix('.') == Some(domain) {
+                return true;
+            } else if domain.ends_with(d) {
+                if d.starts_with('.') {
+                    // If the first character of d is a dot, that means the first character of domain
+                    // must also be a dot, so we are looking at a subdomain of d and that matches
+                    return true;
+                } else if domain.as_bytes().get(domain_len - d.len() - 1) == Some(&b'.') {
+                    // Given that d is a prefix of domain, if the prior character in domain is a dot
+                    // then that means we must be matching a subdomain of d, and that matches
+                    return true;
+                }
+            } else if d == "*" {
+                return true;
+            }
+        }
+        false
+    }
+}
+
+mod builder {
+    /// A type that can used as a `Builder` value.
+    ///
+    /// Private and sealed, only visible in docs.
+    pub trait IntoValue {
+        #[doc(hidden)]
+        fn into_value(self) -> String;
+    }
+
+    impl IntoValue for String {
+        #[doc(hidden)]
+        fn into_value(self) -> String {
+            self
+        }
+    }
+
+    impl IntoValue for &String {
+        #[doc(hidden)]
+        fn into_value(self) -> String {
+            self.into()
+        }
+    }
+
+    impl IntoValue for &str {
+        #[doc(hidden)]
+        fn into_value(self) -> String {
+            self.into()
+        }
+    }
+}
+
+#[cfg(feature = "client-proxy-system")]
+#[cfg(target_os = "macos")]
+mod mac {
+    use system_configuration::core_foundation::base::{CFType, TCFType, TCFTypeRef};
+    use system_configuration::core_foundation::dictionary::CFDictionary;
+    use system_configuration::core_foundation::number::CFNumber;
+    use system_configuration::core_foundation::string::{CFString, CFStringRef};
+    use system_configuration::dynamic_store::SCDynamicStoreBuilder;
+    use system_configuration::sys::schema_definitions::{
+        kSCPropNetProxiesHTTPEnable, kSCPropNetProxiesHTTPPort, kSCPropNetProxiesHTTPProxy,
+        kSCPropNetProxiesHTTPSEnable, kSCPropNetProxiesHTTPSPort, kSCPropNetProxiesHTTPSProxy,
+    };
+
+    pub(super) fn with_system(builder: &mut super::Builder) {
+        let store = SCDynamicStoreBuilder::new("").build();
+
+        let proxies_map = if let Some(proxies_map) = store.get_proxies() {
+            proxies_map
+        } else {
+            return;
+        };
+
+        if builder.http.is_empty() {
+            let http_proxy_config = parse_setting_from_dynamic_store(
+                &proxies_map,
+                unsafe { kSCPropNetProxiesHTTPEnable },
+                unsafe { kSCPropNetProxiesHTTPProxy },
+                unsafe { kSCPropNetProxiesHTTPPort },
+            );
+            if let Some(http) = http_proxy_config {
+                builder.http = http;
+            }
+        }
+
+        if builder.https.is_empty() {
+            let https_proxy_config = parse_setting_from_dynamic_store(
+                &proxies_map,
+                unsafe { kSCPropNetProxiesHTTPSEnable },
+                unsafe { kSCPropNetProxiesHTTPSProxy },
+                unsafe { kSCPropNetProxiesHTTPSPort },
+            );
+
+            if let Some(https) = https_proxy_config {
+                builder.https = https;
+            }
+        }
+    }
+
+    fn parse_setting_from_dynamic_store(
+        proxies_map: &CFDictionary<CFString, CFType>,
+        enabled_key: CFStringRef,
+        host_key: CFStringRef,
+        port_key: CFStringRef,
+    ) -> Option<String> {
+        let proxy_enabled = proxies_map
+            .find(enabled_key)
+            .and_then(|flag| flag.downcast::<CFNumber>())
+            .and_then(|flag| flag.to_i32())
+            .unwrap_or(0)
+            == 1;
+
+        if proxy_enabled {
+            let proxy_host = proxies_map
+                .find(host_key)
+                .and_then(|host| host.downcast::<CFString>())
+                .map(|host| host.to_string());
+            let proxy_port = proxies_map
+                .find(port_key)
+                .and_then(|port| port.downcast::<CFNumber>())
+                .and_then(|port| port.to_i32());
+
+            return match (proxy_host, proxy_port) {
+                (Some(proxy_host), Some(proxy_port)) => Some(format!("{proxy_host}:{proxy_port}")),
+                (Some(proxy_host), None) => Some(proxy_host),
+                (None, Some(_)) => None,
+                (None, None) => None,
+            };
+        }
+
+        None
+    }
+}
+
+#[cfg(feature = "client-proxy-system")]
+#[cfg(windows)]
+mod win {
+    pub(super) fn with_system(builder: &mut super::Builder) {
+        let settings = if let Ok(settings) = windows_registry::CURRENT_USER
+            .open("Software\\Microsoft\\Windows\\CurrentVersion\\Internet Settings")
+        {
+            settings
+        } else {
+            return;
+        };
+
+        if settings.get_u32("ProxyEnable").unwrap_or(0) == 0 {
+            return;
+        }
+
+        if builder.http.is_empty() {
+            if let Ok(val) = settings.get_string("ProxyServer") {
+                builder.http = val;
+            }
+        }
+
+        if builder.no.is_empty() {
+            if let Ok(val) = settings.get_string("ProxyOverride") {
+                builder.no = val
+                    .split(';')
+                    .map(|s| s.trim())
+                    .collect::<Vec<&str>>()
+                    .join(",")
+                    .replace("*.", "");
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_domain_matcher() {
+        let domains = vec![".foo.bar".into(), "bar.foo".into()];
+        let matcher = DomainMatcher(domains);
+
+        // domains match with leading `.`
+        assert!(matcher.contains("foo.bar"));
+        // subdomains match with leading `.`
+        assert!(matcher.contains("www.foo.bar"));
+
+        // domains match with no leading `.`
+        assert!(matcher.contains("bar.foo"));
+        // subdomains match with no leading `.`
+        assert!(matcher.contains("www.bar.foo"));
+
+        // non-subdomain string prefixes don't match
+        assert!(!matcher.contains("notfoo.bar"));
+        assert!(!matcher.contains("notbar.foo"));
+    }
+
+    #[test]
+    fn test_no_proxy_wildcard() {
+        let no_proxy = NoProxy::from_string("*");
+        assert!(no_proxy.contains("any.where"));
+    }
+
+    #[test]
+    fn test_no_proxy_ip_ranges() {
+        let no_proxy =
+            NoProxy::from_string(".foo.bar, bar.baz,10.42.1.1/24,::1,10.124.7.8,2001::/17");
+
+        let should_not_match = [
+            // random url, not in no_proxy
+            "hyper.rs",
+            // make sure that random non-subdomain string prefixes don't match
+            "notfoo.bar",
+            // make sure that random non-subdomain string prefixes don't match
+            "notbar.baz",
+            // ipv4 address out of range
+            "10.43.1.1",
+            // ipv4 address out of range
+            "10.124.7.7",
+            // ipv6 address out of range
+            "[ffff:db8:a0b:12f0::1]",
+            // ipv6 address out of range
+            "[2005:db8:a0b:12f0::1]",
+        ];
+
+        for host in &should_not_match {
+            assert!(!no_proxy.contains(host), "should not contain {:?}", host);
+        }
+
+        let should_match = [
+            // make sure subdomains (with leading .) match
+            "hello.foo.bar",
+            // make sure exact matches (without leading .) match (also makes sure spaces between entries work)
+            "bar.baz",
+            // make sure subdomains (without leading . in no_proxy) match
+            "foo.bar.baz",
+            // make sure subdomains (without leading . in no_proxy) match - this differs from cURL
+            "foo.bar",
+            // ipv4 address match within range
+            "10.42.1.100",
+            // ipv6 address exact match
+            "[::1]",
+            // ipv6 address match within range
+            "[2001:db8:a0b:12f0::1]",
+            // ipv4 address exact match
+            "10.124.7.8",
+        ];
+
+        for host in &should_match {
+            assert!(no_proxy.contains(host), "should contain {:?}", host);
+        }
+    }
+
+    macro_rules! p {
+        ($($n:ident = $v:expr,)*) => ({Builder {
+            $($n: $v.into(),)*
+            ..Builder::default()
+        }.build()});
+    }
+
+    fn intercept<'a>(p: &'a Matcher, u: &str) -> Intercept {
+        p.intercept(&u.parse().unwrap()).unwrap()
+    }
+
+    #[test]
+    fn test_all_proxy() {
+        let p = p! {
+            all = "http://om.nom",
+        };
+
+        assert_eq!("http://om.nom", intercept(&p, "http://example.com").uri());
+
+        assert_eq!("http://om.nom", intercept(&p, "https://example.com").uri());
+    }
+
+    #[test]
+    fn test_specific_overrides_all() {
+        let p = p! {
+            all = "http://no.pe",
+            http = "http://y.ep",
+        };
+
+        assert_eq!("http://no.pe", intercept(&p, "https://example.com").uri());
+
+        // the http rule is "more specific" than the all rule
+        assert_eq!("http://y.ep", intercept(&p, "http://example.com").uri());
+    }
+
+    #[test]
+    fn test_parse_no_scheme_defaults_to_http() {
+        let p = p! {
+            https = "y.ep",
+            http = "127.0.0.1:8887",
+        };
+
+        assert_eq!(intercept(&p, "https://example.local").uri(), "http://y.ep");
+        assert_eq!(
+            intercept(&p, "http://example.local").uri(),
+            "http://127.0.0.1:8887"
+        );
+    }
+
+    #[test]
+    fn test_parse_http_auth() {
+        let p = p! {
+            all = "http://Aladdin:[e-mail]",
+        };
+
+        let proxy = intercept(&p, "https://example.local");
+        assert_eq!(proxy.uri(), "http://y.ep");
+        assert_eq!(
+            proxy.basic_auth().expect("basic_auth"),
+            "Basic QWxhZGRpbjpvcGVuc2VzYW1l"
+        );
+    }
+
+    #[test]
+    fn test_parse_http_auth_without_scheme() {
+        let p = p! {
+            all = "Aladdin:[e-mail]",
+        };
+
+        let proxy = intercept(&p, "https://example.local");
+        assert_eq!(proxy.uri(), "http://y.ep");
+        assert_eq!(
+            proxy.basic_auth().expect("basic_auth"),
+            "Basic QWxhZGRpbjpvcGVuc2VzYW1l"
+        );
+    }
+
+    #[test]
+    fn test_dont_parse_http_when_is_cgi() {
+        let mut builder = Matcher::builder();
+        builder.is_cgi = true;
+        builder.http = "http://never.gonna.let.you.go".into();
+        let m = builder.build();
+
+        assert!(m.intercept(&"http://rick.roll".parse().unwrap()).is_none());
+    }
+}

src/client/proxy/mod.rs

+//! Proxy utilities
+
+pub mod matcher;

src/rt/io.rs

+use std::marker::Unpin;
+use std::pin::Pin;
+use std::task::Poll;
+
+use futures_util::future;
+use futures_util::ready;
+use hyper::rt::{Read, ReadBuf, Write};
+
+pub(crate) async fn read<T>(io: &mut T, buf: &mut [u8]) -> Result<usize, std::io::Error>
+where
+    T: Read + Unpin,
+{
+    future::poll_fn(move |cx| {
+        let mut buf = ReadBuf::new(buf);
+        ready!(Pin::new(&mut *io).poll_read(cx, buf.unfilled()))?;
+        Poll::Ready(Ok(buf.filled().len()))
+    })
+    .await
+}
+
+pub(crate) async fn write_all<T>(io: &mut T, buf: &[u8]) -> Result<(), std::io::Error>
+where
+    T: Write + Unpin,
+{
+    let mut n = 0;
+    future::poll_fn(move |cx| {
+        while n < buf.len() {
+            n += ready!(Pin::new(&mut *io).poll_write(cx, &buf[n..])?);
+        }
+        Poll::Ready(Ok(()))
+    })
+    .await
+}

tests/proxy.rs

+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+use tower_service::Service;
+
+use hyper_util::client::legacy::connect::proxy::{SocksV4, SocksV5, Tunnel};
+use hyper_util::client::legacy::connect::HttpConnector;
+
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_tunnel_works() {
+    let tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let addr = tcp.local_addr().expect("local_addr");
+
+    let proxy_dst = format!("http://{}", addr).parse().expect("uri");
+    let mut connector = Tunnel::new(proxy_dst, HttpConnector::new());
+    let t1 = tokio::spawn(async move {
+        let _conn = connector
+            .call("https://hyper.rs".parse().unwrap())
+            .await
+            .expect("tunnel");
+    });
+
+    let t2 = tokio::spawn(async move {
+        let (mut io, _) = tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 64];
+        let n = io.read(&mut buf).await.expect("read 1");
+        assert_eq!(
+            &buf[..n],
+            b"CONNECT hyper.rs:443 HTTP/1.1\r\nHost: hyper.rs:443\r\n\r\n"
+        );
+        io.write_all(b"HTTP/1.1 200 OK\r\n\r\n")
+            .await
+            .expect("write 1");
+    });
+
+    t1.await.expect("task 1");
+    t2.await.expect("task 2");
+}
+
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_socks_v5_without_auth_works() {
+    let proxy_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let proxy_addr = proxy_tcp.local_addr().expect("local_addr");
+    let proxy_dst = format!("http://{proxy_addr}").parse().expect("uri");
+
+    let target_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let target_addr = target_tcp.local_addr().expect("local_addr");
+    let target_dst = format!("http://{target_addr}").parse().expect("uri");
+
+    let mut connector = SocksV5::new(proxy_dst, HttpConnector::new());
+
+    // Client
+    //
+    // Will use `SocksV5` to establish proxy tunnel.
+    // Will send "Hello World!" to the target and receive "Goodbye!" back.
+    let t1 = tokio::spawn(async move {
+        let conn = connector.call(target_dst).await.expect("tunnel");
+        let mut tcp = conn.into_inner();
+
+        tcp.write_all(b"Hello World!").await.expect("write 1");
+
+        let mut buf = [0u8; 64];
+        let n = tcp.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], b"Goodbye!");
+    });
+
+    // Proxy
+    //
+    // Will receive CONNECT command from client.
+    // Will connect to target and success code back to client.
+    // Will blindly tunnel between client and target.
+    let t2 = tokio::spawn(async move {
+        let (mut to_client, _) = proxy_tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 513];
+
+        // negotiation req/res
+        let n = to_client.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], [0x05, 0x01, 0x00]);
+
+        to_client.write_all(&[0x05, 0x00]).await.expect("write 1");
+
+        // command req/rs
+        let [p1, p2] = target_addr.port().to_be_bytes();
+        let [ip1, ip2, ip3, ip4] = [0x7f, 0x00, 0x00, 0x01];
+        let message = [0x05, 0x01, 0x00, 0x01, ip1, ip2, ip3, ip4, p1, p2];
+        let n = to_client.read(&mut buf).await.expect("read 2");
+        assert_eq!(&buf[..n], message);
+
+        let mut to_target = TcpStream::connect(target_addr).await.expect("connect");
+
+        let message = [0x05, 0x00, 0x00, 0x01, ip1, ip2, ip3, ip4, p1, p2];
+        to_client.write_all(&message).await.expect("write 2");
+
+        let (from_client, from_target) =
+            tokio::io::copy_bidirectional(&mut to_client, &mut to_target)
+                .await
+                .expect("proxy");
+
+        assert_eq!(from_client, 12);
+        assert_eq!(from_target, 8)
+    });
+
+    // Target server
+    //
+    // Will accept connection from proxy server
+    // Will receive "Hello World!" from the client and return "Goodbye!"
+    let t3 = tokio::spawn(async move {
+        let (mut io, _) = target_tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 64];
+
+        let n = io.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], b"Hello World!");
+
+        io.write_all(b"Goodbye!").await.expect("write 1");
+    });
+
+    t1.await.expect("task - client");
+    t2.await.expect("task - proxy");
+    t3.await.expect("task - target");
+}
+
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_socks_v5_with_auth_works() {
+    let proxy_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let proxy_addr = proxy_tcp.local_addr().expect("local_addr");
+    let proxy_dst = format!("http://{proxy_addr}").parse().expect("uri");
+
+    let target_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let target_addr = target_tcp.local_addr().expect("local_addr");
+    let target_dst = format!("http://{target_addr}").parse().expect("uri");
+
+    let mut connector =
+        SocksV5::new(proxy_dst, HttpConnector::new()).with_auth("user".into(), "pass".into());
+
+    // Client
+    //
+    // Will use `SocksV5` to establish proxy tunnel.
+    // Will send "Hello World!" to the target and receive "Goodbye!" back.
+    let t1 = tokio::spawn(async move {
+        let conn = connector.call(target_dst).await.expect("tunnel");
+        let mut tcp = conn.into_inner();
+
+        tcp.write_all(b"Hello World!").await.expect("write 1");
+
+        let mut buf = [0u8; 64];
+        let n = tcp.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], b"Goodbye!");
+    });
+
+    // Proxy
+    //
+    // Will receive CONNECT command from client.
+    // Will connect to target and success code back to client.
+    // Will blindly tunnel between client and target.
+    let t2 = tokio::spawn(async move {
+        let (mut to_client, _) = proxy_tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 513];
+
+        // negotiation req/res
+        let n = to_client.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], [0x05, 0x01, 0x02]);
+
+        to_client.write_all(&[0x05, 0x02]).await.expect("write 1");
+
+        // auth req/res
+        let n = to_client.read(&mut buf).await.expect("read 2");
+        let [u1, u2, u3, u4] = b"user";
+        let [p1, p2, p3, p4] = b"pass";
+        let message = [0x01, 0x04, *u1, *u2, *u3, *u4, 0x04, *p1, *p2, *p3, *p4];
+        assert_eq!(&buf[..n], message);
+
+        to_client.write_all(&[0x01, 0x00]).await.expect("write 2");
+
+        // command req/res
+        let n = to_client.read(&mut buf).await.expect("read 3");
+        let [p1, p2] = target_addr.port().to_be_bytes();
+        let [ip1, ip2, ip3, ip4] = [0x7f, 0x00, 0x00, 0x01];
+        let message = [0x05, 0x01, 0x00, 0x01, ip1, ip2, ip3, ip4, p1, p2];
+        assert_eq!(&buf[..n], message);
+
+        let mut to_target = TcpStream::connect(target_addr).await.expect("connect");
+
+        let message = [0x05, 0x00, 0x00, 0x01, ip1, ip2, ip3, ip4, p1, p2];
+        to_client.write_all(&message).await.expect("write 3");
+
+        let (from_client, from_target) =
+            tokio::io::copy_bidirectional(&mut to_client, &mut to_target)
+                .await
+                .expect("proxy");
+
+        assert_eq!(from_client, 12);
+        assert_eq!(from_target, 8)
+    });
+
+    // Target server
+    //
+    // Will accept connection from proxy server
+    // Will receive "Hello World!" from the client and return "Goodbye!"
+    let t3 = tokio::spawn(async move {
+        let (mut io, _) = target_tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 64];
+
+        let n = io.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], b"Hello World!");
+
+        io.write_all(b"Goodbye!").await.expect("write 1");
+    });
+
+    t1.await.expect("task - client");
+    t2.await.expect("task - proxy");
+    t3.await.expect("task - target");
+}
+
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_socks_v5_with_server_resolved_domain_works() {
+    let proxy_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let proxy_addr = proxy_tcp.local_addr().expect("local_addr");
+    let proxy_addr = format!("http://{proxy_addr}").parse().expect("uri");
+
+    let mut connector = SocksV5::new(proxy_addr, HttpConnector::new())
+        .with_auth("user".into(), "pass".into())
+        .local_dns(false);
+
+    // Client
+    //
+    // Will use `SocksV5` to establish proxy tunnel.
+    // Will send "Hello World!" to the target and receive "Goodbye!" back.
+    let t1 = tokio::spawn(async move {
+        let _conn = connector
+            .call("https://hyper.rs:443".try_into().unwrap())
+            .await
+            .expect("tunnel");
+    });
+
+    // Proxy
+    //
+    // Will receive CONNECT command from client.
+    // Will connect to target and success code back to client.
+    // Will blindly tunnel between client and target.
+    let t2 = tokio::spawn(async move {
+        let (mut to_client, _) = proxy_tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 513];
+
+        // negotiation req/res
+        let n = to_client.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], [0x05, 0x01, 0x02]);
+
+        to_client.write_all(&[0x05, 0x02]).await.expect("write 1");
+
+        // auth req/res
+        let n = to_client.read(&mut buf).await.expect("read 2");
+        let [u1, u2, u3, u4] = b"user";
+        let [p1, p2, p3, p4] = b"pass";
+        let message = [0x01, 0x04, *u1, *u2, *u3, *u4, 0x04, *p1, *p2, *p3, *p4];
+        assert_eq!(&buf[..n], message);
+
+        to_client.write_all(&[0x01, 0x00]).await.expect("write 2");
+
+        // command req/res
+        let n = to_client.read(&mut buf).await.expect("read 3");
+
+        let host = "hyper.rs";
+        let port: u16 = 443;
+        let mut message = vec![0x05, 0x01, 0x00, 0x03, host.len() as u8];
+        message.extend(host.bytes());
+        message.extend(port.to_be_bytes());
+        assert_eq!(&buf[..n], message);
+
+        let mut message = vec![0x05, 0x00, 0x00, 0x03, host.len() as u8];
+        message.extend(host.bytes());
+        message.extend(port.to_be_bytes());
+        to_client.write_all(&message).await.expect("write 3");
+    });
+
+    t1.await.expect("task - client");
+    t2.await.expect("task - proxy");
+}
+
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_socks_v5_with_locally_resolved_domain_works() {
+    let proxy_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let proxy_addr = proxy_tcp.local_addr().expect("local_addr");
+    let proxy_addr = format!("http://{proxy_addr}").parse().expect("uri");
+
+    let mut connector = SocksV5::new(proxy_addr, HttpConnector::new())
+        .with_auth("user".into(), "pass".into())
+        .local_dns(true);
+
+    // Client
+    //
+    // Will use `SocksV5` to establish proxy tunnel.
+    // Will send "Hello World!" to the target and receive "Goodbye!" back.
+    let t1 = tokio::spawn(async move {
+        let _conn = connector
+            .call("https://hyper.rs:443".try_into().unwrap())
+            .await
+            .expect("tunnel");
+    });
+
+    // Proxy
+    //
+    // Will receive CONNECT command from client.
+    // Will connect to target and success code back to client.
+    // Will blindly tunnel between client and target.
+    let t2 = tokio::spawn(async move {
+        let (mut to_client, _) = proxy_tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 513];
+
+        // negotiation req/res
+        let n = to_client.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], [0x05, 0x01, 0x02]);
+
+        to_client.write_all(&[0x05, 0x02]).await.expect("write 1");
+
+        // auth req/res
+        let n = to_client.read(&mut buf).await.expect("read 2");
+        let [u1, u2, u3, u4] = b"user";
+        let [p1, p2, p3, p4] = b"pass";
+        let message = [0x01, 0x04, *u1, *u2, *u3, *u4, 0x04, *p1, *p2, *p3, *p4];
+        assert_eq!(&buf[..n], message);
+
+        to_client.write_all(&[0x01, 0x00]).await.expect("write 2");
+
+        // command req/res
+        let n = to_client.read(&mut buf).await.expect("read 3");
+        let message = [0x05, 0x01, 0x00, 0x01];
+        assert_eq!(&buf[..4], message);
+        assert_eq!(n, 4 + 4 + 2);
+
+        let message = vec![0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0];
+        to_client.write_all(&message).await.expect("write 3");
+    });
+
+    t1.await.expect("task - client");
+    t2.await.expect("task - proxy");
+}
+
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_socks_v4_works() {
+    let proxy_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let proxy_addr = proxy_tcp.local_addr().expect("local_addr");
+    let proxy_dst = format!("http://{proxy_addr}").parse().expect("uri");
+
+    let target_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let target_addr = target_tcp.local_addr().expect("local_addr");
+    let target_dst = format!("http://{target_addr}").parse().expect("uri");
+
+    let mut connector = SocksV4::new(proxy_dst, HttpConnector::new());
+
+    // Client
+    //
+    // Will use `SocksV4` to establish proxy tunnel.
+    // Will send "Hello World!" to the target and receive "Goodbye!" back.
+    let t1 = tokio::spawn(async move {
+        let conn = connector.call(target_dst).await.expect("tunnel");
+        let mut tcp = conn.into_inner();
+
+        tcp.write_all(b"Hello World!").await.expect("write 1");
+
+        let mut buf = [0u8; 64];
+        let n = tcp.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], b"Goodbye!");
+    });
+
+    // Proxy
+    //
+    // Will receive CONNECT command from client.
+    // Will connect to target and success code back to client.
+    // Will blindly tunnel between client and target.
+    let t2 = tokio::spawn(async move {
+        let (mut to_client, _) = proxy_tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 512];
+
+        let [p1, p2] = target_addr.port().to_be_bytes();
+        let [ip1, ip2, ip3, ip4] = [127, 0, 0, 1];
+        let message = [4, 0x01, p1, p2, ip1, ip2, ip3, ip4, 0, 0];
+        let n = to_client.read(&mut buf).await.expect("read");
+        assert_eq!(&buf[..n], message);
+
+        let mut to_target = TcpStream::connect(target_addr).await.expect("connect");
+
+        let message = [0, 90, p1, p2, ip1, ip2, ip3, ip4];
+        to_client.write_all(&message).await.expect("write");
+
+        let (from_client, from_target) =
+            tokio::io::copy_bidirectional(&mut to_client, &mut to_target)
+                .await
+                .expect("proxy");
+
+        assert_eq!(from_client, 12);
+        assert_eq!(from_target, 8)
+    });
+
+    // Target server
+    //
+    // Will accept connection from proxy server
+    // Will receive "Hello World!" from the client and return "Goodbye!"
+    let t3 = tokio::spawn(async move {
+        let (mut io, _) = target_tcp.accept().await.expect("accept");
+        let mut buf = [0u8; 64];
+
+        let n = io.read(&mut buf).await.expect("read 1");
+        assert_eq!(&buf[..n], b"Hello World!");
+
+        io.write_all(b"Goodbye!").await.expect("write 1");
+    });
+
+    t1.await.expect("task - client");
+    t2.await.expect("task - proxy");
+    t3.await.expect("task - target");
+}
+
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_socks_v5_optimistic_works() {
+    let proxy_tcp = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let proxy_addr = proxy_tcp.local_addr().expect("local_addr");
+    let proxy_dst = format!("http://{proxy_addr}").parse().expect("uri");
+
+    let target_addr = std::net::SocketAddr::new([127, 0, 0, 1].into(), 1234);
+    let target_dst = format!("http://{target_addr}").parse().expect("uri");
+
+    let mut connector = SocksV5::new(proxy_dst, HttpConnector::new())
+        .with_auth("ABC".into(), "XYZ".into())
+        .send_optimistically(true);
+
+    // Client
+    //
+    // Will use `SocksV5` to establish proxy tunnel.
+    // Will send "Hello World!" to the target and receive "Goodbye!" back.
+    let t1 = tokio::spawn(async move {
+        let _ = connector.call(target_dst).await.expect("tunnel");
+    });
+
+    // Proxy
+    //
+    // Will receive SOCKS handshake from client.
+    // Will connect to target and success code back to client.
+    // Will blindly tunnel between client and target.
+    let t2 = tokio::spawn(async move {
+        let (mut to_client, _) = proxy_tcp.accept().await.expect("accept");
+        let [p1, p2] = target_addr.port().to_be_bytes();
+
+        let mut buf = [0; 22];
+        let request = vec![
+            5, 1, 2, // Negotiation
+            1, 3, 65, 66, 67, 3, 88, 89, 90, // Auth ("ABC"/"XYZ")
+            5, 1, 0, 1, 127, 0, 0, 1, p1, p2, // Reply
+        ];
+
+        let response = vec![
+            5, 2, // Negotiation,
+            1, 0, // Auth,
+            5, 0, 0, 1, 127, 0, 0, 1, p1, p2, // Reply
+        ];
+
+        // Accept all handshake messages
+        to_client.read_exact(&mut buf).await.expect("read");
+        assert_eq!(request.as_slice(), buf);
+
+        // Send all handshake messages back
+        to_client
+            .write_all(response.as_slice())
+            .await
+            .expect("write");
+
+        to_client.flush().await.expect("flush");
+    });
+
+    t1.await.expect("task - client");
+    t2.await.expect("task - proxy");
+}

.cargo_vcs_info.json

 {
   "git": {
-    "sha1": "4c4e0622aaaceb43acceb20834a854faef86f65d"
+    "sha1": "8805922b8230160f08ca5f05ae1003c07f9260f4"
   },
   "path_in_vcs": ""
 }

Cargo.lock

@@ -75,10 +75,22 @@ # This file is automatically @generated by Cargo.
  "rustc-demangle",
- "windows-targets",
+ "windows-targets 0.52.6",
 ]
 
 [[package]]
+name = "base64"
+version = "0.22.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+
+[[package]]
+name = "bitflags"
+version = "2.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c8214115b7bf84099f1309324e63141d4c5d7cc26862f97a0a857dbefe165bd"
+
+[[package]]
 name = "bytes"
-version = "1.9.0"
+version = "1.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "325918d6fe32f23b19878fe4b34794ae41fc19ddbe53b10571a4874d44ffd39b"
+checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a"
 
@@ -92,2 +104,18 @@ [[package]]
 [[package]]
+name = "core-foundation"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
+[[package]]
+name = "core-foundation-sys"
+version = "0.8.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
+
+[[package]]
 name = "env_logger"
@@ -107,5 +135,5 @@ version = "0.10.2"
 name = "equivalent"
-version = "1.0.1"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
 
@@ -165,5 +193,5 @@ [[package]]
 name = "h2"
-version = "0.4.7"
+version = "0.4.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ccae279728d634d083c00f6099cb58f01cc99c145b84b8be2f6c74618d79922e"
+checksum = "5017294ff4bb30944501348f6f8e42e6ad28f42c8bbef7a74029aff064a4e3c2"
 dependencies = [
@@ -191,11 +219,11 @@ "atomic-waker",
 name = "hermit-abi"
-version = "0.4.0"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fbf6a919d6cf397374f7dfeeea91d974c7c0a7221d0d0f4f20d859d329e53fcc"
+checksum = "fbd780fe5cc30f81464441920d82ac8740e2e46b29a6fad543ddd075229ce37e"
 
 [[package]]
 name = "http"
-version = "1.2.0"
+version = "1.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f16ca2af56261c99fba8bac40a10251ce8188205a4c448fbb745a2e4daa76fea"
+checksum = "f4a85d31aea989eead29a3aaf9e1115a180df8282431156e533de47660892565"
 dependencies = [
@@ -219,8 +247,8 @@ "bytes",
 name = "http-body-util"
-version = "0.1.2"
+version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "793429d76616a256bcb62c2a2ec2bed781c8307e797e2598c50010f2bee2544f"
+checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a"
 dependencies = [
  "bytes",
- "futures-util",
+ "futures-core",
  "http",
@@ -233,5 +261,5 @@ "http-body",
 name = "httparse"
-version = "1.9.5"
+version = "1.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7d71d3574edd2771538b901e6549113b4006ece66150fb69c0fb6d9a2adae946"
+checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
 
@@ -246,5 +274,5 @@ [[package]]
 name = "humantime"
-version = "2.1.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
+checksum = "9b112acc8b3adf4b107a8ec20977da0273a8c386765a3ec0229bd500a1443f9f"
 
@@ -274,4 +302,5 @@ [[package]]
 name = "hyper-util"
-version = "0.1.11"
+version = "0.1.12"
 dependencies = [
+ "base64",
  "bytes",
@@ -284,3 +313,5 @@ "futures-channel",
  "hyper",
+ "ipnet",
  "libc",
+ "percent-encoding",
  "pin-project-lite",
@@ -290,2 +321,3 @@ "pnet_datalink",
  "socket2",
+ "system-configuration",
  "tokio",
@@ -295,2 +327,3 @@ "tokio-test",
  "tracing",
+ "windows-registry",
 ]
@@ -300,5 +333,5 @@
 name = "indexmap"
-version = "2.7.0"
+version = "2.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "62f822373a4fe84d4bb149bf54e584a7f4abec90e072ed49cda0edea5b95471f"
+checksum = "3954d50fe15b02142bf25d3b8bdadb634ec3948f103d04ffe3031bc8fe9d7058"
 dependencies = [
@@ -310,2 +343,8 @@ "equivalent",
 [[package]]
+name = "ipnet"
+version = "2.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "469fb0b9cefa57e3ef31275ee7cacb78f2fdca44e4765491884a2b119d4eb130"
+
+[[package]]
 name = "ipnetwork"
@@ -321,9 +360,9 @@ version = "0.20.0"
 name = "is-terminal"
-version = "0.4.13"
+version = "0.4.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "261f68e344040fbd0edea105bef17c66edf46f984ddb1115b775ce31be948f4b"
+checksum = "e04d7f318608d35d4b61ddd75cbdaee86b023ebe2bd5a66ee0915f0bf93095a9"
 dependencies = [
  "hermit-abi",
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
@@ -333,5 +372,5 @@
 name = "itoa"
-version = "1.0.14"
+version = "1.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d75a2a4b1b190afb6f5425f10f6a8f959d2ea0b9c2b1d79553551850539e4674"
+checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c"
 
@@ -346,5 +385,5 @@ [[package]]
 name = "log"
-version = "0.4.22"
+version = "0.4.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"
+checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94"
 
@@ -359,5 +398,5 @@ [[package]]
 name = "miniz_oxide"
-version = "0.8.0"
+version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2d80299ef12ff69b16a84bb182e3b9df68b5a91574d3d4fa6e41b65deec4df1"
+checksum = "8e3e04debbb59698c15bacbb6d93584a8c0ca9cc3213cb423d31f760d8843ce5"
 dependencies = [
@@ -386,5 +425,5 @@ "adler2",
 name = "object"
-version = "0.36.5"
+version = "0.36.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aedf0a2d09c573ed1d8d85b30c119153926a2b36dce0ab28322c09a117a4683e"
+checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
 dependencies = [
@@ -396,11 +435,17 @@ "memchr",
 name = "once_cell"
-version = "1.20.2"
+version = "1.21.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1261fe7e33c73b354eab43b1273a57c8f967d0391e80353e51f764ac02cf6775"
+checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d"
 
 [[package]]
+name = "percent-encoding"
+version = "2.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
+
+[[package]]
 name = "pin-project-lite"
-version = "0.2.15"
+version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "915a1e146535de9163f3987b8944ed8cf49a18bb0056bcebcdcece385cece4ff"
+checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b"
 
@@ -457,5 +502,5 @@ [[package]]
 name = "proc-macro2"
-version = "1.0.92"
+version = "1.0.94"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "37d3544b3f2748c54e147655edb5025752e2303145b5aefb3c3ea2c78b973bb0"
+checksum = "a31971752e70b8b2686d7e46ec17fb38dad4051d94024c88df49b667caea9c84"
 dependencies = [
@@ -467,5 +512,5 @@ "unicode-ident",
 name = "quote"
-version = "1.0.37"
+version = "1.0.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b5b9d34b8991d19d98081b46eacdd8eb58c6f2b201139f7c5f643cc155a633af"
+checksum = "1885c039570dc00dcb4ff087a89e185fd56bae234ddc7f056a945bf36467248d"
 dependencies = [
@@ -512,5 +557,5 @@ "proc-macro2",
 name = "serde"
-version = "1.0.216"
+version = "1.0.219"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b9781016e935a97e8beecf0c933758c97a5520d32930e460142b4cd80c6338e"
+checksum = "5f0e2c6ed6606019b4e29e69dbaba95b11854410e5347d525002456dbbb786b6"
 dependencies = [
@@ -522,5 +567,5 @@ "serde_derive",
 name = "serde_derive"
-version = "1.0.216"
+version = "1.0.219"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "46f859dbbf73865c6627ed570e78961cd3ac92407a2d117204c49232485da55e"
+checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00"
 dependencies = [
@@ -552,5 +597,5 @@ "proc-macro2",
 name = "smallvec"
-version = "1.13.2"
+version = "1.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67"
+checksum = "7fcf8323ef1faaee30a44a340193b1ac6814fd9b7b4e88e9d4519a3e4abe1cfd"
 
@@ -569,5 +614,5 @@ [[package]]
 name = "syn"
-version = "2.0.90"
+version = "2.0.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "919d3b74a5dd0ccd15aeb8f93e7006bd9e14c295087c9896a110f490752bcf31"
+checksum = "b09a44accad81e1ba1cd74a32461ba89dee89095ba17b32f5d03683b1b1fc2a0"
 dependencies = [
@@ -580,2 +625,23 @@ "proc-macro2",
 [[package]]
+name = "system-configuration"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b"
+dependencies = [
+ "bitflags",
+ "core-foundation",
+ "system-configuration-sys",
+]
+
+[[package]]
+name = "system-configuration-sys"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8e1d1b10ced5ca923a1fcb8d03e96b8d3268065d724548c0211415ff6ac6bac4"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
+[[package]]
 name = "termcolor"
@@ -591,5 +657,5 @@ version = "1.4.1"
 name = "tokio"
-version = "1.42.0"
+version = "1.44.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5cec9b21b0450273377fc97bd4c33a8acffc8c996c987a7c5b319a0083707551"
+checksum = "f382da615b842244d4b8738c82ed1275e6c5dd90c459a30941cd07080b06c91a"
 dependencies = [
@@ -609,5 +675,5 @@ "backtrace",
 name = "tokio-macros"
-version = "2.4.0"
+version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "693d596312e88961bc67d7f1f97af8a70227d9f90c31bba5806eec004978d752"
+checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8"
 dependencies = [
@@ -645,5 +711,5 @@ "proc-macro2",
 name = "tokio-util"
-version = "0.7.13"
+version = "0.7.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7fcaa8d55a2bdd6b83ace262b016eca0d79ee02818c5c1bcdf0305114081078"
+checksum = "6b9590b93e6fcc1739458317cccd391ad3955e2bde8913edf6f95f9e65a8f034"
 dependencies = [
@@ -690,5 +756,5 @@ "bytes",
 name = "unicode-ident"
-version = "1.0.14"
+version = "1.0.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "adb9e6ca4f869e1180728b7950e35922a7fc6397f7b641499e8f3ef06e50dc83"
+checksum = "5a5f39404a5da50712a4c1eecf25e90dd62b613502b7e925fd4e4d19b5c96512"
 
@@ -742,2 +808,37 @@ [[package]]
 [[package]]
+name = "windows-link"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76840935b766e1b0a05c0066835fb9ec80071d4c09a16f6bd5f7e655e3c14c38"
+
+[[package]]
+name = "windows-registry"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4286ad90ddb45071efd1a66dfa43eb02dd0dfbae1545ad6cc3c51cf34d7e8ba3"
+dependencies = [
+ "windows-result",
+ "windows-strings",
+ "windows-targets 0.53.0",
+]
+
+[[package]]
+name = "windows-result"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c64fd11a4fd95df68efcfee5f44a294fe71b8bc6a91993e2791938abcc712252"
+dependencies = [
+ "windows-link",
+]
+
+[[package]]
+name = "windows-strings"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "87fa48cc5d406560701792be122a10132491cff9d0aeb23583cc2dcafc847319"
+dependencies = [
+ "windows-link",
+]
+
+[[package]]
 name = "windows-sys"
@@ -748,3 +849,3 @@ version = "0.52.0"
 dependencies = [
- "windows-targets",
+ "windows-targets 0.52.6",
 ]
@@ -758,3 +859,3 @@
 dependencies = [
- "windows-targets",
+ "windows-targets 0.52.6",
 ]
@@ -768,13 +869,29 @@
 dependencies = [
- "windows_aarch64_gnullvm",
- "windows_aarch64_msvc",
- "windows_i686_gnu",
- "windows_i686_gnullvm",
- "windows_i686_msvc",
- "windows_x86_64_gnu",
- "windows_x86_64_gnullvm",
- "windows_x86_64_msvc",
+ "windows_aarch64_gnullvm 0.52.6",
+ "windows_aarch64_msvc 0.52.6",
+ "windows_i686_gnu 0.52.6",
+ "windows_i686_gnullvm 0.52.6",
+ "windows_i686_msvc 0.52.6",
+ "windows_x86_64_gnu 0.52.6",
+ "windows_x86_64_gnullvm 0.52.6",
+ "windows_x86_64_msvc 0.52.6",
 ]
 
 [[package]]
+name = "windows-targets"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1e4c7e8ceaaf9cb7d7507c974735728ab453b67ef8f18febdd7c11fe59dca8b"
+dependencies = [
+ "windows_aarch64_gnullvm 0.53.0",
+ "windows_aarch64_msvc 0.53.0",
+ "windows_i686_gnu 0.53.0",
+ "windows_i686_gnullvm 0.53.0",
+ "windows_i686_msvc 0.53.0",
+ "windows_x86_64_gnu 0.53.0",
+ "windows_x86_64_gnullvm 0.53.0",
+ "windows_x86_64_msvc 0.53.0",
+]
+
+[[package]]
 name = "windows_aarch64_gnullvm"
@@ -786,2 +903,8 @@ version = "0.52.6"
 [[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "86b8d5f90ddd19cb4a147a5fa63ca848db3df085e25fee3cc10b39b6eebae764"
+
+[[package]]
 name = "windows_aarch64_msvc"
@@ -793,2 +916,8 @@ version = "0.52.6"
 [[package]]
+name = "windows_aarch64_msvc"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7651a1f62a11b8cbd5e0d42526e55f2c99886c77e007179efff86c2b137e66c"
+
+[[package]]
 name = "windows_i686_gnu"
@@ -800,2 +929,8 @@ version = "0.52.6"
 [[package]]
+name = "windows_i686_gnu"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1dc67659d35f387f5f6c479dc4e28f1d4bb90ddd1a5d3da2e5d97b42d6272c3"
+
+[[package]]
 name = "windows_i686_gnullvm"
@@ -807,2 +942,8 @@ version = "0.52.6"
 [[package]]
+name = "windows_i686_gnullvm"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ce6ccbdedbf6d6354471319e781c0dfef054c81fbc7cf83f338a4296c0cae11"
+
+[[package]]
 name = "windows_i686_msvc"
@@ -814,2 +955,8 @@ version = "0.52.6"
 [[package]]
+name = "windows_i686_msvc"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "581fee95406bb13382d2f65cd4a908ca7b1e4c2f1917f143ba16efe98a589b5d"
+
+[[package]]
 name = "windows_x86_64_gnu"
@@ -821,2 +968,8 @@ version = "0.52.6"
 [[package]]
+name = "windows_x86_64_gnu"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e55b5ac9ea33f2fc1716d1742db15574fd6fc8dadc51caab1c16a3d3b4190ba"
+
+[[package]]
 name = "windows_x86_64_gnullvm"
@@ -828,2 +981,8 @@ version = "0.52.6"
 [[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0a6e035dd0599267ce1ee132e51c27dd29437f63325753051e71dd9e42406c57"
+
+[[package]]
 name = "windows_x86_64_msvc"
@@ -833,1 +992,7 @@ version = "0.52.6"
 checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
+
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "271414315aff87387382ec3d271b52d7ae78726f5d44ac98b4f4030c91880486"

Cargo.toml

@@ -16,3 +16,3 @@ # THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
 name = "hyper-util"
-version = "0.1.11"
+version = "0.1.12"
 authors = ["Sean McArthur <[e-mail]>"]
@@ -63,2 +63,12 @@ build = false
 ]
+client-proxy = [
+    "client",
+    "dep:base64",
+    "dep:ipnet",
+    "dep:percent-encoding",
+]
+client-proxy-system = [
+    "dep:system-configuration",
+    "dep:windows-registry",
+]
 default = []
@@ -134,2 +144,10 @@ full = [
 
+[[test]]
+name = "proxy"
+path = "tests/proxy.rs"
+
+[dependencies.base64]
+version = "0.22"
+optional = true
+
 [dependencies.bytes]
@@ -155,2 +173,6 @@ version = "1.7.1"
 
+[dependencies.ipnet]
+version = "2.9"
+optional = true
+
 [dependencies.libc]
@@ -160,2 +182,6 @@ version = "0.2"
 
+[dependencies.percent-encoding]
+version = "2.3"
+optional = true
+
 [dependencies.pin-project-lite]
@@ -210,1 +236,9 @@ version = "0.2.4"
 version = "0.35.0"
+
+[target.'cfg(target_os = "macos")'.dependencies.system-configuration]
+version = "0.6.1"
+optional = true
+
+[target."cfg(windows)".dependencies.windows-registry]
+version = "0.4"
+optional = true

CHANGELOG.md

@@ -0,1 +1,12 @@
+# 0.1.12 (2025-05-19)
+
+- Add `client::legacy::proxy::Tunnel` connector that wraps another connector with HTTP tunneling.
+- Add `client::legacy::proxy::{SocksV4, SocksV5}` connectors that wraps another connector with SOCKS.
+- Add `client::proxy::matcher::Matcher` type that can use environment variables to match proxy rules.
+- Add `server::graceful::Watcher` type that can be sent to watch a connection in another task.
+- Add `GracefulShutdown::count()` method to get number of currently watched connections.
+- Fix missing `must_use` attributes on `Connection` futures.
+- Fix tracing span in GAI resolver that can cause panics.
+
+
 # 0.1.11 (2025-03-31)
@@ -2,0 +13,0 @@

src/client/legacy/client.rs

@@ -125,5 +125,6 @@ //! The legacy HTTP Client from 0.14.x
     /// use hyper_util::client::legacy::Client;
-    /// use hyper_util::rt::TokioExecutor;
+    /// use hyper_util::rt::{TokioExecutor, TokioTimer};
     ///
     /// let client = Client::builder(TokioExecutor::new())
+    ///     .pool_timer(TokioTimer::new())
     ///     .pool_idle_timeout(Duration::from_secs(30))
@@ -130,0 +131,0 @@ ///     .http2_only(true)

src/client/legacy/connect/dns.rs

@@ -33,3 +33,2 @@ //! DNS Resolution used by the `HttpConnector`.
 use tower_service::Service;
-use tracing::debug_span;
 
@@ -121,5 +120,3 @@ pub(super) use self::sealed::Resolve;
     fn call(&mut self, name: Name) -> Self::Future {
-        let span = debug_span!("resolve", host = %name.host);
         let blocking = tokio::task::spawn_blocking(move || {
-            let _enter = span.enter();
             (&*name.host, 0)
@@ -126,0 +123,0 @@ .to_socket_addrs()

src/client/legacy/connect/mod.rs

@@ -83,2 +83,4 @@ //! Connectors used by the `Client`.
 
+pub mod proxy;
+
 pub(crate) mod capture;
@@ -85,0 +87,0 @@ pub use capture::{capture_connection, CaptureConnection};

src/client/mod.rs

@@ -6,1 +6,4 @@ //! HTTP client utilities
 pub mod legacy;
+
+#[cfg(feature = "client-proxy")]
+pub mod proxy;

src/common/rewind.rs

@@ -40,5 +40,4 @@ use std::{cmp, io};
             if !prefix.is_empty() {
-                let copy_len = cmp::min(prefix.len(), remaining(&mut buf));
-                // TODO: There should be a way to do following two lines cleaner...
-                put_slice(&mut buf, &prefix[..copy_len]);
+                let copy_len = cmp::min(prefix.len(), buf.remaining());
+                buf.put_slice(&prefix[..copy_len]);
                 prefix.advance(copy_len);
@@ -57,29 +56,2 @@ // Put back what's left
 
-fn remaining(cursor: &mut ReadBufCursor<'_>) -> usize {
-    // SAFETY:
-    // We do not uninitialize any set bytes.
-    unsafe { cursor.as_mut().len() }
-}
-
-// Copied from `ReadBufCursor::put_slice`.
-// If that becomes public, we could ditch this.
-fn put_slice(cursor: &mut ReadBufCursor<'_>, slice: &[u8]) {
-    assert!(
-        remaining(cursor) >= slice.len(),
-        "buf.len() must fit in remaining()"
-    );
-
-    let amt = slice.len();
-
-    // SAFETY:
-    // the length is asserted above
-    unsafe {
-        cursor.as_mut()[..amt]
-            .as_mut_ptr()
-            .cast::<u8>()
-            .copy_from_nonoverlapping(slice.as_ptr(), amt);
-        cursor.advance(amt);
-    }
-}
-
 impl<T> Write for Rewind<T>
@@ -86,0 +58,0 @@ where

src/rt/mod.rs

 //! Runtime utilities
 
+#[cfg(feature = "client-legacy")]
+mod io;
+#[cfg(feature = "client-legacy")]
+pub(crate) use self::io::{read, write_all};
+
 #[cfg(feature = "tokio")]
@@ -4,0 +9,0 @@ pub mod tokio;

src/server/conn/auto/mod.rs

@@ -69,2 +69,8 @@ //! Http1 or Http2 connection.
 
+impl<E: Default> Default for Builder<E> {
+    fn default() -> Self {
+        Self::new(E::default())
+    }
+}
+
 impl<E> Builder<E> {
@@ -322,3 +328,8 @@ /// Create a new auto connection builder.
 pin_project! {
-    /// Connection future.
+    /// A [`Future`](core::future::Future) representing an HTTP/1 connection, returned from
+    /// [`Builder::serve_connection`](struct.Builder.html#method.serve_connection).
+    ///
+    /// To drive HTTP on this connection this future **must be polled**, typically with
+    /// `.await`. If it isn't polled, no progress will be made on this connection.
+    #[must_use = "futures do nothing unless polled"]
     pub struct Connection<'a, I, S, E>
@@ -495,3 +506,8 @@ where
 pin_project! {
-    /// Connection future.
+    /// An upgradable [`Connection`], returned by
+    /// [`Builder::serve_upgradable_connection`](struct.Builder.html#method.serve_connection_with_upgrades).
+    ///
+    /// To drive HTTP on this connection this future **must be polled**, typically with
+    /// `.await`. If it isn't polled, no progress will be made on this connection.
+    #[must_use = "futures do nothing unless polled"]
     pub struct UpgradeableConnection<'a, I, S, E>
@@ -498,0 +514,0 @@ where

src/server/graceful.rs

@@ -20,2 +20,3 @@ //! Utility to gracefully shutdown a server.
 /// A graceful shutdown utility
+// Purposefully not `Clone`, see `watcher()` method for why.
 pub struct GracefulShutdown {
@@ -25,2 +26,11 @@ tx: watch::Sender<()>,
 
+/// A watcher side of the graceful shutdown.
+///
+/// This type can only watch a connection, it cannot trigger a shutdown.
+///
+/// Call [`GracefulShutdown::watcher()`] to construct one of these.
+pub struct Watcher {
+    rx: watch::Receiver<()>,
+}
+
 impl GracefulShutdown {
@@ -35,10 +45,18 @@ /// Create a new graceful shutdown helper.
     pub fn watch<C: GracefulConnection>(&self, conn: C) -> impl Future<Output = C::Output> {
-        let mut rx = self.tx.subscribe();
-        GracefulConnectionFuture::new(conn, async move {
-            let _ = rx.changed().await;
-            // hold onto the rx until the watched future is completed
-            rx
-        })
+        self.watcher().watch(conn)
     }
 
+    /// Create an owned type that can watch a connection.
+    ///
+    /// This method allows created an owned type that can be sent onto another
+    /// task before calling [`Watcher::watch()`].
+    // Internal: this function exists because `Clone` allows footguns.
+    // If the `tx` were cloned (or the `rx`), race conditions can happens where
+    // one task starting a shutdown is scheduled and interwined with a task
+    // starting to watch a connection, and the "watch version" is one behind.
+    pub fn watcher(&self) -> Watcher {
+        let rx = self.tx.subscribe();
+        Watcher { rx }
+    }
+
     /// Signal shutdown for all watched connections.
@@ -56,2 +74,7 @@ ///
     }
+
+    /// Returns the number of the watching connections.
+    pub fn count(&self) -> usize {
+        self.tx.receiver_count()
+    }
 }
@@ -71,2 +94,20 @@
 
+impl Watcher {
+    /// Wrap a future for graceful shutdown watching.
+    pub fn watch<C: GracefulConnection>(self, conn: C) -> impl Future<Output = C::Output> {
+        let Watcher { mut rx } = self;
+        GracefulConnectionFuture::new(conn, async move {
+            let _ = rx.changed().await;
+            // hold onto the rx until the watched future is completed
+            rx
+        })
+    }
+}
+
+impl Debug for Watcher {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("GracefulWatcher").finish()
+    }
+}
+
 pin_project! {
@@ -73,0 +114,0 @@ struct GracefulConnectionFuture<C, F: Future> {

src/service/glue.rs

@@ -10,3 +10,11 @@ use pin_project_lite::pin_project;
 
-/// A tower service converted into a hyper service.
+/// A tower [`Service`][tower-svc] converted into a hyper [`Service`][hyper-svc].
+///
+/// This wraps an inner tower service `S` in a [`hyper::service::Service`] implementation. See
+/// the module-level documentation of [`service`][crate::service] for more information about using
+/// [`tower`][tower] services and middleware with [`hyper`].
+///
+/// [hyper-svc]: hyper::service::Service
+/// [tower]: https://docs.rs/tower/latest/tower/
+/// [tower-svc]: https://docs.rs/tower/latest/tower/trait.Service.html
 #[derive(Debug, Copy, Clone)]
@@ -18,3 +26,3 @@ pub struct TowerToHyperService<S> {
 impl<S> TowerToHyperService<S> {
-    /// Create a new `TowerToHyperService` from a tower service.
+    /// Create a new [`TowerToHyperService`] from a tower service.
     pub fn new(tower_service: S) -> Self {
@@ -44,2 +52,5 @@ Self {
     /// Response future for [`TowerToHyperService`].
+    ///
+    /// This future is acquired by [`call`][hyper::service::Service::call]ing a
+    /// [`TowerToHyperService`].
     pub struct TowerToHyperServiceFuture<S, R>
@@ -46,0 +57,0 @@ where

src/service/mod.rs

 //! Service utilities.
+//!
+//! [`hyper::service`] provides a [`Service`][hyper-svc] trait, representing an asynchronous
+//! function from a `Request` to a `Response`. This provides an interface allowing middleware for
+//! network application to be written in a modular and reusable way.
+//!
+//! This submodule provides an assortment of utilities for working with [`Service`][hyper-svc]s.
+//! See the module-level documentation of [`hyper::service`] for more information.
+//!
+//! # Tower
+//!
+//! While [`hyper`] uses its own notion of a [`Service`][hyper-svc] internally, many other
+//! libraries use a library such as [`tower`][tower] to provide the fundamental model of an
+//! asynchronous function.
+//!
+//! The [`TowerToHyperService`] type provided by this submodule can be used to bridge these
+//! ecosystems together. By wrapping a [`tower::Service`][tower-svc] in [`TowerToHyperService`],
+//! it can be passed into [`hyper`] interfaces that expect a [`hyper::service::Service`].
+//!
+//! [hyper-svc]: hyper::service::Service
+//! [tower]: https://docs.rs/tower/latest/tower/
+//! [tower-svc]: https://docs.rs/tower/latest/tower/trait.Service.html
 
@@ -3,0 +24,0 @@ #[cfg(feature = "service")]

No alert changes

No metric changes