fix(client): don't reject CONNECT tunnel responses split across reads

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

@@ -224,12 +224,33 @@ where
         // else read more
         } else if recvd.starts_with(b"HTTP/1.1 407") {
             return Err(TunnelError::ProxyAuthRequired);
+        } else if is_prefix_of_expected_status_line(recvd) {
+            // Not enough bytes have arrived yet to tell which (if any) of the
+            // expected status lines this is going to be. This happens when
+            // the proxy's response is split across multiple reads (e.g. the
+            // status line arrives in more than one TCP segment). Keep
+            // reading instead of bailing out on a partial match.
+            if pos == buf.len() {
+                return Err(TunnelError::ProxyHeadersTooLong);
+            }
         } else {
             return Err(TunnelError::TunnelUnsuccessful);
         }
     }
 }
 
+/// Returns true if `recvd` (the bytes read so far) could still turn into one
+/// of the status lines we recognize once more bytes arrive, i.e. `recvd` is
+/// a prefix of that status line. This lets the caller distinguish "not
+/// enough bytes yet" from "this is never going to match".
+fn is_prefix_of_expected_status_line(recvd: &[u8]) -> bool {
+    const EXPECTED: [&[u8]; 3] = [b"HTTP/1.1 200", b"HTTP/1.0 200", b"HTTP/1.1 407"];
+
+    EXPECTED
+        .iter()
+        .any(|status_line| recvd.len() < status_line.len() && status_line.starts_with(recvd))
+}
+
 impl std::fmt::Display for TunnelError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("tunnel error: ")?;

tests/proxy.rs

@@ -39,6 +39,87 @@ async fn test_tunnel_works() {
     t2.await.expect("task 2");
 }
 
+// REPRO: proxy sends the "200 OK" status line and header terminator in two
+// separate writes (simulating two TCP segments arriving as separate reads).
+// This should still succeed since the full response is well-formed HTTP,
+// just split across reads.
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_tunnel_works_with_split_response() {
+    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"
+        );
+        // Write the response in two chunks with a flush+delay between them
+        // so the client's `read()` calls return the bytes as two separate
+        // reads instead of one.
+        io.write_all(b"HTTP/1.1 2").await.expect("write 1a");
+        io.flush().await.expect("flush 1a");
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        io.write_all(b"00 OK\r\n\r\n").await.expect("write 1b");
+    });
+
+    t1.await.expect("task 1");
+    t2.await.expect("task 2");
+}
+
+// A genuinely bad response, also split across reads, should still be
+// rejected (and not be mistaken for "not enough bytes yet").
+#[cfg(not(miri))]
+#[tokio::test]
+async fn test_tunnel_fails_on_split_bad_response() {
+    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 err = connector
+            .call("https://hyper.rs".parse().unwrap())
+            .await
+            .expect_err("tunnel should fail");
+        assert_eq!(err.to_string(), "tunnel error: unsuccessful");
+    });
+
+    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"
+        );
+        // "HTTP/1.1 500" is not a prefix of any accepted status line, so
+        // this must be rejected as soon as enough bytes have arrived to
+        // tell, even though it's still split across two reads.
+        io.write_all(b"HTTP/1.1 5").await.expect("write 1a");
+        io.flush().await.expect("flush 1a");
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        io.write_all(b"00 Internal Server Error\r\n\r\n")
+            .await
+            .expect("write 1b");
+    });
+
+    t1.await.expect("task 1");
+    t2.await.expect("task 2");
+}
+
 #[cfg(not(miri))]
 #[tokio::test]
 async fn test_socks_v5_without_auth_works() {