Add an online test for an http connect proxy_url

ngrok/Cargo.toml

@@ -56,7 +56,8 @@ windows-sys = { version = "0.45.0", features = ["Win32_Foundation"] }
 anyhow = "1.0.66"
 axum = { version = "0.7.4", features = ["tokio"] }
 flate2 = "1.0.25"
-hyper = { version = "1.1.0" }
+http-body-util = "0.1.3"
+hyper = { version = "1.1.0", features = [ "client" ] }
 hyper-util = { version = "0.1.3", features = [
 	"tokio",
 	"server",
@@ -75,7 +76,6 @@ tower = { version = "0.5", features = ["util"] }
 tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
 tracing-test = "0.2.3"
 
-
 [[example]]
 name = "tls"
 required-features = ["axum"]

ngrok/src/online_tests.rs

@@ -893,6 +893,188 @@ fn tls_client_config() -> Result<Arc<ClientConfig>, &'static io::Error> {
     Ok(CONFIG.as_ref()?.clone())
 }
 
+#[traced_test]
+#[test]
+async fn connect_proxy_http() -> Result<(), BoxError> {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
+    let addr = listener.local_addr()?;
+    let (tx, mut rx) = mpsc::channel::<u64>(1);
+    let shutdown = tokio_util::sync::CancellationToken::new();
+
+    let ln_shutdown = shutdown.clone();
+    tokio::spawn(async move {
+        let res = connect_proxy::run_proxy(listener, ln_shutdown).await;
+        tx.send(res).await.unwrap();
+    });
+
+    let sess = Session::builder()
+        .authtoken_from_env()
+        .proxy_url(format!("http://{addr}").parse().unwrap())
+        .unwrap()
+        .connect()
+        .await?;
+
+    tracing::debug!("{}", sess.id());
+
+    shutdown.cancel();
+    // verify we got a request
+    let conns = rx.recv().await;
+
+    assert_eq!(Some(1), conns);
+    Ok(())
+}
+
+// connect_proxy contains code for connect_proxy tests
+// This code is adapted from https://github.com/hyperium/hyper/blob/c449528a33d266a8ca1210baca11e5d649ca6c27/examples/http_proxy.rs#L37
+// Used under the terms of the MIT license, Copyright (c) 2014-2025 Sean McArthur
+mod connect_proxy {
+    use bytes::Bytes;
+    use http_body_util::{
+        combinators::BoxBody,
+        BodyExt,
+        Empty,
+        Full,
+    };
+    use hyper::{
+        client::conn::http1::Builder,
+        http,
+        server::conn::http1,
+        service::service_fn,
+        upgrade::Upgraded,
+        Method,
+        Request,
+        Response,
+    };
+    use hyper_util::rt::TokioIo;
+    use tokio::net::TcpStream;
+    use tokio_util::sync::CancellationToken;
+
+    pub async fn run_proxy(listener: tokio::net::TcpListener, shutdown: CancellationToken) -> u64 {
+        // count requests so our caller can test that we received a request
+        let mut req_count = 0;
+        loop {
+            let (stream, _) = match shutdown.run_until_cancelled(listener.accept()).await {
+                None => {
+                    return req_count;
+                }
+                Some(r) => r.unwrap(),
+            };
+            let io = TokioIo::new(stream);
+            req_count += 1;
+
+            tokio::task::spawn(async move {
+                if let Err(err) = http1::Builder::new()
+                    .preserve_header_case(true)
+                    .title_case_headers(true)
+                    .serve_connection(io, service_fn(proxy))
+                    .with_upgrades()
+                    .await
+                {
+                    println!("Failed to serve connection: {:?}", err);
+                }
+            });
+        }
+    }
+
+    async fn proxy(
+        req: Request<hyper::body::Incoming>,
+    ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+        println!("req: {:?}", req);
+
+        if Method::CONNECT == req.method() {
+            // Received an HTTP request like:
+            // ```
+            // CONNECT www.domain.com:443 HTTP/1.1
+            // Host: www.domain.com:443
+            // Proxy-Connection: Keep-Alive
+            // ```
+            //
+            // When HTTP method is CONNECT we should return an empty body
+            // then we can eventually upgrade the connection and talk a new protocol.
+            //
+            // Note: only after client received an empty body with STATUS_OK can the
+            // connection be upgraded, so we can't return a response inside
+            // `on_upgrade` future.
+            if let Some(addr) = host_addr(req.uri()) {
+                tokio::task::spawn(async move {
+                    match hyper::upgrade::on(req).await {
+                        Ok(upgraded) => {
+                            if let Err(e) = tunnel(upgraded, addr).await {
+                                eprintln!("server io error: {}", e);
+                            };
+                        }
+                        Err(e) => eprintln!("upgrade error: {}", e),
+                    }
+                });
+
+                Ok(Response::new(empty()))
+            } else {
+                eprintln!("CONNECT host is not socket addr: {:?}", req.uri());
+                let mut resp = Response::new(full("CONNECT must be to a socket address"));
+                *resp.status_mut() = http::StatusCode::BAD_REQUEST;
+
+                Ok(resp)
+            }
+        } else {
+            let host = req.uri().host().expect("uri has no host");
+            let port = req.uri().port_u16().unwrap_or(80);
+
+            let stream = TcpStream::connect((host, port)).await.unwrap();
+            let io = TokioIo::new(stream);
+
+            let (mut sender, conn) = Builder::new()
+                .preserve_header_case(true)
+                .title_case_headers(true)
+                .handshake(io)
+                .await?;
+            tokio::task::spawn(async move {
+                if let Err(err) = conn.await {
+                    println!("Connection failed: {:?}", err);
+                }
+            });
+
+            let resp = sender.send_request(req).await?;
+            Ok(resp.map(|b| b.boxed()))
+        }
+    }
+
+    fn host_addr(uri: &http::Uri) -> Option<String> {
+        uri.authority().map(|auth| auth.to_string())
+    }
+
+    fn empty() -> BoxBody<Bytes, hyper::Error> {
+        Empty::<Bytes>::new()
+            .map_err(|never| match never {})
+            .boxed()
+    }
+
+    fn full<T: Into<Bytes>>(chunk: T) -> BoxBody<Bytes, hyper::Error> {
+        Full::new(chunk.into())
+            .map_err(|never| match never {})
+            .boxed()
+    }
+
+    // Create a TCP connection to host:port, build a tunnel between the connection and
+    // the upgraded connection
+    async fn tunnel(upgraded: Upgraded, addr: String) -> std::io::Result<()> {
+        // Connect to remote server
+        let mut server = TcpStream::connect(addr).await?;
+        let mut upgraded = TokioIo::new(upgraded);
+
+        // Proxying data
+        let (from_client, from_server) =
+            tokio::io::copy_bidirectional(&mut upgraded, &mut server).await?;
+
+        // Print message when done
+        println!(
+            "client wrote {} bytes and received {} bytes",
+            from_client, from_server
+        );
+
+        Ok(())
+    }
+}
+
 #[traced_test]
 #[cfg_attr(not(feature = "paid-tests"), ignore)]
 #[test]