diff --git a/codex-rs/codex-api/src/endpoint/mod.rs b/codex-rs/codex-api/src/endpoint/mod.rs index 2e4ef9e766..53a2b72b88 100644 --- a/codex-rs/codex-api/src/endpoint/mod.rs +++ b/codex-rs/codex-api/src/endpoint/mod.rs @@ -30,5 +30,4 @@ pub use responses_websocket::ResponsesWebsocketClient; pub use responses_websocket::ResponsesWebsocketClose; pub use responses_websocket::ResponsesWebsocketConnection; pub use responses_websocket::ResponsesWebsocketProbe; -pub use search::SEARCH_ENDPOINT; pub use search::SearchClient; diff --git a/codex-rs/codex-api/src/endpoint/search.rs b/codex-rs/codex-api/src/endpoint/search.rs index b8535fe99c..2de48fe7db 100644 --- a/codex-rs/codex-api/src/endpoint/search.rs +++ b/codex-rs/codex-api/src/endpoint/search.rs @@ -12,7 +12,7 @@ use serde_json::to_value; use std::sync::Arc; /// The provider-relative endpoint for standalone web search. -pub const SEARCH_ENDPOINT: &str = "alpha/search"; +const SEARCH_ENDPOINT: &str = "alpha/search"; pub struct SearchClient { session: EndpointSession, diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 9f408adc86..165bf6cc50 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -70,7 +70,6 @@ pub use crate::endpoint::ResponsesWebsocketClient; pub use crate::endpoint::ResponsesWebsocketClose; pub use crate::endpoint::ResponsesWebsocketConnection; pub use crate::endpoint::ResponsesWebsocketProbe; -pub use crate::endpoint::SEARCH_ENDPOINT; pub use crate::endpoint::SearchClient; pub use crate::endpoint::session_update_session_json; pub use crate::error::ApiError; diff --git a/codex-rs/core/tests/suite/web_search_system_proxy_tests.rs b/codex-rs/core/tests/suite/web_search_system_proxy_tests.rs index c182fba9b5..eb5b8a56c4 100644 --- a/codex-rs/core/tests/suite/web_search_system_proxy_tests.rs +++ b/codex-rs/core/tests/suite/web_search_system_proxy_tests.rs @@ -1,4 +1,4 @@ -//! Standalone web search must use the configured system proxy route. +//! Standalone web search must resolve the configured route for every redirect destination. use anyhow::Result; use codex_core::config::Config; @@ -12,6 +12,8 @@ use codex_web_search_extension::install as install_web_search_extension; use core_test_support::responses; use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; +use http::HeaderMap; +use http::StatusCode; use pretty_assertions::assert_eq; use serde_json::json; use std::sync::Arc; @@ -20,16 +22,18 @@ use tokio::process::Command; use tokio::time::timeout; use wiremock::Mock; use wiremock::ResponseTemplate; +use wiremock::matchers::header; use wiremock::matchers::method; use wiremock::matchers::path; const TEST_NAME: &str = - "suite::web_search_system_proxy::standalone_web_search_honors_respect_system_proxy"; + "suite::web_search_system_proxy::standalone_web_search_resolves_redirect_routes"; const TEST_SUBPROCESS_ENV_VAR: &str = "CODEX_WEB_SEARCH_SYSTEM_PROXY_TEST_SUBPROCESS"; const API_BASE_URL: &str = "http://web-search-proxy.invalid/v1"; +const REDIRECT_URL: &str = "http://web-search-redirect.invalid/v1/alpha/search"; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn standalone_web_search_honors_respect_system_proxy() -> Result<()> { +async fn standalone_web_search_resolves_redirect_routes() -> Result<()> { skip_if_no_network!(Ok(())); if std::env::var_os(TEST_SUBPROCESS_ENV_VAR).is_none() { @@ -58,14 +62,28 @@ async fn standalone_web_search_honors_respect_system_proxy() -> Result<()> { } let proxy = responses::start_mock_server().await; + let redirect_proxy = responses::start_mock_server().await; Mock::given(method("POST")) .and(path("/v1/alpha/search")) - .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "output": "Search result through system proxy", - }))) - .expect(1) + .and(header("host", "web-search-proxy.invalid")) + .respond_with( + ResponseTemplate::new(StatusCode::TEMPORARY_REDIRECT.as_u16()) + .insert_header("location", REDIRECT_URL), + ) + .expect(/*r*/ 1) .mount(&proxy) .await; + Mock::given(method("POST")) + .and(path("/v1/alpha/search")) + .and(header("host", "web-search-redirect.invalid")) + .respond_with( + ResponseTemplate::new(StatusCode::OK.as_u16()).set_body_json(json!({ + "output": "Search result through system proxy", + })), + ) + .expect(/*r*/ 1) + .mount(&redirect_proxy) + .await; responses::mount_sse_once( &proxy, responses::sse(vec![ @@ -117,9 +135,42 @@ async fn standalone_web_search_honors_respect_system_proxy() -> Result<()> { for endpoint in ["responses", "alpha/search"] { cache_system_proxy_route_for_test(&format!("{API_BASE_URL}/{endpoint}"), proxy.uri()); } + cache_system_proxy_route_for_test(REDIRECT_URL, redirect_proxy.uri()); test.submit_turn("search the web through the system proxy") .await?; + let requests = proxy + .received_requests() + .await + .expect("initial proxy requests"); + let initial_search = requests + .iter() + .find(|request| request.url.path() == "/v1/alpha/search") + .expect("initial proxy should receive search"); + let redirected_requests = redirect_proxy + .received_requests() + .await + .expect("redirect proxy requests"); + let redirected_search = redirected_requests + .first() + .expect("redirect proxy should receive search"); + assert_eq!(initial_search.body, redirected_search.body); + let default_headers = codex_login::default_client::default_headers(); + let observed_default_headers = [initial_search, redirected_search].map(|request| { + default_headers + .keys() + .filter_map(|name| { + request + .headers + .get(name) + .map(|value| (name.clone(), value.clone())) + }) + .collect::() + }); + assert_eq!( + observed_default_headers, + [default_headers.clone(), default_headers] + ); assert_eq!( follow_up_mock .single_request() diff --git a/codex-rs/ext/web-search/src/tool.rs b/codex-rs/ext/web-search/src/tool.rs index 603491d046..411b405bd8 100644 --- a/codex-rs/ext/web-search/src/tool.rs +++ b/codex-rs/ext/web-search/src/tool.rs @@ -1,5 +1,3 @@ -use codex_api::ReqwestTransport; -use codex_api::SEARCH_ENDPOINT; use codex_api::SearchClient; use codex_api::SearchCommands; use codex_api::SearchQuery; @@ -21,9 +19,8 @@ use codex_extension_items::web_search::WebSearchAction; use codex_extension_items::web_search::WebSearchItem; use codex_http_client::ClientRouteClass; use codex_http_client::HttpClientFactory; -use codex_login::default_client::ClientRedirectPolicy; use codex_login::default_client::add_originator_header; -use codex_login::default_client::create_client_for_route_async; +use codex_login::default_client::create_transport_for_routes_async; use codex_model_provider::SharedModelProvider; use codex_protocol::models::WebSearchAction as CoreWebSearchAction; use codex_protocol::protocol::EventMsg; @@ -113,19 +110,13 @@ impl WebSearchTool { .api_auth() .await .map_err(|err| FunctionCallError::Fatal(err.to_string()))?; - let http_client = create_client_for_route_async( + let transport = create_transport_for_routes_async( self.http_client_factory.clone(), - provider.url_for_path(SEARCH_ENDPOINT), ClientRouteClass::Api, - ClientRedirectPolicy::Default, ) .await .map_err(|err| FunctionCallError::Fatal(err.to_string()))?; - let client = SearchClient::new( - ReqwestTransport::from_http_client(http_client), - provider, - auth, - ); + let client = SearchClient::new(transport, provider, auth); let request = SearchRequest { id: self.session_id.clone(), model: call.model.clone(), diff --git a/codex-rs/http-client/src/client.rs b/codex-rs/http-client/src/client.rs index 420d5ad1b7..b5b1ba9145 100644 --- a/codex-rs/http-client/src/client.rs +++ b/codex-rs/http-client/src/client.rs @@ -1,4 +1,4 @@ -//! Reusable HTTP client and request-builder wrappers. +//! Reusable HTTP client wrappers and request construction shared by fixed and route-aware transports. use http::Error as HttpRequestBuildError; use http::HeaderMap; @@ -249,6 +249,10 @@ impl RequestBuilder { self.map(|builder| builder.body(body)) } + pub(crate) fn build(self) -> Result { + self.builder.build() + } + pub async fn send(self) -> Result { let headers = trace_headers(); diff --git a/codex-rs/http-client/src/client_builder.rs b/codex-rs/http-client/src/client_builder.rs index db5ef2ccbe..8b7ae3de5c 100644 --- a/codex-rs/http-client/src/client_builder.rs +++ b/codex-rs/http-client/src/client_builder.rs @@ -105,6 +105,10 @@ impl HttpClientBuilder { self.follow_redirects } + pub(crate) fn request_logging_enabled(&self) -> bool { + self.request_logging == RequestLogging::Enabled + } + pub(crate) fn with_rustls_tls(mut self) -> Self { self.tls_backend = TlsBackend::Rustls; self diff --git a/codex-rs/http-client/src/lib.rs b/codex-rs/http-client/src/lib.rs index f9cdb1fb20..e2f81e823e 100644 --- a/codex-rs/http-client/src/lib.rs +++ b/codex-rs/http-client/src/lib.rs @@ -7,6 +7,7 @@ mod error; mod network_policy; mod outbound_proxy; mod request; +mod request_draft; mod route_aware_client_pool; mod route_aware_redirect; mod tls_backend_fallback; 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 index 8aefa93f1e..70aea74094 100644 --- a/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs +++ b/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs @@ -1,4 +1,10 @@ use super::*; +use crate::HttpTransport; +use crate::Request; +use crate::ReqwestTransport; +use crate::TransportError; +use futures::TryStreamExt; +use http::Method; use pretty_assertions::assert_eq; use std::sync::Arc; use std::sync::Mutex; @@ -6,6 +12,134 @@ use std::time::Duration; use tracing_subscriber::Layer; use tracing_subscriber::layer::SubscriberExt; +#[tokio::test] +async fn transport_resolves_routes_for_execute_and_stream_redirects() { + let initial_url = "http://transport-start.test/start"; + let destination_url = "http://transport-final.test/final"; + let (destination_addr, destination_thread) = spawn_http_listener(vec![ + "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_string(); + 2 + ]); + let (initial_addr, initial_thread) = spawn_http_listener(vec![ + format!( + "HTTP/1.1 307 Temporary Redirect\r\nLocation: {destination_url}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + ); + 2 + ]); + for (url, address) in [ + (initial_url, initial_addr), + (destination_url, destination_addr), + ] { + cache_system_proxy_decision( + url, + SystemProxyDecision::Proxy { + url: format!("http://{address}"), + }, + ); + } + let transport = + ReqwestTransport::from_route_aware_client_pool(crate::RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + )); + let mut request = Request::new(Method::POST, initial_url.to_string()) + .with_json(&serde_json::json!({"input": "redirected request"})); + request.timeout = Some(Duration::from_secs(/*secs*/ 2)); + + let response = transport + .execute(request.clone()) + .await + .expect("execute should follow the routed redirect"); + let streamed = transport + .stream(request) + .await + .expect("stream should follow the routed redirect"); + let streamed_body = streamed + .bytes + .try_collect::>() + .await + .expect("streamed body should be readable") + .concat(); + + assert_eq!( + (response.body.as_ref(), streamed_body.as_slice()), + (&b"ok"[..], &b"ok"[..]) + ); + let initial_requests = initial_thread.join().expect("initial proxy should finish"); + let destination_requests = destination_thread + .join() + .expect("destination proxy should finish"); + assert_eq!((initial_requests.len(), destination_requests.len()), (2, 2)); + for (initial, destination) in initial_requests.iter().zip(&destination_requests) { + assert!(initial.starts_with("POST http://transport-start.test/start HTTP/1.1\r\n")); + assert!(destination.starts_with("POST http://transport-final.test/final HTTP/1.1\r\n")); + assert_eq!( + ( + initial.split_once("\r\n\r\n").map(|(_, body)| body), + destination.split_once("\r\n\r\n").map(|(_, body)| body), + ), + ( + Some(r#"{"input":"redirected request"}"#), + Some(r#"{"input":"redirected request"}"#), + ) + ); + } +} + +#[tokio::test] +async fn transport_classifies_route_construction_failures_as_build_errors() { + let request_url = "http://transport-invalid-proxy.test/request"; + cache_system_proxy_decision( + request_url, + SystemProxyDecision::Proxy { + url: "://invalid".to_string(), + }, + ); + let transport = + ReqwestTransport::from_route_aware_client_pool(crate::RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + )); + + let error = transport + .execute(Request::new(Method::GET, request_url.to_string())) + .await + .expect_err("invalid proxy configuration should fail"); + + assert!(matches!(error, TransportError::Build(_))); +} + +#[tokio::test] +async fn transport_classifies_connection_failures_without_request_urls() { + let listener = + std::net::TcpListener::bind(("127.0.0.1", 0)).expect("unavailable proxy port should bind"); + let address = listener.local_addr().expect("proxy should have an address"); + drop(listener); + let request_url = "http://transport-unavailable-proxy.test/request?token=url-secret"; + cache_system_proxy_decision( + request_url, + SystemProxyDecision::Proxy { + url: format!("http://{address}"), + }, + ); + let transport = + ReqwestTransport::from_route_aware_client_pool(crate::RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ClientRouteClass::Api, + )); + + let error = match transport + .stream(Request::new(Method::GET, request_url.to_string())) + .await + { + Err(TransportError::Connection(error)) => error, + Err(error) => panic!("expected a connection failure, got {error}"), + Ok(_) => panic!("unavailable proxy should not return a response"), + }; + + assert!(!error.to_string().contains("url-secret")); +} + #[tokio::test] async fn route_aware_pool_strips_credentials_on_cross_origin_redirect() { let (destination_addr, destination_thread) = spawn_proxy_listener(); @@ -87,6 +221,32 @@ async fn route_aware_pool_retains_credentials_for_same_origin_and_route() { #[tokio::test] async fn route_aware_pool_sanitizes_redirected_failure_logs() { + const SUBPROCESS_ENV_VAR: &str = "CODEX_HTTP_CLIENT_REDIRECT_FAILURE_LOG_TEST"; + if std::env::var_os(SUBPROCESS_ENV_VAR).is_none() { + // Bazel runs unit tests in one process. Isolate the tracing subscriber and callsite + // cache from other tests that install different subscribers while requests are in flight. + let test_module = module_path!() + .split_once("::") + .expect("test module should include the crate name") + .1; + let test_name = + format!("{test_module}::route_aware_pool_sanitizes_redirected_failure_logs"); + let output = std::process::Command::new( + std::env::current_exe().expect("test executable should be available"), + ) + .args(["--exact", &test_name, "--nocapture"]) + .env(SUBPROCESS_ENV_VAR, "1") + .output() + .expect("logging test subprocess should finish"); + assert!( + output.status.success(), + "logging test subprocess failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + return; + } + let log_buffer = Arc::new(Mutex::new(Vec::new())); let subscriber = tracing_subscriber::registry().with( tracing_subscriber::fmt::layer() @@ -115,12 +275,16 @@ async fn route_aware_pool_sanitizes_redirected_failure_logs() { ClientRouteClass::Api, ); - enabled_pool + let error = enabled_pool .get(&enabled_initial_url) .timeout(Duration::from_secs(2)) .send() .await .expect_err("redirect target should fail"); + assert!( + matches!(error, crate::RouteAwareRequestError::Request(_)), + "expected a transport error from the redirect target, got {error:?}", + ); only_request(enabled_redirect_thread, "enabled redirect"); enabled_failure_thread .join() @@ -153,7 +317,10 @@ async fn route_aware_pool_sanitizes_redirected_failure_logs() { 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(&enabled_initial_url), + "captured logs:\n{logs}" + ); assert!(!logs.contains(&disabled_initial_url)); assert_eq!(logs.matches("Request failed").count(), 1); assert!(logs.contains("is_timeout")); diff --git a/codex-rs/http-client/src/request_draft.rs b/codex-rs/http-client/src/request_draft.rs new file mode 100644 index 0000000000..3d62437b1c --- /dev/null +++ b/codex-rs/http-client/src/request_draft.rs @@ -0,0 +1,62 @@ +//! Request drafts shared by fixed clients and route-aware pools. +//! +//! Replay header operations in order so reqwest owns URL authentication and header precedence. +//! Header operations are the sole header state until build; the draft's request holds no headers. +//! Redirects and TLS retries use the resulting request without repeating this construction. + +use http::HeaderMap; +use http::HeaderName; +use http::HeaderValue; +use http::Method; +use reqwest::IntoUrl; + +use crate::HttpClient; + +/// Request inputs awaiting construction through reqwest with a selected HTTP client. +/// +/// Route-aware pools select a client for the destination before calling [`Self::build`]; fixed +/// transports use their existing client. Building applies URL authentication and the queued header +/// operations once, then redirects and TLS retries reuse the resulting request. +pub(crate) struct RequestDraft { + pub(crate) request: reqwest::Request, + pub(crate) headers: Vec, +} + +pub(crate) enum HeaderUpdate { + Replace(HeaderMap), + Append(HeaderName, HeaderValue), +} + +impl From for RequestDraft { + fn from(mut request: reqwest::Request) -> Self { + let headers = vec![HeaderUpdate::Replace(std::mem::take(request.headers_mut()))]; + Self { request, headers } + } +} + +impl RequestDraft { + pub(crate) fn new(method: Method, url: U) -> Result { + url.into_url() + .map(|url| Self::from(reqwest::Request::new(method, url))) + } + + pub(crate) fn extend_headers(&mut self, headers: HeaderMap) { + self.headers.push(HeaderUpdate::Replace(headers)); + } + + pub(crate) fn build(self, client: &HttpClient) -> Result { + let mut request = self.request; + let mut builder = client.request(request.method().clone(), request.url().clone()); + for update in self.headers { + builder = match update { + HeaderUpdate::Replace(headers) => builder.headers(headers), + HeaderUpdate::Append(name, value) => builder.header(name, value), + }; + } + let mut built = builder.build()?; + *built.body_mut() = request.body_mut().take(); + *built.timeout_mut() = request.timeout().copied(); + *built.version_mut() = request.version(); + Ok(built) + } +} 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 9ad9b2c567..9fdeada936 100644 --- a/codex-rs/http-client/src/route_aware_client_pool.rs +++ b/codex-rs/http-client/src/route_aware_client_pool.rs @@ -27,6 +27,8 @@ use crate::HttpClientFactory; use crate::OutboundProxyPolicy; use crate::OutboundProxyRoute; use crate::RouteFailureClass; +use crate::request_draft::HeaderUpdate; +use crate::request_draft::RequestDraft; use crate::tls_backend_fallback::RustlsClientCache; const MAX_CACHED_ROUTES: usize = 16; @@ -46,8 +48,8 @@ enum SelectedTlsBackend { /// 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. Redirects are followed through the pool as new requests, so +/// Resolves the initial route from the complete input URL, before reqwest handles URL credentials. +/// 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 { @@ -183,7 +185,7 @@ impl RouteAwareRequestError { #[must_use = "requests are not sent unless `send` is awaited"] pub struct RouteAwareRequestBuilder { pool: RouteAwareClientPool, - request: Result, + request: Result, } impl fmt::Debug for RouteAwareRequestBuilder { @@ -192,7 +194,7 @@ impl fmt::Debug for RouteAwareRequestBuilder { formatter .debug_struct("RouteAwareRequestBuilder") .field("pool", &self.pool) - .field("method", &request.map(reqwest::Request::method)) + .field("method", &request.map(|request| request.request.method())) .field("url", &request.map(|_| "")) .finish_non_exhaustive() } @@ -203,16 +205,13 @@ impl RouteAwareRequestBuilder { where U: IntoUrl, { - let request = url - .into_url() - .map(|url| reqwest::Request::new(method, url)) - .map_err(RouteAwareRequestError::Request); + let request = RequestDraft::new(method, url).map_err(RouteAwareRequestError::Request); Self { pool, request } } pub fn headers(mut self, headers: HeaderMap) -> Self { if let Ok(request) = &mut self.request { - request.headers_mut().extend(headers); + request.extend_headers(headers); } self } @@ -234,7 +233,7 @@ impl RouteAwareRequestBuilder { }); match header { Ok((key, value)) => { - request.headers_mut().append(key, value); + request.headers.push(HeaderUpdate::Append(key, value)); } Err(error) => { self.request = Err(RouteAwareRequestError::Build(error.to_string())); @@ -252,7 +251,7 @@ impl RouteAwareRequestBuilder { /// bounded. pub fn timeout(mut self, timeout: Duration) -> Self { if let Ok(request) = &mut self.request { - *request.timeout_mut() = Some(timeout); + *request.request.timeout_mut() = Some(timeout); } self } @@ -264,12 +263,17 @@ impl RouteAwareRequestBuilder { if let Ok(request) = &mut self.request { match serde_json::to_vec(value) { Ok(body) => { - if !request.headers().contains_key(CONTENT_TYPE) { - request - .headers_mut() - .insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + let has_content_type = request.headers.iter().any(|update| match update { + HeaderUpdate::Replace(headers) => headers.contains_key(CONTENT_TYPE), + HeaderUpdate::Append(name, _) => *name == CONTENT_TYPE, + }); + if !has_content_type { + request.headers.push(HeaderUpdate::Append( + CONTENT_TYPE, + HeaderValue::from_static("application/json"), + )); } - *request.body_mut() = Some(body.into()); + *request.request.body_mut() = Some(body.into()); } Err(error) => { self.request = Err(RouteAwareRequestError::Build(error.to_string())); @@ -284,7 +288,7 @@ impl RouteAwareRequestBuilder { B: Into, { if let Ok(request) = &mut self.request { - *request.body_mut() = Some(body.into()); + *request.request.body_mut() = Some(body.into()); } self } @@ -297,7 +301,7 @@ impl RouteAwareRequestBuilder { Bytes: From, { if let Ok(request) = &mut self.request { - *request.body_mut() = Some(reqwest::Body::wrap_stream(stream)); + *request.request.body_mut() = Some(reqwest::Body::wrap_stream(stream)); } self } @@ -308,6 +312,10 @@ impl RouteAwareRequestBuilder { } impl RouteAwareClientPool { + pub(crate) fn request_logging_enabled(&self) -> bool { + self.client_builder.request_logging_enabled() + } + pub fn outbound_proxy_policy(&self) -> OutboundProxyPolicy { self.http_client_factory.outbound_proxy_policy() } @@ -421,11 +429,26 @@ impl RouteAwareClientPool { pub fn with_chatgpt_cloudflare_cookies( http_client_factory: HttpClientFactory, route_class: ClientRouteClass, + ) -> Self { + Self::with_chatgpt_cloudflare_cookies_and_default_headers( + http_client_factory, + route_class, + HeaderMap::new(), + ) + } + + /// Creates a pool with ChatGPT Cloudflare cookies and default headers on every route. + pub fn with_chatgpt_cloudflare_cookies_and_default_headers( + http_client_factory: HttpClientFactory, + route_class: ClientRouteClass, + default_headers: HeaderMap, ) -> Self { Self::with_builder( http_client_factory, route_class, - HttpClientBuilder::new().with_chatgpt_cloudflare_cookie_store(), + HttpClientBuilder::new() + .default_headers(default_headers) + .with_chatgpt_cloudflare_cookie_store(), ) } diff --git a/codex-rs/http-client/src/route_aware_client_pool/execution.rs b/codex-rs/http-client/src/route_aware_client_pool/execution.rs index 3c91ff7675..df4d11045e 100644 --- a/codex-rs/http-client/src/route_aware_client_pool/execution.rs +++ b/codex-rs/http-client/src/route_aware_client_pool/execution.rs @@ -9,6 +9,7 @@ use super::RouteAwareClientPool; use super::RouteAwareRequestError; use super::SelectedTlsBackend; use crate::OutboundProxyRoute; +use crate::request_draft::RequestDraft; use crate::route_aware_redirect::MAX_REDIRECTS; use crate::route_aware_redirect::insert_referer; use crate::route_aware_redirect::is_redirect; @@ -18,9 +19,9 @@ use crate::route_aware_redirect::remove_sensitive_headers; use crate::tls_backend_fallback::should_retry_with_rustls; impl RouteAwareClientPool { - pub(super) async fn send( + pub(crate) async fn send( &self, - request: reqwest::Request, + request: RequestDraft, ) -> Result { let http_client_factory = self.http_client_factory.clone(); self.send_with_resolver(request, move |request_url| { @@ -36,15 +37,18 @@ impl RouteAwareClientPool { pub(super) async fn send_with_resolver( &self, - mut request: reqwest::Request, + request: impl Into, resolve_route: F, ) -> Result where F: Fn(String) -> Fut, Fut: Future>, { + let draft = request.into(); + let mut request = draft.request; + let mut draft_headers = Some(draft.headers); let request_method = request.method().clone(); - let request_url = request.url().to_string(); + let mut request_url = request.url().to_string(); let follows_redirects_manually = self.follows_redirects_manually(); let timeout_deadline = request .timeout() @@ -53,19 +57,23 @@ impl RouteAwareClientPool { let mut redirects = 0; let mut previous_route = None; loop { - let current_url = request.url().clone(); let (current_route, client, selected_tls_backend) = match timeout_deadline { Some(timeout_deadline) => tokio::time::timeout_at( timeout_deadline, - self.client_for_url_with_resolver(current_url.as_str(), &resolve_route), + self.client_for_url_with_resolver(request.url().as_str(), &resolve_route), ) .await .map_err(|_| RouteAwareRequestError::Timeout)??, None => { - self.client_for_url_with_resolver(current_url.as_str(), &resolve_route) + self.client_for_url_with_resolver(request.url().as_str(), &resolve_route) .await? } }; + if let Some(headers) = draft_headers.take() { + request = RequestDraft { request, headers }.build(&client)?; + request_url = request.url().to_string(); + } + let current_url = request.url().clone(); if previous_route .as_ref() .is_some_and(|previous_route| previous_route != ¤t_route) 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 8f903f4d3c..4906f571ea 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 @@ -20,6 +20,206 @@ use tracing_subscriber::layer::SubscriberExt; use super::*; use crate::OutboundProxyPolicy; +#[test] +fn request_draft_preserves_reqwest_url_auth_and_header_precedence() { + let pool = manual_redirect_pool(); + let client = HttpClientBuilder::new() + .build_direct() + .expect("client should build"); + for url in [ + "https://example.com/path", + "https://user:p%40ss@example.com/path", + "https://:%E2%98%83@example.com/path", + "https://%FF:password@example.com/path", + ] { + let headers = HeaderMap::from_iter([ + ( + http::header::AUTHORIZATION, + HeaderValue::from_static("Bearer replacement"), + ), + ( + CONTENT_TYPE, + HeaderValue::from_static("application/custom+json"), + ), + ]); + for (actual, expected) in [ + (pool.post(url), client.post(url)), + ( + pool.post(url) + .header(http::header::AUTHORIZATION, "Bearer appended"), + client + .post(url) + .header(http::header::AUTHORIZATION, "Bearer appended"), + ), + ( + pool.post(url) + .header(http::header::AUTHORIZATION, "Bearer first") + .headers(headers.clone()) + .header(http::header::AUTHORIZATION, "Bearer last"), + client + .post(url) + .header(http::header::AUTHORIZATION, "Bearer first") + .headers(headers) + .header(http::header::AUTHORIZATION, "Bearer last"), + ), + ( + pool.post(url) + .header(CONTENT_TYPE, "application/custom+json") + .headers(HeaderMap::new()), + client + .post(url) + .header(CONTENT_TYPE, "application/custom+json") + .headers(HeaderMap::new()), + ), + ( + pool.post(url).json(&serde_json::json!({"previous": true})), + client + .post(url) + .json(&serde_json::json!({"previous": true})), + ), + ] { + let body = serde_json::json!({"query": "test"}); + let timeout = Duration::from_secs(7); + let actual = actual + .json(&body) + .timeout(timeout) + .request + .expect("request inputs should be valid") + .build(&client) + .expect("request should build"); + let expected = expected + .json(&body) + .timeout(timeout) + .build() + .expect("reqwest should build"); + assert_eq!( + ( + actual.url(), + actual.method(), + actual.headers(), + actual.body().and_then(reqwest::Body::as_bytes), + actual.timeout() + ), + ( + expected.url(), + expected.method(), + expected.headers(), + expected.body().and_then(reqwest::Body::as_bytes), + expected.timeout() + ), + ); + assert_eq!( + actual + .headers() + .values() + .map(HeaderValue::is_sensitive) + .collect::>(), + expected + .headers() + .values() + .map(HeaderValue::is_sensitive) + .collect::>(), + ); + } + } +} + +#[tokio::test] +async fn initial_url_auth_is_built_once_after_routing_and_retained_only_on_same_origin_redirects() { + let (first_proxy, first_server) = spawn_response_server(vec![ + "HTTP/1.1 302 Found\r\nLocation: http://other:secret@origin.test/same?route=2\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string(), + "HTTP/1.1 302 Found\r\nLocation: http://other:secret@other.test/final?route=3\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string(), + ]); + let (last_proxy, last_server) = spawn_response_server(vec![ + "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string(), + ]); + let urls = [ + "http://user:pass@origin.test/start?route=1", + "http://other:secret@origin.test/same?route=2", + "http://other:secret@other.test/final?route=3", + ]; + let resolver = FakeRouteResolver::new( + urls.iter() + .zip([first_proxy, first_proxy, last_proxy]) + .map(|(url, proxy)| { + ( + url.to_string(), + OutboundProxyRoute::Proxy { + url: format!("http://{proxy}"), + no_proxy: None, + }, + ) + }) + .collect(), + ); + let pool = manual_redirect_pool(); + let request = pool + .get(urls[0]) + .timeout(Duration::from_secs(3)) + .request + .expect("valid request"); + let response = pool + .send_with_resolver(request, |url| resolver.resolve(url)) + .await + .expect("request should follow both redirects"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(resolver.observed_urls(), urls.map(str::to_string)); + let mut requests = first_server.join().expect("first proxy should finish"); + requests.extend(last_server.join().expect("last proxy should finish")); + let auth = requests + .iter() + .map(|request| { + request + .lines() + .filter_map(|line| line.split_once(':')) + .filter(|(name, _)| name.eq_ignore_ascii_case("authorization")) + .map(|(_, value)| value.trim()) + .collect::>() + }) + .collect::>(); + assert_eq!( + auth, + vec![ + vec!["Basic dXNlcjpwYXNz"], + vec!["Basic dXNlcjpwYXNz"], + vec![] + ] + ); +} + +#[tokio::test] +async fn fixed_transport_preserves_url_auth_and_explicit_authorization() { + use crate::HttpTransport; + use crate::ReqwestTransport; + for authorization in [None, Some("Bearer explicit")] { + let (address, server) = spawn_response_server(vec![ + "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string(), + ]); + let client = HttpClientBuilder::new() + .build_direct() + .expect("client should build"); + let transport = ReqwestTransport::from_http_client(client); + let mut request = crate::Request::new(Method::GET, format!("http://user:pass@{address}/")); + if let Some(authorization) = authorization { + request + .headers + .insert(http::header::AUTHORIZATION, authorization.parse().unwrap()); + } + transport + .execute(request) + .await + .expect("transport should send"); + let requests = server.join().expect("server should finish"); + let auth = requests[0] + .lines() + .filter_map(|line| line.split_once(':')) + .filter(|(name, _)| name.eq_ignore_ascii_case("authorization")) + .map(|(_, value)| value.trim()) + .collect::>(); + assert_eq!(auth, vec![authorization.unwrap_or("Basic dXNlcjpwYXNz")]); + } +} + #[tokio::test] async fn request_failures_classify_real_untrusted_certificate_handshakes() { codex_utils_rustls_provider::ensure_rustls_crypto_provider(); diff --git a/codex-rs/http-client/src/transport.rs b/codex-rs/http-client/src/transport.rs index 13bddd0489..08792b9e74 100644 --- a/codex-rs/http-client/src/transport.rs +++ b/codex-rs/http-client/src/transport.rs @@ -1,14 +1,16 @@ -//! Shared HTTP transport with optional limits on complete response bodies. +//! Shared HTTP transport for fixed clients and route-aware pools, with optional response limits. //! //! Each request's limit applies to observed body bytes, including unsuccessful responses; //! declared content lengths are only an additional early-rejection check. +use crate::RouteAwareClientPool; +use crate::RouteAwareRequestError; use crate::client::HttpClient; -use crate::client::RequestBuilder; use crate::error::TransportError; use crate::request::Request; use crate::request::RequestBody; use crate::request::Response; +use crate::request_draft::RequestDraft; use bytes::Bytes; use bytes::BytesMut; use futures::StreamExt; @@ -41,21 +43,34 @@ pub trait HttpTransport: Send + Sync { #[derive(Clone, Debug)] pub struct ReqwestTransport { - client: HttpClient, + client: TransportClient, +} + +#[derive(Clone, Debug)] +enum TransportClient { + Fixed(HttpClient), + RouteAware(Box), } impl ReqwestTransport { pub fn new(client: reqwest::Client) -> Self { - Self { - client: HttpClient::new(client), - } + Self::from_http_client(HttpClient::new(client)) } pub fn from_http_client(client: HttpClient) -> Self { - Self { client } + Self { + client: TransportClient::Fixed(client), + } } - fn build(&self, req: Request) -> Result { + /// Uses the pool to resolve the route for every request and redirect URL. + pub fn from_route_aware_client_pool(client: RouteAwareClientPool) -> Self { + Self { + client: TransportClient::RouteAware(Box::new(client)), + } + } + + async fn send(&self, req: Request) -> Result { let prepared = req.prepare_body_for_send().map_err(TransportError::Build)?; let Request { @@ -68,20 +83,22 @@ impl ReqwestTransport { response_body_limit_bytes: _, } = req; - let mut builder = self.client.request( - Method::from_bytes(method.as_str().as_bytes()).unwrap_or(Method::GET), - &url, - ); + let method = Method::from_bytes(method.as_str().as_bytes()).unwrap_or(Method::GET); + let mut request = RequestDraft::new(method, url).map_err(Self::map_error)?; + request.extend_headers(prepared.headers); + *request.request.timeout_mut() = timeout; + *request.request.body_mut() = prepared.body.map(Into::into); - if let Some(timeout) = timeout { - builder = builder.timeout(timeout); + match &self.client { + TransportClient::Fixed(client) => { + let request = request.build(client).map_err(Self::map_error)?; + client.execute(request).await.map_err(Self::map_error) + } + TransportClient::RouteAware(client) => client + .send(request) + .await + .map_err(Self::map_route_aware_error), } - - builder = builder.headers(prepared.headers); - if let Some(body) = prepared.body { - builder = builder.body(body); - } - Ok(builder) } fn map_error(err: reqwest::Error) -> TransportError { @@ -95,8 +112,25 @@ impl ReqwestTransport { } } + fn map_route_aware_error(error: RouteAwareRequestError) -> TransportError { + match error.without_url() { + RouteAwareRequestError::Request(error) => Self::map_error(error), + RouteAwareRequestError::Timeout => TransportError::Timeout, + RouteAwareRequestError::Route(error) => TransportError::Build(error.to_string()), + RouteAwareRequestError::Build(error) => TransportError::Build(error), + error @ (RouteAwareRequestError::UnsupportedRedirectScheme(_) + | RouteAwareRequestError::TooManyRedirects) => { + TransportError::Network(error.to_string()) + } + } + } + fn trace_request(&self, req: &Request) { - if self.client.request_logging_enabled() && enabled!(Level::TRACE) { + let request_logging_enabled = match &self.client { + TransportClient::Fixed(client) => client.request_logging_enabled(), + TransportClient::RouteAware(client) => client.request_logging_enabled(), + }; + if request_logging_enabled && enabled!(Level::TRACE) { trace!( "{} to {}: {}", req.method, @@ -124,8 +158,7 @@ impl HttpTransport for ReqwestTransport { let url = req.url.clone(); let response_body_limit_bytes = req.response_body_limit_bytes; - let builder = self.build(req)?; - let resp = builder.send().await.map_err(Self::map_error)?; + let resp = self.send(req).await?; let status = resp.status(); let headers = resp.headers().clone(); let bytes = match response_body_limit_bytes { @@ -159,8 +192,7 @@ impl HttpTransport for ReqwestTransport { let url = req.url.clone(); let response_body_limit_bytes = req.response_body_limit_bytes; - let builder = self.build(req)?; - let resp = builder.send().await.map_err(Self::map_error)?; + let resp = self.send(req).await?; let status = resp.status(); let headers = resp.headers().clone(); if !status.is_success() { diff --git a/codex-rs/login/src/auth/default_client.rs b/codex-rs/login/src/auth/default_client.rs index b02a3c0d33..9a5ace9502 100644 --- a/codex-rs/login/src/auth/default_client.rs +++ b/codex-rs/login/src/auth/default_client.rs @@ -1,5 +1,5 @@ //! Default Codex HTTP client: shared `User-Agent`, `originator`, optional residency header, and -//! `HttpClient` construction. +//! HTTP client and transport construction. //! //! Use [`crate::default_client`] or [`codex_login::default_client`] from other crates in this //! workspace. @@ -11,6 +11,8 @@ use codex_http_client::HttpClientBuilder; use codex_http_client::HttpClientFactory; use codex_http_client::OutboundProxyPolicy; pub use codex_http_client::RequestBuilder as CodexRequestBuilder; +use codex_http_client::ReqwestTransport; +use codex_http_client::RouteAwareClientPool; use codex_terminal_detection::user_agent; use http::HeaderMap; use http::HeaderValue; @@ -296,6 +298,40 @@ pub async fn create_client_for_route_async( .map_err(std::io::Error::other)? } +/// Builds the default Codex transport without blocking the async runtime worker. +/// +/// Route-aware proxy handling resolves each request and redirect destination. When it is disabled, +/// or the client is running inside the Codex sandbox, this preserves the default client's behavior. +pub async fn create_transport_for_routes_async( + http_client_factory: HttpClientFactory, + route_class: ClientRouteClass, +) -> std::io::Result { + let permit = ROUTE_AWARE_CLIENT_BUILD_PERMIT + .acquire() + .await + .map_err(std::io::Error::other)?; + tokio::task::spawn_blocking(move || { + let _permit = permit; + if matches!( + http_client_factory.outbound_proxy_policy(), + OutboundProxyPolicy::ReqwestDefault + ) || is_sandboxed() + { + return ReqwestTransport::from_http_client(create_client()); + } + + ReqwestTransport::from_route_aware_client_pool( + RouteAwareClientPool::with_chatgpt_cloudflare_cookies_and_default_headers( + http_client_factory, + route_class, + default_headers(), + ), + ) + }) + .await + .map_err(std::io::Error::other) +} + fn default_http_client_builder() -> HttpClientBuilder { HttpClientBuilder::new() .default_headers(default_headers())