From 69124cd8fe6da2f06a652267353a65efd7fa7d2a Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Thu, 9 Jul 2026 14:59:01 -0700 Subject: [PATCH] http-client: cover route-aware proxy and privacy behavior --- .../http-client/src/outbound_proxy_tests.rs | 177 ++++++++++++++++-- .../http-client/src/route_aware_redirect.rs | 4 + .../src/route_aware_redirect_tests.rs | 64 +++++++ .../tests/route_aware_client_pool.rs | 131 +++++++++++++ 4 files changed, 361 insertions(+), 15 deletions(-) create mode 100644 codex-rs/http-client/src/route_aware_redirect_tests.rs create mode 100644 codex-rs/http-client/tests/route_aware_client_pool.rs diff --git a/codex-rs/http-client/src/outbound_proxy_tests.rs b/codex-rs/http-client/src/outbound_proxy_tests.rs index bec3999416..6f9ff21fed 100644 --- a/codex-rs/http-client/src/outbound_proxy_tests.rs +++ b/codex-rs/http-client/src/outbound_proxy_tests.rs @@ -1,6 +1,7 @@ //! Shared outbound proxy policy tests. use super::*; +use http::header::COOKIE; use pretty_assertions::assert_eq; use std::io::Read; use std::io::Write; @@ -10,6 +11,76 @@ struct MapEnv { values: HashMap, } +fn spawn_proxy_listener() -> ( + std::net::SocketAddr, + std::thread::JoinHandle>, +) { + let listener = + std::net::TcpListener::bind(("127.0.0.1", 0)).expect("local proxy listener should bind"); + let proxy_addr = listener + .local_addr() + .expect("local proxy listener should have an address"); + listener + .set_nonblocking(true) + .expect("proxy listener should become nonblocking"); + let proxy_thread = std::thread::spawn(move || { + let deadline = Instant::now() + Duration::from_secs(2); + loop { + match listener.accept() { + Ok((mut stream, _)) => { + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("proxy stream should get a read timeout"); + let mut buffer = [0_u8; 4096]; + let size = stream.read(&mut buffer).expect("proxy should read request"); + stream + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok", + ) + .expect("proxy should write response"); + break Some(String::from_utf8_lossy(&buffer[..size]).into_owned()); + } + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + if Instant::now() >= deadline { + break None; + } + std::thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("proxy should accept a request: {error}"), + } + } + }); + (proxy_addr, proxy_thread) +} + +fn spawn_redirect_listener( + location: &str, +) -> (std::net::SocketAddr, std::thread::JoinHandle) { + let listener = + std::net::TcpListener::bind(("127.0.0.1", 0)).expect("redirect listener should bind"); + let address = listener + .local_addr() + .expect("redirect listener should have an address"); + let location = location.to_string(); + let thread = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("redirect server should accept"); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("redirect stream should get a read timeout"); + let mut buffer = [0_u8; 4096]; + let size = stream + .read(&mut buffer) + .expect("redirect request should read"); + write!( + stream, + "HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + ) + .expect("redirect response should write"); + String::from_utf8_lossy(&buffer[..size]).into_owned() + }); + (address, thread) +} + #[test] fn websocket_route_uses_http_equivalent_for_system_resolution() { let env = MapEnv { @@ -169,20 +240,7 @@ async fn async_resolution_uses_cached_route_before_global_permit() { #[tokio::test] async fn enabled_environment_proxy_routes_request_through_proxy() { - let listener = - std::net::TcpListener::bind(("127.0.0.1", 0)).expect("local proxy listener should bind"); - let proxy_addr = listener - .local_addr() - .expect("local proxy listener should have an address"); - let proxy_thread = std::thread::spawn(move || { - let (mut stream, _) = listener.accept().expect("proxy should accept a request"); - let mut buffer = [0_u8; 4096]; - let size = stream.read(&mut buffer).expect("proxy should read request"); - stream - .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") - .expect("proxy should write response"); - String::from_utf8_lossy(&buffer[..size]).into_owned() - }); + let (proxy_addr, proxy_thread) = spawn_proxy_listener(); let env = MapEnv { values: HashMap::from([("HTTP_PROXY".to_string(), format!("http://{proxy_addr}"))]), }; @@ -206,7 +264,10 @@ async fn enabled_environment_proxy_routes_request_through_proxy() { .send() .await .expect("request should use local proxy"); - let proxy_request = proxy_thread.join().expect("proxy thread should finish"); + let proxy_request = proxy_thread + .join() + .expect("proxy thread should finish") + .expect("proxy should receive request before timeout"); assert_eq!(response.status(), reqwest::StatusCode::OK); assert_eq!( @@ -215,6 +276,92 @@ async fn enabled_environment_proxy_routes_request_through_proxy() { ); } +#[tokio::test] +async fn route_aware_pool_uses_respect_system_proxy_route_for_exact_url() { + let (proxy_addr, proxy_thread) = spawn_proxy_listener(); + let request_url = "http://route-aware-proxy.test/proxy-check?pac=exact"; + cache_system_proxy_decision( + request_url, + SystemProxyDecision::Proxy { + url: format!("http://{proxy_addr}"), + }, + ); + let pool = crate::RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + ); + + let response = tokio::time::timeout(Duration::from_secs(2), pool.get(request_url).send()) + .await + .expect("proxy request should finish") + .expect("request should use local proxy"); + let proxy_request = proxy_thread + .join() + .expect("proxy thread should finish") + .expect("proxy should receive request before timeout"); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + assert_eq!( + proxy_request.lines().next(), + Some("GET http://route-aware-proxy.test/proxy-check?pac=exact HTTP/1.1") + ); +} + +#[tokio::test] +async fn route_aware_pool_resolves_each_redirect_hop_and_strips_credentials() { + let (proxy_addr, proxy_thread) = spawn_proxy_listener(); + let redirected_url = "http://redirect-target.test/final?pac=redirect"; + let (redirect_addr, redirect_thread) = spawn_redirect_listener(redirected_url); + let initial_url = format!("http://{redirect_addr}/start"); + cache_system_proxy_decision(&initial_url, SystemProxyDecision::Direct); + cache_system_proxy_decision( + redirected_url, + SystemProxyDecision::Proxy { + url: format!("http://{proxy_addr}"), + }, + ); + let pool = crate::RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + ); + + let response = tokio::time::timeout( + Duration::from_secs(2), + pool.get(&initial_url) + .bearer_auth("redirect-secret") + .header(COOKIE, "session=redirect-secret") + .send(), + ) + .await + .expect("redirected request should finish") + .expect("redirected request should use selected routes"); + let initial_request = redirect_thread + .join() + .expect("redirect thread should finish"); + let proxy_request = proxy_thread + .join() + .expect("proxy thread should finish") + .expect("redirect target should reach proxy"); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + assert_eq!(response.url().as_str(), redirected_url); + assert!( + initial_request + .to_ascii_lowercase() + .contains("authorization: bearer redirect-secret") + ); + assert_eq!( + proxy_request.lines().next(), + Some("GET http://redirect-target.test/final?pac=redirect HTTP/1.1") + ); + assert!( + !proxy_request + .to_ascii_lowercase() + .contains("authorization:") + ); + assert!(!proxy_request.contains("redirect-secret")); +} + #[test] fn parses_pac_proxy_tokens() { assert_eq!( diff --git a/codex-rs/http-client/src/route_aware_redirect.rs b/codex-rs/http-client/src/route_aware_redirect.rs index 5cefc8bf48..208294041d 100644 --- a/codex-rs/http-client/src/route_aware_redirect.rs +++ b/codex-rs/http-client/src/route_aware_redirect.rs @@ -94,3 +94,7 @@ pub(super) fn remove_sensitive_headers( headers.remove("cookie2"); } } + +#[cfg(test)] +#[path = "route_aware_redirect_tests.rs"] +mod tests; diff --git a/codex-rs/http-client/src/route_aware_redirect_tests.rs b/codex-rs/http-client/src/route_aware_redirect_tests.rs new file mode 100644 index 0000000000..15711f46e0 --- /dev/null +++ b/codex-rs/http-client/src/route_aware_redirect_tests.rs @@ -0,0 +1,64 @@ +use http::HeaderValue; +use http::header::CONTENT_LENGTH; +use http::header::CONTENT_TYPE; +use pretty_assertions::assert_eq; + +use super::*; + +#[test] +fn redirects_match_reqwest_method_and_body_rules() { + let url = reqwest::Url::parse("https://example.com/next").expect("redirect URL should parse"); + let mut original = reqwest::Request::new(Method::POST, url.clone()); + original + .headers_mut() + .insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + original + .headers_mut() + .insert(CONTENT_LENGTH, HeaderValue::from_static("2")); + *original.body_mut() = Some("{}".into()); + + let found = redirect_request( + StatusCode::FOUND, + original.method().clone(), + original.headers().clone(), + original.version(), + original.timeout().copied(), + original.try_clone(), + url.clone(), + ) + .expect("POST redirect should be followed"); + assert_eq!( + ( + found.method(), + found.body().is_some(), + found.headers().contains_key(CONTENT_TYPE), + found.headers().contains_key(CONTENT_LENGTH), + ), + (&Method::GET, false, false, false) + ); + + let temporary = redirect_request( + StatusCode::TEMPORARY_REDIRECT, + original.method().clone(), + original.headers().clone(), + original.version(), + original.timeout().copied(), + original.try_clone(), + url, + ) + .expect("replayable temporary redirect should be followed"); + assert_eq!( + ( + temporary.method(), + temporary.body().is_some(), + temporary.headers().get(CONTENT_TYPE), + temporary.headers().get(CONTENT_LENGTH), + ), + ( + &Method::POST, + true, + Some(&HeaderValue::from_static("application/json")), + Some(&HeaderValue::from_static("2")), + ) + ); +} diff --git a/codex-rs/http-client/tests/route_aware_client_pool.rs b/codex-rs/http-client/tests/route_aware_client_pool.rs new file mode 100644 index 0000000000..e25c516b64 --- /dev/null +++ b/codex-rs/http-client/tests/route_aware_client_pool.rs @@ -0,0 +1,131 @@ +use std::io; +use std::io::Read; +use std::io::Write; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use codex_http_client::RouteAwareClientPool; +use http::StatusCode; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::SubscriberExt; + +#[tokio::test] +async fn disabled_pool_logging_does_not_expose_request_or_response_data() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("HTTP listener should bind"); + let address = listener + .local_addr() + .expect("HTTP listener should have an address"); + let server_thread = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("HTTP listener should accept"); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("HTTP stream should get a read timeout"); + let mut buffer = [0_u8; 4096]; + let _size = stream.read(&mut buffer).expect("HTTP request should read"); + stream + .write_all( + b"HTTP/1.1 200 OK\r\nx-sensitive-response: response-secret-value\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok", + ) + .expect("HTTP response should write"); + }); + let endpoint = format!( + "http://auth-user:password-secret-value@{address}/token?client_secret=query-secret-value" + ); + let pool = RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging( + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ClientRouteClass::Api, + ); + let buffer = Arc::new(Mutex::new(Vec::new())); + let subscriber = tracing_subscriber::registry().with( + tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_writer(TestLogWriter { + buffer: Arc::clone(&buffer), + }) + .with_filter( + tracing_subscriber::filter::Targets::new() + .with_target("codex_http_client", tracing::Level::TRACE), + ), + ); + let _guard = tracing::subscriber::set_default(subscriber); + tracing::debug!(target: "codex_http_client", "log capture sentinel"); + + let response = pool + .post(&endpoint) + .header("x-sensitive-request", "request-header-secret-value") + .body("request-body-secret-value") + .send() + .await + .expect("route-aware request should succeed"); + assert_eq!(response.status(), StatusCode::OK); + server_thread.join().expect("server thread should finish"); + + let unresponsive_listener = + std::net::TcpListener::bind("127.0.0.1:0").expect("unresponsive listener should bind"); + let unresponsive_address = unresponsive_listener + .local_addr() + .expect("unresponsive listener should have an address"); + let unresponsive_endpoint = format!( + "http://auth-user:failure-password-secret-value@{unresponsive_address}/token?client_secret=failure-query-secret-value" + ); + let error = pool + .post(&unresponsive_endpoint) + .timeout(Duration::from_millis(100)) + .send() + .await + .expect_err("request to an unresponsive listener should time out"); + assert!(error.is_timeout()); + + let logs = String::from_utf8(buffer.lock().expect("log buffer lock").clone()) + .expect("logs should be UTF-8"); + assert!(logs.contains("log capture sentinel")); + for secret in [ + "password-secret-value", + "query-secret-value", + "request-header-secret-value", + "request-body-secret-value", + "response-secret-value", + "failure-password-secret-value", + "failure-query-secret-value", + ] { + assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}"); + } +} + +#[derive(Clone)] +struct TestLogWriter { + buffer: Arc>>, +} + +struct TestLogSink { + buffer: Arc>>, +} + +impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for TestLogWriter { + type Writer = TestLogSink; + + fn make_writer(&'a self) -> Self::Writer { + TestLogSink { + buffer: Arc::clone(&self.buffer), + } + } +} + +impl Write for TestLogSink { + fn write(&mut self, buffer: &[u8]) -> io::Result { + let mut log_buffer = self + .buffer + .lock() + .map_err(|_| io::Error::other("log buffer lock was poisoned"))?; + log_buffer.extend(buffer); + Ok(buffer.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +}