diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index e0d17659da..7be935c01f 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -563,7 +563,7 @@ pub enum ExecServerError { Server { code: i64, message: String }, #[error("environment registry request failed ({status}{code_suffix}): {message}", code_suffix = .code.as_ref().map(|code| format!(", {code}")).unwrap_or_default())] EnvironmentRegistryHttp { - status: reqwest::StatusCode, + status: http::StatusCode, code: Option, message: String, }, @@ -572,7 +572,7 @@ pub enum ExecServerError { #[error("environment registry authentication error: {0}")] EnvironmentRegistryAuth(String), #[error("environment registry request failed: {0}")] - EnvironmentRegistryRequest(#[from] reqwest::Error), + EnvironmentRegistryRequest(#[from] codex_http_client::RouteAwareRequestError), #[error("exec-server connection attempt failed: {0}")] ConnectionAttempt(#[source] Arc), } diff --git a/codex-rs/exec-server/src/client_recovery.rs b/codex-rs/exec-server/src/client_recovery.rs index 8b670aeb25..58c7fdd910 100644 --- a/codex-rs/exec-server/src/client_recovery.rs +++ b/codex-rs/exec-server/src/client_recovery.rs @@ -597,12 +597,12 @@ fn is_retryable_registry_error(error: &ExecServerError) -> bool { error, ExecServerError::EnvironmentRegistryHttp { status, code, .. } if status.is_server_error() - || *status == reqwest::StatusCode::REQUEST_TIMEOUT - || *status == reqwest::StatusCode::TOO_MANY_REQUESTS + || *status == http::StatusCode::REQUEST_TIMEOUT + || *status == http::StatusCode::TOO_MANY_REQUESTS // TODO: Replace this coarse retry with an explicit registry/presence // recovery FSM so `environment_offline` is retried only while the // executor is expected to reconnect. - || (*status == reqwest::StatusCode::CONFLICT + || (*status == http::StatusCode::CONFLICT && code.as_deref() == Some("environment_offline")) ) } diff --git a/codex-rs/exec-server/src/environment.rs b/codex-rs/exec-server/src/environment.rs index 0394a6afca..7c1cacf9bb 100644 --- a/codex-rs/exec-server/src/environment.rs +++ b/codex-rs/exec-server/src/environment.rs @@ -226,6 +226,7 @@ impl EnvironmentManager { local_runtime_paths: Option, http_client_factory: HttpClientFactory, ) -> Result { + let connect_provider = config.into_connect_provider(http_client_factory.clone())?; let manager = Self { default_environment: Some(REMOTE_ENVIRONMENT_ID.to_string()), environments: RwLock::new(HashMap::new()), @@ -233,10 +234,7 @@ impl EnvironmentManager { local_runtime_paths, http_client_factory, }; - manager.upsert_noise_environment( - REMOTE_ENVIRONMENT_ID.to_string(), - config.connect_provider(), - )?; + manager.upsert_noise_environment(REMOTE_ENVIRONMENT_ID.to_string(), connect_provider)?; Ok(manager) } @@ -584,13 +582,13 @@ fn noise_environment_config_from_values( } }; - let config = NoiseRendezvousEnvironmentConfig::new( + NoiseRendezvousEnvironmentConfig::new( registry_url, environment_id, auth_token, chatgpt_account_id, - )?; - Ok(Some(config)) + ) + .map(Some) } fn optional_environment_value(name: &str) -> Option { @@ -939,10 +937,14 @@ mod tests { let manager = EnvironmentManager::from_noise_environment_config( config, /*local_runtime_paths*/ None, - legacy_http_client_factory(), + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), ) .expect("build environment manager"); + assert_eq!( + manager.http_client_factory().outbound_proxy_policy(), + OutboundProxyPolicy::RespectSystemProxy + ); assert_eq!( manager.default_environment_id(), Some(REMOTE_ENVIRONMENT_ID) diff --git a/codex-rs/exec-server/src/environment_bootstrap_tests.rs b/codex-rs/exec-server/src/environment_bootstrap_tests.rs index 70fae51c9a..f4c01df664 100644 --- a/codex-rs/exec-server/src/environment_bootstrap_tests.rs +++ b/codex-rs/exec-server/src/environment_bootstrap_tests.rs @@ -12,6 +12,7 @@ use crate::LOCAL_ENVIRONMENT_ID; use crate::REMOTE_ENVIRONMENT_ID; use crate::environment_provider::EnvironmentDefault; use crate::environment_provider::EnvironmentProviderSnapshot; +use crate::remote::NoiseRendezvousEnvironmentConfig; #[test] fn prepared_remote_environment_is_detected_without_starting_a_connection() { @@ -35,6 +36,70 @@ fn prepared_remote_environment_is_detected_without_starting_a_connection() { ); } +#[test] +fn prepared_noise_environment_is_detected_before_http_policy_is_resolved() { + let config = NoiseRendezvousEnvironmentConfig::new( + "https://registry-user:registry-password@registry.example/api?access_token=query-secret#fragment-secret" + .to_string(), + "environment-requested".to_string(), + "registry-token".to_string(), + Some("workspace-123".to_string()), + ) + .expect("Noise environment configuration"); + let prepared = PreparedEnvironmentManager { + source: PreparedEnvironmentSource::Noise(config), + }; + + assert!(prepared.default_environment_is_remote()); + + let debug = format!("{prepared:?}"); + assert!(debug.contains("")); + assert!(!debug.contains("registry-token")); + assert!(!debug.contains("workspace-123")); + assert!(!debug.contains("registry-user")); + assert!(!debug.contains("registry-password")); + assert!(!debug.contains("registry.example")); + assert!(!debug.contains("query-secret")); + assert!(!debug.contains("fragment-secret")); +} + +#[test] +fn prepared_noise_environment_rejects_invalid_configuration() { + let invalid_configs = [ + ("", "environment-requested", "registry-token", None), + ("https://registry.example", "", "registry-token", None), + ( + "https://registry.example", + "environment-requested", + "", + None, + ), + ( + "https://registry.example", + "environment-requested", + "registry\ntoken", + None, + ), + ( + "https://registry.example", + "environment-requested", + "registry-token", + Some("workspace\n123"), + ), + ]; + + for (registry_url, environment_id, auth_token, chatgpt_account_id) in invalid_configs { + let result = NoiseRendezvousEnvironmentConfig::new( + registry_url.to_string(), + environment_id.to_string(), + auth_token.to_string(), + chatgpt_account_id.map(str::to_string), + ); + + assert!(result.is_err()); + } +} + #[test] fn prepared_local_and_disabled_environments_are_not_remote() { let local = PreparedEnvironmentManager { diff --git a/codex-rs/exec-server/src/remote.rs b/codex-rs/exec-server/src/remote.rs index 0b5d995874..8a3c1703e0 100644 --- a/codex-rs/exec-server/src/remote.rs +++ b/codex-rs/exec-server/src/remote.rs @@ -4,7 +4,10 @@ use std::time::Instant; use codex_api::AuthProvider; use codex_api::SharedAuthProvider; +use codex_http_client::ClientRouteClass; use codex_http_client::HttpClientFactory; +use codex_http_client::HttpResponse; +use codex_http_client::RouteAwareClientPool; use futures::FutureExt; use http::HeaderMap; use http::HeaderName; @@ -49,7 +52,7 @@ const NOISE_RELAY_SECURITY_PROFILE: &str = "noise_hybrid_ik_v1"; struct EnvironmentRegistryClient { base_url: String, auth_provider: SharedAuthProvider, - http: reqwest::Client, + http: RouteAwareClientPool, connect_timeout: Duration, telemetry: ExecServerTelemetry, } @@ -66,21 +69,28 @@ impl std::fmt::Debug for EnvironmentRegistryClient { impl EnvironmentRegistryClient { #[cfg(test)] fn new(base_url: String, auth_provider: SharedAuthProvider) -> Result { - Self::new_with_telemetry(base_url, auth_provider, ExecServerTelemetry::default()) + Self::new_with_telemetry( + base_url, + auth_provider, + ExecServerTelemetry::default(), + HttpClientFactory::new(codex_http_client::OutboundProxyPolicy::ReqwestDefault), + ) } fn new_with_telemetry( base_url: String, auth_provider: SharedAuthProvider, telemetry: ExecServerTelemetry, + http_client_factory: HttpClientFactory, ) -> Result { let base_url = normalize_base_url(base_url)?; Ok(Self { base_url, auth_provider, - http: reqwest::Client::builder() - .redirect(reqwest::redirect::Policy::none()) - .build()?, + http: RouteAwareClientPool::new_without_redirects_or_request_logging( + http_client_factory, + ClientRouteClass::Api, + ), connect_timeout: DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT, telemetry, }) @@ -206,15 +216,15 @@ impl EnvironmentRegistryClient { }) } - async fn parse_json_response( - &self, - response: reqwest::Response, - ) -> Result + async fn parse_json_response(&self, response: HttpResponse) -> Result where R: for<'de> Deserialize<'de>, { if response.status().is_success() { - return response.json::().await.map_err(ExecServerError::from); + return response + .json::() + .await + .map_err(|error| ExecServerError::EnvironmentRegistryRequest(error.into())); } let status = response.status(); @@ -276,7 +286,8 @@ impl HarnessKeyValidator for RegistryHarnessKeyValidator { } let response = response .json::() - .await?; + .await + .map_err(|error| ExecServerError::EnvironmentRegistryRequest(error.into()))?; if !response.valid { return Err(ExecServerError::Protocol( "environment registry rejected Noise relay harness key".to_string(), @@ -288,17 +299,22 @@ impl HarnessKeyValidator for RegistryHarnessKeyValidator { /// Noise connection configuration for a Codex harness. /// -/// The provider holds the authenticated registry client so every reconnect -/// receives fresh URL and harness-key authorization material. +/// Configuration stays inert until the effective outbound HTTP policy is known. +/// Its connection provider then holds the authenticated registry client so every +/// reconnect receives fresh URL and harness-key authorization material. #[derive(Clone)] pub(crate) struct NoiseRendezvousEnvironmentConfig { - provider: Arc, + base_url: String, + environment_id: String, + auth_provider: SharedAuthProvider, } impl std::fmt::Debug for NoiseRendezvousEnvironmentConfig { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("NoiseRendezvousEnvironmentConfig") - .field("provider", &"") + .field("base_url", &"") + .field("environment_id", &self.environment_id) + .field("auth_provider", &"") .finish() } } @@ -310,23 +326,30 @@ impl NoiseRendezvousEnvironmentConfig { bearer_token: String, chatgpt_account_id: Option, ) -> Result { + let base_url = normalize_base_url(base_url)?; let environment_id = normalize_environment_id(environment_id)?; let auth_provider = static_bearer_auth_provider(bearer_token, chatgpt_account_id)?; - let client = EnvironmentRegistryClient::new_with_telemetry( - base_url, - auth_provider, - ExecServerTelemetry::default(), - )?; Ok(Self { - provider: Arc::new(EnvironmentRegistryNoiseConnectProvider { - client, - environment_id, - }), + base_url, + environment_id, + auth_provider, }) } - pub(crate) fn connect_provider(&self) -> Arc { - Arc::clone(&self.provider) + pub(crate) fn into_connect_provider( + self, + http_client_factory: HttpClientFactory, + ) -> Result, ExecServerError> { + let client = EnvironmentRegistryClient::new_with_telemetry( + self.base_url, + self.auth_provider, + ExecServerTelemetry::default(), + http_client_factory, + )?; + Ok(Arc::new(EnvironmentRegistryNoiseConnectProvider { + client, + environment_id: self.environment_id, + })) } } @@ -475,6 +498,7 @@ pub async fn run_remote_environment( config.base_url.clone(), config.auth_provider.clone(), config.telemetry.clone(), + config.http_client_factory.clone(), )?; let processor = ConnectionProcessor::new_with_telemetry( runtime_paths, @@ -685,6 +709,8 @@ mod tests { use opentelemetry::trace::TracerProvider as _; use opentelemetry_sdk::trace::SdkTracerProvider; use pretty_assertions::assert_eq; + use tokio::io::AsyncWriteExt; + use tokio::net::TcpListener; use tracing::Instrument; use tracing_subscriber::prelude::*; use wiremock::Mock; @@ -805,7 +831,10 @@ mod tests { .expect("noise configuration"); let bundle = config - .connect_provider() + .into_connect_provider(HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + )) + .expect("Noise connect provider") .connect_bundle(harness_public_key) .await .expect("Noise connect bundle"); @@ -850,6 +879,52 @@ mod tests { )); } + #[tokio::test] + async fn connect_environment_times_out_when_registry_response_body_stalls() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("registry listener should bind"); + let registry_url = format!( + "http://{}", + listener + .local_addr() + .expect("registry listener should have an address") + ); + tokio::spawn(async move { + let (mut stream, _) = listener + .accept() + .await + .expect("registry request should connect"); + stream + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 256\r\n\r\n{", + ) + .await + .expect("registry response headers should write"); + sleep(Duration::from_secs(1)).await; + }); + let mut client = + EnvironmentRegistryClient::new(registry_url, static_registry_auth_provider()) + .expect("client"); + client.connect_timeout = Duration::from_millis(50); + let harness_public_key = NoiseChannelIdentity::generate() + .expect("identity") + .public_key(); + + let error = match client + .connect_environment("environment-requested", harness_public_key) + .await + { + Ok(_) => panic!("stalled connect response body should time out"), + Err(error) => error, + }; + + assert!(matches!( + error, + ExecServerError::EnvironmentRegistryRequest(error) if error.is_timeout() + )); + } + #[tokio::test] async fn register_environment_does_not_follow_redirects_with_auth_headers() { let server = MockServer::start().await; diff --git a/codex-rs/exec-server/src/remote/noise_tests.rs b/codex-rs/exec-server/src/remote/noise_tests.rs index 4aaecba334..c59328227a 100644 --- a/codex-rs/exec-server/src/remote/noise_tests.rs +++ b/codex-rs/exec-server/src/remote/noise_tests.rs @@ -1,4 +1,6 @@ +use std::io::Write; use std::sync::Arc; +use std::sync::Mutex; use std::time::Duration; use anyhow::Result; @@ -6,6 +8,7 @@ use codex_api::AuthProvider; use codex_api::SharedAuthProvider; use codex_http_client::HttpClientFactory; use codex_http_client::OutboundProxyPolicy; +use codex_http_client::cache_system_proxy_route_for_test; use http::HeaderMap; use http::HeaderValue; use tokio::io::AsyncReadExt; @@ -13,6 +16,8 @@ use tokio::io::AsyncWriteExt; use tokio::net::TcpListener; use tokio::time::timeout; use tokio_tungstenite::accept_async; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::SubscriberExt; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; @@ -20,6 +25,7 @@ use wiremock::matchers::body_partial_json; use wiremock::matchers::header; use wiremock::matchers::method; use wiremock::matchers::path; +use wiremock::matchers::query_param; use super::*; @@ -41,6 +47,123 @@ fn static_registry_auth_provider() -> SharedAuthProvider { Arc::new(StaticRegistryAuthProvider) } +#[tokio::test(flavor = "current_thread")] +async fn registry_requests_do_not_log_sensitive_urls_or_response_headers() -> 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 || RegistryLogWriter(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", "registry log capture sentinel"); + + let server = MockServer::start().await; + let harness_public_key = NoiseChannelIdentity::generate()?.public_key(); + let executor_public_key = NoiseChannelIdentity::generate()?.public_key(); + for (operation, response, cookie_secret, location_secret) in [ + ( + "register", + serde_json::json!({ + "environment_id": "environment-requested", + "url": "wss://rendezvous.test/environment", + "security_profile": NOISE_RELAY_SECURITY_PROFILE, + "executor_registration_id": "registration-1", + }), + "register-cookie-secret", + "register-location-secret", + ), + ( + "connect", + serde_json::json!({ + "environment_id": "environment-requested", + "url": "wss://rendezvous.test/harness", + "security_profile": NOISE_RELAY_SECURITY_PROFILE, + "executor_registration_id": "registration-1", + "executor_public_key": executor_public_key.clone(), + "harness_key_authorization": HARNESS_KEY_AUTHORIZATION, + }), + "connect-cookie-secret", + "connect-location-secret", + ), + ( + "validate", + serde_json::json!({ "valid": true }), + "validate-cookie-secret", + "validate-location-secret", + ), + ] { + Mock::given(method("POST")) + .and(path("/registry-path-secret")) + .and(query_param( + "registry_token", + format!( + "registry-query-secret/cloud/environment/environment-requested/{operation}" + ), + )) + .respond_with( + ResponseTemplate::new(200) + .insert_header("set-cookie", format!("session={cookie_secret}")) + .insert_header( + "location", + format!("https://registry.example/private?token={location_secret}"), + ) + .set_body_json(response), + ) + .expect(1) + .mount(&server) + .await; + } + + let registry_url = + server + .uri() + .replacen("http://", "http://registry-user:registry-password@", 1); + let registry_url = + format!("{registry_url}/registry-path-secret?registry_token=registry-query-secret"); + let client = EnvironmentRegistryClient::new(registry_url, static_registry_auth_provider())?; + client + .register_environment("environment-requested", &executor_public_key) + .await?; + client + .connect_environment("environment-requested", harness_public_key.clone()) + .await?; + RegistryHarnessKeyValidator { + client, + environment_id: "environment-requested".to_string(), + executor_registration_id: "registration-1".to_string(), + } + .validate_harness_key(&harness_public_key, HARNESS_KEY_AUTHORIZATION) + .await?; + + let logs = String::from_utf8(log_buffer.lock().expect("log buffer lock").clone())?; + assert!(logs.contains("registry log capture sentinel")); + for secret in [ + "registry-user", + "registry-password", + "registry-path-secret", + "registry-query-secret", + "registry-token", + HARNESS_KEY_AUTHORIZATION, + "register-cookie-secret", + "register-location-secret", + "connect-cookie-secret", + "connect-location-secret", + "validate-cookie-secret", + "validate-location-secret", + ] { + assert!(!logs.contains(secret), "logs exposed {secret}:\n{logs}"); + } + + Ok(()) +} + #[tokio::test] async fn reconnect_reuses_registration_until_url_is_rejected() -> Result<()> { let listener = TcpListener::bind("127.0.0.1:0").await?; @@ -95,6 +218,56 @@ async fn reconnect_reuses_registration_until_url_is_rejected() -> Result<()> { Ok(()) } +#[tokio::test] +async fn noise_connect_provider_uses_supplied_system_proxy_policy() -> Result<()> { + let proxy = MockServer::start().await; + let registry_url = "http://registry-policy-proxy.test"; + let request_url = format!("{registry_url}/cloud/environment/environment-requested/connect"); + cache_system_proxy_route_for_test(&request_url, proxy.uri()); + + let harness_public_key = NoiseChannelIdentity::generate()?.public_key(); + let executor_public_key = NoiseChannelIdentity::generate()?.public_key(); + Mock::given(method("POST")) + .and(path("/cloud/environment/environment-requested/connect")) + .and(header("authorization", "Bearer registry-token")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "environment_id": "environment-requested", + "url": "wss://rendezvous.test/cloud-agent/default/ws/environment/environment-requested", + "security_profile": NOISE_RELAY_SECURITY_PROFILE, + "executor_registration_id": "registration-1", + "executor_public_key": executor_public_key.clone(), + "harness_key_authorization": HARNESS_KEY_AUTHORIZATION, + }))) + .expect(1) + .mount(&proxy) + .await; + + let provider = NoiseRendezvousEnvironmentConfig::new( + registry_url.to_string(), + "environment-requested".to_string(), + "registry-token".to_string(), + /*chatgpt_account_id*/ None, + )? + .into_connect_provider(HttpClientFactory::new( + OutboundProxyPolicy::RespectSystemProxy, + ))?; + let bundle = timeout( + Duration::from_secs(5), + provider.connect_bundle(harness_public_key), + ) + .await??; + let requests = proxy + .received_requests() + .await + .expect("proxy request recording should be enabled"); + + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].url.as_str(), request_url); + assert_eq!(bundle.executor_public_key, executor_public_key); + + Ok(()) +} + #[tokio::test] async fn validate_harness_key_requires_explicit_valid_response() { let server = MockServer::start().await; @@ -164,3 +337,19 @@ async fn validate_harness_key_does_not_expose_error_body() { if message == "environment registry harness key validation failed" )); } + +struct RegistryLogWriter(Arc>>); + +impl Write for RegistryLogWriter { + fn write(&mut self, buffer: &[u8]) -> std::io::Result { + self.0 + .lock() + .expect("log buffer lock") + .extend_from_slice(buffer); + Ok(buffer.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} 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 f05a69cbe8..c2a0de7853 100644 --- a/codex-rs/http-client/src/route_aware_client_pool.rs +++ b/codex-rs/http-client/src/route_aware_client_pool.rs @@ -241,6 +241,20 @@ impl RouteAwareClientPool { ) } + /// Creates a no-redirect pool without request URL or response-header diagnostics. + pub fn new_without_redirects_or_request_logging( + http_client_factory: HttpClientFactory, + route_class: ClientRouteClass, + ) -> Self { + Self::with_builder( + http_client_factory, + route_class, + HttpClientBuilder::new() + .without_redirects() + .without_request_logging(), + ) + } + /// Creates a pool whose clients limit only connection establishment. /// /// The timeout applies to every client built for a resolved route, including redirect hops.