feat(client): add proxy::Tunnel legacy util

src/client/legacy/connect/mod.rs

@@ -74,6 +74,8 @@ pub mod dns;
 #[cfg(feature = "tokio")]
 mod http;
 
+pub mod proxy;
+
 pub(crate) mod capture;
 pub use capture::{capture_connection, CaptureConnection};
 

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

@@ -0,0 +1,5 @@
+//! Proxy helpers
+
+mod tunnel;
+
+pub use self::tunnel::Tunnel;

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

@@ -0,0 +1,181 @@
+use std::future::Future;
+use std::marker::{PhantomData, Unpin};
+use std::pin::Pin;
+use std::task::{self, Poll};
+
+use http::{HeaderValue, Uri};
+use hyper::rt::{Read, Write};
+use pin_project_lite::pin_project;
+use tower_service::Service;
+
+/// Tunnel Proxy via HTTP CONNECT
+#[derive(Debug)]
+pub struct Tunnel<C> {
+ auth: Option<HeaderValue>,
+ inner: C,
+ proxy_dst: Uri,
+}
+
+#[derive(Debug)]
+pub enum TunnelError<C> {
+ Inner(C),
+ 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, E> {
+ #[pin]
+ fut: BoxTunneling<T, E>,
+ _marker: PhantomData<F>,
+ }
+}
+
+type BoxTunneling<T, E> = Pin<Box<dyn Future<Output = Result<T, TunnelError<E>>> + Send>>;
+
+impl<C> Tunnel<C> {
+ /// Create a new Tunnel service.
+ pub fn new(proxy_dst: Uri, connector: C) -> Self {
+ Self {
+ auth: None,
+ 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);
+ self.auth = Some(auth);
+ 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: Send + 'static,
+{
+ type Response = C::Response;
+ type Error = TunnelError<C::Error>;
+ type Future = Tunneling<C::Future, C::Response, C::Error>;
+
+ fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
+ futures_util::ready!(self.inner.poll_ready(cx)).map_err(TunnelError::Inner)?;
+ Poll::Ready(Ok(()))
+ }
+
+ fn call(&mut self, dst: Uri) -> Self::Future {
+ let connecting = self.inner.call(self.proxy_dst.clone());
+
+ Tunneling {
+ fut: Box::pin(async move {
+ let conn = connecting.await.map_err(TunnelError::Inner)?;
+ tunnel(
+ conn,
+ dst.host().ok_or(TunnelError::MissingHost)?,
+ dst.port().map(|p| p.as_u16()).unwrap_or(443),
+ None,
+ None,
+ )
+ .await
+ }),
+ _marker: PhantomData,
+ }
+ }
+}
+
+impl<F, T, E> Future for Tunneling<F, T, E>
+where
+ F: Future<Output = Result<T, E>>,
+{
+ type Output = Result<T, TunnelError<E>>;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
+ self.project().fut.poll(cx)
+ }
+}
+
+async fn tunnel<T, E>(
+ mut conn: T,
+ host: &str,
+ port: u16,
+ user_agent: Option<HeaderValue>,
+ auth: Option<HeaderValue>,
+) -> Result<T, TunnelError<E>>
+where
+ T: Read + Write + Unpin,
+{
+ let mut buf = format!(
+ "\
+ CONNECT {host}:{port} HTTP/1.1\r\n\
+ Host: {host}:{port}\r\n\
+ "
+ )
+ .into_bytes();
+
+ // user-agent
+ if let Some(user_agent) = user_agent {
+ buf.extend_from_slice(b"User-Agent: ");
+ buf.extend_from_slice(user_agent.as_bytes());
+ buf.extend_from_slice(b"\r\n");
+ }
+
+ // proxy-authorization
+ if let Some(value) = auth {
+ //log::debug!("tunnel to {host}:{port} using basic auth");
+ buf.extend_from_slice(b"Proxy-Authorization: ");
+ buf.extend_from_slice(value.as_bytes());
+ buf.extend_from_slice(b"\r\n");
+ }
+
+ // 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);
+ }
+ }
+}

src/rt/io.rs

@@ -0,0 +1,33 @@
+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
+}

src/rt/mod.rs

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

tests/proxy.rs

@@ -0,0 +1,36 @@
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+use tower_service::Service;
+
+use hyper_util::client::legacy::connect::{proxy::Tunnel, HttpConnector};
+
+#[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");
+}