fix(client): parse proxy CONNECT response with httparse

Cargo.toml

@@ -24,6 +24,7 @@ futures-channel = { version = "0.3", optional = true }
 futures-util = { version = "0.3.16", default-features = false, optional = true }
 http = "1.0"
 http-body = "1.0.0"
+httparse = { version = "1", optional = true }
 hyper = "1.9.0"
 ipnet = { version = "2.9", optional = true }
 libc = { version = "0.2", optional = true }
@@ -75,7 +76,7 @@ full = [
 ]
 
 client = ["hyper/client", "dep:tracing", "dep:futures-channel", "dep:tower-service"]
-client-legacy = ["client", "tokio/net", "dep:socket2", "tokio/sync", "dep:libc", "dep:futures-util"]
+client-legacy = ["client", "tokio/net", "dep:socket2", "tokio/sync", "dep:libc", "dep:futures-util", "dep:httparse"]
 client-pool = ["client", "dep:futures-util", "dep:tower-layer", "tokio/sync"]
 client-proxy = ["client", "dep:base64", "dep:ipnet", "dep:percent-encoding"]
 client-proxy-system = ["dep:system-configuration", "dep:windows-registry"]

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

@@ -212,23 +212,26 @@ where
         }
         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);
+        let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
+        let mut res = httparse::Response::new(&mut headers);
+        match res.parse(&buf[..pos]) {
+            Ok(httparse::Status::Complete(_)) => match res.code {
+                Some(200) => return Ok(conn),
+                Some(407) => return Err(TunnelError::ProxyAuthRequired),
+                _ => return Err(TunnelError::TunnelUnsuccessful),
+            },
+            Ok(httparse::Status::Partial) => {
+                if pos == buf.len() {
+                    return Err(TunnelError::ProxyHeadersTooLong);
+                }
             }
-            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);
+            Err(_) => return Err(TunnelError::TunnelUnsuccessful),
         }
     }
 }
 
+const MAX_HEADERS: usize = 100;
+
 impl std::fmt::Display for TunnelError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("tunnel error: ")?;
@@ -254,3 +257,64 @@ impl std::error::Error for TunnelError {
         }
     }
 }
+
+#[cfg(all(test, feature = "tokio"))]
+mod tests {
+    use std::time::Duration;
+
+    use tokio::io::AsyncWriteExt;
+
+    use super::{Headers, TunnelError, tunnel};
+    use crate::rt::TokioIo;
+
+    async fn handshake(response: &'static [u8]) -> Result<(), TunnelError> {
+        let (client, mut server) = tokio::io::duplex(1024);
+        tokio::spawn(async move {
+            server.write_all(response).await.unwrap();
+        });
+
+        tokio::time::timeout(
+            Duration::from_secs(1),
+            tunnel(TokioIo::new(client), "example.com", 443, &Headers::Empty),
+        )
+        .await
+        .expect("handshake should not hang")
+        .map(drop)
+    }
+
+    #[tokio::test]
+    async fn established() {
+        handshake(b"HTTP/1.1 200 Connection established\r\n\r\n")
+            .await
+            .expect("200 response should establish the tunnel");
+    }
+
+    #[tokio::test]
+    async fn established_with_early_data() {
+        handshake(b"HTTP/1.1 200 OK\r\n\r\nHELLO")
+            .await
+            .expect("early data must not prevent establishing the tunnel");
+    }
+
+    #[tokio::test]
+    async fn proxy_auth_required() {
+        let err = handshake(b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n")
+            .await
+            .unwrap_err();
+        assert!(matches!(err, TunnelError::ProxyAuthRequired));
+    }
+
+    #[tokio::test]
+    async fn non_200_is_unsuccessful() {
+        let err = handshake(b"HTTP/1.1 500 Internal Server Error\r\n\r\n")
+            .await
+            .unwrap_err();
+        assert!(matches!(err, TunnelError::TunnelUnsuccessful));
+    }
+
+    #[tokio::test]
+    async fn malformed_status_is_rejected() {
+        let err = handshake(b"HTTP/1.1 2000 OK\r\n\r\n").await.unwrap_err();
+        assert!(matches!(err, TunnelError::TunnelUnsuccessful));
+    }
+}