Support credential brokering in plaintext HTTP tunnels (#44089)

## Why

Configured HTTP credential destinations need dummy credentials translated inside `CONNECT` and SOCKS5 tunnels, where credential interception previously detected only TLS.

## What changed

- Detect plaintext HTTP for configured credential destinations and proxy requests with URL-scoped credential substitution.
- Keep requests bound to the authorized tunnel destination, rejecting mismatched authorities and nested `CONNECT` requests.
- Support HTTP/2 clients and preserve HTTP upgrades, including `h2c`, and opaque traffic in full mode.
- Enforce limited-mode method restrictions on plaintext HTTP and reject upgrades and opaque traffic.

## Testing

Add end-to-end tests for both tunnel transports covering credential URL scope, request bodies, HTTP/2, destination mismatches, upgrades, lossless opaque forwarding, and limited-mode enforcement.

GitOrigin-RevId: 89262c996ea4ee156c628f9ed8e72b52504b2f6b

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

@@ -1,10 +1,23 @@
-//! Detect broker-required TLS without consuming bytes needed for opaque forwarding.
+//! Detect configured tunnel protocols without consuming bytes needed for opaque forwarding.
+//! HTTP requests retain their authorized tunnel destination; upgraded streams stay opaque.
 
 use anyhow::Context;
 use anyhow::Result;
+use rama_core::Service;
 use rama_core::extensions::ExtensionsMut;
+use rama_core::extensions::ExtensionsRef;
 use rama_core::stream::PeekStream;
 use rama_core::stream::Stream;
+use rama_http::HeaderValue;
+use rama_http::Method;
+use rama_http::Request;
+use rama_http::Response;
+use rama_http::StatusCode;
+use rama_http::Uri;
+use rama_http::header::CONNECTION;
+use rama_http::header::PROXY_AUTHENTICATE;
+use rama_http::header::UPGRADE;
+use rama_http::io::upgrade::OnUpgrade;
 use std::io::Cursor;
 use std::time::Duration;
 use tokio::io::AsyncReadExt;
@@ -13,15 +26,18 @@ use tokio::time::timeout;
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(crate) struct BrokeredProtocols {
     pub(crate) tls: bool,
+    pub(crate) http: bool,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(crate) enum TunnelProtocol {
     Tls,
+    Http,
     Opaque,
 }
 
 const PREFIX_TIMEOUT: Duration = Duration::from_millis(250);
+const MAX_REQUEST_LINE: usize = 8192;
 
 /// Detect only configured protocols, replaying every byte for opaque forwarding.
 pub(crate) async fn peek_protocol<S>(
@@ -55,12 +71,147 @@ where
         } else {
             TunnelProtocol::Opaque
         }
+    } else if protocols.http && !prefix.is_empty() {
+        // A bounded probe avoids stalling client-first opaque protocols that wait for a reply.
+        let probe = async {
+            let mut buf = [0_u8; 1024];
+            loop {
+                if let Some(end) = prefix.iter().position(|byte| *byte == b'\n') {
+                    let line = std::str::from_utf8(&prefix[..=end]).ok()?;
+                    let mut parts = line.strip_suffix("\r\n")?.split(' ');
+                    let method = parts.next()?;
+                    let target = parts.next()?;
+                    let version = parts.next()?;
+                    return (parts.next().is_none()
+                        && Method::from_bytes(method.as_bytes()).is_ok()
+                        && target.parse::<Uri>().is_ok()
+                        && (matches!(version, "HTTP/1.0" | "HTTP/1.1")
+                            || (method == "PRI" && target == "*" && version == "HTTP/2.0")))
+                        .then_some(());
+                }
+                if prefix.len() >= MAX_REQUEST_LINE
+                    || prefix
+                        .iter()
+                        .any(|byte| !matches!(byte, b'\r' | b'\t' | 0x20..=0x7e))
+                {
+                    return None;
+                }
+                let remaining = buf.len().min(MAX_REQUEST_LINE - prefix.len());
+                let read = stream.read(&mut buf[..remaining]).await.ok()?;
+                if read == 0 {
+                    return None;
+                }
+                prefix.extend_from_slice(&buf[..read]);
+            }
+        };
+        if matches!(timeout(PREFIX_TIMEOUT, probe).await, Ok(Some(()))) {
+            TunnelProtocol::Http
+        } else {
+            TunnelProtocol::Opaque
+        }
     } else {
         TunnelProtocol::Opaque
     };
     Ok((protocol, PeekStream::new(Cursor::new(prefix), stream)))
 }
 
+/// Preserve HTTP/1 upgrades on an already-authorized tunnel without interpreting upgraded bytes.
+pub(crate) async fn forward_http_request(
+    mut req: Request,
+    upstream: &crate::upstream::UpstreamClient,
+) -> Result<Response> {
+    let upgrade = req.headers().get(UPGRADE).cloned().filter(|_| {
+        req.headers().get_all(CONNECTION).iter().any(|value| {
+            value.to_str().is_ok_and(|value| {
+                value
+                    .split(',')
+                    .any(|token| token.trim().eq_ignore_ascii_case("upgrade"))
+            })
+        })
+    });
+    let downstream_upgrade = req.extensions().get::<OnUpgrade>().cloned();
+    let h2c_settings = if upgrade.as_ref().is_some_and(|value| {
+        value.to_str().is_ok_and(|value| {
+            value
+                .split(',')
+                .any(|protocol| protocol.trim().eq_ignore_ascii_case("h2c"))
+        })
+    }) {
+        req.headers()
+            .get_all("http2-settings")
+            .iter()
+            .cloned()
+            .collect::<Vec<_>>()
+    } else {
+        Vec::new()
+    };
+    crate::http_proxy::remove_hop_by_hop_request_headers(req.headers_mut());
+    if let Some(upgrade) = &upgrade {
+        req.headers_mut().insert(UPGRADE, upgrade.clone());
+        // The upgraded stream is relayed unchanged, including the client's h2c negotiation.
+        let connection = if h2c_settings.is_empty() {
+            "upgrade"
+        } else {
+            "upgrade, http2-settings"
+        };
+        if !h2c_settings.is_empty() {
+            req.headers_mut().remove("http2-settings");
+        }
+        for settings in h2c_settings {
+            req.headers_mut().append("http2-settings", settings);
+        }
+        req.headers_mut()
+            .insert(CONNECTION, HeaderValue::from_static(connection));
+    }
+    let mut response = upstream.serve(req).await?;
+    let response_upgrade = if response.status() == StatusCode::SWITCHING_PROTOCOLS {
+        let expected = upgrade
+            .as_ref()
+            .context("unsolicited HTTP tunnel upgrade")?
+            .to_str()?;
+        let selected = response
+            .headers()
+            .get(UPGRADE)
+            .context("missing HTTP upgrade protocol")?
+            .clone();
+        anyhow::ensure!(
+            expected.split(',').any(|protocol| protocol
+                .trim()
+                .eq_ignore_ascii_case(selected.to_str().unwrap_or_default())),
+            "HTTP tunnel upgrade protocol mismatch"
+        );
+        let downstream = downstream_upgrade.context("missing downstream HTTP upgrade")?;
+        let upstream = response
+            .extensions()
+            .get::<OnUpgrade>()
+            .cloned()
+            .context("missing upstream HTTP upgrade")?;
+        tokio::spawn(async move {
+            let result = async {
+                let (mut downstream, mut upstream) = tokio::try_join!(downstream, upstream)?;
+                tokio::io::copy_bidirectional(&mut downstream, &mut upstream).await?;
+                anyhow::Ok(())
+            }
+            .await;
+            if let Err(err) = result {
+                tracing::warn!("HTTP tunnel upgrade failed: {err}");
+            }
+        });
+        Some(selected)
+    } else {
+        None
+    };
+    crate::http_proxy::remove_hop_by_hop_request_headers(response.headers_mut());
+    response.headers_mut().remove(PROXY_AUTHENTICATE);
+    if let Some(upgrade) = response_upgrade {
+        response.headers_mut().insert(UPGRADE, upgrade);
+        response
+            .headers_mut()
+            .insert(CONNECTION, HeaderValue::from_static("upgrade"));
+    }
+    Ok(response)
+}
+
 #[cfg(test)]
 #[path = "brokered_tunnel_tests.rs"]
 mod tests;

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

@@ -1,10 +1,26 @@
-//! Regression coverage for fragmented TLS detection and lossless prefix replay.
+//! End-to-end checks for credential scope and protocol compatibility on CONNECT and SOCKS5.
 
+use crate::CredentialProviderConfig;
+use crate::NetworkMode;
+use crate::NetworkProxyConfig;
+use crate::NetworkProxyConstraints;
+use crate::NetworkProxyState;
+use crate::build_config_state;
+use crate::connection_lifecycle::ProxyListeners;
+use crate::runtime::ConfigReloader;
+use crate::runtime::ConfigReloaderFuture;
+use crate::runtime::ConfigState;
 use pretty_assertions::assert_eq;
 use rama_core::extensions::Extensions;
 use rama_core::extensions::ExtensionsMut;
 use rama_core::extensions::ExtensionsRef;
+use std::collections::BTreeMap;
+use std::collections::HashMap;
+use std::net::Ipv4Addr;
+use std::net::SocketAddr;
+use std::net::TcpListener as StdTcpListener;
 use std::pin::Pin;
+use std::sync::Arc;
 use std::task::Context;
 use std::task::Poll;
 use tokio::io::AsyncRead;
@@ -13,7 +29,10 @@ use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;
 use tokio::io::DuplexStream;
 use tokio::io::ReadBuf;
+use tokio::net::TcpListener;
+use tokio::net::TcpStream;
 use tokio::time::Duration;
+use tokio::time::timeout;
 
 struct TestStream {
     inner: DuplexStream,
@@ -81,7 +100,10 @@ async fn tls_prefix_detection_accumulates_fragmented_reads() {
 
     let (protocol, mut stream) = crate::brokered_tunnel::peek_protocol(
         TestStream::new(reader),
-        crate::brokered_tunnel::BrokeredProtocols { tls: true },
+        crate::brokered_tunnel::BrokeredProtocols {
+            tls: true,
+            http: true,
+        },
     )
     .await
     .unwrap();
@@ -92,3 +114,431 @@ async fn tls_prefix_detection_accumulates_fragmented_reads() {
     assert_eq!(replayed, tls_prefix);
     writer_task.await.unwrap();
 }
+
+const TOKEN: &str = "local_abcdefghijklmnopqrstuvwxyzabcdef";
+
+struct StaticReloader(ConfigState);
+
+impl ConfigReloader for StaticReloader {
+    fn source_label(&self) -> String {
+        "HTTP tunnel test".to_string()
+    }
+
+    fn maybe_reload(&self) -> ConfigReloaderFuture<'_, Option<ConfigState>> {
+        Box::pin(async { Ok(None) })
+    }
+
+    fn reload_now(&self) -> ConfigReloaderFuture<'_, ConfigState> {
+        Box::pin(async { Ok(self.0.clone()) })
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+enum Transport {
+    Connect,
+    Socks5,
+}
+
+struct TestProxy {
+    addr: SocketAddr,
+    listeners: ProxyListeners,
+    dummy: String,
+    transport: Transport,
+}
+
+impl Drop for TestProxy {
+    fn drop(&mut self) {
+        self.listeners.cancel();
+    }
+}
+
+impl TestProxy {
+    fn start(target: SocketAddr, transport: Transport, mode: NetworkMode) -> Self {
+        let mut config = NetworkProxyConfig {
+            enabled: true,
+            mode,
+            mitm: true,
+            credential_broker: true,
+            allow_local_binding: true,
+            credential_providers: BTreeMap::from([(
+                "local".to_string(),
+                CredentialProviderConfig {
+                    env: vec!["LOCAL_TOKEN".to_string()],
+                    patterns: vec!["^local_[a-z]{32}$".to_string()],
+                    url_prefixes: vec![format!("http://{target}/v1")],
+                    ..CredentialProviderConfig::default()
+                },
+            )]),
+            ..NetworkProxyConfig::default()
+        };
+        config.set_allowed_domains(vec!["127.0.0.1".to_string()]);
+        let config_state = build_config_state(config, NetworkProxyConstraints::default()).unwrap();
+        let state = Arc::new(NetworkProxyState::with_reloader(
+            config_state.clone(),
+            Arc::new(StaticReloader(config_state)),
+        ));
+        let mut env = HashMap::from([("LOCAL_TOKEN".to_string(), TOKEN.to_string())]);
+        state.virtualize_snapshot_credentials(&mut env, /*environment_id*/ None);
+        let dummy = env.remove("LOCAL_TOKEN").unwrap();
+        assert_ne!(dummy, TOKEN);
+        let listener = StdTcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
+        let addr = listener.local_addr().unwrap();
+        let mut listeners = ProxyListeners::new();
+        match transport {
+            Transport::Connect => listeners.spawn(move |guard| {
+                crate::http_proxy::run_http_proxy_with_std_listener(
+                    state, listener, /*policy_decider*/ None, /*environment_id*/ None,
+                    guard,
+                )
+            }),
+            Transport::Socks5 => listeners.spawn(move |guard| {
+                crate::socks5::run_socks5_with_std_listener(
+                    state, listener, /*policy_decider*/ None, /*environment_id*/ None,
+                    /*enable_socks5_udp*/ false, guard,
+                )
+            }),
+        }
+        Self {
+            addr,
+            listeners,
+            dummy,
+            transport,
+        }
+    }
+
+    async fn connect(&self, target: SocketAddr) -> TcpStream {
+        let mut stream = TcpStream::connect(self.addr).await.unwrap();
+        match self.transport {
+            Transport::Connect => {
+                stream
+                    .write_all(
+                        format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n").as_bytes(),
+                    )
+                    .await
+                    .unwrap();
+                assert!(read_headers(&mut stream).await.starts_with("HTTP/1.1 200"));
+            }
+            Transport::Socks5 => {
+                stream.write_all(&[5, 1, 0]).await.unwrap();
+                let mut greeting = [0; 2];
+                stream.read_exact(&mut greeting).await.unwrap();
+                assert_eq!(greeting, [5, 0]);
+                let mut request = vec![5, 1, 0, 1, 127, 0, 0, 1];
+                request.extend_from_slice(&target.port().to_be_bytes());
+                stream.write_all(&request).await.unwrap();
+                let mut response = [0; 10];
+                stream.read_exact(&mut response).await.unwrap();
+                assert_eq!(&response[..4], &[5, 0, 0, 1]);
+            }
+        }
+        stream
+    }
+}
+
+async fn read_headers(stream: &mut TcpStream) -> String {
+    timeout(Duration::from_secs(3), async {
+        let mut bytes = Vec::new();
+        while !bytes.ends_with(b"\r\n\r\n") {
+            assert!(bytes.len() < 16384);
+            bytes.push(stream.read_u8().await.unwrap());
+        }
+        String::from_utf8(bytes).unwrap()
+    })
+    .await
+    .unwrap()
+}
+
+#[tokio::test]
+async fn plaintext_tunnels_translate_only_the_configured_url_prefix() {
+    for transport in [Transport::Connect, Transport::Socks5] {
+        let target = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+        let addr = target.local_addr().unwrap();
+        let proxy = TestProxy::start(addr, transport, NetworkMode::Full);
+        let mut client = proxy.connect(addr).await;
+        for (method, path, body, expected) in [
+            ("GET", "/v1/models", "", TOKEN),
+            ("OPTIONS", "*", "", proxy.dummy.as_str()),
+            ("POST", "/v1/responses", "request body", TOKEN),
+            ("GET", "/v10/models", "", proxy.dummy.as_str()),
+            ("GET", "/v1/%2e%2e/private", "", proxy.dummy.as_str()),
+        ] {
+            client.write_all(format!("{method} {path} HTTP/1.1\r\nHost: {addr}\r\nAuthorization: Bearer {}\r\nContent-Length: {}\r\n\r\n{body}", proxy.dummy, body.len()).as_bytes()).await.unwrap();
+            let (mut upstream, _) = timeout(Duration::from_secs(3), target.accept())
+                .await
+                .unwrap()
+                .unwrap();
+            let request = read_headers(&mut upstream).await;
+            assert!(request.starts_with(&format!("{method} {path} HTTP/1.1\r\n")));
+            assert!(
+                request
+                    .lines()
+                    .any(|line| line
+                        .eq_ignore_ascii_case(&format!("authorization: Bearer {expected}"))),
+                "{transport:?}, {path}: {request:?}"
+            );
+            let mut received_body = vec![0; body.len()];
+            timeout(
+                Duration::from_secs(3),
+                upstream.read_exact(&mut received_body),
+            )
+            .await
+            .unwrap()
+            .unwrap();
+            assert_eq!(received_body, body.as_bytes());
+            upstream
+                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .await
+                .unwrap();
+            assert!(read_headers(&mut client).await.starts_with("HTTP/1.1 200"));
+        }
+    }
+}
+
+#[tokio::test]
+async fn plaintext_tunnels_accept_http2_clients() {
+    use rama_core::Service;
+    use rama_core::error::BoxError;
+    use rama_core::service::service_fn;
+    use rama_http::Body;
+    use rama_http::Request;
+    use rama_http::Version;
+    use rama_http_backend::client::HttpConnector;
+    use rama_net::client::EstablishedClientConnection;
+    for transport in [Transport::Connect, Transport::Socks5] {
+        let target = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+        let addr = target.local_addr().unwrap();
+        let proxy = TestProxy::start(addr, transport, NetworkMode::Full);
+        let stream = Arc::new(tokio::sync::Mutex::new(Some(rama_tcp::TcpStream::from(
+            proxy.connect(addr).await,
+        ))));
+        let (captured_tx, mut captured_rx) = tokio::sync::mpsc::channel(1);
+        let target_task = tokio::spawn(async move {
+            let (stream, _) = target.accept().await.unwrap();
+            rama_http_backend::server::HttpServer::auto(rama_core::rt::Executor::default())
+                .service(service_fn(move |request: Request| {
+                    let captured_tx = captured_tx.clone();
+                    async move {
+                        captured_tx
+                            .send(request.headers().get("authorization").cloned())
+                            .await
+                            .unwrap();
+                        Ok::<_, std::convert::Infallible>(rama_http::Response::new(Body::empty()))
+                    }
+                }))
+                .serve(rama_tcp::TcpStream::from(stream))
+                .await
+                .unwrap();
+        });
+        let connector = HttpConnector::<_, Body>::new(service_fn(move |req: Request| {
+            let stream = stream.clone();
+            async move {
+                Ok::<_, BoxError>(EstablishedClientConnection {
+                    input: req,
+                    conn: stream.lock().await.take().unwrap(),
+                })
+            }
+        }));
+        let request = Request::builder()
+            .version(Version::HTTP_2)
+            .uri(format!("http://{addr}/v1/models"))
+            .header("authorization", format!("Bearer {}", proxy.dummy))
+            .body(Body::empty())
+            .unwrap();
+        let EstablishedClientConnection { input, conn } = connector.serve(request).await.unwrap();
+        let response = tokio::spawn(async move { conn.serve(input).await });
+        assert_eq!(
+            timeout(Duration::from_secs(3), captured_rx.recv())
+                .await
+                .unwrap()
+                .unwrap(),
+            Some(rama_http::HeaderValue::from_str(&format!("Bearer {TOKEN}")).unwrap())
+        );
+        assert_eq!(
+            timeout(Duration::from_secs(3), response)
+                .await
+                .unwrap()
+                .unwrap()
+                .unwrap()
+                .status(),
+            rama_http::StatusCode::OK
+        );
+        target_task.abort();
+    }
+}
+
+#[tokio::test]
+async fn plaintext_tunnels_reject_retargeting_and_nested_connect() {
+    for transport in [Transport::Connect, Transport::Socks5] {
+        let target = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+        let addr = target.local_addr().unwrap();
+        let proxy = TestProxy::start(addr, transport, NetworkMode::Full);
+        for (method, uri, host, status) in [
+            ("GET", "/v1".to_string(), "127.0.0.1:1".to_string(), 400),
+            (
+                "GET",
+                "http://127.0.0.1:1/v1".to_string(),
+                addr.to_string(),
+                400,
+            ),
+            ("GET", format!("https://{addr}/v1"), addr.to_string(), 400),
+            ("CONNECT", addr.to_string(), addr.to_string(), 405),
+        ] {
+            let mut client = proxy.connect(addr).await;
+            client.write_all(format!("{method} {uri} HTTP/1.1\r\nHost: {host}\r\nAuthorization: Bearer {}\r\nConnection: close\r\n\r\n", proxy.dummy).as_bytes()).await.unwrap();
+            let response = read_headers(&mut client).await;
+            assert!(
+                response.starts_with(&format!("HTTP/1.1 {status}")),
+                "{transport:?}: {response:?}"
+            );
+        }
+        assert!(
+            timeout(Duration::from_millis(50), target.accept())
+                .await
+                .is_err()
+        );
+    }
+}
+
+#[tokio::test]
+async fn plaintext_tunnels_preserve_server_first_and_client_first_opaque_bytes() {
+    for transport in [Transport::Connect, Transport::Socks5] {
+        for server_first in [true, false] {
+            let target = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+            let addr = target.local_addr().unwrap();
+            let proxy = TestProxy::start(addr, transport, NetworkMode::Full);
+            let mut client = proxy.connect(addr).await;
+            if !server_first {
+                client.write_all(b"PING").await.unwrap();
+            }
+            let (mut upstream, _) = timeout(Duration::from_secs(3), target.accept())
+                .await
+                .unwrap()
+                .unwrap();
+            upstream.write_all(b"SSH-2.0-server\r\n").await.unwrap();
+            let mut banner = [0; 16];
+            timeout(Duration::from_secs(3), client.read_exact(&mut banner))
+                .await
+                .unwrap()
+                .unwrap();
+            assert_eq!(&banner, b"SSH-2.0-server\r\n");
+            if server_first {
+                client.write_all(b"PING").await.unwrap();
+            }
+            let mut request = [0; 4];
+            timeout(Duration::from_secs(3), upstream.read_exact(&mut request))
+                .await
+                .unwrap()
+                .unwrap();
+            assert_eq!(&request, b"PING");
+        }
+    }
+}
+
+#[tokio::test]
+async fn plaintext_tunnels_preserve_http_upgrades() {
+    for (transport, protocol) in [Transport::Connect, Transport::Socks5]
+        .into_iter()
+        .flat_map(|transport| ["test-echo", "h2c"].map(move |protocol| (transport, protocol)))
+    {
+        let target = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+        let addr = target.local_addr().unwrap();
+        let proxy = TestProxy::start(addr, transport, NetworkMode::Full);
+        let mut client = proxy.connect(addr).await;
+        client.write_all(format!("GET /v1 HTTP/1.1\r\nHost: {addr}\r\nAuthorization: Bearer {}\r\nConnection: upgrade, HTTP2-Settings, X-Remove\r\nUpgrade: {protocol}\r\nHTTP2-Settings: AAMAAABk\r\nX-Remove: hop-local\r\n\r\n", proxy.dummy).as_bytes()).await.unwrap();
+        let (mut upstream, _) = timeout(Duration::from_secs(3), target.accept())
+            .await
+            .unwrap()
+            .unwrap();
+        let request = read_headers(&mut upstream).await;
+        assert!(
+            request
+                .lines()
+                .any(|line| line.eq_ignore_ascii_case(&format!("Authorization: Bearer {TOKEN}")))
+        );
+        assert!(
+            request
+                .lines()
+                .any(|line| line.eq_ignore_ascii_case(&format!("Upgrade: {protocol}")))
+        );
+        assert_eq!(
+            request
+                .lines()
+                .any(|line| line.eq_ignore_ascii_case("HTTP2-Settings: AAMAAABk")),
+            protocol == "h2c"
+        );
+        let connection = if protocol == "h2c" {
+            "upgrade, http2-settings"
+        } else {
+            "upgrade"
+        };
+        assert!(
+            request
+                .lines()
+                .any(|line| line.eq_ignore_ascii_case(&format!("Connection: {connection}")))
+        );
+        assert!(!request.to_ascii_lowercase().contains("x-remove"));
+        upstream.write_all(format!("HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\nUpgrade: {protocol}\r\n\r\nbanner").as_bytes()).await.unwrap();
+        assert!(read_headers(&mut client).await.starts_with("HTTP/1.1 101"));
+        let mut banner = [0; 6];
+        timeout(Duration::from_secs(3), client.read_exact(&mut banner))
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(&banner, b"banner");
+        client.write_all(b"PING").await.unwrap();
+        let mut payload = [0; 4];
+        timeout(Duration::from_secs(3), upstream.read_exact(&mut payload))
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(&payload, b"PING");
+    }
+}
+
+#[tokio::test]
+async fn limited_plaintext_tunnels_allow_reads_but_reject_writes_and_opaque_traffic() {
+    for transport in [Transport::Connect, Transport::Socks5] {
+        let target = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+        let addr = target.local_addr().unwrap();
+        let proxy = TestProxy::start(addr, transport, NetworkMode::Limited);
+        for method in ["GET", "HEAD", "OPTIONS"] {
+            let mut client = proxy.connect(addr).await;
+            client.write_all(format!("{method} /v1 HTTP/1.1\r\nHost: {addr}\r\nAuthorization: Bearer {}\r\nConnection: close\r\n\r\n", proxy.dummy).as_bytes()).await.unwrap();
+            let (mut upstream, _) = timeout(Duration::from_secs(3), target.accept())
+                .await
+                .unwrap()
+                .unwrap();
+            let request = read_headers(&mut upstream).await;
+            assert!(
+                request.lines().any(
+                    |line| line.eq_ignore_ascii_case(&format!("Authorization: Bearer {TOKEN}"))
+                )
+            );
+            upstream
+                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .await
+                .unwrap();
+            assert!(read_headers(&mut client).await.starts_with("HTTP/1.1 200"));
+        }
+        for (method, extra_headers) in [
+            ("POST", ""),
+            ("GET", "Upgrade: test-echo\r\nConnection: upgrade\r\n"),
+        ] {
+            let mut client = proxy.connect(addr).await;
+            client.write_all(format!("{method} /v1 HTTP/1.1\r\nHost: {addr}\r\nAuthorization: Bearer {}\r\nContent-Length: 0\r\n{extra_headers}\r\n", proxy.dummy).as_bytes()).await.unwrap();
+            assert!(read_headers(&mut client).await.starts_with("HTTP/1.1 403"));
+        }
+        let mut client = proxy.connect(addr).await;
+        client.write_all(b"PING").await.unwrap();
+        let mut byte = [0];
+        let result = timeout(Duration::from_secs(3), client.read(&mut byte))
+            .await
+            .unwrap();
+        assert!(matches!(result, Ok(0) | Err(_)));
+        assert!(
+            timeout(Duration::from_millis(50), target.accept())
+                .await
+                .is_err()
+        );
+    }
+}

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

@@ -1094,15 +1094,25 @@ impl CredentialBroker {
     ) -> crate::brokered_tunnel::BrokeredProtocols {
         let normalized_host = normalize_host(host);
         let state = self.read_state();
-        crate::brokered_tunnel::BrokeredProtocols {
-            tls: state.enabled
-                && state.credentials.iter().any(|credential| {
-                    credential.belongs_to_environment(environment_id)
-                        && credential
-                            .host_bindings()
-                            .any(|binding| binding.requires_mitm(&normalized_host, port))
-                }),
+        let mut protocols = crate::brokered_tunnel::BrokeredProtocols::default();
+        if state.enabled {
+            for credential in state
+                .credentials
+                .iter()
+                .filter(|credential| credential.belongs_to_environment(environment_id))
+            {
+                for binding in credential.host_bindings() {
+                    protocols.tls |= binding.requires_mitm(&normalized_host, port);
+                    if let providers::CredentialHostBinding::ConfiguredHosts(destinations) = binding
+                    {
+                        protocols.http |= destinations.iter().any(|destination| {
+                            destination.requires_http_interception(&normalized_host, port)
+                        });
+                    }
+                }
+            }
         }
+        protocols
     }
 
     pub(crate) fn virtualize_text(&self, text: &mut String, env: &HashMap<String, String>) -> bool {

codex-rs/network-proxy/src/credential_broker/destination.rs

@@ -131,6 +131,10 @@ impl CredentialDestination {
         self.scheme == CredentialDestinationScheme::Https && self.matches_host(host, port)
     }
 
+    pub(super) fn requires_http_interception(&self, host: &str, port: u16) -> bool {
+        self.scheme == CredentialDestinationScheme::Http && self.matches_host(host, port)
+    }
+
     pub(super) fn matches_request(&self, host: &str, request: Option<&Url>) -> bool {
         let Some(request) = request else {
             return false;

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

@@ -307,7 +307,8 @@ async fn http_connect_accept(
             return Err(text_response(StatusCode::INTERNAL_SERVER_ERROR, "error"));
         }
     };
-    let connect_mitm_mode = if mode == NetworkMode::Limited {
+    let brokered_http = matches!(host_mitm_requirement, HostMitmRequirement::Credential(protocols) if protocols.http);
+    let connect_mitm_mode = if mode == NetworkMode::Limited && !brokered_http {
         ConnectMitmMode::Enabled
     } else {
         match host_mitm_requirement {
@@ -321,7 +322,7 @@ async fn http_connect_accept(
 
     if connect_mitm_mode == ConnectMitmMode::Enabled && mitm_state.is_none() {
         // Limited-mode enforcement and host-specific hooks require interception. Credential-only
-        // interception is deferred until the upgraded stream presents a TLS ClientHello.
+        // interception is deferred until the upgraded stream presents a supported protocol.
         emit_http_block_decision_audit_event(
             &app_state,
             BlockDecisionAuditEventArgs {
@@ -381,6 +382,11 @@ async fn http_connect_accept(
 }
 
 async fn http_connect_proxy(upgraded: Upgraded) -> Result<(), Infallible> {
+    let mode = upgraded
+        .extensions()
+        .get::<NetworkMode>()
+        .copied()
+        .unwrap_or(NetworkMode::Full);
     let connect_mitm_mode = upgraded
         .extensions()
         .get::<ConnectMitmMode>()
@@ -394,10 +400,25 @@ async fn http_connect_proxy(upgraded: Upgraded) -> Result<(), Infallible> {
                 Ok((crate::brokered_tunnel::TunnelProtocol::Tls, stream)) => {
                     mitm_connect_tunnel(stream).await
                 }
-                Ok((crate::brokered_tunnel::TunnelProtocol::Opaque, stream)) => {
-                    forward_connect_tunnel(stream).await
+                Ok((crate::brokered_tunnel::TunnelProtocol::Http, stream)) => {
+                    mitm::mitm_stream(stream, rama_http::uri::Scheme::HTTP)
+                        .await
+                        .map_err(|err| {
+                            OpaqueError::from_display(format!("HTTP tunnel error: {err}"))
+                        })
                 }
-                Err(err) => Err(OpaqueError::from_display(format!("detect TLS: {err:#}"))),
+                Ok((crate::brokered_tunnel::TunnelProtocol::Opaque, stream)) => {
+                    if mode == NetworkMode::Limited {
+                        Err(OpaqueError::from_display(
+                            "opaque tunnels are not allowed in limited mode",
+                        ))
+                    } else {
+                        forward_connect_tunnel(stream).await
+                    }
+                }
+                Err(err) => Err(OpaqueError::from_display(format!(
+                    "detect tunnel protocol: {err:#}"
+                ))),
             }
         }
     };
@@ -430,7 +451,7 @@ where
     }
 
     info!("CONNECT MITM enabled (host={host}, port={port}, mode={mode:?})");
-    mitm::mitm_stream(stream)
+    mitm::mitm_stream(stream, rama_http::uri::Scheme::HTTPS)
         .await
         .map_err(|err| OpaqueError::from_display(format!("MITM tunnel error: {err}")))
 }
@@ -1019,7 +1040,7 @@ fn validate_absolute_form_host_header(
 
     Ok(())
 }
-fn remove_hop_by_hop_request_headers(headers: &mut HeaderMap) {
+pub(crate) fn remove_hop_by_hop_request_headers(headers: &mut HeaderMap) {
     while let Some(raw_connection) = headers.get(header::CONNECTION).cloned() {
         headers.remove(header::CONNECTION);
         if let Ok(raw_connection) = raw_connection.to_str() {
@@ -1333,7 +1354,10 @@ mod tests {
         assert_eq!(
             request.extensions().get::<ConnectMitmMode>().copied(),
             Some(ConnectMitmMode::DetectProtocol(
-                crate::brokered_tunnel::BrokeredProtocols { tls: true }
+                crate::brokered_tunnel::BrokeredProtocols {
+                    tls: true,
+                    http: false
+                }
             ))
         );
     }
@@ -1406,6 +1430,25 @@ mod tests {
             state.virtualize_child_credentials(&mut env);
             let dummy = env.get("PROVIDER_TOKEN").expect("dummy provider token");
             assert_ne!(dummy, real_token);
+            assert_eq!(
+                state.host_mitm_requirement(host, port).await.unwrap(),
+                HostMitmRequirement::Credential(crate::brokered_tunnel::BrokeredProtocols {
+                    tls: !prefix.is_empty(),
+                    http: prefix.is_empty()
+                })
+            );
+            assert_eq!(
+                state.host_mitm_requirement(host, port + 1).await.unwrap(),
+                HostMitmRequirement::None
+            );
+            assert_eq!(
+                state
+                    .for_environment_id(Some("other"))
+                    .host_mitm_requirement(host, port)
+                    .await
+                    .unwrap(),
+                HostMitmRequirement::None
+            );
             for (uri, inject) in [
                 (format!("http://{host}:{port}/v1/models"), prefix.is_empty()),
                 (format!("http://{host}:{}/v1/models", port + 1), false),

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

@@ -38,6 +38,7 @@ use rama_http::Uri;
 use rama_http::header::HOST;
 use rama_http::layer::remove_header::RemoveRequestHeaderLayer;
 use rama_http::layer::remove_header::RemoveResponseHeaderLayer;
+use rama_http::uri::Scheme;
 use rama_http_backend::server::HttpServer;
 use rama_net::proxy::ProxyTarget;
 use rama_net::stream::SocketInfo;
@@ -68,6 +69,7 @@ pub(crate) struct MitmUpstreamConfig {
 struct MitmPolicyContext {
     target_host: String,
     target_port: u16,
+    scheme: Scheme,
     mode: NetworkMode,
     app_state: Arc<NetworkProxyState>,
 }
@@ -132,8 +134,8 @@ impl MitmState {
     }
 }
 
-/// Terminate a raw client stream with a generated leaf cert and proxy inner HTTPS traffic.
-pub(crate) async fn mitm_stream<S>(stream: S) -> Result<()>
+/// Proxy inner HTTP traffic, terminating TLS with a generated leaf certificate for HTTPS.
+pub(crate) async fn mitm_stream<S>(stream: S, scheme: Scheme) -> Result<()>
 where
     S: Stream + Unpin + ExtensionsMut,
 {
@@ -155,7 +157,11 @@ where
         .clone();
     let target_host = normalize_host(&target.host.to_string());
     let target_port = target.port;
-    let acceptor_data = mitm.tls_acceptor_data_for_host(&target_host)?;
+    let acceptor_data = if scheme == Scheme::HTTPS {
+        Some(mitm.tls_acceptor_data_for_host(&target_host)?)
+    } else {
+        None
+    };
     let mode = stream
         .extensions()
         .get::<NetworkMode>()
@@ -176,6 +182,7 @@ where
         policy: MitmPolicyContext {
             target_host,
             target_port,
+            scheme,
             mode,
             app_state,
         },
@@ -191,8 +198,9 @@ where
 
     let http_service = HttpServer::auto(executor).service(
         (
-            RemoveResponseHeaderLayer::hop_by_hop(),
-            RemoveRequestHeaderLayer::hop_by_hop(),
+            (request_ctx.policy.scheme == Scheme::HTTPS)
+                .then(RemoveResponseHeaderLayer::hop_by_hop),
+            (request_ctx.policy.scheme == Scheme::HTTPS).then(RemoveRequestHeaderLayer::hop_by_hop),
         )
             .into_layer(service_fn({
                 let request_ctx = request_ctx.clone();
@@ -203,14 +211,17 @@ where
             })),
     );
 
-    let https_service = TlsAcceptorLayer::new(acceptor_data)
-        .with_store_client_hello(true)
-        .into_layer(http_service);
-
-    https_service
-        .serve(stream)
-        .await
-        .map_err(|err| anyhow!("MITM serve error: {err}"))?;
+    match acceptor_data {
+        Some(acceptor_data) => {
+            TlsAcceptorLayer::new(acceptor_data)
+                .with_store_client_hello(true)
+                .into_layer(http_service)
+                .serve(stream)
+                .await
+        }
+        None => http_service.serve(stream).await,
+    }
+    .map_err(|err| anyhow!("MITM serve error: {err}"))?;
     Ok(())
 }
 
@@ -243,12 +254,22 @@ async fn forward_request(req: Request, request_ctx: &MitmRequestContext) -> Resu
     let log_path = path_for_log(req.uri());
 
     let (mut parts, body) = req.into_parts();
-    let authority = authority_header_value(&target_host, target_port);
-    parts.uri = build_https_uri(&authority, &path)?;
+    let authority = authority_header_value(&target_host, target_port, &request_ctx.policy.scheme);
+    parts.uri = Uri::builder()
+        .scheme(request_ctx.policy.scheme.clone())
+        .authority(authority.as_str())
+        .path_and_query(path.as_str())
+        .build()?;
+    // Server-wide OPTIONS must not borrow authority from a narrower credential URL prefix.
+    let credential_path = if path == "*" { "/" } else { &path };
+    let credential_destination = format!(
+        "{}://{authority}{credential_path}",
+        request_ctx.policy.scheme
+    );
     request_ctx
         .policy
         .app_state
-        .inject_request_credentials(&parts.uri.to_string(), &mut parts.headers);
+        .inject_request_credentials(&credential_destination, &mut parts.headers);
     apply_mitm_hook_actions(&mut parts.headers, hook_actions.as_ref());
     parts
         .headers
@@ -271,7 +292,11 @@ async fn forward_request(req: Request, request_ctx: &MitmRequestContext) -> Resu
     };
 
     let upstream_req = Request::from_parts(parts, body);
-    let upstream_resp = request_ctx.upstream.serve(upstream_req).await?;
+    let upstream_resp = if request_ctx.policy.scheme == Scheme::HTTP {
+        crate::brokered_tunnel::forward_http_request(upstream_req, &request_ctx.upstream).await?
+    } else {
+        request_ctx.upstream.serve(upstream_req).await?
+    };
     respond_with_inspection(
         upstream_resp,
         inspect,
@@ -297,6 +322,34 @@ async fn evaluate_mitm_policy(
     req: &Request,
     policy: &MitmPolicyContext,
 ) -> Result<MitmPolicyDecision> {
+    if policy.scheme == Scheme::HTTP {
+        let matches_target = |authority: &rama_http::uri::Authority| {
+            !authority.as_str().contains('@')
+                && normalize_host(authority.host()) == policy.target_host
+                && authority.port_u16().unwrap_or(80) == policy.target_port
+        };
+        let invalid_destination = req
+            .uri()
+            .scheme()
+            .is_some_and(|scheme| scheme != &Scheme::HTTP)
+            || req
+                .uri()
+                .authority()
+                .is_some_and(|authority| !matches_target(authority))
+            || req.headers().get_all(HOST).iter().any(|value| {
+                value
+                    .to_str()
+                    .ok()
+                    .and_then(|value| value.parse::<rama_http::uri::Authority>().ok())
+                    .is_none_or(|authority| !matches_target(&authority))
+            });
+        if invalid_destination {
+            return Ok(MitmPolicyDecision::Block(text_response(
+                StatusCode::BAD_REQUEST,
+                "tunnel destination mismatch",
+            )));
+        }
+    }
     if req.method().as_str() == "CONNECT" {
         return Ok(MitmPolicyDecision::Block(text_response(
             StatusCode::METHOD_NOT_ALLOWED,
@@ -343,7 +396,7 @@ async fn evaluate_mitm_policy(
                 client: client.clone(),
                 method: Some(method.clone()),
                 mode: Some(policy.mode),
-                protocol: "https".to_string(),
+                protocol: policy.scheme.to_string(),
                 decision: None,
                 source: None,
                 port: Some(policy.target_port),
@@ -371,7 +424,7 @@ async fn evaluate_mitm_policy(
                     client: client.clone(),
                     method: Some(method.clone()),
                     mode: Some(policy.mode),
-                    protocol: "https".to_string(),
+                    protocol: policy.scheme.to_string(),
                     decision: None,
                     source: None,
                     port: Some(policy.target_port),
@@ -388,7 +441,10 @@ async fn evaluate_mitm_policy(
         HookEvaluation::NoHooksForHost => None,
     };
 
-    if !policy.mode.allows_method(&method) {
+    let opaque_http_upgrade = policy.scheme == Scheme::HTTP
+        && policy.mode == NetworkMode::Limited
+        && req.headers().contains_key(rama_http::header::UPGRADE);
+    if !policy.mode.allows_method(&method) || opaque_http_upgrade {
         let _ = policy
             .app_state
             .record_blocked(BlockedRequest::new(BlockedRequestArgs {
@@ -397,7 +453,7 @@ async fn evaluate_mitm_policy(
                 client: client.clone(),
                 method: Some(method.clone()),
                 mode: Some(policy.mode),
-                protocol: "https".to_string(),
+                protocol: policy.scheme.to_string(),
                 decision: None,
                 source: None,
                 port: Some(policy.target_port),
@@ -544,26 +600,22 @@ fn extract_request_host(req: &Request) -> Option<String> {
         .or_else(|| req.uri().authority().map(|a| a.as_str().to_string()))
 }
 
-fn authority_header_value(host: &str, port: u16) -> String {
+fn authority_header_value(host: &str, port: u16, scheme: &Scheme) -> String {
     // Host header / URI authority formatting.
+    let default_port = if scheme == &Scheme::HTTP { 80 } else { 443 };
     if host.contains(':') {
-        if port == 443 {
+        if port == default_port {
             format!("[{host}]")
         } else {
             format!("[{host}]:{port}")
         }
-    } else if port == 443 {
+    } else if port == default_port {
         host.to_string()
     } else {
         format!("{host}:{port}")
     }
 }
 
-fn build_https_uri(authority: &str, path: &str) -> Result<Uri> {
-    let target = format!("https://{authority}{path}");
-    Ok(target.parse()?)
-}
-
 fn path_and_query(uri: &Uri) -> String {
     uri.path_and_query()
         .map(rama_http::uri::PathAndQuery::as_str)

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

@@ -45,6 +45,7 @@ fn policy_ctx(
     MitmPolicyContext {
         target_host: target_host.to_string(),
         target_port,
+        scheme: Scheme::HTTPS,
         mode,
         app_state,
     }

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

@@ -907,7 +907,7 @@ impl NetworkProxyState {
         );
         Ok(if host_has_mitm_hooks {
             HostMitmRequirement::Always
-        } else if protocols.tls {
+        } else if protocols.tls || protocols.http {
             HostMitmRequirement::Credential(protocols)
         } else {
             HostMitmRequirement::None

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

@@ -271,10 +271,17 @@ async fn handle_socks5_tcp(
             return Err(io::Error::other("proxy error").into());
         }
     };
-    // SOCKS5 only exposes host and port, so only the default HTTPS port is identifiable as a
-    // TLS stream that the HTTPS MITM path can safely terminate.
+    let host_mitm_requirement = match app_state.host_mitm_requirement(&host, port).await {
+        Ok(requirement) => requirement,
+        Err(err) => {
+            error!("failed to inspect MITM requirements for {host}: {err}");
+            return Err(io::Error::other("proxy error").into());
+        }
+    };
+    // Otherwise retain the existing limited-mode restriction to the default HTTPS port.
+    let brokered_http = matches!(host_mitm_requirement, HostMitmRequirement::Credential(protocols) if protocols.http);
     let socks5_tcp_target_is_https = port == 443;
-    if mode == NetworkMode::Limited && !socks5_tcp_target_is_https {
+    if mode == NetworkMode::Limited && !socks5_tcp_target_is_https && !brokered_http {
         emit_socks_block_decision_audit_event(
             &app_state,
             NetworkDecisionSource::ModeGuard,
@@ -364,13 +371,6 @@ async fn handle_socks5_tcp(
         }
     }
 
-    let host_mitm_requirement = match app_state.host_mitm_requirement(&host, port).await {
-        Ok(requirement) => requirement,
-        Err(err) => {
-            error!("failed to inspect MITM requirements for {host}: {err}");
-            return Err(io::Error::other("proxy error").into());
-        }
-    };
     let mitm_state = match app_state.mitm_state().await {
         Ok(state) => state,
         Err(err) => {
@@ -378,7 +378,7 @@ async fn handle_socks5_tcp(
             return Err(io::Error::other("proxy error").into());
         }
     };
-    let socks_mitm_mode = if mode == NetworkMode::Limited {
+    let socks_mitm_mode = if mode == NetworkMode::Limited && !brokered_http {
         SocksMitmMode::Enabled
     } else {
         match host_mitm_requirement {
@@ -599,7 +599,9 @@ async fn proxy_socks5_tcp(
             source.extensions_mut().insert(ProxyTarget(target));
             source.extensions_mut().insert(mode);
             source.extensions_mut().insert(mitm);
-            mitm::mitm_stream(source).await.map_err(Into::into)
+            mitm::mitm_stream(source, rama_http::uri::Scheme::HTTPS)
+                .await
+                .map_err(Into::into)
         }
         Socks5TcpConnection::DetectProtocol {
             protocols,
@@ -617,9 +619,23 @@ async fn proxy_socks5_tcp(
                 .map_err(|err| -> BoxError { err.into() })?;
             match protocol {
                 crate::brokered_tunnel::TunnelProtocol::Tls => {
-                    mitm::mitm_stream(source).await.map_err(Into::into)
+                    mitm::mitm_stream(source, rama_http::uri::Scheme::HTTPS)
+                        .await
+                        .map_err(Into::into)
+                }
+                crate::brokered_tunnel::TunnelProtocol::Http => {
+                    mitm::mitm_stream(source, rama_http::uri::Scheme::HTTP)
+                        .await
+                        .map_err(Into::into)
                 }
                 crate::brokered_tunnel::TunnelProtocol::Opaque => {
+                    if mode == NetworkMode::Limited {
+                        return Err(io::Error::new(
+                            io::ErrorKind::PermissionDenied,
+                            "opaque tunnels are not allowed in limited mode",
+                        )
+                        .into());
+                    }
                     info!("SOCKS opaque upstream dial started (target={target})");
                     let connect_started_at = Instant::now();
                     let EstablishedClientConnection { conn: upstream, .. } =