rama/http/client/

proxy_connector.rs

use crate::utils::macros::define_inner_service_accessors;
use crate::{
    Layer, Service,
    error::{BoxError, BoxErrorExt},
    extensions::{Extensions, ExtensionsRef},
    http::client::proxy::layer::{
        HttpProxyConnector, HttpProxyConnectorLayer, MaybeHttpProxiedConnection,
    },
    io::Io,
    net::{
        AuthorityInputExt, HttpVersionInputExt, Protocol, ProtocolInputExt,
        TargetHttpVersionInputExt,
        client::{
            ConnectionError, ConnectionErrorKind, ConnectorService, EstablishedClientConnection,
            EstablishedProxyRoute, ProxyRoute,
        },
    },
    proxy::socks5::{Socks5ProxyConnector, Socks5ProxyConnectorLayer},
    telemetry::tracing,
};
#[cfg(feature = "tls")]
use crate::{net::client::ConnectionPolicyScope, tls::TlsTunnel};
use pin_project_lite::pin_project;
use std::{
    fmt::Debug,
    pin::Pin,
    task::{self, Poll},
};
use tokio::io::{AsyncRead, AsyncWrite};

/// Proxy connector which supports http(s), masque and socks5(h) proxy address
///
/// Connector will look at [`ProxyRoute`] to determine which proxy
/// connector to use if one is configured
#[derive(Debug, Clone, ExtensionsRef)]
pub struct ProxyConnector<S> {
    inner: S,
    socks: Socks5ProxyConnector<S>,
    http: HttpProxyConnector<S>,
    required: bool,
}

impl<S: Clone> ProxyConnector<S> {
    define_inner_service_accessors!();

    /// Creates a new [`ProxyConnector`].
    fn new(
        inner: S,
        socks_proxy_layer: Socks5ProxyConnectorLayer,
        http_proxy_layer: HttpProxyConnectorLayer,
        required: bool,
    ) -> Self {
        Self {
            socks: socks_proxy_layer.into_layer(inner.clone()),
            http: http_proxy_layer.into_layer(inner.clone()),
            inner,
            required,
        }
    }

    #[inline]
    /// Creates a new required [`ProxyConnector`].
    ///
    /// This connector will fail unless a proxied [`ProxyRoute`] is configured.
    pub fn required(
        inner: S,
        socks_proxy_layer: Socks5ProxyConnectorLayer,
        http_proxy_layer: HttpProxyConnectorLayer,
    ) -> Self {
        Self::new(inner, socks_proxy_layer, http_proxy_layer, true)
    }

    #[inline]
    /// Creates a new optional [`ProxyConnector`].
    ///
    /// This connector will forward to the inner connector for a direct or missing [`ProxyRoute`].
    pub fn optional(
        inner: S,
        socks_proxy_layer: Socks5ProxyConnectorLayer,
        http_proxy_layer: HttpProxyConnectorLayer,
    ) -> Self {
        Self::new(inner, socks_proxy_layer, http_proxy_layer, false)
    }
}

impl<Input, S> Service<Input> for ProxyConnector<S>
where
    S: ConnectorService<Input, Connection: Io + Unpin>,
    Input: AuthorityInputExt
        + ProtocolInputExt
        + HttpVersionInputExt
        + TargetHttpVersionInputExt
        + Send
        + ExtensionsRef
        + 'static,
{
    type Output = EstablishedClientConnection<MaybeProxiedConnection<S::Connection>, Input>;
    type Error = ConnectionError;

    async fn serve(&self, input: Input) -> Result<Self::Output, Self::Error> {
        let route = input.extensions().get_ref::<ProxyRoute>();
        let route_requested = route.is_some();

        match route {
            None | Some(ProxyRoute::Direct) => {
                if self.required {
                    return Err(ConnectionError::local(
                        BoxError::from_static_str("proxy required but none is defined"),
                        ConnectionErrorKind::InvalidInput,
                    ));
                }
                tracing::trace!("no proxy detected in ctx, using inner connector");
                let EstablishedClientConnection { input, conn } = self.inner.connect(input).await?;

                if route_requested {
                    conn.extensions().insert(EstablishedProxyRoute::Direct);
                }

                let conn = MaybeProxiedConnection::direct(conn);
                Ok(EstablishedClientConnection { input, conn })
            }
            Some(ProxyRoute::Proxy(proxy)) => {
                let protocol = proxy.protocol.as_ref();
                tracing::trace!(?protocol, "proxy detected in ctx");

                let protocol = protocol.unwrap_or_else(|| {
                    tracing::trace!("no protocol detected, using http as protocol");
                    &Protocol::HTTP
                });

                if protocol.is_socks5() {
                    tracing::trace!(
                        target = %&proxy.address,
                        "using socks proxy connector",
                    );

                    // SOCKS does not supply a tunnel TLS policy of its own.
                    // An explicit request policy must not poison shared routes.
                    #[cfg(feature = "tls")]
                    let request_tunnel = input.extensions().contains::<TlsTunnel>();
                    let established = self.socks.connect(input).await;
                    #[cfg(feature = "tls")]
                    let established = established.map_err(|error| {
                        if request_tunnel {
                            error.with_policy_scope(ConnectionPolicyScope::Request)
                        } else {
                            error
                        }
                    });
                    let EstablishedClientConnection { input, conn } = established?;

                    let conn = MaybeProxiedConnection::socks(conn);
                    Ok(EstablishedClientConnection { input, conn })
                } else if protocol.is_http() || protocol.is_masque() {
                    // A MASQUE proxy is an HTTPS proxy for TCP, and receives its own requests
                    // as a forward proxy; see `ProxyRoute::addresses_proxy`.
                    tracing::trace!(
                        target = %&proxy.address,
                        "using http proxy connector"
                    );

                    let EstablishedClientConnection { input, conn } =
                        self.http.connect(input).await?;

                    let conn = MaybeProxiedConnection::http(conn);
                    Ok(EstablishedClientConnection { input, conn })
                } else {
                    Err(ConnectionError::transport(
                        BoxError::from_static_str("received unsupported proxy protocol"),
                        ConnectionErrorKind::Protocol,
                    )
                    .context_debug_field("protocol", protocol.clone()))
                }
            }
        }
    }
}

pin_project! {
    /// A connection which will be proxied if a proxied [`ProxyRoute`] was configured.
    pub struct MaybeProxiedConnection<S> {
        #[pin]
        inner: Connection<S>,
    }
}

impl<S: ExtensionsRef> MaybeProxiedConnection<S> {
    pub fn direct(conn: S) -> Self {
        Self {
            inner: Connection::Direct { conn },
        }
    }

    pub fn socks(conn: S) -> Self {
        Self {
            inner: Connection::Socks { conn },
        }
    }

    pub fn http(conn: MaybeHttpProxiedConnection<S>) -> Self {
        Self {
            inner: Connection::Http { conn },
        }
    }
}

impl<S: Debug> Debug for MaybeProxiedConnection<S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MaybeProxiedConnection")
            .field("inner", &self.inner)
            .finish()
    }
}

impl<S: ExtensionsRef> ExtensionsRef for MaybeProxiedConnection<S> {
    fn extensions(&self) -> &Extensions {
        match &self.inner {
            Connection::Direct { conn } | Connection::Socks { conn } => conn.extensions(),
            Connection::Http { conn } => conn.extensions(),
        }
    }
}

pin_project! {
    #[project = ConnectionProj]
    enum Connection<S> {
        Direct{ #[pin] conn: S },
        Socks{ #[pin] conn: S },
        Http{ #[pin] conn: MaybeHttpProxiedConnection<S> },

    }
}

impl<S: Debug> Debug for Connection<S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Direct { conn } => f.debug_struct("Direct").field("conn", conn).finish(),
            Self::Socks { conn } => f.debug_struct("Socks").field("conn", conn).finish(),
            Self::Http { conn } => f.debug_struct("Http").field("conn", conn).finish(),
        }
    }
}

#[warn(clippy::missing_trait_methods)]
impl<Conn: AsyncWrite> AsyncWrite for MaybeProxiedConnection<Conn> {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, std::io::Error>> {
        match self.project().inner.project() {
            ConnectionProj::Direct { conn } | ConnectionProj::Socks { conn } => {
                conn.poll_write(cx, buf)
            }
            ConnectionProj::Http { conn } => conn.poll_write(cx, buf),
        }
    }

    fn poll_flush(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
    ) -> Poll<Result<(), std::io::Error>> {
        match self.project().inner.project() {
            ConnectionProj::Direct { conn } | ConnectionProj::Socks { conn } => conn.poll_flush(cx),
            ConnectionProj::Http { conn } => conn.poll_flush(cx),
        }
    }

    fn poll_shutdown(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
    ) -> Poll<Result<(), std::io::Error>> {
        match self.project().inner.project() {
            ConnectionProj::Direct { conn } | ConnectionProj::Socks { conn } => {
                conn.poll_shutdown(cx)
            }
            ConnectionProj::Http { conn } => conn.poll_shutdown(cx),
        }
    }

    fn is_write_vectored(&self) -> bool {
        match &self.inner {
            Connection::Direct { conn } | Connection::Socks { conn } => conn.is_write_vectored(),
            Connection::Http { conn } => conn.is_write_vectored(),
        }
    }

    fn poll_write_vectored(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
        bufs: &[std::io::IoSlice<'_>],
    ) -> Poll<Result<usize, std::io::Error>> {
        match self.project().inner.project() {
            ConnectionProj::Direct { conn } | ConnectionProj::Socks { conn } => {
                conn.poll_write_vectored(cx, bufs)
            }
            ConnectionProj::Http { conn } => conn.poll_write_vectored(cx, bufs),
        }
    }
}

#[warn(clippy::missing_trait_methods)]
impl<Conn: AsyncRead> AsyncRead for MaybeProxiedConnection<Conn> {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        match self.project().inner.project() {
            ConnectionProj::Direct { conn } | ConnectionProj::Socks { conn } => {
                conn.poll_read(cx, buf)
            }
            ConnectionProj::Http { conn } => conn.poll_read(cx, buf),
        }
    }
}

/// Proxy connector layer which supports http(s), masque and socks5(h) proxy address
///
/// Connector will look at [`ProxyRoute`] to determine which proxy
/// connector to use if one is configured
pub struct ProxyConnectorLayer {
    socks_layer: Socks5ProxyConnectorLayer,
    http_layer: HttpProxyConnectorLayer,
    required: bool,
}

impl ProxyConnectorLayer {
    #[must_use]
    /// Creates a new required [`ProxyConnectorLayer`].
    ///
    /// This connector will fail unless a proxied [`ProxyRoute`] is configured.
    pub fn required(
        socks_proxy_layer: Socks5ProxyConnectorLayer,
        http_proxy_layer: HttpProxyConnectorLayer,
    ) -> Self {
        Self {
            socks_layer: socks_proxy_layer,
            http_layer: http_proxy_layer,
            required: true,
        }
    }

    #[must_use]
    /// Creates a new optional [`ProxyConnectorLayer`].
    ///
    /// This connector will forward to the inner connector for a direct or missing [`ProxyRoute`].
    pub fn optional(
        socks_proxy_layer: Socks5ProxyConnectorLayer,
        http_proxy_layer: HttpProxyConnectorLayer,
    ) -> Self {
        Self {
            socks_layer: socks_proxy_layer,
            http_layer: http_proxy_layer,
            required: false,
        }
    }
}

impl<S: Clone> Layer<S> for ProxyConnectorLayer {
    type Service = ProxyConnector<S>;

    fn layer(&self, inner: S) -> Self::Service {
        ProxyConnector::new(
            inner,
            self.socks_layer.clone(),
            self.http_layer.clone(),
            self.required,
        )
    }

    fn into_layer(self, inner: S) -> Self::Service {
        ProxyConnector::new(inner, self.socks_layer, self.http_layer, self.required)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        net::{proxy::IoForwardService, test_utils::client::MockSocket},
        proxy::socks5::{
            Socks5ProxyConnectorLayer,
            server::{Connector as EagerSocks5Connector, Socks5Connector},
        },
        tcp::client::service::TcpConnector,
    };

    #[cfg(feature = "tls")]
    use {
        crate::{
            net::{
                address::HostWithPort,
                address::ProxyAddress,
                client::{
                    ConnectRequest, EstablishedProxyRoute, ProxyRouteFailureCache,
                    ProxyRouteFailureCacheConfig, ProxyRouteFailureCacheConnector,
                    ProxyRouteFailureCacheScope,
                },
            },
            service::service_fn,
        },
        std::sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
    };

    fn assert_socks5_connector<S, C: Socks5Connector<S>>(_: &C) {}

    #[cfg(feature = "tls")]
    #[tokio::test]
    async fn a_masque_proxy_receives_its_own_requests_directly_as_forwarded() {
        let dialed = Arc::new(parking_lot::Mutex::new(None));
        let transport = service_fn({
            let dialed = dialed.clone();
            move |input: ConnectRequest| {
                *dialed.lock() = Some(input.authority.clone());
                async move {
                    let (stream, _peer) = tokio::io::duplex(64);
                    Ok::<_, ConnectionError>(EstablishedClientConnection {
                        input,
                        conn: MockSocket::new(stream),
                    })
                }
            }
        });
        let connector = ProxyConnector::required(
            transport,
            Socks5ProxyConnectorLayer::required(),
            HttpProxyConnectorLayer::required(),
        );
        let proxy: ProxyAddress = "masque://user:secret@proxy.example:443".parse().unwrap();
        let input = ConnectRequest::new(HostWithPort::try_from("proxy.example:443").unwrap())
            .with_application_protocol(Protocol::HTTPS);
        input.extensions().insert(ProxyRoute::Proxy(proxy.clone()));
        let established = connector.serve(input).await.unwrap();
        assert_eq!(
            dialed.lock().take().unwrap(),
            HostWithPort::try_from("proxy.example:443").unwrap()
        );
        assert_eq!(
            established
                .conn
                .extensions()
                .get_ref::<EstablishedProxyRoute>(),
            Some(&EstablishedProxyRoute::Forward(proxy))
        );
    }

    #[cfg(feature = "tls")]
    #[tokio::test]
    async fn request_tunnel_failure_does_not_poison_socks_proxy_route() {
        for scope in [
            ProxyRouteFailureCacheScope::PerDestination,
            ProxyRouteFailureCacheScope::PerProxy,
        ] {
            let attempts = Arc::new(AtomicUsize::new(0));
            let transport = service_fn({
                let attempts = attempts.clone();
                move |input: ConnectRequest| {
                    attempts.fetch_add(1, Ordering::Relaxed);
                    async move {
                        Err::<EstablishedClientConnection<MockSocket, ConnectRequest>, _>(
                            ConnectionError::transport(
                                BoxError::from_static_str("scripted proxy failure"),
                                if input.extensions().contains::<TlsTunnel>() {
                                    ConnectionErrorKind::Protocol
                                } else {
                                    ConnectionErrorKind::Unavailable
                                },
                            ),
                        )
                    }
                }
            });
            let mut config = ProxyRouteFailureCacheConfig::default();
            config.scope = scope;
            let connector = ProxyRouteFailureCacheConnector::new(
                ProxyConnector::required(
                    transport,
                    Socks5ProxyConnectorLayer::required(),
                    HttpProxyConnectorLayer::required(),
                ),
                ProxyRouteFailureCache::try_new(config).unwrap(),
            );
            let request = || {
                let input = ConnectRequest::new(HostWithPort::example_domain_http())
                    .with_application_protocol(Protocol::HTTP);
                input.extensions().insert(ProxyRoute::Proxy(
                    "socks5://127.0.0.1:1080".parse().unwrap(),
                ));
                input
            };
            let customized = request();
            customized.extensions().insert(TlsTunnel {
                server_identity: Some("proxy.example".parse().unwrap()),
                application_protocol: None,
                alpn: None,
            });
            let error = connector.serve(customized).await.unwrap_err();
            assert_eq!(error.kind(), ConnectionErrorKind::Protocol);
            assert_eq!(error.policy_scope(), ConnectionPolicyScope::Request);

            let error = connector.serve(request()).await.unwrap_err();
            assert_eq!(error.kind(), ConnectionErrorKind::Unavailable);
            assert_eq!(error.policy_scope(), ConnectionPolicyScope::Unknown);
            assert_eq!(attempts.load(Ordering::Relaxed), 2);
            _ = connector.serve(request()).await.unwrap_err();
            assert_eq!(
                attempts.load(Ordering::Relaxed),
                2,
                "ordinary failures remain cached"
            );
        }
    }

    #[test]
    fn eager_socks5_accepts_combined_proxy_connection() {
        let proxy_connector = ProxyConnectorLayer::optional(
            Socks5ProxyConnectorLayer::optional(),
            HttpProxyConnectorLayer::optional(),
        )
        .into_layer(TcpConnector::new());
        let connector = EagerSocks5Connector::new(proxy_connector, IoForwardService::default());

        assert_socks5_connector::<MockSocket, _>(&connector);
    }
}