diff --git a/codex-rs/http-client/src/default_client.rs b/codex-rs/http-client/src/default_client.rs index 153f9372d7..99cbe39d7d 100644 --- a/codex-rs/http-client/src/default_client.rs +++ b/codex-rs/http-client/src/default_client.rs @@ -118,6 +118,18 @@ impl HttpClient { ); } } + pub(crate) fn log_error_summary(&self, method: &Method, url: &str, error: &reqwest::Error) { + if self.request_logging == RequestLogging::Enabled { + tracing::debug!( + method = %method, + url = %url, + status = error.status().map(|status| status.as_u16()), + is_timeout = error.is_timeout(), + is_connect = error.is_connect(), + "Request failed" + ); + } + } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] diff --git a/codex-rs/http-client/src/lib.rs b/codex-rs/http-client/src/lib.rs index e06535eab7..4f23da776c 100644 --- a/codex-rs/http-client/src/lib.rs +++ b/codex-rs/http-client/src/lib.rs @@ -6,6 +6,7 @@ mod error; mod outbound_proxy; mod request; mod route_aware_client_pool; +mod route_aware_redirect; mod transport; pub use crate::chatgpt_cloudflare_cookies::with_chatgpt_cloudflare_cookie_store; diff --git a/codex-rs/http-client/src/outbound_proxy.rs b/codex-rs/http-client/src/outbound_proxy.rs index 29797e5527..f0d1eaaa4a 100644 --- a/codex-rs/http-client/src/outbound_proxy.rs +++ b/codex-rs/http-client/src/outbound_proxy.rs @@ -819,6 +819,10 @@ fn proxy_env_value(env: &dyn EnvSource, upper: &str) -> Option { .filter(|value| !value.is_empty()) } +#[cfg(test)] +#[path = "route_aware_redirect_integration_tests.rs"] +mod redirect_integration_tests; + #[cfg(test)] #[path = "outbound_proxy_tests.rs"] mod tests; diff --git a/codex-rs/http-client/src/route_aware_client_pool.rs b/codex-rs/http-client/src/route_aware_client_pool.rs index 661fd0d878..0536533f4e 100644 --- a/codex-rs/http-client/src/route_aware_client_pool.rs +++ b/codex-rs/http-client/src/route_aware_client_pool.rs @@ -12,6 +12,7 @@ use http::HeaderValue; use http::Method; use http::StatusCode; use http::header::CONTENT_TYPE; +use http::header::PROXY_AUTHORIZATION; use reqwest::IntoUrl; use serde::Serialize; @@ -21,6 +22,12 @@ use crate::HttpClient; use crate::HttpClientFactory; use crate::OutboundProxyPolicy; use crate::OutboundProxyRoute; +use crate::route_aware_redirect::MAX_REDIRECTS; +use crate::route_aware_redirect::insert_referer; +use crate::route_aware_redirect::is_redirect; +use crate::route_aware_redirect::redirect_request; +use crate::route_aware_redirect::redirect_url; +use crate::route_aware_redirect::remove_sensitive_headers; use crate::with_chatgpt_cloudflare_cookie_store; const MAX_CACHED_ROUTES: usize = 16; @@ -28,7 +35,8 @@ const MAX_CACHED_ROUTES: usize = 16; /// Reuses transport clients by resolved route while selecting a route for every request URL. /// /// Request creation stays on the pool so the URL used for PAC or system-proxy resolution cannot -/// differ from the URL that is sent. +/// differ from the URL that is sent. Redirects are followed through the pool as new requests, so +/// each hop gets its own route decision while connections are still reused by route. #[derive(Clone)] pub struct RouteAwareClientPool { http_client_factory: HttpClientFactory, @@ -73,18 +81,28 @@ pub enum RouteAwareRequestError { Route(#[from] RouteAwareClientPoolError), #[error("failed to build route-aware request: {0}")] Build(String), + #[error("redirect target uses unsupported URL scheme: {0}")] + UnsupportedRedirectScheme(String), + #[error("too many redirects")] + TooManyRedirects, + #[error("route-aware request timed out")] + Timeout, } impl RouteAwareRequestError { pub fn status(&self) -> Option { match self { Self::Request(error) => error.status(), - Self::Route(_) | Self::Build(_) => None, + Self::Route(_) + | Self::Build(_) + | Self::UnsupportedRedirectScheme(_) + | Self::TooManyRedirects + | Self::Timeout => None, } } pub fn is_timeout(&self) -> bool { - matches!(self, Self::Request(error) if error.is_timeout()) + matches!(self, Self::Timeout) || matches!(self, Self::Request(error) if error.is_timeout()) } pub fn is_connect(&self) -> bool { @@ -308,28 +326,137 @@ impl RouteAwareClientPool { &self, request: reqwest::Request, ) -> Result { - let client = self.client_for_url(request.url().as_str()).await?; - Ok(client.execute(request).await?) - } - - async fn client_for_url( - &self, - request_url: &str, - ) -> Result { let http_client_factory = self.http_client_factory.clone(); - self.client_for_url_with_resolver(request_url, move |request_url| async move { - http_client_factory - .resolve_proxy_route_async(request_url) - .await + self.send_with_resolver(request, move |request_url| { + let http_client_factory = http_client_factory.clone(); + async move { + http_client_factory + .resolve_proxy_route_async(request_url) + .await + } }) .await } + async fn send_with_resolver( + &self, + mut request: reqwest::Request, + resolve_route: F, + ) -> Result + where + F: Fn(String) -> Fut, + Fut: Future>, + { + let request_method = request.method().clone(); + let request_url = request.url().to_string(); + let follows_redirects_manually = self.http_client_factory.outbound_proxy_policy() + == OutboundProxyPolicy::RespectSystemProxy; + let timeout_deadline = request + .timeout() + .copied() + .map(|timeout| tokio::time::Instant::now() + timeout); + let mut redirects = 0; + let mut previous_route = None; + loop { + let current_url = request.url().clone(); + let (current_route, client) = match timeout_deadline { + Some(timeout_deadline) => tokio::time::timeout_at( + timeout_deadline, + self.client_for_url_with_resolver(current_url.as_str(), &resolve_route), + ) + .await + .map_err(|_| RouteAwareRequestError::Timeout)??, + None => { + self.client_for_url_with_resolver(current_url.as_str(), &resolve_route) + .await? + } + }; + if previous_route + .as_ref() + .is_some_and(|previous_route| previous_route != ¤t_route) + { + request.headers_mut().remove(PROXY_AUTHORIZATION); + } + previous_route = Some(current_route); + if let Some(timeout_deadline) = timeout_deadline { + let remaining = timeout_deadline + .checked_duration_since(tokio::time::Instant::now()) + .ok_or(RouteAwareRequestError::Timeout)?; + if remaining.is_zero() { + return Err(RouteAwareRequestError::Timeout); + } + *request.timeout_mut() = Some(remaining); + } + let method = request.method().clone(); + let headers = request.headers().clone(); + let version = request.version(); + let timeout = request.timeout().copied(); + let replay = request.try_clone(); + let execute_request = async { + if follows_redirects_manually { + client.execute_without_request_logging(request).await + } else { + client.execute(request).await + } + }; + let response = match match timeout_deadline { + Some(timeout_deadline) => { + tokio::time::timeout_at(timeout_deadline, execute_request) + .await + .map_err(|_| RouteAwareRequestError::Timeout)? + } + None => execute_request.await, + } { + Ok(response) => response, + Err(error) => { + if follows_redirects_manually { + client.log_error_summary(&request_method, &request_url, &error); + } + return Err(error.into()); + } + }; + let status = response.status(); + if !is_redirect(status) { + if follows_redirects_manually { + client.log_response(&request_method, &request_url, &response); + } + return Ok(response); + } + let Some(next_url) = redirect_url(&response) else { + if follows_redirects_manually { + client.log_response(&request_method, &request_url, &response); + } + return Ok(response); + }; + let Some(mut next_request) = + redirect_request(status, method, headers, version, timeout, replay, next_url) + else { + if follows_redirects_manually { + client.log_response(&request_method, &request_url, &response); + } + return Ok(response); + }; + let next_request_url = next_request.url().clone(); + if !matches!(next_request_url.scheme(), "http" | "https") { + return Err(RouteAwareRequestError::UnsupportedRedirectScheme( + next_request_url.scheme().to_string(), + )); + } + if redirects >= MAX_REDIRECTS { + return Err(RouteAwareRequestError::TooManyRedirects); + } + remove_sensitive_headers(next_request.headers_mut(), ¤t_url, &next_request_url); + insert_referer(next_request.headers_mut(), ¤t_url, &next_request_url); + request = next_request; + redirects += 1; + } + } + async fn client_for_url_with_resolver( &self, request_url: &str, resolve_route: F, - ) -> Result + ) -> Result<(OutboundProxyRoute, HttpClient), RouteAwareClientPoolError> where F: FnOnce(String) -> Fut, Fut: Future>, @@ -342,7 +469,7 @@ impl RouteAwareClientPool { Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"), }; if let Some(client) = clients.get(&route) { - return Ok(client.clone()); + return Ok((route, client.clone())); } drop(clients); @@ -365,15 +492,15 @@ impl RouteAwareClientPool { Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"), }; if let Some(existing_client) = clients.get(&route) { - return Ok(existing_client.clone()); + return Ok((route, existing_client.clone())); } if clients.len() >= MAX_CACHED_ROUTES && let Some(route_to_evict) = clients.keys().next().cloned() { clients.remove(&route_to_evict); } - clients.insert(route, client.clone()); - Ok(client) + clients.insert(route.clone(), client.clone()); + Ok((route, client)) } } diff --git a/codex-rs/http-client/src/route_aware_client_pool_tests.rs b/codex-rs/http-client/src/route_aware_client_pool_tests.rs index ff74a4e92e..caca3454fe 100644 --- a/codex-rs/http-client/src/route_aware_client_pool_tests.rs +++ b/codex-rs/http-client/src/route_aware_client_pool_tests.rs @@ -249,8 +249,10 @@ async fn resolve_with( request_url: &str, ) -> Result { let resolver = resolver.clone(); - pool.client_for_url_with_resolver(request_url, move |request_url| async move { - resolver.resolve(request_url).await - }) - .await + let (_, client) = pool + .client_for_url_with_resolver(request_url, move |request_url| async move { + resolver.resolve(request_url).await + }) + .await?; + Ok(client) } diff --git a/codex-rs/http-client/src/route_aware_redirect.rs b/codex-rs/http-client/src/route_aware_redirect.rs new file mode 100644 index 0000000000..25ee9ed2b9 --- /dev/null +++ b/codex-rs/http-client/src/route_aware_redirect.rs @@ -0,0 +1,138 @@ +//! Redirect semantics for requests whose proxy route is selected from the complete URL. +//! +//! Reqwest normally follows redirects inside one `Client::execute` call. That is safe for +//! `ReqwestDefault`, where reqwest owns proxy selection, but not for `RespectSystemProxy`: Codex +//! resolves PAC and operating-system proxy settings before building a client pinned to the +//! resulting direct or concrete-proxy route. An internal reqwest redirect would reuse that client +//! for the new URL without giving [`crate::RouteAwareClientPool`] an opportunity to resolve the +//! redirect destination. +//! +//! The pool therefore disables reqwest redirects for `RespectSystemProxy`, observes each redirect +//! response, and uses these helpers to construct the next request before resolving its route. The +//! helpers preserve the relevant reqwest method/body behavior while applying stricter +//! origin-sensitive credential and `Referer` rules at the boundary between requests. + +use std::time::Duration; + +use http::HeaderMap; +use http::Method; +use http::StatusCode; +use http::header::AUTHORIZATION; +use http::header::CONTENT_ENCODING; +use http::header::CONTENT_LENGTH; +use http::header::CONTENT_TYPE; +use http::header::COOKIE; +use http::header::LOCATION; +use http::header::PROXY_AUTHORIZATION; +use http::header::REFERER; +use http::header::TRANSFER_ENCODING; +use http::header::WWW_AUTHENTICATE; + +pub(super) const MAX_REDIRECTS: usize = 10; + +pub(super) fn is_redirect(status: StatusCode) -> bool { + matches!( + status, + StatusCode::MOVED_PERMANENTLY + | StatusCode::FOUND + | StatusCode::SEE_OTHER + | StatusCode::TEMPORARY_REDIRECT + | StatusCode::PERMANENT_REDIRECT + ) +} + +pub(super) fn redirect_url(response: &reqwest::Response) -> Option { + let location = response.headers().get(LOCATION)?.to_str().ok()?; + response.url().join(location).ok() +} + +pub(super) fn redirect_request( + status: StatusCode, + mut method: Method, + mut headers: HeaderMap, + version: http::Version, + timeout: Option, + replay: Option, + next_url: reqwest::Url, +) -> Option { + let drop_body = match status { + StatusCode::MOVED_PERMANENTLY | StatusCode::FOUND if method == Method::POST => { + method = Method::GET; + true + } + StatusCode::SEE_OTHER => { + if method != Method::HEAD { + method = Method::GET; + } + true + } + StatusCode::MOVED_PERMANENTLY + | StatusCode::FOUND + | StatusCode::TEMPORARY_REDIRECT + | StatusCode::PERMANENT_REDIRECT => false, + _ => return None, + }; + + if drop_body { + for header in [ + CONTENT_TYPE, + CONTENT_LENGTH, + CONTENT_ENCODING, + TRANSFER_ENCODING, + ] { + headers.remove(header); + } + let mut request = reqwest::Request::new(method, next_url); + *request.headers_mut() = headers; + *request.version_mut() = version; + *request.timeout_mut() = timeout; + Some(request) + } else { + replay.map(|mut request| { + *request.url_mut() = next_url; + request + }) + } +} + +pub(super) fn remove_sensitive_headers( + headers: &mut HeaderMap, + previous: &reqwest::Url, + next: &reqwest::Url, +) { + if !same_origin(previous, next) { + for header in [AUTHORIZATION, COOKIE, PROXY_AUTHORIZATION, WWW_AUTHENTICATE] { + headers.remove(header); + } + headers.remove("cookie2"); + } +} + +pub(super) fn insert_referer( + headers: &mut HeaderMap, + previous: &reqwest::Url, + next: &reqwest::Url, +) { + headers.remove(REFERER); + if next.scheme() == "http" && previous.scheme() == "https" { + return; + } + + let mut referer = previous.clone(); + let _ = referer.set_username(""); + let _ = referer.set_password(None); + referer.set_fragment(None); + if !same_origin(previous, next) { + referer.set_path("/"); + referer.set_query(None); + } + if let Ok(value) = referer.as_str().parse() { + headers.insert(REFERER, value); + } +} + +fn same_origin(previous: &reqwest::Url, next: &reqwest::Url) -> bool { + previous.scheme() == next.scheme() + && previous.host_str() == next.host_str() + && previous.port_or_known_default() == next.port_or_known_default() +} diff --git a/codex-rs/http-client/src/route_aware_redirect_integration_tests.rs b/codex-rs/http-client/src/route_aware_redirect_integration_tests.rs new file mode 100644 index 0000000000..91119ca08c --- /dev/null +++ b/codex-rs/http-client/src/route_aware_redirect_integration_tests.rs @@ -0,0 +1,168 @@ +use std::io; +use std::io::Read; +use std::io::Write; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; +use std::time::Instant; + +use http::header::AUTHORIZATION; +use http::header::COOKIE; +use http::header::PROXY_AUTHORIZATION; +use pretty_assertions::assert_eq; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::SubscriberExt; + +use super::*; +use crate::RouteAwareClientPool; + +#[tokio::test] +async fn route_aware_pool_re_resolves_redirects_and_logs_only_final_outcome() { + let (proxy_addr, proxy_thread) = + spawn_response("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok"); + let (redirect_addr, redirect_thread) = spawn_response( + "HTTP/1.1 302 Found\r\nLocation: /final?token=redirect-target-secret\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ); + let initial_url = format!("http://{redirect_addr}/start"); + let redirected_url = format!("http://{redirect_addr}/final?token=redirect-target-secret"); + cache_system_proxy_decision(&initial_url, SystemProxyDecision::Direct); + cache_system_proxy_decision( + &redirected_url, + SystemProxyDecision::Proxy { + url: format!("http://{proxy_addr}"), + }, + ); + let pool = 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); + + 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("redirected request should use both selected routes"); + redirect_thread + .join() + .expect("redirect listener should finish"); + let redirected_request = proxy_thread.join().expect("proxy listener should finish"); + assert_eq!( + redirected_request.lines().next(), + Some(format!("GET {redirected_url} HTTP/1.1").as_str()) + ); + assert!(has_header(&redirected_request, "authorization")); + assert!(has_header(&redirected_request, "cookie")); + assert!(!has_header(&redirected_request, "proxy-authorization")); + + let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone()) + .expect("logs should be UTF-8"); + assert!(logs.contains(&initial_url)); + assert_eq!(logs.matches("Request completed").count(), 1); + assert!(!logs.contains("redirect-target-secret")); +} + +fn spawn_response( + response: &'static str, +) -> (std::net::SocketAddr, std::thread::JoinHandle) { + let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("bind listener"); + let address = listener.local_addr().expect("HTTP listener address"); + listener.set_nonblocking(true).expect("set 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, "request timed out"); + 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("read timeout"); + let request = read_http_headers(&mut stream); + stream + .write_all(response.as_bytes()) + .expect("write response"); + request + }); + (address, thread) +} + +fn read_http_headers(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 headers should read"); + if bytes_read == 0 { + break; + } + buffer.extend_from_slice(&chunk[..bytes_read]); + if buffer.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + String::from_utf8_lossy(&buffer).into_owned() +} + +fn has_header(request: &str, expected_name: &str) -> bool { + request.lines().any(|line| { + line.split_once(':') + .is_some_and(|(name, _)| name.eq_ignore_ascii_case(expected_name)) + }) +} + +#[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(()) + } +}