vey-proxy: add http connect-udp support to proxy_http escapers

Author: zh-jq

lib/vey-http/src/connect/client.rs

@@ -17,7 +17,7 @@ pub async fn http_connect_to<S>(
 where
     S: AsyncBufRead + AsyncWrite + Unpin,
 {
-    let mut req = HttpConnectRequest::new(addr, &[]);
+    let mut req = HttpConnectRequest::new(&[]);
 
     match auth {
         HttpAuth::None => {}
@@ -27,7 +27,7 @@ where
         }
     }
 
-    req.send(buf_stream)
+    req.send(addr, buf_stream)
         .await
         .map_err(HttpConnectError::WriteFailed)?;
 
@@ -50,7 +50,6 @@ mod tests {
     async fn http_connect_to_success_no_auth() {
         let stream = Builder::new()
             .write(b"CONNECT example.com:8080 HTTP/1.1\r\n")
-            .write(b"Host: example.com:8080\r\n")
             .write(b"Connection: keep-alive\r\n")
             .write(b"\r\n")
             .read(b"HTTP/1.1 200 OK\r\n\r\n")
@@ -68,7 +67,6 @@ mod tests {
     async fn http_connect_to_success_basic_auth() {
         let stream = Builder::new()
             .write(b"CONNECT example.com:8080 HTTP/1.1\r\n")
-            .write(b"Host: example.com:8080\r\n")
             .write(b"Connection: keep-alive\r\n")
             .write(b"Proxy-Authorization: Basic dXNlcjpwYXNz\r\n")
             .write(b"\r\n")
@@ -108,7 +106,6 @@ mod tests {
     async fn http_connect_to_read_failed() {
         let stream = Builder::new()
             .write(b"CONNECT example.com:8080 HTTP/1.1\r\n")
-            .write(b"Host: example.com:8080\r\n")
             .write(b"Connection: keep-alive\r\n")
             .write(b"\r\n")
             .read_error(io::Error::new(
@@ -131,7 +128,6 @@ mod tests {
     async fn http_connect_to_unexpected_status_code() {
         let stream = Builder::new()
             .write(b"CONNECT example.com:8080 HTTP/1.1\r\n")
-            .write(b"Host: example.com:8080\r\n")
             .write(b"Connection: keep-alive\r\n")
             .write(b"\r\n")
             .read(b"HTTP/1.1 404 Not Found\r\n\r\n")

lib/vey-http/src/connect/request.rs

@@ -1,25 +1,24 @@
 /*
  * SPDX-License-Identifier: Apache-2.0
  * SPDX-FileCopyrightText: 2023-2025 ByteDance and/or its affiliates.
+ * SPDX-FileCopyrightText: 2026 VEY-OSS Developers.
  */
 
-use std::io;
+use std::io::{self, Write};
 
-use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter};
+use tokio::io::{AsyncWrite, AsyncWriteExt};
 
 use vey_types::net::UpstreamAddr;
 
 /// the extra header lines should end with \r\n
 pub struct HttpConnectRequest<'a> {
-    host: &'a UpstreamAddr,
     static_headers: &'a [String],
     dyn_headers: Vec<String>,
 }
 
 impl<'a> HttpConnectRequest<'a> {
-    pub fn new(host: &'a UpstreamAddr, static_headers: &'a [String]) -> Self {
+    pub fn new(static_headers: &'a [String]) -> Self {
         HttpConnectRequest {
-            host,
             static_headers,
             dyn_headers: Vec::new(),
         }
@@ -30,28 +29,23 @@ impl<'a> HttpConnectRequest<'a> {
         self.dyn_headers.push(line);
     }
 
-    /// the extra header lines should end with \r\n
-    pub async fn send<W>(&'a self, writer: &mut W) -> io::Result<()>
+    pub async fn send<W>(&'a self, target: &UpstreamAddr, writer: &mut W) -> io::Result<()>
     where
         W: AsyncWrite + Unpin,
     {
-        let mut buf_writer = BufWriter::new(writer);
-        buf_writer
-            .write_all(format!("CONNECT {} HTTP/1.1\r\n", self.host).as_bytes())
-            .await?;
-        buf_writer
-            .write_all(format!("Host: {}\r\n", self.host).as_bytes())
-            .await?;
-        buf_writer.write_all(b"Connection: keep-alive\r\n").await?;
+        let mut buf = Vec::with_capacity(256);
+        write!(&mut buf, "CONNECT {target} HTTP/1.1\r\n")?;
+        buf.extend_from_slice(b"Connection: keep-alive\r\n");
         for line in self.static_headers {
             debug_assert!(line.ends_with("\r\n"));
-            buf_writer.write_all(line.as_bytes()).await?;
+            buf.extend_from_slice(line.as_bytes());
         }
         for line in &self.dyn_headers {
-            buf_writer.write_all(line.as_bytes()).await?;
+            buf.extend_from_slice(line.as_bytes());
         }
-        buf_writer.write_all(b"\r\n").await?;
-        buf_writer.flush().await
+        buf.extend_from_slice(b"\r\n");
+        writer.write_all(&buf).await?;
+        writer.flush().await
     }
 }
 
@@ -60,166 +54,104 @@ mod tests {
     use super::*;
     use std::str::FromStr;
 
-    // Helper function to create a test UpstreamAddr
-    fn create_test_upstream_addr() -> UpstreamAddr {
+    fn test_upstream_addr() -> UpstreamAddr {
         UpstreamAddr::from_str("example.com:8080").unwrap()
     }
 
+    async fn render_request(request: &HttpConnectRequest<'_>, target: &UpstreamAddr) -> String {
+        let mut buffer = Vec::new();
+        request.send(target, &mut buffer).await.unwrap();
+        String::from_utf8(buffer).unwrap()
+    }
+
     #[test]
-    fn new() {
-        let host = create_test_upstream_addr();
+    fn new_starts_with_static_headers_only() {
         let static_headers = vec![
             "User-Agent: test-agent\r\n".to_string(),
             "Accept: */*\r\n".to_string(),
         ];
 
-        let request = HttpConnectRequest::new(&host, &static_headers);
+        let request = HttpConnectRequest::new(&static_headers);
 
-        assert_eq!(request.host.host_str(), "example.com");
-        assert_eq!(request.host.port(), 8080);
         assert_eq!(request.static_headers.len(), 2);
         assert!(request.dyn_headers.is_empty());
     }
 
     #[test]
-    fn append_dyn_header() {
-        let host = create_test_upstream_addr();
+    fn append_dyn_header_preserves_order() {
         let static_headers: Vec<String> = Vec::new();
-        let mut request = HttpConnectRequest::new(&host, &static_headers);
+        let mut request = HttpConnectRequest::new(&static_headers);
 
-        // Appending a single dynamic header
         request.append_dyn_header("X-Custom-Header: value1\r\n".to_string());
-        assert_eq!(request.dyn_headers.len(), 1);
-        assert_eq!(request.dyn_headers[0], "X-Custom-Header: value1\r\n");
-
-        // Appending multiple dynamic headers
         request.append_dyn_header("X-Another-Header: value2\r\n".to_string());
+
         assert_eq!(request.dyn_headers.len(), 2);
+        assert_eq!(request.dyn_headers[0], "X-Custom-Header: value1\r\n");
         assert_eq!(request.dyn_headers[1], "X-Another-Header: value2\r\n");
     }
 
     #[tokio::test]
-    async fn send_no_headers() {
-        let host = create_test_upstream_addr();
+    async fn send_without_extra_headers() {
+        let target = test_upstream_addr();
         let static_headers: Vec<String> = Vec::new();
-        let request = HttpConnectRequest::new(&host, &static_headers);
-
-        let mut buffer = Vec::new();
-        request.send(&mut buffer).await.unwrap();
-
-        let output = String::from_utf8(buffer).unwrap();
-        let expected = "CONNECT example.com:8080 HTTP/1.1\r\n\
-                       Host: example.com:8080\r\n\
-                       Connection: keep-alive\r\n\
-                       \r\n";
-
-        assert_eq!(output, expected);
+        let request = HttpConnectRequest::new(&static_headers);
+
+        assert_eq!(
+            render_request(&request, &target).await,
+            "CONNECT example.com:8080 HTTP/1.1\r\n\
+             Connection: keep-alive\r\n\
+             \r\n"
+        );
     }
 
     #[tokio::test]
-    async fn send_with_static_headers() {
-        let host = create_test_upstream_addr();
+    async fn send_with_static_and_dynamic_headers() {
+        let target = test_upstream_addr();
         let static_headers = vec![
             "User-Agent: test-agent\r\n".to_string(),
             "Accept: */*\r\n".to_string(),
         ];
-
-        let request = HttpConnectRequest::new(&host, &static_headers);
-
-        let mut buffer = Vec::new();
-        request.send(&mut buffer).await.unwrap();
-
-        let output = String::from_utf8(buffer).unwrap();
-        let expected = "CONNECT example.com:8080 HTTP/1.1\r\n\
-                       Host: example.com:8080\r\n\
-                       Connection: keep-alive\r\n\
-                       User-Agent: test-agent\r\n\
-                       Accept: */*\r\n\
-                       \r\n";
-
-        assert_eq!(output, expected);
-    }
-
-    #[tokio::test]
-    async fn send_with_dynamic_headers() {
-        let host = create_test_upstream_addr();
-        let static_headers: Vec<String> = Vec::new();
-        let mut request = HttpConnectRequest::new(&host, &static_headers);
+        let mut request = HttpConnectRequest::new(&static_headers);
 
         request.append_dyn_header("X-Custom-Header: value1\r\n".to_string());
         request.append_dyn_header("X-Another-Header: value2\r\n".to_string());
 
-        let mut buffer = Vec::new();
-        request.send(&mut buffer).await.unwrap();
-
-        let output = String::from_utf8(buffer).unwrap();
-        let expected = "CONNECT example.com:8080 HTTP/1.1\r\n\
-                       Host: example.com:8080\r\n\
-                       Connection: keep-alive\r\n\
-                       X-Custom-Header: value1\r\n\
-                       X-Another-Header: value2\r\n\
-                       \r\n";
-
-        assert_eq!(output, expected);
-    }
-
-    #[tokio::test]
-    async fn send_with_both_header_types() {
-        let host = create_test_upstream_addr();
-        let static_headers = vec![
-            "User-Agent: test-agent\r\n".to_string(),
-            "Accept: */*\r\n".to_string(),
-        ];
-
-        let mut request = HttpConnectRequest::new(&host, &static_headers);
-        request.append_dyn_header("X-Custom-Header: value1\r\n".to_string());
-        request.append_dyn_header("X-Another-Header: value2\r\n".to_string());
-
-        let mut buffer = Vec::new();
-        request.send(&mut buffer).await.unwrap();
-
-        let output = String::from_utf8(buffer).unwrap();
-        let expected = "CONNECT example.com:8080 HTTP/1.1\r\n\
-                       Host: example.com:8080\r\n\
-                       Connection: keep-alive\r\n\
-                       User-Agent: test-agent\r\n\
-                       Accept: */*\r\n\
-                       X-Custom-Header: value1\r\n\
-                       X-Another-Header: value2\r\n\
-                       \r\n";
-
-        assert_eq!(output, expected);
+        assert_eq!(
+            render_request(&request, &target).await,
+            "CONNECT example.com:8080 HTTP/1.1\r\n\
+             Connection: keep-alive\r\n\
+             User-Agent: test-agent\r\n\
+             Accept: */*\r\n\
+             X-Custom-Header: value1\r\n\
+             X-Another-Header: value2\r\n\
+             \r\n"
+        );
     }
 
     #[tokio::test]
     async fn send_with_ipv6_host() {
-        let host = UpstreamAddr::from_str("[2001:db8::1]:8080").unwrap();
+        let target = UpstreamAddr::from_str("[2001:db8::1]:8080").unwrap();
         let static_headers: Vec<String> = Vec::new();
-        let request = HttpConnectRequest::new(&host, &static_headers);
-
-        let mut buffer = Vec::new();
-        request.send(&mut buffer).await.unwrap();
-
-        let output = String::from_utf8(buffer).unwrap();
-        let expected = "CONNECT [2001:db8::1]:8080 HTTP/1.1\r\n\
-                       Host: [2001:db8::1]:8080\r\n\
-                       Connection: keep-alive\r\n\
-                       \r\n";
-
-        assert_eq!(output, expected);
+        let request = HttpConnectRequest::new(&static_headers);
+
+        assert_eq!(
+            render_request(&request, &target).await,
+            "CONNECT [2001:db8::1]:8080 HTTP/1.1\r\n\
+             Connection: keep-alive\r\n\
+             \r\n"
+        );
     }
 
     #[tokio::test]
-    async fn send_flush_behavior() {
-        let host = create_test_upstream_addr();
+    async fn send_propagates_writer_error() {
+        let target = test_upstream_addr();
         let static_headers: Vec<String> = Vec::new();
-        let request = HttpConnectRequest::new(&host, &static_headers);
-
-        let mut buffer = Vec::new();
-        let result = request.send(&mut buffer).await;
+        let request = HttpConnectRequest::new(&static_headers);
+        let mut writer = tokio_test::io::Builder::new()
+            .write_error(io::Error::new(io::ErrorKind::BrokenPipe, "write failed"))
+            .build();
 
-        assert!(result.is_ok());
-        // Verify that data was actually written to the buffer
-        assert!(!buffer.is_empty());
+        let err = request.send(&target, &mut writer).await.unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::BrokenPipe);
     }
 }

lib/vey-http/src/connect/response.rs

@@ -177,7 +177,7 @@ impl HttpConnectResponse {
         if self.code >= 200 && self.code < 300 {
             Ok(())
         } else if self.code == 504 || self.code == 522 || self.code == 524 {
-            // Peer tells us it timeout
+            // Peer tells us it times out
             Err(HttpConnectError::PeerTimeout(self.code))
         } else {
             Err(HttpConnectError::UnexpectedStatusCode(

lib/vey-http/src/lib.rs

@@ -18,4 +18,5 @@ pub mod client;
 pub mod connect;
 pub mod header;
 pub mod server;
+pub mod upgrade;
 pub mod uri;

lib/vey-http/src/upgrade/error.rs

@@ -0,0 +1,44 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * SPDX-FileCopyrightText: 2026 VEY-OSS Developers.
+ */
+
+use std::io;
+
+use thiserror::Error;
+
+use crate::HttpLineParseError;
+
+#[derive(Debug, Error)]
+pub enum HttpUpgradeResponseError {
+    #[error("too large header, should be less than {0}")]
+    TooLargeHeader(usize),
+    #[error("invalid status line: {0}")]
+    InvalidStatusLine(HttpLineParseError),
+    #[error("invalid header line: {0}")]
+    InvalidHeaderLine(HttpLineParseError),
+    #[error("unsupported value in header {0}")]
+    UnsupportedHeaderValue(&'static str),
+    #[error("upgrade token not match")]
+    UpgradeTokenNotMatch,
+    #[error("invalid chunked transfer-encoding")]
+    InvalidChunkedTransferEncoding,
+    #[error("invalid content length")]
+    InvalidContentLength,
+}
+
+#[derive(Debug, Error)]
+pub enum HttpUpgradeError {
+    #[error("remote closed")]
+    RemoteClosed,
+    #[error("read failed: {0:?}")]
+    ReadFailed(io::Error),
+    #[error("write failed: {0:?}")]
+    WriteFailed(io::Error),
+    #[error("invalid response: {0}")]
+    InvalidResponse(#[from] HttpUpgradeResponseError),
+    #[error("unexpected status code {0} {1}")]
+    UnexpectedStatusCode(u16, String),
+    #[error("peer timeout with status code {0}")]
+    PeerTimeout(u16),
+}