Reject duplicate Host headers in the network proxy (#53506)

## Why

Policy evaluation and upstream parsers must not select different `Host` field lines from the same request.

## What changed

Return `400 Bad Request` for multiple `Host` headers, including identical values, before policy evaluation, upstream forwarding, or Unix socket routing. Apply the check to both plain HTTP requests and `CONNECT` requests.

## Testing

Add regression tests covering identical and conflicting headers in origin-form, absolute-form, and `CONNECT` requests, verifying rejection without an upstream connection. Also cover single-header forwarding and rejection before Unix socket routing.

GitOrigin-RevId: 8ec95171e120f784629d3121b9a1bedbaf88e7cd

codex-rs/network-proxy/src/http_proxy.rs

@@ -190,6 +190,12 @@ async fn http_connect_accept(
     environment_id: Option<String>,
     mut req: Request,
 ) -> Result<(Response, Request), Response> {
+    if req.headers().get_all(header::HOST).iter().nth(1).is_some() {
+        return Err(text_response(
+            StatusCode::BAD_REQUEST,
+            "multiple Host headers",
+        ));
+    }
     let started_at = Instant::now();
     let app_state = req
         .extensions()
@@ -559,6 +565,13 @@ async fn http_plain_proxy(
     environment_id: Option<String>,
     mut req: Request,
 ) -> Result<Response, Infallible> {
+    // Policy and upstream parsers must not select different Host field lines.
+    if req.headers().get_all(header::HOST).iter().nth(1).is_some() {
+        return Ok(text_response(
+            StatusCode::BAD_REQUEST,
+            "multiple Host headers",
+        ));
+    }
     let started_at = Instant::now();
     let app_state = match req.extensions().get::<Arc<NetworkProxyState>>().cloned() {
         Some(state) => state,
@@ -1261,6 +1274,10 @@ struct BlockedResponse<'a> {
     message: Option<String>,
 }
 
+#[cfg(test)]
+#[path = "http_proxy/host_header_tests.rs"]
+mod host_header_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;

codex-rs/network-proxy/src/http_proxy/host_header_tests.rs

@@ -0,0 +1,168 @@
+//! Checks Host field cardinality before policy evaluation and upstream forwarding.
+
+use super::http_plain_proxy;
+use crate::NetworkProxy;
+use crate::NetworkProxyConfig;
+use crate::runtime::network_proxy_state_for_policy;
+use pretty_assertions::assert_eq;
+use rama_core::extensions::ExtensionsMut;
+use rama_http::Body;
+use rama_http::Request;
+use rama_http::StatusCode;
+use std::net::Ipv4Addr;
+use std::sync::Arc;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpListener;
+use tokio::net::TcpStream;
+use tokio::sync::oneshot;
+use tokio::time::Duration;
+use tokio::time::timeout;
+
+#[derive(Clone, Copy, Debug)]
+enum RequestTarget {
+    Origin,
+    Absolute,
+    Connect,
+}
+
+#[derive(Clone, Copy, Debug)]
+enum HostFields {
+    Single,
+    Identical,
+    Conflicting,
+}
+
+async fn exchange(target: RequestTarget, hosts: HostFields) -> (String, Option<String>) {
+    let origin = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+    let origin_addr = origin.local_addr().unwrap();
+    let (stop_origin, stop_requested) = oneshot::channel::<()>();
+    let origin_task = tokio::spawn(async move {
+        let (mut stream, _) = tokio::select! {
+            biased;
+            accepted = origin.accept() => accepted.unwrap(),
+            _ = stop_requested => return None,
+        };
+        if matches!(target, RequestTarget::Connect) {
+            stream.write_all(b"fixture tunnel\r\n").await.unwrap();
+            return Some("CONNECT tunnel established".to_string());
+        }
+        let mut headers = Vec::new();
+        loop {
+            let byte = stream.read_u8().await.unwrap();
+            headers.push(byte);
+            assert!(headers.len() < 8192);
+            if headers.ends_with(b"\r\n\r\n") {
+                break;
+            }
+        }
+        stream
+            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nOK")
+            .await
+            .unwrap();
+        Some(String::from_utf8(headers).unwrap())
+    });
+
+    let mut network = NetworkProxyConfig {
+        allow_local_binding: Some(true),
+        allow_upstream_proxy: false,
+        enable_socks5: false,
+        ..NetworkProxyConfig::default()
+    };
+    network.set_allowed_domains(vec!["127.0.0.1".to_string()]);
+    network.set_denied_domains(vec!["localhost".to_string()]);
+    let state = Arc::new(network_proxy_state_for_policy(network));
+    let proxy = NetworkProxy::builder()
+        .state(state)
+        .managed_proxy_routing(crate::ManagedProxyRouting::DedicatedListeners)
+        .build()
+        .await
+        .unwrap();
+    let handle = proxy.run().await.unwrap();
+    let mut client = TcpStream::connect(proxy.http_addr()).await.unwrap();
+    let first_host = format!("Host: {origin_addr}\r\n");
+    let second_host = match hosts {
+        HostFields::Single => String::new(),
+        HostFields::Identical => format!("hOsT: {origin_addr}\r\n"),
+        HostFields::Conflicting => format!("hOsT: localhost:{}\r\n", origin_addr.port()),
+    };
+    let request_line = match target {
+        RequestTarget::Origin => "GET / HTTP/1.1".to_string(),
+        RequestTarget::Absolute => format!("GET http://{origin_addr}/ HTTP/1.1"),
+        RequestTarget::Connect => format!("CONNECT {origin_addr} HTTP/1.1"),
+    };
+    let request = format!("{request_line}\r\n{first_host}{second_host}Connection: close\r\n\r\n");
+    client.write_all(request.as_bytes()).await.unwrap();
+    let mut response = Vec::new();
+    timeout(Duration::from_secs(5), client.read_to_end(&mut response))
+        .await
+        .expect("proxy should complete the response")
+        .unwrap();
+    drop(client);
+    let _ = stop_origin.send(());
+    let origin_request = timeout(Duration::from_secs(5), origin_task)
+        .await
+        .expect("origin should stop")
+        .unwrap();
+    timeout(Duration::from_secs(5), handle.shutdown())
+        .await
+        .expect("proxy should shut down")
+        .unwrap();
+    (String::from_utf8(response).unwrap(), origin_request)
+}
+
+#[tokio::test]
+async fn rejects_duplicate_host_fields_before_upstream() {
+    let mut observed = Vec::new();
+    for target in [
+        RequestTarget::Origin,
+        RequestTarget::Absolute,
+        RequestTarget::Connect,
+    ] {
+        for hosts in [HostFields::Identical, HostFields::Conflicting] {
+            let (response, origin_request) = exchange(target, hosts).await;
+            observed.push((
+                response.lines().next().unwrap().to_string(),
+                origin_request.is_some(),
+            ));
+        }
+    }
+    assert_eq!(
+        observed,
+        vec![("HTTP/1.1 400 Bad Request".to_string(), false); 6]
+    );
+}
+
+#[tokio::test]
+async fn preserves_single_host_forwarding() {
+    for target in [RequestTarget::Origin, RequestTarget::Absolute] {
+        let (response, origin_request) = exchange(target, HostFields::Single).await;
+        assert!(response.starts_with("HTTP/1.1 200 OK\r\n"));
+        let host_count = origin_request
+            .as_ref()
+            .unwrap()
+            .lines()
+            .filter(|line| line.to_ascii_lowercase().starts_with("host:"))
+            .count();
+        assert_eq!(host_count, 1);
+    }
+}
+
+#[tokio::test]
+async fn rejects_duplicate_hosts_before_unix_socket_routing() {
+    let state = Arc::new(network_proxy_state_for_policy(NetworkProxyConfig::default()));
+    let mut request = Request::builder()
+        .uri("/")
+        .header("host", "allowed.test")
+        .header("host", "denied.test")
+        .header("x-unix-socket", "/not-an-allowed-socket")
+        .body(Body::empty())
+        .unwrap();
+    request.extensions_mut().insert(state);
+    let response = http_plain_proxy(
+        /*policy_decider*/ None, /*environment_id*/ None, request,
+    )
+    .await
+    .unwrap();
+    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+}