diff --git a/codex-rs/app-server/src/config_manager_service_tests.rs b/codex-rs/app-server/src/config_manager_service_tests.rs index e87ef08a39..639839d706 100644 --- a/codex-rs/app-server/src/config_manager_service_tests.rs +++ b/codex-rs/app-server/src/config_manager_service_tests.rs @@ -736,6 +736,7 @@ async fn psp_feature_configures_first_party_routing() -> Result<()> { assert_eq!( config.http_client_factory(), HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) + .with_system_proxy_fallback() .with_chatgpt_cookies([HeaderValue::from_static("oai-chat-psp=true")]) ); assert_eq!( diff --git a/codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs b/codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs index aff213843f..3eedf082f5 100644 --- a/codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs +++ b/codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs @@ -413,7 +413,7 @@ impl AccountRequestProcessor { _ = auth_changes.wait_for(|state| state.owner_generation != auth_state.owner_generation) => { return Err(WorkspaceRoutingError::AccountChanged.into()); } - result = tokio::time::timeout(Duration::from_secs(/*secs*/ 10), read) => { + result = tokio::time::timeout(Duration::from_secs(/*secs*/ 15), read) => { result.map_err(|_| WorkspaceRoutingError::DiscoveryTimeout)? } }; diff --git a/codex-rs/app-server/tests/suite/v2/account_system_proxy_tests.rs b/codex-rs/app-server/tests/suite/v2/account_system_proxy_tests.rs new file mode 100644 index 0000000000..1770bc7086 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/account_system_proxy_tests.rs @@ -0,0 +1,297 @@ +//! Browser login must finish enterprise bootstrap when only the system proxy can reach it. + +use anyhow::Context; +use anyhow::Result; +use app_test_support::ChatGptIdTokenClaims; +use app_test_support::MockResponsesConfig; +use app_test_support::encode_id_token; +use app_test_support::mount_workspace_routing; +use codex_app_server::in_process; +use codex_app_server::in_process::InProcessServerEvent; +use codex_app_server::in_process::InProcessStartArgs; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::LoginAccountParams; +use codex_app_server_protocol::LoginAccountResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::UserInput; +use codex_arg0::Arg0DispatchPaths; +use codex_config::CloudConfigBundleLoader; +use codex_config::LoaderOverrides; +use codex_core::config::ConfigBuilder; +use codex_exec_server::EnvironmentManager; +use codex_feedback::CodexFeedback; +use codex_http_client::HttpClientBuilder; +use codex_http_client::cache_system_proxy_route_for_test; +use codex_protocol::protocol::SessionSource; +use core_test_support::responses; +use core_test_support::skip_if_no_network; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::sync::Arc; +use std::time::Duration; +use tempfile::TempDir; +use tokio::process::Command; +use tokio::time::timeout; +use url::Url; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::body_string_contains; +use wiremock::matchers::method; +use wiremock::matchers::path; + +const TEST_NAME: &str = + "suite::v2::account_system_proxy::browser_login_bootstraps_through_system_proxy"; +const TEST_MODE: &str = "CODEX_APP_SERVER_SYSTEM_PROXY_TEST_MODE"; +const BLOCKED_ORIGIN: &str = "http://127.0.0.1:0"; +const READ_TIMEOUT: Duration = Duration::from_secs(60); + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[serial_test::serial(login_port)] +async fn browser_login_bootstraps_through_system_proxy() -> Result<()> { + skip_if_no_network!(Ok(())); + + let Ok(mode) = std::env::var(TEST_MODE) else { + // The in-process server shares the proxy cache. Isolate environment-derived + // routing and the login issuer from the other integration tests. + for mode in ["bootstrap-only", "cloud-enables-proxy"] { + let mut command = Command::new(std::env::current_exe()?); + command.arg("--exact").arg(TEST_NAME); + for key in codex_network_proxy::PROXY_ENV_KEYS { + command.env_remove(key); + } + let output = timeout( + Duration::from_secs(120), + command + .kill_on_drop(true) + .env(TEST_MODE, mode) + .env("CODEX_APP_SERVER_LOGIN_ISSUER", BLOCKED_ORIGIN) + .env_remove("OPENAI_API_KEY") + .env_remove("CODEX_API_KEY") + .output(), + ) + .await??; + assert!( + output.status.success(), + "{mode} failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + } + return Ok(()); + }; + + let cloud_enables_proxy = mode == "cloud-enables-proxy"; + let proxy = MockServer::start().await; + let provider = MockServer::start().await; + let codex_home = TempDir::new()?; + let model_catalog = codex_models_manager::bundled_models_response()?; + let model_catalog_path = codex_home.path().join("models.json"); + std::fs::write(&model_catalog_path, serde_json::to_vec(&model_catalog)?)?; + MockResponsesConfig::new(&provider.uri()) + .with_model(&model_catalog.models[0].slug) + .with_provider_base_url(&format!( + "{}/v1", + if cloud_enables_proxy { + BLOCKED_ORIGIN.to_string() + } else { + provider.uri() + } + )) + .with_provider_config("requires_openai_auth = true\nsupports_websockets = false") + .with_root_config(&format!( + "chatgpt_base_url = \"{BLOCKED_ORIGIN}/backend-api\"\ncli_auth_credentials_store = \"file\"\nmodel_catalog_json = {}", + serde_json::to_string(&model_catalog_path)?, + )) + .write(codex_home.path())?; + let id_token = encode_id_token( + &ChatGptIdTokenClaims::new() + .email("proxy@example.com") + .plan_type("enterprise") + .chatgpt_user_id("proxy-user") + .chatgpt_account_id("proxy-workspace"), + )?; + Mock::given(method("POST")) + .and(path("/oauth/token")) + .and(body_string_contains("grant_type=authorization_code")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "id_token": id_token, + "access_token": "proxy-access-token", + "refresh_token": "proxy-refresh-token", + }))) + .expect(1) + .mount(&proxy) + .await; + Mock::given(method("POST")) + .and(path("/oauth/token")) + .and(body_string_contains("requested_token=openai-api-key")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "access_token": "proxy-api-key", + }))) + .expect(1) + .mount(&proxy) + .await; + let cloud_config = if cloud_enables_proxy { + json!({"config_toml": {"enterprise_managed": [{ + "id": "proxy-rollout", + "name": "System proxy rollout", + "contents": "[features]\nrespect_system_proxy = true\n", + }]}}) + } else { + json!({}) + }; + Mock::given(method("GET")) + .and(path("/backend-api/wham/config/bundle")) + .respond_with(ResponseTemplate::new(200).set_body_json(cloud_config)) + .expect(1..) + .mount(&proxy) + .await; + mount_workspace_routing(&proxy).await; + for route in [ + "/oauth/token", + "/backend-api/wham/config/bundle", + "/backend-api/wham/accounts/check", + "/v1/responses", + ] { + cache_system_proxy_route_for_test(&format!("{BLOCKED_ORIGIN}{route}"), proxy.uri()); + } + let model_requests = responses::mount_sse_once( + if cloud_enables_proxy { + &proxy + } else { + &provider + }, + responses::sse(vec![responses::ev_completed("proxy-response")]), + ) + .await; + + let loader_overrides = LoaderOverrides::without_managed_config_for_tests(); + let config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .fallback_cwd(Some(codex_home.path().to_path_buf())) + .loader_overrides(loader_overrides.clone()) + .build() + .await?; + assert!(!config.respect_system_proxy); + let mut client = in_process::start(InProcessStartArgs { + arg0_paths: Arg0DispatchPaths::default(), + config: Arc::new(config), + cli_overrides: Vec::new(), + loader_overrides, + strict_config: false, + cloud_config_bundle: CloudConfigBundleLoader::default(), + thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader), + feedback: CodexFeedback::new(), + log_db: None, + state_db: None, + environment_manager: Arc::new(EnvironmentManager::default_for_tests()), + config_warnings: Vec::new(), + session_source: SessionSource::Cli, + enable_codex_api_key_env: false, + initialize: InitializeParams { + client_info: ClientInfo { + name: "codex-app-server-tests".to_string(), + title: None, + version: "0.1.0".to_string(), + }, + capabilities: None, + }, + channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, + }) + .await?; + let login: LoginAccountResponse = serde_json::from_value( + client + .request(ClientRequest::LoginAccount { + request_id: RequestId::Integer(1), + params: LoginAccountParams::Chatgpt { + codex_streamlined_login: false, + use_hosted_login_success_page: false, + app_brand: None, + }, + }) + .await? + .expect("browser login starts"), + )?; + let LoginAccountResponse::Chatgpt { login_id, auth_url } = login else { + anyhow::bail!("unexpected login response: {login:?}"); + }; + let auth_url = Url::parse(&auth_url)?; + let params = auth_url + .query_pairs() + .collect::>(); + let mut callback = Url::parse(params.get("redirect_uri").context("redirect URI")?)?; + callback + .query_pairs_mut() + .append_pair("code", "proxy-code") + .append_pair("state", params.get("state").context("OAuth state")?); + let browser = HttpClientBuilder::new().build_direct()?; + let response = timeout(READ_TIMEOUT, browser.get(callback).send()).await??; + assert_eq!(response.status(), 200); + timeout(READ_TIMEOUT, async { + loop { + let event = client + .next_event() + .await + .context("login completion event")?; + if let InProcessServerEvent::ServerNotification(notification) = event + && let ServerNotification::AccountLoginCompleted(completed) = notification.as_ref() + { + assert_eq!(completed.login_id.as_deref(), Some(login_id.as_str())); + assert!(completed.success, "login failed: {completed:?}"); + return anyhow::Ok(()); + } + } + }) + .await??; + let saved_auth: serde_json::Value = + serde_json::from_slice(&std::fs::read(codex_home.path().join("auth.json"))?)?; + assert_eq!(saved_auth["OPENAI_API_KEY"], "proxy-api-key"); + + let started: ThreadStartResponse = serde_json::from_value( + client + .request(ClientRequest::ThreadStart { + request_id: RequestId::Integer(2), + params: ThreadStartParams::default(), + }) + .await? + .expect("first thread starts without restarting"), + )?; + client + .request(ClientRequest::TurnStart { + request_id: RequestId::Integer(3), + params: TurnStartParams { + thread_id: started.thread.id.clone(), + input: vec![UserInput::Text { + text: "Hello through the proxy".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + }) + .await? + .expect("first turn starts without restarting"); + timeout(READ_TIMEOUT, async { + loop { + let event = client.next_event().await.context("turn completion event")?; + if let InProcessServerEvent::ServerNotification(notification) = event + && let ServerNotification::TurnCompleted(completed) = notification.as_ref() + && completed.thread_id == started.thread.id + { + assert_eq!(completed.turn.status, TurnStatus::Completed); + return anyhow::Ok(()); + } + } + }) + .await??; + model_requests.single_request(); + client.shutdown().await?; + proxy.verify().await; + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 325dff2f8a..91c6ddab00 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -1,4 +1,6 @@ mod account; +#[path = "account_system_proxy_tests.rs"] +mod account_system_proxy; mod account_thread_usage; mod analytics; mod app_installed; diff --git a/codex-rs/app-server/tests/suite/v2/workspace_routing.rs b/codex-rs/app-server/tests/suite/v2/workspace_routing.rs index d7a22d40ed..c87f31c711 100644 --- a/codex-rs/app-server/tests/suite/v2/workspace_routing.rs +++ b/codex-rs/app-server/tests/suite/v2/workspace_routing.rs @@ -18,6 +18,7 @@ use serde_json::json; use std::sync::Arc; use std::time::Duration; use tempfile::TempDir; +use test_case::test_case; use tokio::sync::Notify; use tokio::time::timeout; use wiremock::Mock; @@ -84,8 +85,15 @@ fn routing(account: &str, origin: &str, routing: &str) -> Value { json!({"chatgptAccountId": account, "backendOrigin": origin, "accountRoutingOverride": routing}) } +#[test_case(0, 0, "pro"; "immediate_responses")] +#[test_case(6, 0, "pro"; "slow_accounts_response")] +#[test_case(0, 12, "enterprise"; "slow_cloud_bundle_response")] #[tokio::test] -async fn saved_workspace_is_discovered_once_and_not_the_default_account() -> Result<()> { +async fn saved_workspace_is_discovered_once_and_not_the_default_account( + accounts_delay_secs: u64, + bundle_delay_secs: u64, + plan_type: &str, +) -> Result<()> { let backend = MockServer::start().await; Mock::given(method("GET")).and(path("/backend-api/wham/accounts/check")) .and(header("chatgpt-account-id", "selected")) @@ -94,18 +102,43 @@ async fn saved_workspace_is_discovered_once_and_not_the_default_account() -> Res {"id": "other", "workspace_backend_origin": "https://other.example", "account_routing_override": "us"}, {"id": "selected", "workspace_backend_origin": "https://gov.chatgpt.com", "account_routing_override": "NO_CONSTRAINT"} ], "default_account_id": "other" - }))).expect(1).mount(&backend).await; + })).set_delay(Duration::from_secs(accounts_delay_secs))) + .expect(if accounts_delay_secs > 5 { 2 } else { 1 }) + .mount(&backend).await; let home = TempDir::new()?; config(&home, &backend).await?; + if bundle_delay_secs > 0 { + Mock::given(method("GET")) + .and(path("/backend-api/wham/config/bundle")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(json!({})) + .set_delay(Duration::from_secs(bundle_delay_secs)), + ) + .with_priority(1) + .expect(2) + .mount(&backend) + .await; + } write_chatgpt_auth( home.path(), ChatGptAuthFixture::new("token") .account_id("selected") .email("user@example.com") - .plan_type("pro"), + .plan_type(plan_type), AuthCredentialsStoreMode::File, )?; - let mut server = start(&home).await?; + let mut env_overrides = vec![("OPENAI_API_KEY", None), ("CODEX_API_KEY", None)]; + env_overrides.extend( + codex_network_proxy::PROXY_ENV_KEYS + .iter() + .map(|key| (*key, None)), + ); + let mut server = TestAppServer::builder() + .with_codex_home(home.path()) + .with_env_overrides(&env_overrides) + .build_initialized_with_timeout(READ_TIMEOUT) + .await?; let notification = timeout( READ_TIMEOUT, server.read_stream_until_notification_message("account/updated"), @@ -113,10 +146,10 @@ async fn saved_workspace_is_discovered_once_and_not_the_default_account() -> Res .await??; assert_eq!( notification.params, - Some(json!({"authMode": "chatgpt", "planType": "pro"})) + Some(json!({"authMode": "chatgpt", "planType": plan_type})) ); let expected = json!({ - "account": {"type": "chatgpt", "email": "user@example.com", "planType": "pro"}, + "account": {"type": "chatgpt", "email": "user@example.com", "planType": plan_type}, "requiresOpenaiAuth": true, "workspaceRouting": routing("selected", "https://gov.chatgpt.com", "NO_CONSTRAINT") }); assert_eq!(read(&mut server).await?, expected); diff --git a/codex-rs/backend-client/Cargo.toml b/codex-rs/backend-client/Cargo.toml index 512d960617..8a8948a9a2 100644 --- a/codex-rs/backend-client/Cargo.toml +++ b/codex-rs/backend-client/Cargo.toml @@ -24,6 +24,7 @@ codex-http-client = { workspace = true } codex-login = { workspace = true } codex-model-provider = { workspace = true } codex-protocol = { workspace = true } +tokio = { workspace = true, features = ["time"] } [dev-dependencies] pretty_assertions = "1" diff --git a/codex-rs/backend-client/src/client.rs b/codex-rs/backend-client/src/client.rs index 6b5fcbf9fd..06d57377b1 100644 --- a/codex-rs/backend-client/src/client.rs +++ b/codex-rs/backend-client/src/client.rs @@ -12,8 +12,10 @@ use anyhow::Result; use codex_api::SharedAuthProvider; use codex_http_client::ClientRouteClass; use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_http_client::RouteAwareClientPool; use codex_http_client::RouteAwareRequestBuilder; +use codex_http_client::RouteAwareRequestError; use codex_login::CodexAuth; use codex_login::default_client::get_codex_user_agent; use codex_protocol::account::PlanType as AccountPlanType; @@ -33,6 +35,7 @@ use http::header::USER_AGENT; use serde::Serialize; use serde::de::DeserializeOwned; use std::fmt; +use std::time::Duration; pub(crate) mod analytics; mod chatgpt_turn_cost; @@ -320,6 +323,39 @@ impl Client { Ok((body, content_type)) } + async fn exec_bootstrap_get( + &self, + url: &str, + ) -> std::result::Result<(String, String), RequestError> { + let request = self.request(Method::GET, url).headers(self.headers()); + if !self.http.allows_system_proxy_fallback() { + return self.exec_request_detailed(request, "GET", url).await; + } + + // Bound the complete GET, including its body, to leave time for proxy discovery and retry + // within the cloud loader's startup budget. GETs are safe to retry after a response timeout. + match tokio::time::timeout( + Duration::from_secs(5), + self.exec_request_detailed(request, "GET", url), + ) + .await + { + Ok(Err(RequestError::Other(error))) + if error + .downcast_ref::() + .is_some_and(|error| error.is_connect() || error.is_timeout()) => {} + Err(_) => {} + Ok(response) => return response, + } + + let http = self + .http + .clone() + .with_outbound_proxy_policy(OutboundProxyPolicy::RespectSystemProxy); + let request = http.get(url).headers(self.headers()); + self.exec_request_detailed(request, "GET", url).await + } + fn decode_json(&self, url: &str, ct: &str, body: &str) -> Result { match serde_json::from_str::(body) { Ok(v) => Ok(v), @@ -349,8 +385,7 @@ impl Client { PathStyle::CodexApi => format!("{}/api/codex/accounts/check", self.base_url), PathStyle::ChatGptApi => format!("{}/wham/accounts/check", self.base_url), }; - let req = self.request(Method::GET, &url).headers(self.headers()); - let (body, _) = self.exec_request_detailed(req, "GET", &url).await?; + let (body, _) = self.exec_bootstrap_get(&url).await?; serde_json::from_str(&body) .map_err(|_| RequestError::Other(anyhow::anyhow!("Invalid accounts response."))) } @@ -480,8 +515,7 @@ impl Client { PathStyle::CodexApi => format!("{}/api/codex/config/bundle", self.base_url), PathStyle::ChatGptApi => format!("{}/wham/config/bundle", self.base_url), }; - let req = self.request(Method::GET, &url).headers(self.headers()); - let (body, ct) = self.exec_request_detailed(req, "GET", &url).await?; + let (body, ct) = self.exec_bootstrap_get(&url).await?; self.decode_json::(&url, &ct, &body) .map_err(RequestError::from) } diff --git a/codex-rs/backend-client/src/client_request_tests.rs b/codex-rs/backend-client/src/client_request_tests.rs index 7c10a2a9f8..4dfe774fec 100644 --- a/codex-rs/backend-client/src/client_request_tests.rs +++ b/codex-rs/backend-client/src/client_request_tests.rs @@ -1,11 +1,21 @@ +use std::io::BufRead; use std::io::Read; use std::io::Write; +use std::net::TcpListener; +use std::process::Command; use std::sync::Arc; use std::time::Duration; use codex_http_client::HttpClientFactory; use codex_http_client::OutboundProxyPolicy; +use codex_http_client::cache_system_proxy_route_for_test; use pretty_assertions::assert_eq; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::header; +use wiremock::matchers::method; +use wiremock::matchers::path; use super::*; @@ -142,3 +152,197 @@ async fn migrated_requests_preserve_query_auth_and_json_body() { ); assert!(requests[1].ends_with(r#"{"prompt":"hello"}"#)); } + +const BLOCKED_ORIGIN: &str = "http://127.0.0.1:0"; + +fn run_without_environment_proxies(test_name: &str) -> bool { + const CHILD: &str = "CODEX_BOOTSTRAP_PROXY_TEST_CHILD"; + if std::env::var_os(CHILD).is_some() { + return false; + } + let mut command = Command::new(std::env::current_exe().expect("test executable")); + command.arg("--exact").arg(test_name).env(CHILD, "1"); + for key in [ + "HTTP_PROXY", + "http_proxy", + "HTTPS_PROXY", + "https_proxy", + "ALL_PROXY", + "all_proxy", + "NO_PROXY", + "no_proxy", + ] { + command.env_remove(key); + } + let output = command.output().expect("run isolated proxy test"); + assert!( + output.status.success(), + "{test_name} failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + true +} + +fn fallback_client(base_url: &str) -> Client { + Client::new( + base_url, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault).with_system_proxy_fallback(), + ) + .with_auth_provider(Arc::new(codex_model_provider::BearerAuthProvider::new( + "bootstrap-token".to_string(), + ))) + .with_chatgpt_account_id("workspace-123") + .with_user_agent("bootstrap-test") +} + +#[tokio::test] +async fn bootstrap_gets_resolve_each_fallback_destination_and_preserve_headers() { + if run_without_environment_proxies( + "client::request_tests::bootstrap_gets_resolve_each_fallback_destination_and_preserve_headers", + ) { + return; + } + for (base_suffix, api_prefix) in [("", "/api/codex"), ("/backend-api", "/wham")] { + let bundle_proxy = MockServer::start().await; + let accounts_proxy = MockServer::start().await; + let base_url = format!("{BLOCKED_ORIGIN}{base_suffix}"); + let client = fallback_client(&base_url); + for (endpoint, proxy) in [ + ("config/bundle", &bundle_proxy), + ("accounts/check", &accounts_proxy), + ] { + let endpoint_path = format!("{base_suffix}{api_prefix}/{endpoint}"); + cache_system_proxy_route_for_test( + &format!("{base_url}{api_prefix}/{endpoint}"), + proxy.uri(), + ); + Mock::given(method("GET")) + .and(path(endpoint_path)) + .and(header("authorization", "Bearer bootstrap-token")) + .and(header("chatgpt-account-id", "workspace-123")) + .and(header("user-agent", "bootstrap-test")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({}))) + .expect(1) + .mount(proxy) + .await; + } + client.get_config_bundle().await.expect("cloud bundle"); + client.get_accounts_check().await.expect("accounts check"); + bundle_proxy.verify().await; + accounts_proxy.verify().await; + } +} + +#[tokio::test] +async fn bootstrap_gets_keep_default_responses_without_proxy_retry() { + if run_without_environment_proxies( + "client::request_tests::bootstrap_gets_keep_default_responses_without_proxy_retry", + ) { + return; + } + for status in [200, 401] { + let issuer = MockServer::start().await; + let proxy = MockServer::start().await; + for endpoint in ["config/bundle", "accounts/check"] { + cache_system_proxy_route_for_test( + &format!("{}/api/codex/{endpoint}", issuer.uri()), + proxy.uri(), + ); + } + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(status).set_body_json(serde_json::json!({}))) + .expect(2) + .mount(&issuer) + .await; + let client = fallback_client(&issuer.uri()); + let bundle = client.get_config_bundle().await; + let accounts = client.get_accounts_check().await; + if status == 200 { + bundle.expect("default cloud route succeeds"); + accounts.expect("default accounts route succeeds"); + } else { + assert_eq!( + [ + bundle.expect_err("cloud status error").status(), + accounts.expect_err("accounts status error").status(), + ], + [Some(http::StatusCode::UNAUTHORIZED); 2] + ); + } + issuer.verify().await; + assert_eq!(proxy.received_requests().await.unwrap_or_default().len(), 0); + } +} + +#[tokio::test] +async fn bootstrap_gets_honor_disabled_fallback() { + if run_without_environment_proxies( + "client::request_tests::bootstrap_gets_honor_disabled_fallback", + ) { + return; + } + let proxy = MockServer::start().await; + let base_url = BLOCKED_ORIGIN; + for endpoint in ["config/bundle", "accounts/check"] { + cache_system_proxy_route_for_test(&format!("{base_url}/api/codex/{endpoint}"), proxy.uri()); + } + let client = Client::new( + base_url, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ); + assert!(client.get_config_bundle().await.is_err()); + assert!(client.get_accounts_check().await.is_err()); + assert_eq!(proxy.received_requests().await.unwrap_or_default().len(), 0); +} + +#[tokio::test] +async fn bootstrap_get_recovers_from_stalled_body_before_cloud_startup_timeout() { + if run_without_environment_proxies( + "client::request_tests::bootstrap_get_recovers_from_stalled_body_before_cloud_startup_timeout", + ) { + return; + } + let issuer = TcpListener::bind(("127.0.0.1", 0)).expect("bind stalled issuer"); + let issuer_url = format!("http://{}", issuer.local_addr().expect("issuer address")); + let (finish, finished) = std::sync::mpsc::channel(); + let issuer_thread = std::thread::spawn(move || { + let (mut stream, _) = issuer.accept().expect("accept bootstrap GET"); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("set read timeout"); + let mut request = String::new(); + std::io::BufReader::new(&stream) + .read_line(&mut request) + .expect("read request line"); + stream + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\n\r\n{", + ) + .expect("send headers and incomplete body"); + finished + .recv_timeout(Duration::from_secs(20)) + .expect("proxy request finishes while issuer body is pending"); + }); + let proxy = MockServer::start().await; + cache_system_proxy_route_for_test( + &format!("{issuer_url}/api/codex/config/bundle"), + proxy.uri(), + ); + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({}))) + .expect(1) + .mount(&proxy) + .await; + let result = tokio::time::timeout( + Duration::from_secs(15), + fallback_client(&issuer_url).get_config_bundle(), + ) + .await; + finish.send(()).expect("release stalled issuer"); + issuer_thread.join().expect("issuer finishes"); + result + .expect("proxy retry must fit inside the cloud startup budget") + .expect("cloud bundle via proxy"); + proxy.verify().await; +} diff --git a/codex-rs/cloud-config/src/service.rs b/codex-rs/cloud-config/src/service.rs index 29641f14eb..6df148010a 100644 --- a/codex-rs/cloud-config/src/service.rs +++ b/codex-rs/cloud-config/src/service.rs @@ -32,7 +32,7 @@ use tokio::sync::OnceCell; use tokio::time::sleep; use tokio::time::timeout; -pub(crate) const CLOUD_CONFIG_BUNDLE_TIMEOUT: Duration = Duration::from_secs(15); +pub(crate) const CLOUD_CONFIG_BUNDLE_TIMEOUT: Duration = Duration::from_secs(20); const CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS: usize = 5; const CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60); const CLOUD_CONFIG_BUNDLE_TIMEOUT_RETRY_INTERVAL: Duration = Duration::from_secs(5); diff --git a/codex-rs/config/src/loader/mod.rs b/codex-rs/config/src/loader/mod.rs index e6ceed2f90..8dbaad22f2 100644 --- a/codex-rs/config/src/loader/mod.rs +++ b/codex-rs/config/src/loader/mod.rs @@ -1153,8 +1153,10 @@ fn sanitize_project_config( { ignored_keys.push("features.shell_snapshot".to_string()); } - if features.remove("respect_system_proxy").is_some() { - ignored_keys.push("features.respect_system_proxy".to_string()); + for key in ["respect_system_proxy", "system_proxy_fallback"] { + if features.remove(key).is_some() { + ignored_keys.push(format!("features.{key}")); + } } if credential_broker_configured && features diff --git a/codex-rs/config/src/loader/tests.rs b/codex-rs/config/src/loader/tests.rs index 77c32f3351..e1ae693f30 100644 --- a/codex-rs/config/src/loader/tests.rs +++ b/codex-rs/config/src/loader/tests.rs @@ -1056,3 +1056,25 @@ async fn local_layers_keep_raw_paths_order_and_legacy_requirements() { 0 ); } + +#[test] +fn project_config_cannot_change_system_proxy_routing() { + for key in ["respect_system_proxy", "system_proxy_fallback"] { + for enabled in [false, true] { + let mut config: TomlValue = + toml::from_str(&format!("[features]\n{key} = {enabled}\nplugins = true")) + .expect("valid project config"); + let ignored = sanitize_project_config( + &mut config, + CredentialBrokerProjectState::Unconfigured, + &[], + ); + assert_eq!(ignored, vec![format!("features.{key}")]); + assert_eq!( + config, + toml::from_str::("[features]\nplugins = true") + .expect("valid expected config"), + ); + } + } +} diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 08f1ed5774..15c109cabf 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -1037,6 +1037,9 @@ "step_model_switching": { "type": "boolean" }, + "system_proxy_fallback": { + "type": "boolean" + }, "telepathy": { "type": "boolean" }, @@ -6790,6 +6793,9 @@ "step_model_switching": { "type": "boolean" }, + "system_proxy_fallback": { + "type": "boolean" + }, "telepathy": { "type": "boolean" }, diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index d1670d580e..0e4a13835c 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -2055,6 +2055,10 @@ respect_system_proxy = true .outbound_proxy_policy(), codex_http_client::OutboundProxyPolicy::ReqwestDefault ); + assert!( + resolve_bootstrap_http_client_factory(&configured, Some(&disabled))? + .allows_system_proxy_fallback() + ); let configured = ConfigToml::default(); let enabled = Sourced::new( @@ -2073,6 +2077,115 @@ respect_system_proxy = true .outbound_proxy_policy(), codex_http_client::OutboundProxyPolicy::RespectSystemProxy ); + assert!( + !resolve_bootstrap_http_client_factory(&configured, Some(&enabled))? + .allows_system_proxy_fallback() + ); + assert!( + resolve_bootstrap_http_client_factory(&configured, /*feature_requirements*/ None)? + .allows_system_proxy_fallback() + ); + Ok(()) +} + +#[tokio::test] +async fn system_proxy_fallback_config_matches_bootstrap() -> std::io::Result<()> { + for (features, expected_fallback, expected_policy) in [ + ( + "", + true, + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ), + ( + "respect_system_proxy = false", + true, + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ), + ( + "system_proxy_fallback = false", + false, + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ), + ( + "respect_system_proxy = true", + false, + codex_http_client::OutboundProxyPolicy::RespectSystemProxy, + ), + ( + "respect_system_proxy = true\nsystem_proxy_fallback = false", + false, + codex_http_client::OutboundProxyPolicy::RespectSystemProxy, + ), + ] { + let codex_home = TempDir::new()?; + std::fs::write( + codex_home.path().join(CONFIG_TOML_FILE), + format!("[features]\n{features}\n"), + )?; + let config = ConfigBuilder::without_managed_config_for_tests() + .codex_home(codex_home.path().to_path_buf()) + .build() + .await?; + let bootstrap = ConfigToml { + features: Some(toml::from_str(features).expect("valid feature configuration")), + ..Default::default() + }; + let factory = + resolve_bootstrap_http_client_factory(&bootstrap, /*feature_requirements*/ None)?; + assert_eq!( + factory.allows_system_proxy_fallback(), + expected_fallback, + "{features}" + ); + assert_eq!( + factory.outbound_proxy_policy(), + expected_policy, + "{features}" + ); + assert_eq!(config.http_client_factory(), factory, "{features}"); + assert_eq!( + config.auth_route_config().http_client_factory(), + &factory, + "{features}" + ); + } + Ok(()) +} + +#[tokio::test] +async fn system_proxy_fallback_honors_feature_requirements() -> std::io::Result<()> { + for (configured, required) in [(true, false), (false, true)] { + let cfg = ConfigToml { + features: Some( + toml::from_str(&format!("system_proxy_fallback = {configured}")) + .expect("valid features"), + ), + ..Default::default() + }; + let requirements = Sourced::new( + FeatureRequirementsToml { + entries: BTreeMap::from([("system_proxy_fallback".to_string(), required)]), + }, + RequirementSource::Unknown, + ); + let bootstrap = resolve_bootstrap_http_client_factory(&cfg, Some(&requirements))?; + assert_eq!(bootstrap.allows_system_proxy_fallback(), required); + let codex_home = TempDir::new()?; + std::fs::write( + codex_home.path().join(CONFIG_TOML_FILE), + format!("[features]\nsystem_proxy_fallback = {configured}\n"), + )?; + let config = ConfigBuilder::without_managed_config_for_tests() + .codex_home(codex_home.path().to_path_buf()) + .cloud_config_bundle( + CloudConfigBundleFixture::loader_with_enterprise_requirement(format!( + "[features]\nsystem_proxy_fallback = {required}\n" + )), + ) + .build() + .await?; + assert_eq!(config.http_client_factory(), bootstrap); + } Ok(()) } diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index ffce510b61..2d7625cd6f 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1650,7 +1650,10 @@ impl Config { } else { OutboundProxyPolicy::ReqwestDefault }; - let factory = HttpClientFactory::new(outbound_proxy_policy); + let mut factory = HttpClientFactory::new(outbound_proxy_policy); + if !self.respect_system_proxy && self.features.enabled(Feature::SystemProxyFallback) { + factory = factory.with_system_proxy_fallback(); + } if self.features.enabled(Feature::Psp) { factory.with_chatgpt_cookies([HeaderValue::from_static("oai-chat-psp=true")]) } else { @@ -2976,17 +2979,8 @@ pub fn resolve_bootstrap_respect_system_proxy( cfg: &ConfigToml, feature_requirements: Option<&Sourced>, ) -> std::io::Result { - let configured_features = Features::from_sources( - FeatureConfigSource { - features: cfg.features.as_ref(), - experimental_use_unified_exec_tool: cfg.experimental_use_unified_exec_tool, - }, - FeatureConfigSource::default(), - FeatureOverrides::default(), - ); - let features = - ManagedFeatures::from_configured(configured_features, feature_requirements.cloned())?; - Ok(features.get().enabled(Feature::RespectSystemProxy)) + resolve_bootstrap_http_client_factory(cfg, feature_requirements) + .map(|factory| factory.outbound_proxy_policy() == OutboundProxyPolicy::RespectSystemProxy) } /// Resolves auth route settings for the initial cloud-config bootstrap. @@ -3003,14 +2997,28 @@ pub fn resolve_bootstrap_http_client_factory( cfg: &ConfigToml, feature_requirements: Option<&Sourced>, ) -> std::io::Result { - resolve_bootstrap_respect_system_proxy(cfg, feature_requirements).map(|respect_system_proxy| { - let outbound_proxy_policy = if respect_system_proxy { - OutboundProxyPolicy::RespectSystemProxy - } else { - OutboundProxyPolicy::ReqwestDefault - }; - HttpClientFactory::new(outbound_proxy_policy) - }) + let configured_features = Features::from_sources( + FeatureConfigSource { + features: cfg.features.as_ref(), + experimental_use_unified_exec_tool: cfg.experimental_use_unified_exec_tool, + }, + FeatureConfigSource::default(), + FeatureOverrides::default(), + ); + let features = + ManagedFeatures::from_configured(configured_features, feature_requirements.cloned())?; + let outbound_proxy_policy = if features.enabled(Feature::RespectSystemProxy) { + OutboundProxyPolicy::RespectSystemProxy + } else { + OutboundProxyPolicy::ReqwestDefault + }; + let mut factory = HttpClientFactory::new(outbound_proxy_policy); + if outbound_proxy_policy == OutboundProxyPolicy::ReqwestDefault + && features.enabled(Feature::SystemProxyFallback) + { + factory = factory.with_system_proxy_fallback(); + } + Ok(factory) } pub(crate) fn resolve_web_search_mode_for_turn( diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index ac8f675e13..83337a6c89 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -199,6 +199,8 @@ pub enum Feature { Worktrees, /// Respect host system proxy settings for Codex-owned network clients. RespectSystemProxy, + /// Retry eligible bootstrap requests through the system proxy after normal routing fails. + SystemProxyFallback, /// Enable collab tools. Collab, /// Enable task-path-based multi-agent routing. @@ -1291,6 +1293,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::UnderDevelopment, default_enabled: false, }, + FeatureSpec { + id: Feature::SystemProxyFallback, + key: "system_proxy_fallback", + stage: Stage::Stable, + default_enabled: true, + }, FeatureSpec { id: Feature::Collab, key: "multi_agent", diff --git a/codex-rs/http-client/src/client_builder.rs b/codex-rs/http-client/src/client_builder.rs index 9f374293a1..db5ef2ccbe 100644 --- a/codex-rs/http-client/src/client_builder.rs +++ b/codex-rs/http-client/src/client_builder.rs @@ -7,6 +7,8 @@ use http::HeaderMap; use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; use std::time::Duration; use codex_utils_rustls_provider::ensure_rustls_crypto_provider; @@ -31,6 +33,7 @@ use crate::with_chatgpt_cloudflare_cookie_store; pub struct HttpClientBuilder { default_headers: Option, follow_redirects: bool, + redirect_observed: Option>, connect_timeout: Option, chatgpt_cloudflare_cookie_store: bool, chatgpt_cookie_store: Option>, @@ -91,6 +94,13 @@ impl HttpClientBuilder { self } + /// Marks the supplied flag when a redirect is encountered, preserving the default policy. + /// Use a fresh flag for each operation whose retry safety depends on its redirect history. + pub fn with_redirect_tracking(mut self, redirect_observed: Arc) -> Self { + self.redirect_observed = Some(redirect_observed); + self + } + pub(crate) fn follows_redirects(&self) -> bool { self.follow_redirects } @@ -284,6 +294,11 @@ impl HttpClientBuilder { } if !self.follow_redirects { builder = builder.redirect(reqwest::redirect::Policy::none()); + } else if let Some(redirect_observed) = self.redirect_observed { + builder = builder.redirect(reqwest::redirect::Policy::custom(move |attempt| { + redirect_observed.store(/*val*/ true, Ordering::Relaxed); + reqwest::redirect::Policy::default().redirect(attempt) + })); } if let Some(connect_timeout) = self.connect_timeout { builder = builder.connect_timeout(connect_timeout); @@ -303,6 +318,7 @@ impl Default for HttpClientBuilder { Self { default_headers: None, follow_redirects: true, + redirect_observed: None, connect_timeout: None, chatgpt_cloudflare_cookie_store: false, chatgpt_cookie_store: None, diff --git a/codex-rs/http-client/src/outbound_proxy.rs b/codex-rs/http-client/src/outbound_proxy.rs index 64713478ec..429711f0bb 100644 --- a/codex-rs/http-client/src/outbound_proxy.rs +++ b/codex-rs/http-client/src/outbound_proxy.rs @@ -166,6 +166,7 @@ impl fmt::Debug for OutboundProxyRoute { #[derive(Clone)] pub struct HttpClientFactory { outbound_proxy_policy: OutboundProxyPolicy, + system_proxy_fallback: bool, chatgpt_cookie_store: Option>, network_policy: NetworkPolicy, } @@ -173,6 +174,7 @@ pub struct HttpClientFactory { impl PartialEq for HttpClientFactory { fn eq(&self, other: &Self) -> bool { self.outbound_proxy_policy == other.outbound_proxy_policy + && self.system_proxy_fallback == other.system_proxy_fallback && self.network_policy == other.network_policy && self .chatgpt_cookie_store @@ -191,6 +193,7 @@ impl fmt::Debug for HttpClientFactory { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("HttpClientFactory") .field("outbound_proxy_policy", &self.outbound_proxy_policy) + .field("system_proxy_fallback", &self.system_proxy_fallback) .field("network_policy", &self.network_policy) .finish() } @@ -201,6 +204,7 @@ impl HttpClientFactory { pub const fn new(outbound_proxy_policy: OutboundProxyPolicy) -> Self { Self { outbound_proxy_policy, + system_proxy_fallback: false, chatgpt_cookie_store: None, network_policy: NetworkPolicy::unmanaged(), } @@ -216,6 +220,24 @@ impl HttpClientFactory { &self.network_policy } + /// Changes the route policy while retaining the configured cookies and network policy. + pub fn with_outbound_proxy_policy(mut self, policy: OutboundProxyPolicy) -> Self { + self.outbound_proxy_policy = policy; + self + } + + /// Allows bootstrap callers to retry through the system proxy when their request is safe + /// to replay. This does not change routing or enable retries for ordinary HTTP requests. + pub fn with_system_proxy_fallback(mut self) -> Self { + self.system_proxy_fallback = true; + self + } + + pub fn allows_system_proxy_fallback(&self) -> bool { + self.system_proxy_fallback + && self.outbound_proxy_policy == OutboundProxyPolicy::ReqwestDefault + } + /// Adds process-scoped cookies to requests made by ChatGPT cookie-store clients. pub fn with_chatgpt_cookies(mut self, cookies: impl IntoIterator) -> Self { let cookies = cookies.into_iter().collect::>(); 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 3c688c78b4..9ad9b2c567 100644 --- a/codex-rs/http-client/src/route_aware_client_pool.rs +++ b/codex-rs/http-client/src/route_aware_client_pool.rs @@ -312,6 +312,16 @@ impl RouteAwareClientPool { self.http_client_factory.outbound_proxy_policy() } + pub fn allows_system_proxy_fallback(&self) -> bool { + self.http_client_factory.allows_system_proxy_fallback() + } + + /// Changes routing while preserving transport settings and clients cached by resolved route. + pub fn with_outbound_proxy_policy(mut self, policy: OutboundProxyPolicy) -> Self { + self.http_client_factory = self.http_client_factory.with_outbound_proxy_policy(policy); + self + } + /// Creates a pool with the shared default HTTP transport settings. pub fn new(http_client_factory: HttpClientFactory, route_class: ClientRouteClass) -> Self { Self::with_builder(http_client_factory, route_class, HttpClientBuilder::new()) 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 8918175de4..8f903f4d3c 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 @@ -125,6 +125,7 @@ fn request_builder_debug_redacts_url_secrets() { concat!( "RouteAwareRequestBuilder { pool: RouteAwareClientPool { ", "http_client_factory: HttpClientFactory { outbound_proxy_policy: ReqwestDefault, ", + "system_proxy_fallback: false, ", "network_policy: NetworkPolicy { managed: false, .. } }, ", "route_class: Api, .. }, method: Some(GET), ", "url: Some(\"\"), .. }" diff --git a/codex-rs/login/src/device_code_auth.rs b/codex-rs/login/src/device_code_auth.rs index a909da40f2..263abd5aaf 100644 --- a/codex-rs/login/src/device_code_auth.rs +++ b/codex-rs/login/src/device_code_auth.rs @@ -201,7 +201,7 @@ pub async fn complete_device_code_login( }; let redirect_uri = format!("{base_url}/deviceauth/callback"); - let tokens = crate::server::exchange_code_for_tokens( + let (tokens, _) = crate::server::exchange_code_for_tokens( base_url, &opts.client_id, &redirect_uri, diff --git a/codex-rs/login/src/server.rs b/codex-rs/login/src/server.rs index ca77cf7a16..084a8ba574 100644 --- a/codex-rs/login/src/server.rs +++ b/codex-rs/login/src/server.rs @@ -21,6 +21,8 @@ use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::LazyLock; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; use std::thread; use std::time::Duration; @@ -30,7 +32,6 @@ use crate::auth::save_auth; use crate::callback_params::LIFE_SCIENCES_OAUTH_STATE_SUFFIX; use crate::callback_params::LoginCallbackResult; use crate::callback_params::LoginOnboardingEntrypoint; -use crate::default_client::create_raw_auth_client; use crate::default_client::originator; use crate::oauth::AuthorizationCodeGrant; use crate::oauth::AuthorizationRequest; @@ -55,6 +56,11 @@ use crate::token_data::TokenData; use crate::token_data::parse_chatgpt_jwt_claims; use chrono::Utc; use codex_config::types::AuthCredentialsStoreMode; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClient; +use codex_http_client::HttpClientBuilder; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_protocol::auth::AuthMode; use codex_utils_template::Template; use serde_json::Value as JsonValue; @@ -418,7 +424,7 @@ async fn process_request( ) .await { - Ok(tokens) => { + Ok((tokens, client)) => { if let Err(message) = ensure_workspace_allowed( opts.forced_chatgpt_workspace_id.as_deref(), &tokens.id_token, @@ -432,14 +438,10 @@ async fn process_request( ); } // Obtain API key via token-exchange and persist - let api_key = obtain_api_key( - &opts.issuer, - &opts.client_id, - &tokens.id_token, - &opts.auth_route_config, - ) - .await - .ok(); + let api_key = + obtain_api_key(&client, &opts.issuer, &opts.client_id, &tokens.id_token) + .await + .ok(); if let Err(err) = persist_tokens_async( &opts.codex_home, api_key.clone(), @@ -698,7 +700,7 @@ pub(crate) struct ExchangedTokens { pub refresh_token: String, } -/// Exchanges an authorization code for tokens. +/// Exchanges an authorization code for tokens and returns the client for further token exchanges. /// /// The returned error remains suitable for user-facing CLI/browser surfaces, so backend-provided /// non-JSON error text is preserved there. Structured logging stays narrower: it logs reviewed @@ -711,38 +713,59 @@ pub(crate) async fn exchange_code_for_tokens( pkce: &PkceCodes, code: &str, auth_route_config: &AuthRouteConfig, -) -> io::Result { - // Reuse the route selected for the issuer, rather than resolving the token path again. - let client = create_raw_auth_client(issuer.trim_end_matches('/'), auth_route_config)?; - let endpoint = format!("{}/oauth/token", issuer.trim_end_matches('/')); - let oauth = OAuthClient::new( - &client, - TokenEndpoint { - url: &endpoint, - client_id, - encoding: TokenEncoding::Form, - timeout: None, - error_body_limit: ErrorBodyLimit::Unlimited, - }, - ); +) -> io::Result<(ExchangedTokens, HttpClient)> { + let token_endpoint = format!("{}/oauth/token", issuer.trim_end_matches('/')); + let factory = auth_route_config.http_client_factory(); + let allows_fallback = factory.allows_system_proxy_fallback(); + let redirect_observed = Arc::new(AtomicBool::new(false)); + let mut builder = HttpClientBuilder::new().without_request_logging(); + if allows_fallback { + // Only bound connection establishment. A response timeout could mean the one-time code + // was consumed, so it must never cause another token POST. + builder = builder + .with_redirect_tracking(Arc::clone(&redirect_observed)) + .connect_timeout(Duration::from_secs(10)); + } info!( issuer = %sanitize_url_for_logging(issuer), - token_endpoint = %sanitize_url_for_logging(&endpoint), + token_endpoint = %sanitize_url_for_logging(&token_endpoint), %redirect_uri, "starting oauth token exchange" ); - match oauth - .exchange_code(AuthorizationCodeGrant { - code, + let (mut client, mut result) = send_code_exchange_request( + factory, + builder.clone(), + &token_endpoint, + client_id, + redirect_uri, + pkce, + code, + ) + .await?; + // A redirect means the original POST reached the server and may have consumed the code. + if matches!(&result, Err(OAuthError::Transport(error)) if error.is_connect()) + && allows_fallback + && !redirect_observed.load(Ordering::Relaxed) + { + info!("oauth token connection failed; retrying with system proxy"); + let factory = factory + .clone() + .with_outbound_proxy_policy(OutboundProxyPolicy::RespectSystemProxy); + (client, result) = send_code_exchange_request( + &factory, + builder, + &token_endpoint, + client_id, redirect_uri, pkce, - resource: None, - }) - .await - { + code, + ) + .await?; + } + match result { Ok(tokens) => { info!("oauth token exchange succeeded"); - Ok(tokens) + Ok((tokens, client)) } Err(OAuthError::Rejected(rejection)) => { if let Some(error) = rejection.body_read_error { @@ -769,6 +792,41 @@ pub(crate) async fn exchange_code_for_tokens( } } +async fn send_code_exchange_request( + factory: &HttpClientFactory, + builder: HttpClientBuilder, + token_endpoint: &str, + client_id: &str, + redirect_uri: &str, + pkce: &PkceCodes, + code: &str, +) -> io::Result<(HttpClient, Result)> { + let client = builder.build_respecting_outbound_proxy_policy( + factory, + token_endpoint, + ClientRouteClass::Auth, + )?; + let oauth = OAuthClient::new( + &client, + TokenEndpoint { + url: token_endpoint, + client_id, + encoding: TokenEncoding::Form, + timeout: None, + error_body_limit: ErrorBodyLimit::Unlimited, + }, + ); + let result = oauth + .exchange_code(AuthorizationCodeGrant { + code, + redirect_uri, + pkce, + resource: None, + }) + .await; + Ok((client, result)) +} + /// Persists exchanged credentials using the configured local auth store. pub(crate) async fn persist_tokens_async( codex_home: &Path, @@ -953,10 +1011,10 @@ fn html_escape(input: &str) -> String { /// Exchanges an authenticated ID token for an API-key style access token. pub(crate) async fn obtain_api_key( + client: &HttpClient, issuer: &str, client_id: &str, id_token: &str, - auth_route_config: &AuthRouteConfig, ) -> io::Result { // Token exchange for an API key access token #[derive(serde::Deserialize)] @@ -964,7 +1022,6 @@ pub(crate) async fn obtain_api_key( access_token: String, } let token_endpoint = format!("{}/oauth/token", issuer.trim_end_matches('/')); - let client = create_raw_auth_client(&token_endpoint, auth_route_config)?; let resp = client .post(token_endpoint) .header("Content-Type", "application/x-www-form-urlencoded") diff --git a/codex-rs/login/tests/suite/login_proxy_fallback.rs b/codex-rs/login/tests/suite/login_proxy_fallback.rs new file mode 100644 index 0000000000..e05fdf8a04 --- /dev/null +++ b/codex-rs/login/tests/suite/login_proxy_fallback.rs @@ -0,0 +1,225 @@ +//! Browser callback coverage for one-time OAuth codes and system-proxy fallback. + +use std::io::Read; +use std::net::TcpListener; +use std::process::Command; +use std::time::Duration; + +use anyhow::Result; +use base64::Engine; +use codex_config::types::AuthCredentialsStoreMode; +use codex_http_client::HttpClientBuilder; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use codex_http_client::cache_system_proxy_route_for_test; +use codex_login::AuthKeyringBackendKind; +use codex_login::AuthRouteConfig; +use codex_login::ServerOptions; +use codex_login::run_login_server; +use core_test_support::skip_if_no_network; +use pretty_assertions::assert_eq; +use tempfile::tempdir; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; + +const CHILD_CASE: &str = "CODEX_TOKEN_PROXY_FALLBACK_TEST_CASE"; +const CHILD_PROXY: &str = "CODEX_TOKEN_PROXY_FALLBACK_TEST_PROXY"; +const CHILD_ISSUER: &str = "CODEX_TOKEN_PROXY_FALLBACK_TEST_ISSUER"; +const BLOCKED_ORIGIN: &str = "http://127.0.0.1:0"; +const PROXY_ENV_KEYS: [&str; 8] = [ + "HTTP_PROXY", + "http_proxy", + "HTTPS_PROXY", + "https_proxy", + "ALL_PROXY", + "all_proxy", + "NO_PROXY", + "no_proxy", +]; + +fn run_case(test_name: &str, issuer: &str, proxy: &str) -> Result<()> { + let mut command = Command::new(std::env::current_exe()?); + command.arg("--exact").arg(test_name); + for key in PROXY_ENV_KEYS { + command.env_remove(key); + } + command + .env(CHILD_CASE, "1") + .env(CHILD_PROXY, proxy) + .env(CHILD_ISSUER, issuer); + let output = command.output()?; + assert!( + output.status.success(), + "{test_name} child failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + Ok(()) +} + +enum CallbackOutcome { + Succeeded, + Failed, +} + +async fn exercise_callback(expected: CallbackOutcome) -> Result<()> { + let issuer = std::env::var(CHILD_ISSUER)?; + let proxy = std::env::var(CHILD_PROXY)?; + cache_system_proxy_route_for_test(&format!("{issuer}/oauth/token"), proxy); + + let tmp = tempdir()?; + let mut opts = ServerOptions::new( + tmp.path().to_path_buf(), + codex_login::CLIENT_ID.to_string(), + /*forced_chatgpt_workspace_id*/ None, + AuthCredentialsStoreMode::File, + AuthKeyringBackendKind::Direct, + AuthRouteConfig::from_http_client_factory( + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) + .with_system_proxy_fallback(), + ), + ); + opts.issuer = issuer; + opts.port = 0; + opts.open_browser = false; + opts.force_state = Some("proxy-fallback".to_string()); + let server = run_login_server(opts)?; + let callback = HttpClientBuilder::new() + .without_redirects() + .build_direct()? + .get(format!( + "http://127.0.0.1:{}/auth/callback?code=abc&state=proxy-fallback", + server.actual_port + )) + .timeout(Duration::from_secs(20)) + .send() + .await?; + + match expected { + CallbackOutcome::Succeeded => { + assert_eq!(callback.status(), 302); + let success_url = callback.headers()["location"].to_str()?; + HttpClientBuilder::new() + .build_direct()? + .get(success_url) + .send() + .await? + .error_for_status()?; + tokio::time::timeout(Duration::from_secs(20), server.block_until_done()).await??; + let auth: serde_json::Value = + serde_json::from_slice(&std::fs::read(tmp.path().join("auth.json"))?)?; + assert_eq!(auth["tokens"]["access_token"], "redirect-access"); + } + CallbackOutcome::Failed => { + assert!(callback.text().await?.contains("Token exchange failed")); + let error = tokio::time::timeout(Duration::from_secs(20), server.block_until_done()) + .await? + .expect_err("the callback should fail after the token exchange fails"); + assert!(error.to_string().contains("Token exchange failed")); + assert!(!tmp.path().join("auth.json").exists()); + } + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn token_exchange_http_responses_do_not_retry_one_time_code() -> Result<()> { + skip_if_no_network!(Ok(())); + if std::env::var_os(CHILD_CASE).is_some() { + return exercise_callback(CallbackOutcome::Failed).await; + } + for response in [ + ResponseTemplate::new(400).set_body_string("invalid_grant"), + ResponseTemplate::new(307).insert_header("Location", BLOCKED_ORIGIN), + ] { + let issuer = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(response) + .expect(1) + .mount(&issuer) + .await; + let proxy = MockServer::start().await; + run_case( + "suite::login_proxy_fallback::token_exchange_http_responses_do_not_retry_one_time_code", + &issuer.uri(), + &proxy.uri(), + )?; + issuer.verify().await; + assert_eq!(proxy.received_requests().await.unwrap_or_default().len(), 0); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn token_exchange_does_not_retry_after_post_is_sent() -> Result<()> { + skip_if_no_network!(Ok(())); + if std::env::var_os(CHILD_CASE).is_some() { + return exercise_callback(CallbackOutcome::Failed).await; + } + + let issuer = TcpListener::bind(("127.0.0.1", 0))?; + let issuer_url = format!("http://{}", issuer.local_addr()?); + let issuer_thread = std::thread::spawn(move || -> std::io::Result { + let (mut stream, _) = issuer.accept()?; + stream.set_read_timeout(Some(Duration::from_secs(10)))?; + let mut request = [0_u8; 4096]; + let count = stream.read(&mut request)?; + Ok(String::from_utf8_lossy(&request[..count]).into_owned()) + }); + let proxy = MockServer::start().await; + run_case( + "suite::login_proxy_fallback::token_exchange_does_not_retry_after_post_is_sent", + &issuer_url, + &proxy.uri(), + )?; + let request = issuer_thread.join().expect("issuer thread should finish")?; + assert!(request.starts_with("POST /oauth/token HTTP/1.1")); + assert_eq!(proxy.received_requests().await.unwrap_or_default().len(), 0); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn token_exchange_preserves_redirects() -> Result<()> { + skip_if_no_network!(Ok(())); + if std::env::var_os(CHILD_CASE).is_some() { + return exercise_callback(CallbackOutcome::Succeeded).await; + } + for (status, redirected_method) in [(302, "GET"), (307, "POST")] { + let issuer = MockServer::start().await; + let target = MockServer::start().await; + let proxy = MockServer::start().await; + let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(serde_json::to_vec( + &serde_json::json!({ + "email": "proxy@example.com", + "https://api.openai.com/auth": {"chatgpt_account_id": "proxy-workspace"} + }), + )?); + Mock::given(method("POST")) + .and(path("/oauth/token")) + .respond_with(ResponseTemplate::new(status).insert_header("Location", target.uri())) + .expect(2) + .mount(&issuer) + .await; + Mock::given(method(redirected_method)) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "id_token": format!("e30.{payload}.sig"), + "access_token": "redirect-access", + "refresh_token": "redirect-refresh", + }))) + .expect(2) + .mount(&target) + .await; + run_case( + "suite::login_proxy_fallback::token_exchange_preserves_redirects", + &issuer.uri(), + &proxy.uri(), + )?; + issuer.verify().await; + target.verify().await; + assert_eq!(proxy.received_requests().await.unwrap_or_default().len(), 0); + } + Ok(()) +} diff --git a/codex-rs/login/tests/suite/mod.rs b/codex-rs/login/tests/suite/mod.rs index 3c3bb24d62..fb90143fd4 100644 --- a/codex-rs/login/tests/suite/mod.rs +++ b/codex-rs/login/tests/suite/mod.rs @@ -1,5 +1,6 @@ // Aggregates all former standalone integration tests as modules. mod auth_refresh; mod device_code_login; +mod login_proxy_fallback; mod login_server_e2e; mod logout;