From 634ebc1865c6ac840ed3ba118f040d527bf4b55d Mon Sep 17 00:00:00 2001 From: Winston Howes Date: Wed, 9 Sep 2026 06:53:46 +0000 Subject: [PATCH] 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 | 153 +++++- .../src/brokered_tunnel_tests.rs | 454 +++++++++++++++++- .../network-proxy/src/credential_broker.rs | 26 +- .../src/credential_broker/destination.rs | 4 + codex-rs/network-proxy/src/http_proxy.rs | 59 ++- codex-rs/network-proxy/src/mitm.rs | 110 +++-- codex-rs/network-proxy/src/mitm_tests.rs | 1 + codex-rs/network-proxy/src/runtime.rs | 2 +- codex-rs/network-proxy/src/socks5.rs | 42 +- 9 files changed, 789 insertions(+), 62 deletions(-) diff --git a/codex-rs/network-proxy/src/brokered_tunnel.rs b/codex-rs/network-proxy/src/brokered_tunnel.rs index 9dd6ce6520..1ef4566c99 100644 --- a/codex-rs/network-proxy/src/brokered_tunnel.rs +++ b/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( @@ -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::().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 { + 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::().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::>() + } 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::() + .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; diff --git a/codex-rs/network-proxy/src/brokered_tunnel_tests.rs b/codex-rs/network-proxy/src/brokered_tunnel_tests.rs index 9fbc768860..72affafb9a 100644 --- a/codex-rs/network-proxy/src/brokered_tunnel_tests.rs +++ b/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> { + 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() + ); + } +} diff --git a/codex-rs/network-proxy/src/credential_broker.rs b/codex-rs/network-proxy/src/credential_broker.rs index e36b1deb66..39b545f33c 100644 --- a/codex-rs/network-proxy/src/credential_broker.rs +++ b/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) -> bool { diff --git a/codex-rs/network-proxy/src/credential_broker/destination.rs b/codex-rs/network-proxy/src/credential_broker/destination.rs index a72db716ba..116ed8d083 100644 --- a/codex-rs/network-proxy/src/credential_broker/destination.rs +++ b/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; diff --git a/codex-rs/network-proxy/src/http_proxy.rs b/codex-rs/network-proxy/src/http_proxy.rs index 0dd011b784..52263eafcf 100644 --- a/codex-rs/network-proxy/src/http_proxy.rs +++ b/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::() + .copied() + .unwrap_or(NetworkMode::Full); let connect_mitm_mode = upgraded .extensions() .get::() @@ -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::().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), diff --git a/codex-rs/network-proxy/src/mitm.rs b/codex-rs/network-proxy/src/mitm.rs index 16eb36bef6..79bfec5d48 100644 --- a/codex-rs/network-proxy/src/mitm.rs +++ b/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, } @@ -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(stream: S) -> Result<()> +/// Proxy inner HTTP traffic, terminating TLS with a generated leaf certificate for HTTPS. +pub(crate) async fn mitm_stream(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::() @@ -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 { + 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::().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 { .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 { - 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) diff --git a/codex-rs/network-proxy/src/mitm_tests.rs b/codex-rs/network-proxy/src/mitm_tests.rs index 2e423737ae..ce02f0e013 100644 --- a/codex-rs/network-proxy/src/mitm_tests.rs +++ b/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, } diff --git a/codex-rs/network-proxy/src/runtime.rs b/codex-rs/network-proxy/src/runtime.rs index fd3131ad21..dad445d622 100644 --- a/codex-rs/network-proxy/src/runtime.rs +++ b/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 diff --git a/codex-rs/network-proxy/src/socks5.rs b/codex-rs/network-proxy/src/socks5.rs index 642305fa88..82a7093dab 100644 --- a/codex-rs/network-proxy/src/socks5.rs +++ b/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, .. } =