diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 85ab7d1882..34626523b5 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3836,6 +3836,7 @@ dependencies = [ "codex-api", "codex-config", "codex-exec-server", + "codex-http-client", "codex-keyring-store", "codex-protocol", "codex-secrets", diff --git a/codex-rs/app-server/src/bin/exec_server.rs b/codex-rs/app-server/src/bin/exec_server.rs index 0b6b7ce001..ee65d8ffad 100644 --- a/codex-rs/app-server/src/bin/exec_server.rs +++ b/codex-rs/app-server/src/bin/exec_server.rs @@ -6,6 +6,8 @@ //! `codex_self_exe` for sandboxed filesystem and process requests. use codex_exec_server::ExecServerRuntimePaths; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use std::ffi::OsStr; const CODEX_LINUX_SANDBOX_EXE_ENV_VAR: &str = "CODEX_TEST_LINUX_SANDBOX_EXE"; @@ -32,5 +34,7 @@ fn main() -> Result<(), Box> { .block_on(codex_exec_server::run_main( "ws://127.0.0.1:0", runtime_paths, + // This test-only fixture has no application configuration to resolve HTTP policy. + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )) } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index f8b868d55d..45817bd95b 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -1713,6 +1713,7 @@ async fn run_exec_server_command( base_url, environment_id, auth_provider, + config.http_client_factory(), )?; if let Some(name) = cmd.name { remote_config.name = name; @@ -1731,11 +1732,25 @@ async fn run_exec_server_command( config_result.ok() }; let (_otel, telemetry) = exec_server_telemetry::init(config.as_ref()); + let http_client_factory = config + .as_ref() + .map(codex_core::config::Config::http_client_factory) + .unwrap_or_else(|| { + codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ) + }); let listen_url = cmd .listen .unwrap_or_else(|| codex_exec_server::DEFAULT_LISTEN_URL.to_string()); exec_server_telemetry::run_until_shutdown(async move { - codex_exec_server::run_main_with_telemetry(&listen_url, runtime_paths, telemetry).await + codex_exec_server::run_main_with_telemetry( + &listen_url, + runtime_paths, + telemetry, + http_client_factory, + ) + .await }) .await .map_err(anyhow::Error::from_boxed) diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 32b9e067b2..8901c36723 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -45,8 +45,6 @@ use codex_config::types::OAuthCredentialsStoreMode; use codex_connectors::ConnectorRuntimeContext; use codex_connectors::ConnectorRuntimeFetchSource; use codex_exec_server::Environment; -use codex_exec_server::HttpClient; -use codex_exec_server::ReqwestHttpClient; use codex_protocol::mcp::McpServerInfo; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; @@ -1023,7 +1021,7 @@ async fn make_rmcp_client( bearer_token_env_var, } => { let http_client = resolved_environment.as_ref().map_or_else( - || Arc::new(ReqwestHttpClient) as Arc, + || runtime_context.local_http_client(), |environment| environment.get_http_client(), ); let http_client = maybe_with_openai_docs_source_attribution(&url, http_client); diff --git a/codex-rs/codex-mcp/src/runtime.rs b/codex-rs/codex-mcp/src/runtime.rs index 67abb8423c..c96fbc4311 100644 --- a/codex-rs/codex-mcp/src/runtime.rs +++ b/codex-rs/codex-mcp/src/runtime.rs @@ -295,6 +295,12 @@ impl McpRuntimeContext { self.local_stdio_fallback_cwd.clone() } + pub(crate) fn local_http_client(&self) -> Arc { + Arc::new(ReqwestHttpClient::new( + self.environment_manager.http_client_factory().clone(), + )) + } + pub(crate) fn resolve_server_environment( &self, server_name: &str, @@ -334,7 +340,7 @@ impl McpRuntimeContext { Ok(self .resolve_server_environment(server_name, config)? .map_or_else( - || Arc::new(ReqwestHttpClient) as Arc, + || self.local_http_client(), |environment| environment.get_http_client(), )) } diff --git a/codex-rs/exec-server/README.md b/codex-rs/exec-server/README.md index a250ba5886..d98bc304bf 100644 --- a/codex-rs/exec-server/README.md +++ b/codex-rs/exec-server/README.md @@ -400,11 +400,13 @@ The crate exports: - `RemoteEnvironmentConfig` and `run_remote_environment()` for embedding remote registration mode -Callers must pass `ExecServerRuntimePaths` to `run_main()`. The top-level -`codex exec-server` command builds these paths from the `codex` arg0 dispatch -state. `RemoteEnvironmentConfig::new(...)` also takes the auth provider that -remote registration should use; the CLI builds that provider from Codex auth -state before starting remote mode. +Callers must pass `ExecServerRuntimePaths` and an explicitly configured +`HttpClientFactory` to `run_main()`. The top-level `codex exec-server` command +builds these paths from the `codex` arg0 dispatch state and resolves its HTTP +client factory from the effective Codex configuration. +`RemoteEnvironmentConfig::new(...)` also takes the auth provider and HTTP client +factory that remote registration mode should use; the CLI builds the auth +provider from Codex auth state before starting remote mode. ## Example session diff --git a/codex-rs/exec-server/src/client/reqwest_http_client.rs b/codex-rs/exec-server/src/client/reqwest_http_client.rs index 47205874f8..5191c9586f 100644 --- a/codex-rs/exec-server/src/client/reqwest_http_client.rs +++ b/codex-rs/exec-server/src/client/reqwest_http_client.rs @@ -5,12 +5,13 @@ //! - in a remote environment, that means the remote runtime after the //! orchestrator has forwarded `http/request` over JSON-RPC -use std::error::Error as StdError; use std::time::Duration; use codex_exec_server_protocol::JSONRPCErrorError; -use codex_http_client::build_reqwest_client_with_custom_ca; -use codex_http_client::with_chatgpt_cloudflare_cookie_store; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::RouteAwareClientPool; +use codex_http_client::RouteAwareRequestError; use futures::FutureExt; use futures::StreamExt; use futures::future::BoxFuture; @@ -35,10 +36,12 @@ use crate::rpc::RpcNotificationSender; use crate::rpc::internal_error; use crate::rpc::invalid_params; -/// `HttpClient` implementation that performs the actual HTTP request with -/// `reqwest`. -#[derive(Clone, Default)] -pub struct ReqwestHttpClient; +/// HTTP capability implementation backed by the shared route-aware transport. +#[derive(Clone)] +pub struct ReqwestHttpClient { + follow_redirects: RouteAwareClientPool, + stop_redirects: RouteAwareClientPool, +} /// Streaming response state held between the initial HTTP response and /// downstream body-delta forwarding. @@ -50,26 +53,32 @@ pub(crate) struct PendingReqwestHttpBodyStream { /// Validates `http/request` parameters and runs the actual `reqwest` call used /// by the exec-server route and the local [`HttpClient`] backend. pub(crate) struct ReqwestHttpRequestRunner { - client: reqwest::Client, + client: RouteAwareClientPool, } impl ReqwestHttpClient { - fn build_client( - timeout_ms: Option, - redirect_policy: HttpRedirectPolicy, - ) -> Result { - let builder = match timeout_ms { - None => reqwest::Client::builder(), - Some(timeout_ms) => { - reqwest::Client::builder().timeout(Duration::from_millis(timeout_ms)) - } + pub fn new(http_client_factory: HttpClientFactory) -> Self { + Self { + follow_redirects: RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging( + http_client_factory.clone(), + // Delegated HTTP targets arbitrary endpoints; route class only labels diagnostics. + ClientRouteClass::Other, + ), + stop_redirects: + RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_redirects_or_request_logging( + http_client_factory, + // Proxy routing comes from the factory, not this diagnostic-only route class. + ClientRouteClass::Other, + ), + } + } + + pub(crate) fn runner(&self, redirect_policy: HttpRedirectPolicy) -> ReqwestHttpRequestRunner { + let client = match redirect_policy { + HttpRedirectPolicy::Follow => self.follow_redirects.clone(), + HttpRedirectPolicy::Stop => self.stop_redirects.clone(), }; - let builder = match redirect_policy { - HttpRedirectPolicy::Follow => builder, - HttpRedirectPolicy::Stop => builder.redirect(reqwest::redirect::Policy::none()), - }; - build_reqwest_client_with_custom_ca(with_chatgpt_cloudflare_cookie_store(builder)) - .map_err(|error| ExecServerError::HttpRequest(error.to_string())) + ReqwestHttpRequestRunner { client } } } @@ -79,8 +88,7 @@ impl HttpClient for ReqwestHttpClient { params: HttpRequestParams, ) -> BoxFuture<'_, Result> { async move { - let runner = ReqwestHttpRequestRunner::new(params.timeout_ms, params.redirect_policy) - .map_err(|error| ExecServerError::HttpRequest(error.message))?; + let runner = self.runner(params.redirect_policy); let (response, _) = runner .run(HttpRequestParams { stream_response: false, @@ -98,8 +106,7 @@ impl HttpClient for ReqwestHttpClient { params: HttpRequestParams, ) -> BoxFuture<'_, Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError>> { async move { - let runner = ReqwestHttpRequestRunner::new(params.timeout_ms, params.redirect_policy) - .map_err(|error| ExecServerError::HttpRequest(error.message))?; + let runner = self.runner(params.redirect_policy); let (response, pending_stream) = runner .run(HttpRequestParams { stream_response: true, @@ -122,15 +129,6 @@ impl HttpClient for ReqwestHttpClient { } impl ReqwestHttpRequestRunner { - pub(crate) fn new( - timeout_ms: Option, - redirect_policy: HttpRedirectPolicy, - ) -> Result { - let client = ReqwestHttpClient::build_client(timeout_ms, redirect_policy) - .map_err(|error| internal_error(error.to_string()))?; - Ok(Self { client }) - } - pub(crate) async fn run( &self, params: HttpRequestParams, @@ -164,6 +162,9 @@ impl ReqwestHttpRequestRunner { if let Some(body) = params.body { request = request.body(body.into_inner()); } + if let Some(timeout_ms) = params.timeout_ms { + request = request.timeout(Duration::from_millis(timeout_ms)); + } let response = match request.send().instrument(request_span.clone()).await { Ok(response) => response, @@ -301,25 +302,18 @@ impl ReqwestHttpRequestRunner { } } -fn log_send_error(method: &Method, error: reqwest::Error) { - let error = error.without_url(); - let source_chain = error_source_chain(&error); +fn log_send_error(method: &Method, error: RouteAwareRequestError) { + let error_is_timeout = error.is_timeout(); + let error_is_connect = error.is_connect(); + let error = match error { + RouteAwareRequestError::Request(error) => error.without_url().to_string(), + error => error.to_string(), + }; tracing::warn!( http_method = method.as_str(), - error_is_timeout = error.is_timeout(), - error_is_connect = error.is_connect(), + error_is_timeout, + error_is_connect, error = %error, - error_sources = ?source_chain, "http/request send failed" ); } - -fn error_source_chain(error: &reqwest::Error) -> Option { - let mut sources = Vec::new(); - let mut source = error.source(); - while let Some(error) = source { - sources.push(error.to_string()); - source = error.source(); - } - (!sources.is_empty()).then(|| sources.join(": ")) -} diff --git a/codex-rs/exec-server/src/environment.rs b/codex-rs/exec-server/src/environment.rs index 30c2e4445f..0394a6afca 100644 --- a/codex-rs/exec-server/src/environment.rs +++ b/codex-rs/exec-server/src/environment.rs @@ -139,6 +139,7 @@ impl EnvironmentManager { )])), local_environment: Some(Arc::new(Environment::default_for_tests())), local_runtime_paths: None, + // Test-only construction has no application config from which to resolve proxy policy. http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), } } @@ -163,6 +164,7 @@ impl EnvironmentManager { match Self::from_snapshot( provider.snapshot_inner(), local_runtime_paths, + // Test-only construction has no application config from which to resolve proxy policy. HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), ) { Ok(manager) => manager, @@ -249,6 +251,7 @@ impl EnvironmentManager { match Self::from_snapshot( snapshot, Some(local_runtime_paths), + // Test-only construction has no application config from which to resolve proxy policy. HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), ) { Ok(manager) => manager, @@ -274,7 +277,10 @@ impl EnvironmentManager { "local environment requires configured runtime paths".to_string(), ) })?; - let local_environment = Arc::new(Environment::local(local_runtime_paths)); + let local_environment = Arc::new(Environment::local( + local_runtime_paths, + http_client_factory.clone(), + )); environment_map.insert( LOCAL_ENVIRONMENT_ID.to_string(), Arc::clone(&local_environment), @@ -619,7 +625,10 @@ impl Environment { startup_task: Arc::new(Mutex::new(None)), exec_backend: Arc::new(LocalProcess::default()), filesystem: Arc::new(LocalFileSystem::unsandboxed()), - http_client: Arc::new(ReqwestHttpClient), + // Test-only construction has no application config from which to resolve proxy policy. + http_client: Arc::new(ReqwestHttpClient::new(HttpClientFactory::new( + OutboundProxyPolicy::ReqwestDefault, + ))), local_runtime_paths: None, } } @@ -661,13 +670,20 @@ impl Environment { Ok(match exec_server_url { Some(exec_server_url) => Self::remote_inner(exec_server_url, local_runtime_paths), None => match local_runtime_paths { - Some(local_runtime_paths) => Self::local(local_runtime_paths), + Some(local_runtime_paths) => Self::local( + local_runtime_paths, + // This legacy constructor has no resolved application proxy configuration. + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ), None => Self::default_for_tests(), }, }) } - pub(crate) fn local(local_runtime_paths: ExecServerRuntimePaths) -> Self { + pub(crate) fn local( + local_runtime_paths: ExecServerRuntimePaths, + http_client_factory: HttpClientFactory, + ) -> Self { Self { remote_client: None, ready_info: None, @@ -678,7 +694,7 @@ impl Environment { filesystem: Arc::new(LocalFileSystem::with_runtime_paths( local_runtime_paths.clone(), )), - http_client: Arc::new(ReqwestHttpClient), + http_client: Arc::new(ReqwestHttpClient::new(http_client_factory)), local_runtime_paths: Some(local_runtime_paths), } } @@ -1548,7 +1564,7 @@ mod tests { #[tokio::test] async fn local_environment_passes_runtime_paths_to_exec_backend() { - let environment = Environment::local(test_runtime_paths()); + let environment = Environment::local(test_runtime_paths(), legacy_http_client_factory()); #[cfg(unix)] let uri = "file://server/share/checkout"; #[cfg(windows)] diff --git a/codex-rs/exec-server/src/remote.rs b/codex-rs/exec-server/src/remote.rs index 8870e0efe9..0b5d995874 100644 --- a/codex-rs/exec-server/src/remote.rs +++ b/codex-rs/exec-server/src/remote.rs @@ -4,6 +4,7 @@ use std::time::Instant; use codex_api::AuthProvider; use codex_api::SharedAuthProvider; +use codex_http_client::HttpClientFactory; use futures::FutureExt; use http::HeaderMap; use http::HeaderName; @@ -421,6 +422,7 @@ pub struct RemoteEnvironmentConfig { pub name: String, auth_provider: SharedAuthProvider, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, } impl std::fmt::Debug for RemoteEnvironmentConfig { @@ -439,6 +441,7 @@ impl RemoteEnvironmentConfig { base_url: String, environment_id: String, auth_provider: SharedAuthProvider, + http_client_factory: HttpClientFactory, ) -> Result { let environment_id = normalize_environment_id(environment_id)?; Ok(Self { @@ -447,6 +450,7 @@ impl RemoteEnvironmentConfig { name: "codex-exec-server".to_string(), auth_provider, telemetry: ExecServerTelemetry::default(), + http_client_factory, }) } @@ -472,8 +476,11 @@ pub async fn run_remote_environment( config.auth_provider.clone(), config.telemetry.clone(), )?; - let processor = - ConnectionProcessor::new_with_telemetry(runtime_paths, config.telemetry.clone()); + let processor = ConnectionProcessor::new_with_telemetry( + runtime_paths, + config.telemetry.clone(), + config.http_client_factory.clone(), + ); let identity = NoiseChannelIdentity::generate().map_err(|error| { ExecServerError::Protocol(format!("failed to generate Noise relay identity: {error}")) })?; @@ -672,6 +679,7 @@ mod tests { use std::sync::Arc; use codex_api::AuthProvider; + use codex_http_client::OutboundProxyPolicy; use http::HeaderMap; use http::HeaderValue; use opentelemetry::trace::TracerProvider as _; @@ -880,12 +888,29 @@ mod tests { )); } + #[test] + fn remote_environment_config_preserves_http_client_factory_policy() { + let config = RemoteEnvironmentConfig::new( + "https://registry.example".to_string(), + "env-1".to_string(), + static_registry_auth_provider(), + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ) + .expect("config"); + + assert_eq!( + config.http_client_factory.outbound_proxy_policy(), + OutboundProxyPolicy::RespectSystemProxy + ); + } + #[test] fn debug_output_redacts_auth_provider() { let config = RemoteEnvironmentConfig::new( "https://registry.example".to_string(), "env-1".to_string(), static_registry_auth_provider(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), ) .expect("config"); diff --git a/codex-rs/exec-server/src/remote/noise_tests.rs b/codex-rs/exec-server/src/remote/noise_tests.rs index 1c4525e63f..4aaecba334 100644 --- a/codex-rs/exec-server/src/remote/noise_tests.rs +++ b/codex-rs/exec-server/src/remote/noise_tests.rs @@ -4,6 +4,8 @@ use std::time::Duration; use anyhow::Result; use codex_api::AuthProvider; use codex_api::SharedAuthProvider; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use http::HeaderMap; use http::HeaderValue; use tokio::io::AsyncReadExt; @@ -59,6 +61,7 @@ async fn reconnect_reuses_registration_until_url_is_rejected() -> Result<()> { registry.uri(), "environment-requested".to_string(), static_registry_auth_provider(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )?; let environment_task = tokio::spawn(run_remote_environment( config, diff --git a/codex-rs/exec-server/src/server.rs b/codex-rs/exec-server/src/server.rs index 7988e0870b..ba58b36dc9 100644 --- a/codex-rs/exec-server/src/server.rs +++ b/codex-rs/exec-server/src/server.rs @@ -13,12 +13,20 @@ pub use transport::ExecServerListenUrlParseError; use crate::ExecServerRuntimePaths; use crate::ExecServerTelemetry; +use codex_http_client::HttpClientFactory; pub async fn run_main( listen_url: &str, runtime_paths: ExecServerRuntimePaths, + http_client_factory: HttpClientFactory, ) -> Result<(), Box> { - run_main_with_telemetry(listen_url, runtime_paths, ExecServerTelemetry::default()).await + run_main_with_telemetry( + listen_url, + runtime_paths, + ExecServerTelemetry::default(), + http_client_factory, + ) + .await } #[tracing::instrument( @@ -30,12 +38,15 @@ pub async fn run_main_with_telemetry( listen_url: &str, runtime_paths: ExecServerRuntimePaths, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, ) -> Result<(), Box> { - transport::run_transport(listen_url, runtime_paths, telemetry).await + transport::run_transport(listen_url, runtime_paths, telemetry, http_client_factory).await } #[cfg(test)] mod tests { + use codex_http_client::HttpClientFactory; + use codex_http_client::OutboundProxyPolicy; use opentelemetry::trace::TracerProvider as _; use opentelemetry_sdk::trace::InMemorySpanExporter; use opentelemetry_sdk::trace::SdkTracerProvider; @@ -65,6 +76,7 @@ mod tests { ) .expect("runtime paths"), ExecServerTelemetry::default(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), ) .await .expect_err("invalid listen URL should fail"); diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index fd49f4912c..675c6ff1a7 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -5,6 +5,7 @@ use std::sync::atomic::Ordering; use codex_exec_server_protocol::JSONRPCErrorError; use codex_exec_server_protocol::RequestId; +use codex_http_client::HttpClientFactory; use serde_json::to_value; use std::collections::HashSet; use tokio::sync::Mutex; @@ -13,6 +14,7 @@ use tokio_util::task::TaskTracker; use crate::ExecServerRuntimePaths; use crate::client::http_client::PendingReqwestHttpBodyStream; +use crate::client::http_client::ReqwestHttpClient; use crate::client::http_client::ReqwestHttpRequestRunner; use crate::protocol::CapabilityRootsDiscoverParams; use crate::protocol::CapabilityRootsDiscoverResponse; @@ -73,6 +75,7 @@ pub(crate) struct ExecServerHandler { background_tasks: TaskTracker, file_system: FileSystemHandler, runtime_paths: ExecServerRuntimePaths, + http_client: ReqwestHttpClient, initialize_requested: AtomicBool, initialized: AtomicBool, } @@ -82,6 +85,7 @@ impl ExecServerHandler { session_registry: Arc, notifications: RpcNotificationSender, runtime_paths: ExecServerRuntimePaths, + http_client_factory: HttpClientFactory, ) -> Self { Self { session_registry, @@ -92,6 +96,7 @@ impl ExecServerHandler { background_tasks: TaskTracker::new(), file_system: FileSystemHandler::new(runtime_paths.clone()), runtime_paths, + http_client: ReqwestHttpClient::new(http_client_factory), initialize_requested: AtomicBool::new(false), initialized: AtomicBool::new(false), } @@ -222,7 +227,9 @@ impl ExecServerHandler { if stream_response { self.reserve_http_body_stream(&http_request_id).await?; } - let response = ReqwestHttpRequestRunner::new(params.timeout_ms, params.redirect_policy)? + let response = self + .http_client + .runner(params.redirect_policy) .run(params) .await; if response.is_err() && stream_response { diff --git a/codex-rs/exec-server/src/server/handler/tests.rs b/codex-rs/exec-server/src/server/handler/tests.rs index b70b7ffc26..0d3b41768a 100644 --- a/codex-rs/exec-server/src/server/handler/tests.rs +++ b/codex-rs/exec-server/src/server/handler/tests.rs @@ -2,6 +2,8 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_utils_path_uri::PathUri; use pretty_assertions::assert_eq; use tokio::sync::mpsc; @@ -88,6 +90,7 @@ async fn initialized_handler() -> Arc { registry, RpcNotificationSender::new(outgoing_tx), test_runtime_paths(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )); let initialize_response = handler .initialize(InitializeParams { @@ -166,6 +169,7 @@ async fn long_poll_read_fails_after_session_resume() { Arc::clone(®istry), RpcNotificationSender::new(first_tx), test_runtime_paths(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )); let initialize_response = first_handler .initialize(InitializeParams { @@ -206,6 +210,7 @@ async fn long_poll_read_fails_after_session_resume() { registry, RpcNotificationSender::new(second_tx), test_runtime_paths(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )); second_handler .initialize(InitializeParams { @@ -239,6 +244,7 @@ async fn active_session_resume_is_rejected() { Arc::clone(®istry), RpcNotificationSender::new(first_tx), test_runtime_paths(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )); let initialize_response = first_handler .initialize(InitializeParams { @@ -253,6 +259,7 @@ async fn active_session_resume_is_rejected() { registry, RpcNotificationSender::new(second_tx), test_runtime_paths(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )); let err = second_handler .initialize(InitializeParams { @@ -281,6 +288,7 @@ async fn output_and_exit_are_retained_after_notification_receiver_closes() { SessionRegistry::new(crate::ExecServerTelemetry::default()), RpcNotificationSender::new(outgoing_tx), test_runtime_paths(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )); handler .initialize(InitializeParams { diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 681c4bd67a..016147e756 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -21,28 +21,38 @@ use crate::server::registry::build_router; use crate::server::session_registry::SessionRegistry; use crate::telemetry::ConnectionTransport; use crate::telemetry::ExecServerTelemetry; +use codex_http_client::HttpClientFactory; #[derive(Clone)] pub(crate) struct ConnectionProcessor { session_registry: Arc, runtime_paths: ExecServerRuntimePaths, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, } impl ConnectionProcessor { #[cfg(test)] pub(crate) fn new(runtime_paths: ExecServerRuntimePaths) -> Self { - Self::new_with_telemetry(runtime_paths, ExecServerTelemetry::default()) + Self::new_with_telemetry( + runtime_paths, + ExecServerTelemetry::default(), + codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ), + ) } pub(crate) fn new_with_telemetry( runtime_paths: ExecServerRuntimePaths, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, ) -> Self { Self { session_registry: SessionRegistry::new(telemetry.clone()), runtime_paths, telemetry, + http_client_factory, } } @@ -56,6 +66,7 @@ impl ConnectionProcessor { Arc::clone(&self.session_registry), self.runtime_paths.clone(), self.telemetry.clone(), + self.http_client_factory.clone(), transport, ) .await; @@ -71,6 +82,7 @@ async fn run_connection( session_registry: Arc, runtime_paths: ExecServerRuntimePaths, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, transport: ConnectionTransport, ) { let _connection_metrics = telemetry.connection_started(transport); @@ -90,6 +102,7 @@ async fn run_connection( session_registry, notifications, runtime_paths, + http_client_factory, )); let outbound_task = tokio::spawn(async move { @@ -530,6 +543,9 @@ mod tests { registry, test_runtime_paths(), crate::ExecServerTelemetry::default(), + codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ), crate::telemetry::ConnectionTransport::Stdio, )); (client_writer, BufReader::new(client_reader).lines(), task) diff --git a/codex-rs/exec-server/src/server/transport.rs b/codex-rs/exec-server/src/server/transport.rs index 2affab0070..7e8e24b902 100644 --- a/codex-rs/exec-server/src/server/transport.rs +++ b/codex-rs/exec-server/src/server/transport.rs @@ -12,6 +12,7 @@ use axum::response::IntoResponse; use axum::response::Response; use axum::routing::any; use axum::routing::get; +use codex_http_client::HttpClientFactory; use std::io::Write as _; use std::net::SocketAddr; use tokio::io; @@ -83,20 +84,32 @@ pub(crate) async fn run_transport( listen_url: &str, runtime_paths: ExecServerRuntimePaths, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, ) -> Result<(), Box> { match parse_listen_url(listen_url)? { ExecServerListenTransport::WebSocket(bind_address) => { - run_websocket_listener(bind_address, runtime_paths, telemetry).await + run_websocket_listener(bind_address, runtime_paths, telemetry, http_client_factory) + .await + } + ExecServerListenTransport::Stdio => { + run_stdio_connection(runtime_paths, telemetry, http_client_factory).await } - ExecServerListenTransport::Stdio => run_stdio_connection(runtime_paths, telemetry).await, } } async fn run_stdio_connection( runtime_paths: ExecServerRuntimePaths, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, ) -> Result<(), Box> { - run_stdio_connection_with_io(io::stdin(), io::stdout(), runtime_paths, telemetry).await + run_stdio_connection_with_io( + io::stdin(), + io::stdout(), + runtime_paths, + telemetry, + http_client_factory, + ) + .await } async fn run_stdio_connection_with_io( @@ -104,12 +117,14 @@ async fn run_stdio_connection_with_io( writer: W, runtime_paths: ExecServerRuntimePaths, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, ) -> Result<(), Box> where R: AsyncRead + Unpin + Send + 'static, W: AsyncWrite + Unpin + Send + 'static, { - let processor = ConnectionProcessor::new_with_telemetry(runtime_paths, telemetry); + let processor = + ConnectionProcessor::new_with_telemetry(runtime_paths, telemetry, http_client_factory); tracing::info!("codex-exec-server listening on stdio"); processor .run_connection( @@ -126,10 +141,12 @@ async fn run_websocket_listener( bind_address: SocketAddr, runtime_paths: ExecServerRuntimePaths, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, ) -> Result<(), Box> { let listener = TcpListener::bind(bind_address).await?; let local_addr = listener.local_addr()?; - let processor = ConnectionProcessor::new_with_telemetry(runtime_paths, telemetry); + let processor = + ConnectionProcessor::new_with_telemetry(runtime_paths, telemetry, http_client_factory); info!("codex-exec-server listening on ws://{local_addr}"); println!("ws://{local_addr}"); std::io::stdout().flush()?; diff --git a/codex-rs/exec-server/src/server/transport_tests.rs b/codex-rs/exec-server/src/server/transport_tests.rs index 3016effa6c..f25203c6a3 100644 --- a/codex-rs/exec-server/src/server/transport_tests.rs +++ b/codex-rs/exec-server/src/server/transport_tests.rs @@ -62,6 +62,9 @@ async fn stdio_listen_transport_serves_initialize() { server_writer, test_runtime_paths(), crate::ExecServerTelemetry::default(), + codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ), )); let mut client_lines = BufReader::new(client_reader).lines(); diff --git a/codex-rs/exec-server/testing/BUILD.bazel b/codex-rs/exec-server/testing/BUILD.bazel index 2e2cc220d4..93cb4ceab9 100644 --- a/codex-rs/exec-server/testing/BUILD.bazel +++ b/codex-rs/exec-server/testing/BUILD.bazel @@ -40,6 +40,7 @@ rust_binary( visibility = ["//visibility:public"], deps = [ "//codex-rs/exec-server", + "//codex-rs/http-client", "@crates//:tokio", ], ) diff --git a/codex-rs/exec-server/testing/exec_server.rs b/codex-rs/exec-server/testing/exec_server.rs index a0b380a957..36c7685ceb 100644 --- a/codex-rs/exec-server/testing/exec_server.rs +++ b/codex-rs/exec-server/testing/exec_server.rs @@ -5,6 +5,8 @@ //! helper mode because sandboxed process requests re-exec this binary. use codex_exec_server::ExecServerRuntimePaths; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; #[cfg(unix)] use std::ffi::OsStr; @@ -26,5 +28,10 @@ async fn main() -> Result<(), Box> { let current_exe = std::env::current_exe()?; let runtime_paths = ExecServerRuntimePaths::new(current_exe, /*codex_linux_sandbox_exe*/ None)?; - codex_exec_server::run_main("ws://127.0.0.1:0", runtime_paths).await + codex_exec_server::run_main( + "ws://127.0.0.1:0", + runtime_paths, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ) + .await } diff --git a/codex-rs/exec-server/tests/common/mod.rs b/codex-rs/exec-server/tests/common/mod.rs index 599d054fb8..05a19fcba3 100644 --- a/codex-rs/exec-server/tests/common/mod.rs +++ b/codex-rs/exec-server/tests/common/mod.rs @@ -9,6 +9,8 @@ use std::time::Duration; use codex_exec_server::CODEX_ARG0_EXEC_HELPER_ARG1; use codex_exec_server::CODEX_FS_HELPER_ARG1; use codex_exec_server::ExecServerRuntimePaths; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_sandboxing::landlock::CODEX_LINUX_SANDBOX_ARG0; use codex_test_binary_support::TestBinaryDispatchGuard; use codex_test_binary_support::TestBinaryDispatchMode; @@ -19,6 +21,9 @@ pub(crate) mod exec_server; pub(crate) const DELAYED_OUTPUT_AFTER_EXIT_PARENT_ARG: &str = "--codex-test-delayed-output-after-exit-parent"; +pub(crate) const SYSTEM_PROXY_REQUEST_URL_ENV: &str = + "CODEX_EXEC_SERVER_TEST_SYSTEM_PROXY_REQUEST_URL"; +pub(crate) const SYSTEM_PROXY_URL_ENV: &str = "CODEX_EXEC_SERVER_TEST_SYSTEM_PROXY_URL"; const CODEX_WINDOWS_SANDBOX_ARG1: &str = "--run-as-windows-sandbox"; const DELAYED_OUTPUT_AFTER_EXIT_CHILD_ARG: &str = "--codex-test-delayed-output-after-exit-child"; @@ -188,8 +193,27 @@ fn maybe_run_exec_server_from_test_binary(guard: Option<&TestBinaryDispatchGuard std::process::exit(1); } }; - let exit_code = match runtime.block_on(codex_exec_server::run_main(&listen_url, runtime_paths)) - { + let http_client_factory = match ( + env::var(SYSTEM_PROXY_REQUEST_URL_ENV), + env::var(SYSTEM_PROXY_URL_ENV), + ) { + (Ok(request_url), Ok(proxy_url)) => { + codex_http_client::cache_system_proxy_route_for_test(&request_url, proxy_url); + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy) + } + (Err(env::VarError::NotPresent), Err(env::VarError::NotPresent)) => { + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) + } + _ => { + eprintln!("system proxy test configuration requires both request and proxy URLs"); + std::process::exit(1); + } + }; + let exit_code = match runtime.block_on(codex_exec_server::run_main( + &listen_url, + runtime_paths, + http_client_factory, + )) { Ok(()) => 0, Err(error) => { eprintln!("exec-server failed: {error}"); diff --git a/codex-rs/exec-server/tests/http_request.rs b/codex-rs/exec-server/tests/http_request.rs index 1edacc4b13..df787805d5 100644 --- a/codex-rs/exec-server/tests/http_request.rs +++ b/codex-rs/exec-server/tests/http_request.rs @@ -17,8 +17,11 @@ use codex_exec_server_protocol::JSONRPCMessage; use codex_exec_server_protocol::JSONRPCNotification; use codex_exec_server_protocol::JSONRPCResponse; use codex_exec_server_protocol::RequestId; +use common::SYSTEM_PROXY_REQUEST_URL_ENV; +use common::SYSTEM_PROXY_URL_ENV; use common::exec_server::ExecServerHarness; use common::exec_server::exec_server; +use common::exec_server::exec_server_with_env; use pretty_assertions::assert_eq; use serde::de::DeserializeOwned; use serde_json::Value; @@ -110,6 +113,61 @@ async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> Ok(()) } +/// What this tests: a configured system-proxy factory survives the complete +/// executor transport, processor, and handler chain for delegated HTTP. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_server_http_request_uses_configured_system_proxy() -> anyhow::Result<()> { + let proxy_listener = TcpListener::bind("127.0.0.1:0").await?; + let proxy_url = format!("http://{}", proxy_listener.local_addr()?); + let request_url = "http://exec-server-system-proxy.invalid/delegated?route=system"; + let mut server = exec_server_with_env([ + (SYSTEM_PROXY_REQUEST_URL_ENV, request_url), + (SYSTEM_PROXY_URL_ENV, proxy_url.as_str()), + ("HTTP_PROXY", ""), + ("http_proxy", ""), + ("HTTPS_PROXY", ""), + ("https_proxy", ""), + ("ALL_PROXY", ""), + ("all_proxy", ""), + ("NO_PROXY", ""), + ("no_proxy", ""), + ]) + .await?; + initialize_exec_server(&mut server).await?; + + let http_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url: request_url.to_string(), + headers: Vec::new(), + body: None, + timeout_ms: Some(5_000), + redirect_policy: HttpRedirectPolicy::Follow, + request_id: "system-proxy-request".to_string(), + stream_response: false, + })?, + ) + .await?; + + let captured = accept_http_request(&proxy_listener).await?; + assert_eq!( + captured.request_line, + "GET http://exec-server-system-proxy.invalid/delegated?route=system HTTP/1.1" + ); + respond_with_status_and_headers(captured.stream, "200 OK", &[], b"proxied-response").await?; + + let response: HttpRequestResponse = wait_for_response(&mut server, http_request_id).await?; + assert_eq!( + (response.status, response.body.into_inner()), + (200, b"proxied-response".to_vec()) + ); + + server.shutdown().await?; + Ok(()) +} + /// What this tests: OAuth callers can inspect redirect responses without the /// executor following the Location header. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -165,6 +223,69 @@ async fn exec_server_http_request_can_stop_at_redirects() -> anyhow::Result<()> Ok(()) } +/// What this tests: the executor follows redirects when the HTTP request +/// explicitly selects the redirect-following client pool. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_server_http_request_can_follow_redirects() -> anyhow::Result<()> { + let mut server = exec_server().await?; + initialize_exec_server(&mut server).await?; + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let base_url = format!("http://{}", listener.local_addr()?); + let http_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url: format!("{base_url}/redirect"), + headers: Vec::new(), + body: None, + timeout_ms: Some(5_000), + redirect_policy: HttpRedirectPolicy::Follow, + request_id: "follow-redirect-request".to_string(), + stream_response: false, + })?, + ) + .await?; + + let redirect_request = accept_http_request(&listener).await?; + assert_eq!(redirect_request.request_line, "GET /redirect HTTP/1.1"); + respond_with_status_and_headers( + redirect_request.stream, + "302 Found", + &[("location", &format!("{base_url}/final"))], + b"redirect", + ) + .await?; + + let final_request = accept_http_request(&listener).await?; + assert_eq!(final_request.request_line, "GET /final HTTP/1.1"); + respond_with_status_and_headers( + final_request.stream, + "200 OK", + &[("x-mcp-test", "redirected")], + b"final-response-body", + ) + .await?; + + let response: HttpRequestResponse = wait_for_response(&mut server, http_request_id).await?; + assert_eq!( + ( + response.status, + response_header(&response.headers, "x-mcp-test"), + response.body.into_inner(), + ), + ( + 200, + Some("redirected".to_string()), + b"final-response-body".to_vec(), + ) + ); + + server.shutdown().await?; + Ok(()) +} + /// What this tests: a real exec-server websocket `http/request` can return /// response headers immediately and stream the response body as ordered /// `http/request/bodyDelta` notifications. diff --git a/codex-rs/exec-server/tests/http_request_logging.rs b/codex-rs/exec-server/tests/http_request_logging.rs new file mode 100644 index 0000000000..329f155ec8 --- /dev/null +++ b/codex-rs/exec-server/tests/http_request_logging.rs @@ -0,0 +1,174 @@ +use std::io::Write; +use std::sync::Arc; +use std::sync::Mutex; + +use codex_exec_server::HttpClient; +use codex_exec_server::HttpRedirectPolicy; +use codex_exec_server::HttpRequestParams; +use codex_exec_server::ReqwestHttpClient; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use pretty_assertions::assert_eq; +use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncWriteExt; +use tokio::io::BufReader; +use tokio::net::TcpListener; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::SubscriberExt; + +#[tokio::test(flavor = "current_thread")] +async fn delegated_http_success_logs_do_not_expose_sensitive_request_or_response_data() +-> anyhow::Result<()> { + let log_buffer = Arc::new(Mutex::new(Vec::new())); + let writer_buffer = Arc::clone(&log_buffer); + let subscriber = tracing_subscriber::registry().with( + tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_writer(move || TestLogWriter(Arc::clone(&writer_buffer))) + .with_filter( + tracing_subscriber::filter::Targets::new() + .with_target("codex_http_client", tracing::Level::TRACE) + .with_target("codex_exec_server", tracing::Level::TRACE), + ), + ); + let _guard = tracing::subscriber::set_default(subscriber); + tracing::debug!(target: "codex_exec_server", "log capture sentinel"); + let client = + ReqwestHttpClient::new(HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault)); + + for (redirect_policy, status, query_secret, cookie_secret, location_secret) in [ + ( + HttpRedirectPolicy::Follow, + "200 OK", + "follow-query-secret", + "follow-cookie-secret", + "follow-location-secret", + ), + ( + HttpRedirectPolicy::Stop, + "302 Found", + "stop-query-secret", + "stop-cookie-secret", + "stop-location-secret", + ), + ] { + let listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let address = listener.local_addr()?; + let response = format!( + "HTTP/1.1 {status}\r\nSet-Cookie: session={cookie_secret}\r\nLocation: http://127.0.0.1/private?token={location_secret}\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok" + ); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await?; + let mut reader = BufReader::new(stream); + loop { + let mut line = String::new(); + if reader.read_line(&mut line).await? == 0 { + anyhow::bail!("HTTP client disconnected before completing request headers"); + } + if line == "\r\n" { + break; + } + } + reader.get_mut().write_all(response.as_bytes()).await?; + anyhow::Ok(()) + }); + + let response = client + .http_request(HttpRequestParams { + method: "GET".to_string(), + url: format!("http://{address}/delegated?token={query_secret}"), + headers: Vec::new(), + body: None, + timeout_ms: Some(5_000), + redirect_policy, + request_id: "sensitive-request".to_string(), + stream_response: false, + }) + .await?; + let expected_status = match redirect_policy { + HttpRedirectPolicy::Follow => 200, + HttpRedirectPolicy::Stop => 302, + }; + assert_eq!(response.status, expected_status); + server.await??; + } + + let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())?; + assert!(logs.contains("log capture sentinel")); + for secret in [ + "follow-query-secret", + "follow-cookie-secret", + "follow-location-secret", + "stop-query-secret", + "stop-cookie-secret", + "stop-location-secret", + ] { + assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}"); + } + + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn delegated_http_failure_warning_redacts_request_url() -> anyhow::Result<()> { + let log_buffer = Arc::new(Mutex::new(Vec::new())); + let writer_buffer = Arc::clone(&log_buffer); + let subscriber = tracing_subscriber::registry().with( + tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_writer(move || TestLogWriter(Arc::clone(&writer_buffer))) + .with_filter( + tracing_subscriber::filter::Targets::new() + .with_target("codex_http_client", tracing::Level::TRACE) + .with_target("codex_exec_server", tracing::Level::TRACE), + ), + ); + let _guard = tracing::subscriber::set_default(subscriber); + let unavailable_server = std::net::TcpListener::bind(("127.0.0.1", 0))?; + let unavailable_address = unavailable_server.local_addr()?; + drop(unavailable_server); + let client = + ReqwestHttpClient::new(HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault)); + + let error = client + .http_request(HttpRequestParams { + method: "GET".to_string(), + url: format!( + "http://{unavailable_address}/private-path-secret?token=failure-query-secret" + ), + headers: Vec::new(), + body: None, + timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, + request_id: "failed-sensitive-request".to_string(), + stream_response: false, + }) + .await; + assert!(error.is_err(), "request to a closed port should fail"); + + let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())?; + assert!(logs.contains("http/request send failed")); + assert!(logs.contains("error_is_connect=true")); + for secret in ["private-path-secret", "failure-query-secret"] { + assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}"); + } + + Ok(()) +} + +#[derive(Clone)] +struct TestLogWriter(Arc>>); + +impl Write for TestLogWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0 + .lock() + .map_err(|_| std::io::Error::other("log buffer lock"))? + .extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} diff --git a/codex-rs/exec-server/tests/relay.rs b/codex-rs/exec-server/tests/relay.rs index 41b84e4f5b..6ae4d226b1 100644 --- a/codex-rs/exec-server/tests/relay.rs +++ b/codex-rs/exec-server/tests/relay.rs @@ -31,6 +31,8 @@ use codex_exec_server::NoiseRendezvousConnectProvider; use codex_exec_server::ProcessId; use codex_exec_server::RemoteEnvironmentConfig; use codex_exec_server_test_support::environment_manager_without_environments; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_protocol::capabilities::CapabilityRootLocation; use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_utils_path_uri::PathUri; @@ -144,6 +146,7 @@ async fn deferred_noise_environment_connects_and_reconnects_with_fresh_bundle() registry.uri(), ENVIRONMENT_ID.to_string(), static_registry_auth_provider(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )?; let remote_environment = tokio::spawn(codex_exec_server::run_remote_environment( config, @@ -274,6 +277,7 @@ async fn remote_environment_routes_encrypted_exec_server_rpc() -> Result<()> { registry.uri(), ENVIRONMENT_ID.to_string(), static_registry_auth_provider(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), )?; let remote_environment = tokio::spawn(codex_exec_server::run_remote_environment( config, 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 2e3d1614fb..f05a69cbe8 100644 --- a/codex-rs/http-client/src/route_aware_client_pool.rs +++ b/codex-rs/http-client/src/route_aware_client_pool.rs @@ -308,6 +308,21 @@ impl RouteAwareClientPool { ) } + /// Creates a no-redirect ChatGPT Cloudflare-cookie pool without request diagnostics. + pub fn with_chatgpt_cloudflare_cookies_without_redirects_or_request_logging( + http_client_factory: HttpClientFactory, + route_class: ClientRouteClass, + ) -> Self { + Self::with_builder( + http_client_factory, + route_class, + HttpClientBuilder::new() + .with_chatgpt_cloudflare_cookie_store() + .without_redirects() + .without_request_logging(), + ) + } + /// Creates a ChatGPT Cloudflare-cookie pool without URL or response-header diagnostics. pub fn with_chatgpt_cloudflare_cookies_without_request_logging( http_client_factory: HttpClientFactory, diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index a8464f3054..8c6dcd0e56 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -18,6 +18,7 @@ base64 = { workspace = true } codex-api = { workspace = true } codex-config = { workspace = true } codex-exec-server = { workspace = true } +codex-http-client = { workspace = true } codex-keyring-store = { workspace = true } codex-protocol = { workspace = true } codex-secrets = { workspace = true } diff --git a/codex-rs/rmcp-client/src/perform_oauth_login.rs b/codex-rs/rmcp-client/src/perform_oauth_login.rs index 7a9d242bc4..0bb283ead4 100644 --- a/codex-rs/rmcp-client/src/perform_oauth_login.rs +++ b/codex-rs/rmcp-client/src/perform_oauth_login.rs @@ -11,6 +11,8 @@ use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use codex_exec_server::HttpClient; use codex_exec_server::ReqwestHttpClient; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use reqwest::Url; use rmcp::transport::AuthorizationManager; use rmcp::transport::AuthorizationSession; @@ -160,7 +162,9 @@ async fn perform_oauth_login_with_browser_output( let http_context = OAuthHttpContext { http_headers, env_http_headers, - http_client: Arc::new(ReqwestHttpClient), + http_client: Arc::new(ReqwestHttpClient::new(HttpClientFactory::new( + OutboundProxyPolicy::ReqwestDefault, + ))), }; OauthLoginFlow::new( server_name, @@ -209,7 +213,9 @@ pub async fn perform_oauth_login_return_url( timeout_secs, callback_port, callback_url, - Arc::new(ReqwestHttpClient), + Arc::new(ReqwestHttpClient::new(HttpClientFactory::new( + OutboundProxyPolicy::ReqwestDefault, + ))), ) .await } @@ -715,6 +721,8 @@ mod tests { use axum::Router; use axum::routing::get; use codex_exec_server::ReqwestHttpClient; + use codex_http_client::HttpClientFactory; + use codex_http_client::OutboundProxyPolicy; use pretty_assertions::assert_eq; use reqwest::Url; use reqwest::header::HeaderMap; @@ -774,7 +782,9 @@ mod tests { let oauth_state = start_authorization( &format!("{base_url}/mcp"), Arc::new(OAuthHttpClientAdapter::new( - Arc::new(ReqwestHttpClient), + Arc::new(ReqwestHttpClient::new(HttpClientFactory::new( + OutboundProxyPolicy::ReqwestDefault, + ))), HeaderMap::new(), )), &[],