openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 17:28:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add system proxy fallback for login and startup requests (#46562)

Browse Source 
## Why

Login and enterprise configuration bootstrap can fail when their endpoints are reachable only through the system proxy, before cloud configuration can enable `respect_system_proxy`.

## What changed

- Enable `features.system_proxy_fallback` by default, honor managed feature requirements, and prevent project configuration from changing it.
- Retry account discovery and cloud configuration GETs through the system proxy after connection failures or timeouts, bounding the initial request and body read to five seconds.
- Retry OAuth authorization-code exchange only on connection failures before any redirect, with a ten-second connection timeout. Reuse the successful client for the subsequent API-key exchange to avoid changing routes.
- Increase workspace discovery and cloud configuration timeouts to 15 and 20 seconds, respectively, to accommodate fallback.

## Testing

Add coverage for proxy-only browser login through the first completed turn, including cloud configuration enabling `respect_system_proxy` without a restart. Cover stalled GET bodies, preserved request headers, disabled fallback, managed requirements, and project configuration restrictions. Verify that OAuth HTTP errors, redirects followed by connection failures, and failures after sending the POST do not trigger proxy retries, while successful redirects remain supported.

GitOrigin-RevId: 854eeeac186bb4865218ce645019c1e4bff85425
This commit is contained in:
Abhinav
2026-09-19 01:37:14 +00:00
committed by copyberry
parent 888be42a20
commit 3bb0a530d1
23 changed files with 1134 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/config_manager_service_tests.rs
+1
View File
@@ -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!(
codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs
+1 -1
View File
@@ -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)?
}
};
codex-rs/app-server/tests/suite/v2/account_system_proxy_tests.rs
+297
View File
@@ -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::<std::collections::HashMap<_, _>>();
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(())
}
codex-rs/app-server/tests/suite/v2/mod.rs
+2
View File
@@ -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;
codex-rs/app-server/tests/suite/v2/workspace_routing.rs
+39 -6
View File
@@ -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);
codex-rs/backend-client/Cargo.toml
+1
View File
@@ -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"
codex-rs/backend-client/src/client.rs
+38 -4
View File
@@ -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::<RouteAwareRequestError>()
.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<T: DeserializeOwned>(&self, url: &str, ct: &str, body: &str) -> Result<T> {
match serde_json::from_str::<T>(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::<ConfigBundleResponse>(&url, &ct, &body)
.map_err(RequestError::from)
}
codex-rs/backend-client/src/client_request_tests.rs
+204
View File
@@ -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;
}
codex-rs/cloud-config/src/service.rs
+1 -1
View File
@@ -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);
codex-rs/config/src/loader/mod.rs
+4 -2
View File
@@ -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
codex-rs/config/src/loader/tests.rs
+22
View File
@@ -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::<TomlValue>("[features]\nplugins = true")
.expect("valid expected config"),
);
}
}
}
codex-rs/core/config.schema.json
+6
View File
@@ -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"
},
codex-rs/core/src/config/config_tests.rs
+113
View File
@@ -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(())
}
codex-rs/core/src/config/mod.rs
+28 -20
View File
@@ -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<FeatureRequirementsToml>>,
) -> std::io::Result<bool> {
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<FeatureRequirementsToml>>,
) -> std::io::Result<HttpClientFactory> {
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(
codex-rs/features/src/lib.rs
+8
View File
@@ -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",
codex-rs/http-client/src/client_builder.rs
+16
View File
@@ -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<HeaderMap>,
follow_redirects: bool,
redirect_observed: Option<Arc<AtomicBool>>,
connect_timeout: Option<Duration>,
chatgpt_cloudflare_cookie_store: bool,
chatgpt_cookie_store: Option<Arc<ChatGptCookieStore>>,
@@ -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<AtomicBool>) -> 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,
codex-rs/http-client/src/outbound_proxy.rs
+22
View File
@@ -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<Arc<ChatGptCookieStore>>,
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<Item = HeaderValue>) -> Self {
let cookies = cookies.into_iter().collect::<Vec<_>>();
codex-rs/http-client/src/route_aware_client_pool.rs
+10
View File
@@ -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())
codex-rs/http-client/src/route_aware_client_pool_tests.rs
+1
View File
@@ -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(\"<redacted>\"), .. }"
codex-rs/login/src/device_code_auth.rs
+1 -1
View File
@@ -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,
codex-rs/login/src/server.rs
+93 -36
View File
@@ -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<ExchangedTokens> {
// 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<ExchangedTokens, OAuthError>)> {
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<String> {
// 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")
codex-rs/login/tests/suite/login_proxy_fallback.rs
+225
View File
@@ -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<String> {
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(())
}
codex-rs/login/tests/suite/mod.rs
+1
View File
@@ -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;