diff --git a/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs b/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs new file mode 100644 index 0000000000..ed10af742e --- /dev/null +++ b/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs @@ -0,0 +1,217 @@ +use super::*; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::SubscriberExt; + +#[tokio::test] +async fn route_aware_pool_strips_credentials_on_cross_origin_redirect() { + let (destination_addr, destination_thread) = spawn_proxy_listener(); + let destination_url = format!("http://{destination_addr}/final"); + let (redirect_addr, redirect_thread) = spawn_redirect_listener(&destination_url); + let initial_url = format!("http://{redirect_addr}/start"); + cache_system_proxy_decision(&initial_url, SystemProxyDecision::Direct); + cache_system_proxy_decision(&destination_url, SystemProxyDecision::Direct); + let pool = crate::RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + ); + + let response = tokio::time::timeout( + Duration::from_secs(2), + pool.get(&initial_url) + .header(AUTHORIZATION, "Bearer origin-secret") + .header(COOKIE, "session=origin-secret") + .header(PROXY_AUTHORIZATION, "Basic proxy-secret") + .send(), + ) + .await + .expect("redirected request should finish") + .expect("cross-origin redirect should succeed"); + let initial_request = only_request(redirect_thread, "redirect"); + let destination_request = only_request(destination_thread, "destination"); + + assert_eq!(response.url().as_str(), destination_url); + assert_eq!( + [ + credential_headers(&initial_request), + credential_headers(&destination_request), + ], + [(true, true, true), (false, false, false)] + ); +} + +#[tokio::test] +async fn route_aware_pool_retains_credentials_for_same_origin_and_route() { + let (proxy_addr, proxy_thread) = spawn_http_listener(vec![ + "HTTP/1.1 302 Found\r\nLocation: /final\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + .to_string(), + "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_string(), + ]); + let initial_url = "http://same-route.test/start"; + let redirected_url = "http://same-route.test/final"; + let route = SystemProxyDecision::Proxy { + url: format!("http://{proxy_addr}"), + }; + cache_system_proxy_decision(initial_url, route.clone()); + cache_system_proxy_decision(redirected_url, route); + let pool = crate::RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + ); + + let response = tokio::time::timeout( + Duration::from_secs(2), + pool.get(initial_url) + .header(AUTHORIZATION, "Bearer origin-secret") + .header(COOKIE, "session=origin-secret") + .header(PROXY_AUTHORIZATION, "Basic proxy-secret") + .send(), + ) + .await + .expect("redirected request should finish") + .expect("same-route redirect should succeed"); + let requests = proxy_thread.join().expect("proxy thread should finish"); + + assert_eq!(response.url().as_str(), redirected_url); + assert_eq!( + requests + .iter() + .map(|request| credential_headers(request)) + .collect::>(), + vec![(true, true, true), (true, true, true)] + ); +} + +#[tokio::test] +async fn route_aware_pool_sanitizes_redirected_failure_logs() { + let log_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(&log_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 (enabled_failure_addr, enabled_failure_thread) = spawn_failing_listener(); + let enabled_target_url = + format!("http://{enabled_failure_addr}/final?token=enabled-target-secret"); + let (enabled_redirect_addr, enabled_redirect_thread) = + spawn_redirect_listener(&enabled_target_url); + let enabled_initial_url = format!("http://{enabled_redirect_addr}/start"); + cache_system_proxy_decision(&enabled_initial_url, SystemProxyDecision::Direct); + cache_system_proxy_decision(&enabled_target_url, SystemProxyDecision::Direct); + let enabled_pool = crate::RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + ); + + enabled_pool + .get(&enabled_initial_url) + .timeout(Duration::from_secs(2)) + .send() + .await + .expect_err("redirect target should fail"); + only_request(enabled_redirect_thread, "enabled redirect"); + enabled_failure_thread + .join() + .expect("enabled failure thread should finish"); + + let (disabled_failure_addr, disabled_failure_thread) = spawn_failing_listener(); + let disabled_target_url = + format!("http://{disabled_failure_addr}/final?token=disabled-target-secret"); + let (disabled_redirect_addr, disabled_redirect_thread) = + spawn_redirect_listener(&disabled_target_url); + let disabled_initial_url = format!("http://{disabled_redirect_addr}/start"); + cache_system_proxy_decision(&disabled_initial_url, SystemProxyDecision::Direct); + cache_system_proxy_decision(&disabled_target_url, SystemProxyDecision::Direct); + let disabled_pool = crate::RouteAwareClientPool::new_without_request_logging( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + ); + + disabled_pool + .get(&disabled_initial_url) + .timeout(Duration::from_secs(2)) + .send() + .await + .expect_err("redirect target should fail"); + only_request(disabled_redirect_thread, "disabled redirect"); + disabled_failure_thread + .join() + .expect("disabled failure thread should finish"); + + let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone()) + .expect("logs should be UTF-8"); + assert!(logs.contains("log capture sentinel")); + assert!(logs.contains(&enabled_initial_url)); + assert!(!logs.contains(&disabled_initial_url)); + assert_eq!(logs.matches("Request failed").count(), 1); + assert!(logs.contains("is_timeout")); + assert!(logs.contains("is_connect")); + for secret in [ + "enabled-target-secret", + "disabled-target-secret", + enabled_target_url.as_str(), + disabled_target_url.as_str(), + "error=", + ] { + assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}"); + } +} + +fn credential_headers(request: &str) -> (bool, bool, bool) { + let names = request + .lines() + .filter_map(|line| { + line.split_once(':') + .map(|(name, _)| name.to_ascii_lowercase()) + }) + .collect::>(); + ( + names.iter().any(|name| name == "authorization"), + names.iter().any(|name| name == "cookie"), + names.iter().any(|name| name == "proxy-authorization"), + ) +} + +fn spawn_failing_listener() -> (std::net::SocketAddr, std::thread::JoinHandle<()>) { + let listener = + std::net::TcpListener::bind(("127.0.0.1", 0)).expect("failing listener should bind"); + let address = listener + .local_addr() + .expect("failing listener should have an address"); + listener + .set_nonblocking(true) + .expect("failing listener should become nonblocking"); + let thread = std::thread::spawn(move || { + let deadline = Instant::now() + Duration::from_secs(2); + let (mut stream, _) = loop { + match listener.accept() { + Ok(connection) => break connection, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < deadline, + "failing listener should receive a request" + ); + std::thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("failing listener should accept: {error}"), + } + }; + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("failing stream should get a read timeout"); + read_http_message(&mut stream); + }); + (address, thread) +} diff --git a/codex-rs/http-client/src/outbound_proxy_tests.rs b/codex-rs/http-client/src/outbound_proxy_tests.rs index 956b03813e..b928350d7f 100644 --- a/codex-rs/http-client/src/outbound_proxy_tests.rs +++ b/codex-rs/http-client/src/outbound_proxy_tests.rs @@ -1,15 +1,121 @@ //! Shared outbound proxy policy tests. use super::*; +use http::header::AUTHORIZATION; +use http::header::COOKIE; +use http::header::PROXY_AUTHORIZATION; use pretty_assertions::assert_eq; use std::io::Read; use std::io::Write; use std::sync::Arc; +use std::sync::Mutex; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::SubscriberExt; + +#[path = "outbound_proxy_redirect_coverage_tests.rs"] +mod redirect_coverage_tests; struct MapEnv { values: HashMap, } +fn spawn_proxy_listener() -> (std::net::SocketAddr, std::thread::JoinHandle>) { + spawn_http_listener(vec![ + "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_string(), + ]) +} + +fn spawn_redirect_listener( + location: &str, +) -> (std::net::SocketAddr, std::thread::JoinHandle>) { + spawn_http_listener(vec![format!( + "HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + )]) +} + +fn spawn_http_listener( + responses: Vec, +) -> (std::net::SocketAddr, std::thread::JoinHandle>) { + 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"); + listener + .set_nonblocking(true) + .expect("HTTP listener should become nonblocking"); + let thread = std::thread::spawn(move || { + let mut requests = Vec::new(); + for response in responses { + let deadline = Instant::now() + Duration::from_secs(2); + let (mut stream, _) = loop { + match listener.accept() { + Ok(connection) => break connection, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < deadline, + "HTTP listener should receive the next request" + ); + std::thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("HTTP listener should accept: {error}"), + } + }; + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("HTTP stream should get a read timeout"); + requests.push(read_http_message(&mut stream)); + stream + .write_all(response.as_bytes()) + .expect("HTTP listener should write response"); + } + requests + }); + (address, thread) +} + +fn only_request(thread: std::thread::JoinHandle>, source: &str) -> String { + let requests = thread + .join() + .unwrap_or_else(|_| panic!("{source} thread should finish")); + let [request]: [String; 1] = requests.try_into().unwrap_or_else(|requests: Vec| { + panic!( + "{source} should receive one request, got {}", + requests.len() + ) + }); + request +} + +fn read_http_message(stream: &mut impl Read) -> String { + let mut buffer = Vec::new(); + let mut chunk = [0_u8; 1024]; + loop { + let bytes_read = stream.read(&mut chunk).expect("HTTP message should read"); + if bytes_read == 0 { + break; + } + buffer.extend_from_slice(&chunk[..bytes_read]); + if let Some(header_end) = buffer.windows(4).position(|window| window == b"\r\n\r\n") { + let body_start = header_end + 4; + let headers = String::from_utf8_lossy(&buffer[..body_start]); + let content_length = headers + .lines() + .filter_map(|line| line.split_once(':')) + .find_map(|(name, value)| { + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().ok()) + .flatten() + }) + .unwrap_or(0); + if buffer.len() >= body_start + content_length { + break; + } + } + } + String::from_utf8_lossy(&buffer).into_owned() +} + #[test] fn websocket_route_uses_http_equivalent_for_system_resolution() { let env = MapEnv { @@ -194,20 +300,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}"))]), }; @@ -231,7 +324,7 @@ 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 = only_request(proxy_thread, "proxy"); assert_eq!(response.status(), reqwest::StatusCode::OK); assert_eq!( @@ -240,6 +333,88 @@ 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 = only_request(proxy_thread, "proxy"); + + 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_logs_only_the_final_redirect_outcome() { + let (proxy_addr, proxy_thread) = spawn_proxy_listener(); + let redirected_url = "http://redirect-target.test/final?token=redirect-target-secret-value"; + 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 log_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(&log_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 = tokio::time::timeout(Duration::from_secs(2), pool.get(&initial_url).send()) + .await + .expect("redirected request should finish") + .expect("redirected request should use selected routes"); + only_request(redirect_thread, "redirect"); + only_request(proxy_thread, "proxy"); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone()) + .expect("logs should be UTF-8"); + assert!(logs.contains("log capture sentinel")); + assert!(logs.contains(&initial_url)); + assert_eq!(logs.matches("Request completed").count(), 1); + for secret in ["redirect-target-secret-value", redirected_url, "location"] { + assert!( + !logs + .to_ascii_lowercase() + .contains(&secret.to_ascii_lowercase()) + ); + } +} + #[test] fn parses_pac_proxy_tokens() { assert_eq!( @@ -264,6 +439,40 @@ fn unavailable_system_proxy_decision_is_cached() { assert_eq!(cached_system_proxy_decision(request_url), Some(decision)); } +#[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(()) + } +} + #[test] fn system_proxy_resolution_is_single_flight() { let cache = Arc::new(Mutex::new(HashMap::new()));