diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index d3889e2a74..7b8ae40ef4 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3323,6 +3323,7 @@ version = "0.0.0" dependencies = [ "anyhow", "assert_matches", + "chrono", "clap", "codex-api", "codex-app-server", @@ -3343,8 +3344,10 @@ dependencies = [ "codex-utils-cli", "codex-utils-process", "codex-websocket-auth", + "futures", "http 1.4.0", "insta", + "jsonwebtoken", "pretty_assertions", "serde", "serde_json", @@ -3355,6 +3358,7 @@ dependencies = [ "tracing", "tracing-subscriber", "url", + "wiremock", ] [[package]] @@ -3992,6 +3996,7 @@ dependencies = [ "url", "urlencoding", "webbrowser", + "windows-sys 0.61.2", "wiremock", ] diff --git a/codex-rs/cli/tests/exec_server.rs b/codex-rs/cli/tests/exec_server.rs index 59c6512602..4dd347cd68 100644 --- a/codex-rs/cli/tests/exec_server.rs +++ b/codex-rs/cli/tests/exec_server.rs @@ -37,6 +37,7 @@ use codex_exec_server::ProcessId; use codex_http_client::HttpClientFactory; use codex_http_client::OutboundProxyPolicy; use codex_utils_cargo_bin::copy_executable; +use codex_utils_path_uri::PathUri; use futures::SinkExt; use futures::StreamExt; use predicates::prelude::PredicateBooleanExt; @@ -56,6 +57,7 @@ use tokio_tungstenite::tungstenite::handshake::server::Response; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; +use wiremock::matchers::header; use wiremock::matchers::method; use wiremock::matchers::path; @@ -205,6 +207,235 @@ fn local_exec_server_allows_disabled_parent_lifetime_environment_variable() -> R Ok(()) } +enum PatCredentialChange { + Rotate, + SwitchAccount, + Logout, + Unchanged, + UnavailableReplacement, +} + +async fn assert_pat_registry_reload(change: PatCredentialChange) -> Result<()> { + let home = TempDir::new()?; + let config_server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/backend-api/wham/config/bundle")) + .and(header("authorization", "Bearer at-synthetic-executor-pat")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "config_toml": {"enterprise_managed": []}, + "requirements_toml": {"enterprise_managed": []}, + }))) + .expect(1) + .mount(&config_server) + .await; + std::fs::write( + home.path().join("config.toml"), + format!( + "cli_auth_credentials_store = 'file'\nchatgpt_base_url = '{}/backend-api'\n", + config_server.uri(), + ), + )?; + let auth_path = home.path().join("auth.json"); + std::fs::write( + &auth_path, + serde_json::to_vec(&serde_json::json!({ + "personal_access_token": "at-synthetic-executor-pat", + }))?, + )?; + let auth_server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v1/user-auth-credential/whoami")) + .and(header("authorization", "Bearer at-synthetic-executor-pat")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "chatgpt_user_id": "user-pat-test", + "chatgpt_account_id": "account-pat-test", + "chatgpt_plan_type": "enterprise", + "chatgpt_account_is_fedramp": false, + }))) + // Cloud-config bootstrap and executor startup each hydrate once. + // Later registry requests must reuse metadata for the unchanged PAT. + .with_priority(1) + .up_to_n_times(2) + .expect(2) + .mount(&auth_server) + .await; + Mock::given(method("GET")) + .and(path("/v1/user-auth-credential/whoami")) + .respond_with(ResponseTemplate::new(503)) + .with_priority(10) + .expect(0) + .mount(&auth_server) + .await; + if !matches!( + change, + PatCredentialChange::Logout | PatCredentialChange::Unchanged + ) { + let account = match change { + PatCredentialChange::Rotate | PatCredentialChange::UnavailableReplacement => { + "account-pat-test" + } + PatCredentialChange::SwitchAccount => "another-account", + PatCredentialChange::Logout | PatCredentialChange::Unchanged => unreachable!(), + }; + if matches!(change, PatCredentialChange::UnavailableReplacement) { + Mock::given(method("GET")) + .and(path("/v1/user-auth-credential/whoami")) + .and(header( + "authorization", + "Bearer at-synthetic-replacement-pat", + )) + .respond_with(ResponseTemplate::new(503)) + .with_priority(1) + .up_to_n_times(1) + .expect(1) + .mount(&auth_server) + .await; + } + let response = ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "chatgpt_user_id": "user-pat-test", + "chatgpt_account_id": account, + "chatgpt_plan_type": "enterprise", + "chatgpt_account_is_fedramp": false, + })); + Mock::given(method("GET")) + .and(path("/v1/user-auth-credential/whoami")) + .and(header( + "authorization", + "Bearer at-synthetic-replacement-pat", + )) + .respond_with(response) + .with_priority(2) + .expect(1) + .mount(&auth_server) + .await; + } + let expected_registry_requests = match change { + PatCredentialChange::Rotate + | PatCredentialChange::Unchanged + | PatCredentialChange::UnavailableReplacement => 2, + PatCredentialChange::SwitchAccount | PatCredentialChange::Logout => 1, + }; + let second_bearer = if matches!(change, PatCredentialChange::Unchanged) { + "Bearer at-synthetic-executor-pat" + } else { + "Bearer at-synthetic-replacement-pat" + }; + let registry = MockServer::start().await; + let registration_path = "/cloud/environment/environment-pat-test/register"; + let (credential_change_tx, mut credential_change_rx) = tokio::sync::mpsc::unbounded_channel(); + Mock::given(method("POST")) + .and(path(registration_path)) + .and(header("authorization", "Bearer at-synthetic-executor-pat")) + .and(header("chatgpt-account-id", "account-pat-test")) + .respond_with(move |_: &wiremock::Request| { + let result = match change { + PatCredentialChange::Rotate + | PatCredentialChange::SwitchAccount + | PatCredentialChange::UnavailableReplacement => std::fs::write( + &auth_path, + r#"{"personal_access_token":"at-synthetic-replacement-pat"}"#, + ), + PatCredentialChange::Logout => std::fs::remove_file(&auth_path), + PatCredentialChange::Unchanged => Ok(()), + }; + let _ = credential_change_tx.send(result); + ResponseTemplate::new(502).set_body_json(serde_json::json!({ + "detail": {"code": "authentication_service_unavailable"}, + })) + }) + .with_priority(1) + .up_to_n_times(1) + .expect(1) + .mount(®istry) + .await; + Mock::given(method("POST")) + .and(path(registration_path)) + .and(header("authorization", second_bearer)) + .and(header("chatgpt-account-id", "account-pat-test")) + .respond_with(ResponseTemplate::new(403)) + .with_priority(2) + .expect(expected_registry_requests - 1) + .mount(®istry) + .await; + + let mut command = tokio::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?); + command + .current_dir(home.path()) + .env("CODEX_HOME", home.path()) + .env_remove("CODEX_ACCESS_TOKEN") + .env("CODEX_AUTHAPI_BASE_URL", auth_server.uri()) + .env_remove("CODEX_API_KEY") + .env_remove("OPENAI_API_KEY") + .env("NO_PROXY", "127.0.0.1,localhost") + .env("no_proxy", "127.0.0.1,localhost") + .args([ + "exec-server", + "--remote", + ®istry.uri(), + "--environment-id", + "environment-pat-test", + ]) + .kill_on_drop(true); + // A same-owner replacement reaches the second registry request, whose denial + // stops the executor. Logout or a different owner must fail before that request. + let result = tokio::time::timeout(Duration::from_secs(30), command.output()).await; + assert!( + result.is_ok(), + "executor timed out (AuthAPI request count: {:?})", + auth_server + .received_requests() + .await + .map(|requests| requests.len()), + ); + let output = result??; + let credential_update = credential_change_rx.try_recv(); + assert!( + credential_update.is_ok(), + "registration did not update the credential fixture ({}): {}", + output.status, + String::from_utf8_lossy(&output.stderr), + ); + credential_update?.context("update the stored PAT fixture")?; + assert!(!output.status.success()); + registry.verify().await; + auth_server.verify().await; + config_server.verify().await; + assert_eq!( + registry + .received_requests() + .await + .context("registry request recording is disabled")? + .len(), + expected_registry_requests as usize + ); + Ok(()) +} + +#[tokio::test] +async fn remote_exec_server_pat_follows_stored_rotation() -> Result<()> { + assert_pat_registry_reload(PatCredentialChange::Rotate).await +} + +#[tokio::test] +async fn remote_exec_server_pat_rejects_account_switch() -> Result<()> { + assert_pat_registry_reload(PatCredentialChange::SwitchAccount).await +} + +#[tokio::test] +async fn remote_exec_server_pat_stops_using_removed_credentials() -> Result<()> { + assert_pat_registry_reload(PatCredentialChange::Logout).await +} + +#[tokio::test] +async fn remote_exec_server_pat_reuses_metadata_during_authapi_outage() -> Result<()> { + assert_pat_registry_reload(PatCredentialChange::Unchanged).await +} + +#[tokio::test] +async fn remote_exec_server_pat_recovers_after_replacement_metadata_outage() -> Result<()> { + assert_pat_registry_reload(PatCredentialChange::UnavailableReplacement).await +} + #[tokio::test] async fn remote_exec_server_gracefully_stops_when_parent_stdin_closes() -> Result<()> { const ENVIRONMENT_ID: &str = "environment-parent-lifetime"; @@ -248,6 +479,8 @@ async fn remote_exec_server_gracefully_stops_when_parent_stdin_closes() -> Resul codex_home.path().join("config.toml"), format!( r#" +cli_auth_credentials_store = "file" + [analytics] enabled = true @@ -267,10 +500,14 @@ metrics_exporter = {{ otlp-http = {{ endpoint = "{collector_url}/v1/metrics", pr let manifest = package.path().join("codex-package.json"); std::fs::write(&manifest, r#"{"version":"1.2.3-alpha.4"}"#)?; + let stderr_path = codex_home.path().join("executor-stderr.log"); + let stderr = std::fs::File::create(&stderr_path)?; let mut command = tokio::process::Command::new(executable); command + .current_dir(codex_home.path()) .env("CODEX_HOME", codex_home.path()) .env("CODEX_API_KEY", "test-api-key") + .env_remove("CODEX_ACCESS_TOKEN") .env( codex_exec_server::CODEX_EXEC_SERVER_EXIT_ON_STDIN_CLOSE_ENV_VAR, "true", @@ -288,7 +525,7 @@ metrics_exporter = {{ otlp-http = {{ endpoint = "{collector_url}/v1/metrics", pr "2", ]) .stdin(Stdio::piped()) - .stderr(Stdio::inherit()) + .stderr(Stdio::from(stderr)) .kill_on_drop(true); let mut child = command.spawn()?; @@ -315,10 +552,11 @@ metrics_exporter = {{ otlp-http = {{ endpoint = "{collector_url}/v1/metrics", pr .filter(|request| request.url.path().ends_with("/register")) .count() }); - // Inspect only the stage reached, not request headers or bodies containing auth. + // Keep registry request headers and bodies out of the failure report. let child_status = child.try_wait(); + let stderr = std::fs::read_to_string(&stderr_path)?; return Err(error.context(format!( - "remote executor failed to attach (registration requests: {registration_requests:?}, child status: {child_status:?})" + "remote executor failed to attach (registration requests: {registration_requests:?}, child status: {child_status:?}); stderr: {stderr}" ))); } }; @@ -364,6 +602,9 @@ metrics_exporter = {{ otlp-http = {{ endpoint = "{collector_url}/v1/metrics", pr ); let expected_info = EnvironmentInfo { executor_version: "1.2.3-alpha.4".to_string(), + cwd: Some(PathUri::from_host_native_path( + codex_home.path().canonicalize()?, + )?), // The build identity belongs to the spawned CLI, not this test process. provider_id: environment_info.provider_id.clone(), prepend_path_dirs: environment_info.prepend_path_dirs.clone(), diff --git a/codex-rs/cli/tests/exec_server/auth_refresh_tests.rs b/codex-rs/cli/tests/exec_server/auth_refresh_tests.rs index 57db55f9e5..832876c96c 100644 --- a/codex-rs/cli/tests/exec_server/auth_refresh_tests.rs +++ b/codex-rs/cli/tests/exec_server/auth_refresh_tests.rs @@ -16,6 +16,62 @@ use wiremock::ResponseTemplate; use wiremock::matchers::header; use wiremock::matchers::path; +#[tokio::test] +async fn executor_startup_rejects_agent_identity_before_cloud_config() -> Result<()> { + let home = TempDir::new()?; + let server = MockServer::start().await; + Mock::given(wiremock::matchers::any()) + .respond_with(ResponseTemplate::new(503)) + .expect(0) + .mount(&server) + .await; + let manager = codex_login::test_support::auth_manager_with_agent_identity().await?; + let Some(codex_login::CodexAuth::AgentIdentity(agent)) = manager.auth_cached() else { + anyhow::bail!("expected Agent Identity fixture"); + }; + let mut record = agent.record().clone(); + record.task_id = None; + std::fs::write( + home.path().join("auth.json"), + serde_json::to_vec(&serde_json::json!({ + "auth_mode": codex_protocol::auth::AuthMode::AgentIdentity, + "agent_identity": record, + }))?, + )?; + std::fs::write( + home.path().join("config.toml"), + "cli_auth_credentials_store = 'file'\n", + )?; + let mut command = tokio::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?); + command + .current_dir(home.path()) + .env("CODEX_HOME", home.path()) + .env_remove("CODEX_API_KEY") + .env_remove("OPENAI_API_KEY") + .env_remove("CODEX_ACCESS_TOKEN") + .env("CODEX_AGENT_IDENTITY_AUTHAPI_BASE_URL", server.uri()) + .env("NO_PROXY", "127.0.0.1,localhost") + .env("no_proxy", "127.0.0.1,localhost") + .args([ + "exec-server", + "--remote", + &server.uri(), + "--environment-id", + "env", + ]) + .kill_on_drop(true); + let output = + tokio::time::timeout(std::time::Duration::from_secs(20), command.output()).await??; + assert!(!output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("requires ChatGPT authentication or API key authentication"), + "{stderr}" + ); + server.verify().await; + Ok(()) +} + #[tokio::test] async fn noise_executor_refreshes_on_401_and_reloads_rotated_credentials() -> Result<()> { let home = TempDir::new()?; @@ -58,12 +114,12 @@ async fn noise_executor_refreshes_on_401_and_reloads_rotated_credentials() -> Re .mount(®istry) .await; } - // A later registry rejection reloads the rotated token from storage, without - // another OAuth refresh or unconditional reloads on ordinary requests. + // The later registration sees the stored rotation before sending a stale + // request. The initial 401 above still exercises bounded auth recovery. Mock::given(path("/cloud/environment/env/register")) .and(header("authorization", "Bearer refreshed-token")) .respond_with(ResponseTemplate::new(401)) - .expect(1) + .expect(0) .mount(®istry) .await; let mut child = tokio::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?) diff --git a/codex-rs/exec-server-cli/Cargo.toml b/codex-rs/exec-server-cli/Cargo.toml index 2f066185a7..1743ee5278 100644 --- a/codex-rs/exec-server-cli/Cargo.toml +++ b/codex-rs/exec-server-cli/Cargo.toml @@ -29,6 +29,7 @@ codex-exec-server = { workspace = true } codex-http-client = { workspace = true } codex-login = { workspace = true } codex-model-provider = { workspace = true } +codex-protocol = { workspace = true } codex-sandboxing = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-cli = { workspace = true } @@ -47,10 +48,13 @@ tikv-jemallocator = { workspace = true } [dev-dependencies] assert_matches = { workspace = true } -codex-protocol = { workspace = true } +chrono = { workspace = true } codex-utils-cargo-bin = { workspace = true } codex-utils-process = { workspace = true } +futures = { workspace = true } insta = { workspace = true } +jsonwebtoken = { workspace = true } pretty_assertions = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["io-util", "process"] } +wiremock = { workspace = true } diff --git a/codex-rs/exec-server-cli/src/config.rs b/codex-rs/exec-server-cli/src/config.rs index 8ef9b87073..4911f27b96 100644 --- a/codex-rs/exec-server-cli/src/config.rs +++ b/codex-rs/exec-server-cli/src/config.rs @@ -1,7 +1,7 @@ //! Loads executor startup configuration and remote authentication without changing CLI policy. use codex_app_server::in_process::EmbeddedNetworkPolicy; -use codex_cloud_config::chatgpt_config_bundle_loader_for_storage; +use codex_cloud_config::chatgpt_config_bundle_loader; use codex_config::LoaderOverrides; use codex_core::config::ConfigBuilder; use codex_core::config::ConfigLoadOptions; @@ -11,10 +11,20 @@ use codex_core::config::load_config_toml_with_layer_stack; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_login::read_codex_access_token_from_env; +use codex_protocol::auth::AuthMode; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_cli::CliConfigOverrides; use std::sync::Arc; +use crate::exec_server_auth; + +const REMOTE_AUTH_MODES: &[AuthMode] = &[ + AuthMode::ApiKey, + AuthMode::Chatgpt, + AuthMode::ChatgptAuthTokens, + AuthMode::PersonalAccessToken, +]; + pub(super) async fn load_exec_server_remote_auth_provider( config: &codex_core::config::Config, base_url: &str, @@ -53,11 +63,16 @@ pub(super) async fn load_exec_server_remote_auth_provider( validate_api_key_remote_host(base_url)?; } - if auth.is_chatgpt_auth() || auth_manager.is_workload_identity_selected() { + if auth_manager.is_workload_identity_selected() { Ok(codex_model_provider::auth_provider_from_auth_manager( auth_manager, &auth, )) + } else if auth.is_chatgpt_auth() { + Ok(exec_server_auth::chatgpt_auth_provider( + auth_manager, + &auth, + )?) } else { Ok(codex_model_provider::auth_provider_from_auth(&auth)) } @@ -129,12 +144,25 @@ pub(super) async fn load_exec_server_config( }, ) .await?; - let bootstrap_auth_config = bootstrap_auth_config(&codex_home, &bootstrap_config)?; - let cloud_config_bundle = chatgpt_config_bundle_loader_for_storage( - network_policy.bind_bootstrap_auth(bootstrap_auth_config), + let bootstrap_auth_config = network_policy + .bind_bootstrap_auth(bootstrap_auth_config(&codex_home, &bootstrap_config)?); + let auth_manager = AuthManager::shared_from_auth_config_with_auth_modes( + bootstrap_auth_config.clone(), /*enable_codex_api_key_env*/ true, + REMOTE_AUTH_MODES.to_vec(), ) .await?; + let cloud_config_bundle = chatgpt_config_bundle_loader( + auth_manager, + bootstrap_auth_config + .chatgpt_base_url + .unwrap_or_else(|| "https://chatgpt.com/backend-api/".to_string()), + bootstrap_auth_config.codex_home, + bootstrap_auth_config + .auth_route_config + .http_client_factory() + .clone(), + ); builder = builder.cloud_config_bundle(cloud_config_bundle); } let mut config = builder.build().await?; @@ -146,8 +174,12 @@ async fn load_exec_server_remote_auth( config: &codex_core::config::Config, missing_auth_error: &'static str, ) -> anyhow::Result<(Arc, codex_login::CodexAuth)> { - let auth_manager = - AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ true).await?; + let auth_manager = AuthManager::shared_from_config_with_auth_modes( + config, + /*enable_codex_api_key_env*/ true, + REMOTE_AUTH_MODES.to_vec(), + ) + .await?; let auth = match auth_manager.auth().await { Some(auth) => auth, diff --git a/codex-rs/exec-server-cli/src/exec_server_auth.rs b/codex-rs/exec-server-cli/src/exec_server_auth.rs index ba74f98996..75beb1d6d5 100644 --- a/codex-rs/exec-server-cli/src/exec_server_auth.rs +++ b/codex-rs/exec-server-cli/src/exec_server_auth.rs @@ -1,10 +1,17 @@ -//! CLI-only glue between Codex authentication and generic AWS signing. +//! Request-time authentication for remote executors, including AWS signing. //! -//! Keeping this adapter here leaves `codex-api` and `codex-aws-auth` independent. +//! ChatGPT credentials may rotate while an executor is idle. Reload them for +//! registry requests while retaining the identity that registered the executor. +//! Failed credential reads block headers until revalidation, without counting as logout. use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; +use std::time::Duration; use codex_api::AuthError; +use codex_api::AuthHeadersFuture; use codex_api::AuthProvider; use codex_api::SharedAuthProvider; use codex_aws_auth::AwsAuthConfig; @@ -14,7 +21,221 @@ use codex_aws_auth::AwsRequestToSign; use codex_http_client::Request; use codex_http_client::RequestBody; use codex_http_client::RequestCompression; +use codex_login::AuthManager; +use codex_login::CodexAuth; use http::HeaderMap; +use tokio::sync::Semaphore; + +/// Follows stored credential rotation without allowing an executor's owner to change. +pub(super) fn chatgpt_auth_provider( + auth_manager: Arc, + expected_auth: &CodexAuth, +) -> Result { + auth_manager + .restrict_auth_modes(vec![expected_auth.auth_mode()]) + .map_err(|_| AuthError::Build("executor credential policy is unavailable".to_string()))?; + let account_id = expected_auth.get_account_id(); + let user_id = expected_auth.get_chatgpt_user_id(); + let identity_complete = account_id.as_ref().is_some_and(|id| !id.trim().is_empty()) + && user_id.as_ref().is_some_and(|id| !id.trim().is_empty()); + let unidentified_headers = if identity_complete { + None + } else { + Some(codex_model_provider::auth_provider_from_auth(expected_auth).to_auth_headers()?) + }; + Ok(Arc::new(ReloadingChatgptAuthProvider { + managed: codex_model_provider::auth_provider_from_auth_manager( + Arc::clone(&auth_manager), + expected_auth, + ), + auth_manager, + account_id, + user_id, + workspace_account: expected_auth.is_workspace_account(), + fedramp_account: expected_auth.is_fedramp_account(), + unidentified_headers, + resolution_lock: Arc::new(Semaphore::new(/*permits*/ 1)), + uncertain_refresh: Arc::new(Mutex::new(None)), + headers_valid: Arc::new(AtomicBool::new(true)), + })) +} + +#[derive(Clone)] +struct ReloadingChatgptAuthProvider { + managed: SharedAuthProvider, + auth_manager: Arc, + account_id: Option, + user_id: Option, + workspace_account: bool, + fedramp_account: bool, + unidentified_headers: Option, + resolution_lock: Arc, + uncertain_refresh: Arc>>, + headers_valid: Arc, +} + +impl ReloadingChatgptAuthProvider { + fn checked_headers(&self, auth: Option) -> Result { + let auth = auth + .filter(|auth| { + auth.is_chatgpt_auth() + && auth.get_account_id() == self.account_id + && auth.get_chatgpt_user_id() == self.user_id + && auth.is_workspace_account() == self.workspace_account + && auth.is_fedramp_account() == self.fedramp_account + }) + .ok_or_else(|| AuthError::Build("executor authentication changed".to_string()))?; + let token = auth.get_token().map_err(|_| { + AuthError::Build("executor credential has no usable bearer".to_string()) + })?; + if token.trim().is_empty() { + return Err(AuthError::Build( + "executor credential has an empty bearer".to_string(), + )); + } + let uncertain = self.uncertain_refresh.lock().map_err(|_| { + AuthError::Transient("executor authentication is unavailable".to_string()) + })?; + if uncertain + .as_ref() + .is_some_and(|previous| auth.get_token_data().as_ref().ok() == Some(previous)) + { + return Err(AuthError::Transient( + "executor refresh outcome is unknown; replace or renew stored credentials" + .to_string(), + )); + } + let headers = codex_model_provider::auth_provider_from_auth(&auth).to_auth_headers()?; + if self + .unidentified_headers + .as_ref() + .is_some_and(|expected| expected != &headers) + { + return Err(AuthError::Build( + "executor credential rotation requires a complete account and user identity" + .to_string(), + )); + } + Ok(headers) + } +} + +impl AuthProvider for ReloadingChatgptAuthProvider { + fn add_auth_headers(&self, headers: &mut HeaderMap) -> Result<(), AuthError> { + if !self.headers_valid.load(Ordering::Acquire) { + return Err(AuthError::Transient( + "executor credentials have not been successfully revalidated".to_string(), + )); + } + headers.extend(self.checked_headers(self.auth_manager.auth_cached())?); + Ok(()) + } + + fn apply_auth_headers(&self, mut headers: HeaderMap) -> AuthHeadersFuture<'_> { + Box::pin(async move { + headers.extend(self.resolve(/*rejected_headers*/ None).await?); + Ok(headers) + }) + } + + fn refresh_after_unauthorized( + &self, + rejected_headers: HeaderMap, + ) -> codex_api::AuthRefreshFuture<'_> { + Box::pin(async move { + self.resolve(Some(rejected_headers.clone())) + .await + .is_ok_and(|headers| { + headers + .iter() + .any(|(name, value)| rejected_headers.get(name) != Some(value)) + }) + }) + } +} + +impl ReloadingChatgptAuthProvider { + async fn resolve(&self, rejected_headers: Option) -> Result { + let guard = Arc::clone(&self.resolution_lock) + .acquire_owned() + .await + .map_err(|_| { + AuthError::Transient("executor authentication is unavailable".to_string()) + })?; + let provider = self.clone(); + // A failed read leaves AuthManager's last known owner intact. Keep those + // cached credentials unavailable to synchronous callers until revalidated. + provider.headers_valid.store(false, Ordering::Release); + // Outlive a cancelled handshake, but never hold the permit indefinitely. + // A timed-out OAuth exchange may already have rotated the token: quarantine + // that generation until another login/refresh supplies different credentials. + tokio::spawn(async move { + let _guard = guard; + let mut attempted_refresh = None; + let result = tokio::time::timeout(Duration::from_secs(30), async { + provider + .auth_manager + .reload_with_cached_auth() + .await + .map_err(|_| { + AuthError::Transient( + "executor credential storage or hydration is unavailable".to_string(), + ) + })?; + let auth = provider.auth_manager.auth_cached(); + let current = provider.checked_headers(auth.clone())?; + if provider.unidentified_headers.is_some() { + return Ok(current); + } + attempted_refresh = auth.as_ref().and_then(|auth| auth.get_token_data().ok()); + if let Some(rejected) = rejected_headers { + if !current + .iter() + .any(|(name, value)| rejected.get(name) != Some(value)) + { + provider.managed.refresh_after_unauthorized(rejected).await; + } + } else { + provider.auth_manager.auth().await; + } + // A later storage timeout is not an interrupted OAuth exchange. + attempted_refresh = None; + // auth() deliberately falls back to a snapshot on refresh errors. + // Re-read the selected source before dispatch: logout, replacement, + // or failed hydration must never authorize that fallback snapshot. + provider + .auth_manager + .reload_with_cached_auth() + .await + .map_err(|_| { + AuthError::Transient( + "executor credential storage or hydration is unavailable".to_string(), + ) + })?; + provider.checked_headers(provider.auth_manager.auth_cached()) + }) + .await; + let result = match result { + Ok(result) => result, + Err(_) => { + if let Some(attempted) = attempted_refresh + && let Ok(mut uncertain) = provider.uncertain_refresh.lock() + { + *uncertain = Some(attempted); + } + Err(AuthError::Transient( + "executor credential resolution timed out; retry after repairing credentials" + .to_string(), + )) + } + }; + provider.headers_valid.store(result.is_ok(), Ordering::Release); + result + }) + .await + .map_err(|_| AuthError::Transient("executor authentication task failed".to_string()))? + } +} /// Creates a SigV4 provider, preferring an explicit profile over the default credential chain. pub(super) async fn aws_sigv4_auth_provider( diff --git a/codex-rs/exec-server-cli/src/exec_server_auth_tests.rs b/codex-rs/exec-server-cli/src/exec_server_auth_tests.rs index 00985ce28f..6a801fd62f 100644 --- a/codex-rs/exec-server-cli/src/exec_server_auth_tests.rs +++ b/codex-rs/exec-server-cli/src/exec_server_auth_tests.rs @@ -1,14 +1,564 @@ #![cfg(test)] +use std::path::Path; use std::time::Duration; use codex_aws_auth::AwsAccessKeys; +use codex_login::AuthCredentialsStoreMode; +use codex_login::AuthKeyringBackendKind; use http::HeaderValue; use http::Method; use pretty_assertions::assert_eq; +use tempfile::TempDir; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::header; +use wiremock::matchers::method; +use wiremock::matchers::path; use super::*; +fn write_chatgpt_auth(codex_home: &Path, token: &str, account: &str) { + write_chatgpt_auth_with_claims( + codex_home, + token, + Some(account), + serde_json::json!({"chatgpt_user_id": "user", "chatgpt_plan_type": "pro"}), + ); +} + +fn write_chatgpt_auth_with_claims( + codex_home: &Path, + token: &str, + account: Option<&str>, + claims: serde_json::Value, +) { + let id_token = jsonwebtoken::encode( + &jsonwebtoken::Header::default(), + &serde_json::json!({"https://api.openai.com/auth": claims}), + &jsonwebtoken::EncodingKey::from_secret(b"executor-auth-test"), + ) + .unwrap(); + std::fs::write( + codex_home.join("auth.json"), + serde_json::to_vec(&serde_json::json!({ + "auth_mode": "chatgpt", + "tokens": { + "id_token": id_token, + "access_token": token, + "refresh_token": "test-refresh-token", + "account_id": account, + }, + "last_refresh": chrono::Utc::now(), + })) + .unwrap(), + ) + .unwrap(); +} + +async fn stored_chatgpt_provider(codex_home: &Path) -> SharedAuthProvider { + let auth_manager = Arc::new( + AuthManager::new( + codex_home.to_path_buf(), + /*enable_codex_api_key_env*/ false, + AuthCredentialsStoreMode::File, + /*forced_chatgpt_workspace_id*/ None, + /*chatgpt_base_url*/ None, + AuthKeyringBackendKind::default(), + codex_login::test_support::transport_default_auth_route_config(), + ) + .await, + ); + let initial_auth = auth_manager.auth_cached().expect("load initial auth"); + chatgpt_auth_provider(auth_manager, &initial_auth).unwrap() +} + +#[tokio::test] +async fn chatgpt_executor_follows_disk_rotation_without_restart() { + let home = TempDir::new().unwrap(); + write_chatgpt_auth(home.path(), "initial-token", "account"); + let provider = stored_chatgpt_provider(home.path()).await; + for token in ["initial-token", "rotated-token", "rotated-again"] { + write_chatgpt_auth(home.path(), token, "account"); + let headers = provider.apply_auth_headers(HeaderMap::new()).await.unwrap(); + let mut expected = HeaderMap::new(); + expected.insert( + http::header::AUTHORIZATION, + HeaderValue::from_str(&format!("Bearer {token}")).unwrap(), + ); + expected.insert("ChatGPT-Account-ID", HeaderValue::from_static("account")); + assert_eq!(headers, expected); + } +} + +#[tokio::test] +async fn chatgpt_executor_rejects_account_switch_and_recovers_original_identity() { + let home = TempDir::new().unwrap(); + write_chatgpt_auth(home.path(), "initial-token", "account"); + let provider = stored_chatgpt_provider(home.path()).await; + + let rejected = provider.to_auth_headers().unwrap(); + write_chatgpt_auth(home.path(), "other-token", "other-account"); + assert!(!provider.refresh_after_unauthorized(rejected.clone()).await); + assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err()); + assert!(provider.to_auth_headers().is_err()); + + write_chatgpt_auth(home.path(), "restored-token", "account"); + assert!(provider.refresh_after_unauthorized(rejected).await); + assert_eq!( + provider.apply_auth_headers(HeaderMap::new()).await.unwrap()[http::header::AUTHORIZATION], + "Bearer restored-token" + ); +} + +#[tokio::test] +async fn chatgpt_executor_does_not_reuse_credentials_after_logout() { + let home = TempDir::new().unwrap(); + write_chatgpt_auth(home.path(), "initial-token", "account"); + let provider = stored_chatgpt_provider(home.path()).await; + std::fs::remove_file(home.path().join("auth.json")).unwrap(); + assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err()); + assert!(provider.to_auth_headers().is_err()); +} + +#[tokio::test] +async fn executor_rejects_incomplete_or_empty_bearer() { + for missing_refresh in [true, false] { + let home = TempDir::new().unwrap(); + write_chatgpt_auth(home.path(), "initial", "account"); + let provider = stored_chatgpt_provider(home.path()).await; + let path = home.path().join("auth.json"); + let mut doc: serde_json::Value = + serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + if missing_refresh { + doc.as_object_mut().unwrap().remove("last_refresh"); + } else { + doc["tokens"]["access_token"] = serde_json::json!(""); + } + std::fs::write(path, serde_json::to_vec(&doc).unwrap()).unwrap(); + assert!(matches!( + provider.apply_auth_headers(HeaderMap::new()).await, + Err(AuthError::Build(_)) + )); + } +} + +#[tokio::test] +async fn chatgpt_executor_rejects_rotation_with_incomplete_identity() { + for (account, user) in [ + (Some("account"), None), + (None, Some("user")), + (Some("account"), Some("")), + (Some(" "), Some("user")), + ] { + let home = TempDir::new().unwrap(); + let claims = serde_json::json!({"chatgpt_user_id": user}); + write_chatgpt_auth_with_claims(home.path(), "initial", account, claims.clone()); + let provider = stored_chatgpt_provider(home.path()).await; + assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_ok()); + assert!( + !provider + .refresh_after_unauthorized(provider.to_auth_headers().unwrap()) + .await + ); + write_chatgpt_auth_with_claims(home.path(), "replacement", account, claims); + assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err()); + assert!(provider.to_auth_headers().is_err()); + } +} + +#[tokio::test] +async fn chatgpt_executor_rejects_changed_user_and_workspace_policy() { + for claims in [ + serde_json::json!({"chatgpt_user_id": "other-user", "chatgpt_plan_type": "pro"}), + serde_json::json!({"chatgpt_user_id": "user", "chatgpt_plan_type": "enterprise"}), + serde_json::json!({"chatgpt_user_id": "user", "chatgpt_plan_type": "pro", "chatgpt_account_is_fedramp": true}), + ] { + let home = TempDir::new().unwrap(); + write_chatgpt_auth(home.path(), "initial", "account"); + let provider = stored_chatgpt_provider(home.path()).await; + write_chatgpt_auth_with_claims(home.path(), "replacement", Some("account"), claims); + assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err()); + assert!(provider.to_auth_headers().is_err()); + } +} + +#[tokio::test] +async fn chatgpt_executor_rejects_same_owner_agent_identity() { + let home = TempDir::new().unwrap(); + write_chatgpt_auth_with_claims( + home.path(), + "initial", + Some("test-account-id"), + serde_json::json!({"chatgpt_user_id": "test-user-id", "chatgpt_plan_type": "pro"}), + ); + let provider = stored_chatgpt_provider(home.path()).await; + let agent_manager = codex_login::test_support::auth_manager_with_agent_identity() + .await + .unwrap(); + let Some(CodexAuth::AgentIdentity(agent)) = agent_manager.auth_cached() else { + panic!("expected agent identity fixture"); + }; + std::fs::write( + home.path().join("auth.json"), + serde_json::to_vec(&serde_json::json!({ + "auth_mode": codex_protocol::auth::AuthMode::AgentIdentity, + "agent_identity": agent.record(), + })) + .unwrap(), + ) + .unwrap(); + // Ensure the fixture is a valid, loadable credential for the same owner. + assert!(matches!( + CodexAuth::from_auth_storage( + home.path(), + AuthCredentialsStoreMode::File, + /*chatgpt_base_url*/ None, + AuthKeyringBackendKind::default(), + &codex_login::test_support::transport_default_auth_route_config(), + ) + .await + .unwrap(), + Some(CodexAuth::AgentIdentity(_)) + )); + assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err()); + assert!(provider.to_auth_headers().is_err()); +} + +#[tokio::test] +async fn executor_concurrent_rotation_checks_identity_before_refresh() { + const CHILD_MODE: &str = "CODEX_EXECUTOR_AUTH_TEST_MODE"; + let Ok(mode) = std::env::var(CHILD_MODE) else { + // Run in subprocesses so endpoint overrides cannot affect parallel tests. + let server = MockServer::start().await; + for token in [ + "at-initial", + "at-rotated", + "at-retry-original", + "at-retry-new", + ] { + Mock::given(method("GET")) + .and(path("/v1/user-auth-credential/whoami")) + .and(header("authorization", format!("Bearer {token}"))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "chatgpt_user_id": "user", + "chatgpt_account_id": "account", + "chatgpt_plan_type": "pro", + "chatgpt_account_is_fedramp": false, + }))) + .with_priority(2) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + } + Mock::given(method("GET")) + .and(path("/v1/user-auth-credential/whoami")) + .and(header("authorization", "Bearer at-retry-new")) + .respond_with(ResponseTemplate::new(503)) + .with_priority(1) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/oauth/token")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "refreshed", + "refresh_token": "refreshed-refresh-token", + }))) + .with_priority(1) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + Mock::given(wiremock::matchers::any()) + .respond_with(ResponseTemplate::new(503)) + .with_priority(10) + .expect(0) + .mount(&server) + .await; + for mode in ["pat", "chatgpt", "agent", "kind-switch", "pat-outage"] { + let mut child = tokio::process::Command::new(std::env::current_exe().unwrap()); + child + .args([ + "--exact", + "exec_server_auth::tests::executor_concurrent_rotation_checks_identity_before_refresh", + "--nocapture", + ]) + .env(CHILD_MODE, mode) + .env_remove("CODEX_ACCESS_TOKEN") + .env("NO_PROXY", "127.0.0.1,localhost") + .env("no_proxy", "127.0.0.1,localhost") + .env("CODEX_AUTHAPI_BASE_URL", server.uri()) + .env("CODEX_AGENT_IDENTITY_AUTHAPI_BASE_URL", server.uri()) + .env("CODEX_REFRESH_TOKEN_URL_OVERRIDE", format!("{}/oauth/token", server.uri())) + .kill_on_drop(true); + let output = tokio::time::timeout(Duration::from_secs(30), child.output()) + .await + .unwrap() + .unwrap(); + assert!(output.status.success(), "{output:?}"); + assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed")); + } + server.verify().await; + return; + }; + + let home = TempDir::new().unwrap(); + let auth_path = home.path().join("auth.json"); + if mode == "pat" || mode == "pat-outage" { + let token = if mode == "pat" { + "at-initial" + } else { + "at-retry-original" + }; + std::fs::write( + &auth_path, + serde_json::to_vec(&serde_json::json!({"personal_access_token":token})).unwrap(), + ) + .unwrap(); + } else { + write_chatgpt_auth(home.path(), "initial", "account"); + } + let provider = stored_chatgpt_provider(home.path()).await; + if mode == "pat-outage" { + std::fs::write(&auth_path, r#"{"personal_access_token":"at-retry-new"}"#).unwrap(); + assert!(matches!( + provider.apply_auth_headers(HeaderMap::new()).await, + Err(AuthError::Transient(_)) + )); + assert!(provider.to_auth_headers().is_err()); + let headers = provider.apply_auth_headers(HeaderMap::new()).await.unwrap(); + assert_eq!(headers[http::header::AUTHORIZATION], "Bearer at-retry-new"); + assert_eq!(provider.to_auth_headers().unwrap(), headers); + return; + } + if mode == "agent" || mode == "kind-switch" { + if mode == "agent" { + let manager = codex_login::test_support::auth_manager_with_agent_identity() + .await + .unwrap(); + let Some(CodexAuth::AgentIdentity(agent)) = manager.auth_cached() else { + panic!("agent fixture"); + }; + let mut record = agent.record().clone(); + record.task_id = None; + std::fs::write(&auth_path, serde_json::to_vec(&serde_json::json!({"auth_mode":codex_protocol::auth::AuthMode::AgentIdentity, "agent_identity":record})).unwrap()).unwrap(); + } else { + std::fs::write(&auth_path, r#"{"personal_access_token":"at-initial"}"#).unwrap(); + } + assert!(matches!( + provider.apply_auth_headers(HeaderMap::new()).await, + Err(AuthError::Build(_)) + )); + return; + } + let expected_bearer = if mode == "pat" { + std::fs::write(&auth_path, r#"{"personal_access_token":"at-rotated"}"#).unwrap(); + "Bearer at-rotated" + } else { + let expired = jsonwebtoken::encode( + &jsonwebtoken::Header::default(), + &serde_json::json!({"exp": chrono::Utc::now().timestamp() - 60}), + &jsonwebtoken::EncodingKey::from_secret(b"executor-auth-test"), + ) + .unwrap(); + for (account, user) in [ + ("other-account", "user"), + ("account", "other-user"), + ("account", "user"), + ] { + write_chatgpt_auth_with_claims( + home.path(), + &expired, + Some(account), + serde_json::json!({"chatgpt_user_id": user, "chatgpt_plan_type": "pro"}), + ); + let original = std::fs::read(&auth_path).unwrap(); + if account != "account" || user != "user" { + assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err()); + assert_eq!(std::fs::read(&auth_path).unwrap(), original); + } + } + "Bearer refreshed" + }; + let barrier = Arc::new(tokio::sync::Barrier::new(32)); + let results = futures::future::join_all((0..32).map(|_| { + let provider = Arc::clone(&provider); + let barrier = Arc::clone(&barrier); + async move { + barrier.wait().await; + provider.apply_auth_headers(HeaderMap::new()).await + } + })) + .await; + for result in results { + assert_eq!( + result.unwrap()[http::header::AUTHORIZATION], + expected_bearer + ); + } +} + +#[tokio::test] +async fn executor_refresh_survives_cancelled_caller() { + use std::process::Stdio; + use tokio::io::AsyncBufReadExt; + use tokio::io::AsyncReadExt; + use tokio::io::AsyncWriteExt; + + const CHILD_MODE: &str = "CODEX_EXECUTOR_CANCEL_TEST_MODE"; + let Ok(mode) = std::env::var(CHILD_MODE) else { + // Isolate endpoint overrides; release the response only after the caller + // is cancelled, with no sleep or race against the HTTP response. + for mode in ["proactive", "recovery", "logout", "replacement", "timeout"] { + tokio::time::timeout(Duration::from_secs(30), async { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}/token", listener.local_addr().unwrap()); + let mut child = tokio::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "exec_server_auth::tests::executor_refresh_survives_cancelled_caller", "--nocapture"]) + .env(CHILD_MODE, mode) + .env_remove("CODEX_ACCESS_TOKEN") + .env("CODEX_REFRESH_TOKEN_URL_OVERRIDE", endpoint) + .env("NO_PROXY", "127.0.0.1,localhost") + .env("no_proxy", "127.0.0.1,localhost") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn().unwrap(); + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = tokio::io::BufReader::new(&mut stream); + let mut content_length = 0; + loop { + let mut line = String::new(); + assert!(request.read_line(&mut line).await.unwrap() > 0); + if line == "\r\n" { break; } + if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") { + content_length = length.trim().parse().unwrap(); + } + } + request.read_exact(&mut vec![0; content_length]).await.unwrap(); + drop(request); + child.stdin.take().unwrap().write_all(b"cancel\n").await.unwrap(); + let mut stdout = tokio::io::BufReader::new(child.stdout.take().unwrap()); + loop { + let mut line = String::new(); + assert!(stdout.read_line(&mut line).await.unwrap() > 0); + if line.contains("CALLER_CANCELLED") { break; } + } + let body = r#"{"access_token":"rotated-access","refresh_token":"rotated-refresh"}"#; + if mode != "timeout" { + stream.write_all(format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len() + ).as_bytes()).await.unwrap(); + } + // Keep the unanswered timeout connection open until the child finishes. + let mut remaining = String::new(); + stdout.read_to_string(&mut remaining).await.unwrap(); + let output = child.wait_with_output().await.unwrap(); + assert!(output.status.success(), "{remaining}\n{output:?}"); + assert!(remaining.contains("1 passed"), "{remaining}"); + }).await.expect("cancelled refresh must finish persisting"); + } + return; + }; + + let home = TempDir::new().unwrap(); + let token = if mode != "recovery" { + jsonwebtoken::encode( + &jsonwebtoken::Header::default(), + &serde_json::json!({"exp": chrono::Utc::now().timestamp() - 60}), + &jsonwebtoken::EncodingKey::from_secret(b"executor-auth-test"), + ) + .unwrap() + } else { + "initial-access".to_string() + }; + write_chatgpt_auth(home.path(), &token, "account"); + let provider = stored_chatgpt_provider(home.path()).await; + let caller = Arc::clone(&provider); + let rejected = provider.to_auth_headers().unwrap(); + let refresh_mode = mode.clone(); + let task = tokio::spawn(async move { + if refresh_mode != "recovery" { + caller.apply_auth_headers(HeaderMap::new()).await.unwrap(); + } else { + assert!(caller.refresh_after_unauthorized(rejected).await); + } + }); + let mut instruction = String::new(); + tokio::io::BufReader::new(tokio::io::stdin()) + .read_line(&mut instruction) + .await + .unwrap(); + assert_eq!(instruction, "cancel\n"); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + if mode == "logout" { + std::fs::remove_file(home.path().join("auth.json")).unwrap(); + } else if mode == "replacement" { + write_chatgpt_auth(home.path(), "fresh-external-access", "account"); + let path = home.path().join("auth.json"); + let mut stored: serde_json::Value = + serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + stored["tokens"]["refresh_token"] = serde_json::json!("externally-rotated-refresh"); + std::fs::write(path, serde_json::to_vec(&stored).unwrap()).unwrap(); + } + println!("CALLER_CANCELLED"); + if mode == "timeout" { + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(31)).await; + tokio::time::resume(); + assert!(matches!( + provider.apply_auth_headers(HeaderMap::new()).await, + Err(AuthError::Transient(_)) + )); + // Repair is usable without releasing the old response or restarting the executor. + write_chatgpt_auth(home.path(), "fresh-external-access", "account"); + } + if mode == "logout" { + assert!(matches!( + provider.apply_auth_headers(HeaderMap::new()).await, + Err(AuthError::Build(_)) + )); + assert!(!home.path().join("auth.json").exists()); + return; + } + let repaired = mode == "replacement" || mode == "timeout"; + // This waits on the same permit, proving the detached operation kept it + // until both persistence and cache update were complete. + let headers = provider.apply_auth_headers(HeaderMap::new()).await.unwrap(); + assert_eq!( + headers[http::header::AUTHORIZATION], + if repaired { + "Bearer fresh-external-access" + } else { + "Bearer rotated-access" + } + ); + let stored: serde_json::Value = + serde_json::from_slice(&std::fs::read(home.path().join("auth.json")).unwrap()).unwrap(); + assert_eq!( + stored["tokens"]["access_token"], + if repaired { + "fresh-external-access" + } else { + "rotated-access" + } + ); + assert_eq!( + stored["tokens"]["refresh_token"], + match mode.as_str() { + "replacement" => "externally-rotated-refresh", + "timeout" => "test-refresh-token", + _ => "rotated-refresh", + } + ); +} + async fn test_provider() -> AwsSigV4AuthProvider { let context = AwsAuthContext::load_with_access_keys( AwsAuthConfig { diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index 53743bb518..82c3f7ca83 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -2051,7 +2051,6 @@ mod tests { use crate::FileSystemSandboxContext; use crate::ProcessId; use crate::client_api::DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT; - #[cfg(not(windows))] use crate::client_api::DEFAULT_REMOTE_EXEC_SERVER_INITIALIZE_TIMEOUT; use crate::client_api::ExecServerTransportParams; use crate::client_api::RemoteExecServerConnectArgs; @@ -2393,7 +2392,7 @@ mod tests { cwd: None, }, client_name: "stdio-test-client".to_string(), - initialize_timeout: Duration::from_secs(1), + initialize_timeout: DEFAULT_REMOTE_EXEC_SERVER_INITIALIZE_TIMEOUT, resume_session_id: None, }) .await diff --git a/codex-rs/exec-server/src/remote.rs b/codex-rs/exec-server/src/remote.rs index a43f5ac567..118cc81aa5 100644 --- a/codex-rs/exec-server/src/remote.rs +++ b/codex-rs/exec-server/src/remote.rs @@ -307,11 +307,7 @@ impl EnvironmentRegistryClient { .auth_provider .apply_auth_headers(HeaderMap::new()) .await - .map_err(|error| { - ExecServerError::EnvironmentRegistryAuth(format!( - "failed to resolve environment registry authentication: {error}" - )) - })?; + .map_err(auth::registry_auth_error)?; let executor_credential = if self.auth_mode == RegistryAuthMode::ExecutorCredential { // Scoped mode sends exactly one bearer and never accompanying account credentials. let token = (headers.len() == 1) diff --git a/codex-rs/exec-server/src/remote/auth.rs b/codex-rs/exec-server/src/remote/auth.rs index 1f0bbe482c..58bbd006af 100644 --- a/codex-rs/exec-server/src/remote/auth.rs +++ b/codex-rs/exec-server/src/remote/auth.rs @@ -11,6 +11,15 @@ use super::RegistryAuthMode; use super::current_trace_context_headers; use crate::ExecServerError; +pub(super) fn registry_auth_error(error: codex_api::AuthError) -> ExecServerError { + match error { + codex_api::AuthError::Transient(_) => ExecServerError::RequestNotSent( + "environment registry credential resolution is temporarily unavailable".to_string(), + ), + error => super::direct::direct_auth_error(error), + } +} + impl EnvironmentRegistryClient { pub(super) async fn send_authenticated( &self, @@ -32,7 +41,7 @@ impl EnvironmentRegistryClient { self.auth_provider .apply_auth(request.clone()) .await - .map_err(super::direct::direct_auth_error)?, + .map_err(registry_auth_error)?, None, ) }; diff --git a/codex-rs/exec-server/src/remote/registration_retry.rs b/codex-rs/exec-server/src/remote/registration_retry.rs index 92b09f076b..134d8b1821 100644 --- a/codex-rs/exec-server/src/remote/registration_retry.rs +++ b/codex-rs/exec-server/src/remote/registration_retry.rs @@ -27,6 +27,20 @@ impl EnvironmentRegistryClient { .await { Ok(response) => return Ok(response), + // Credential resolution failed before any registration was sent. + // Keep the executor alive without replaying an ambiguous write. + Err(ExecServerError::RequestNotSent(_)) => { + let delay = registry_recovery_retry_delay(&retry_key, attempt); + attempt = attempt.saturating_add(1); + warn!( + noise_event = "registration", + noise_outcome = "retry", + retry_attempt = attempt, + retry_delay_ms = delay.as_millis() as u64, + "Noise executor retrying unavailable credential resolution" + ); + sleep(delay).await; + } Err(ExecServerError::EnvironmentRegistryHttp { status, code, .. }) if matches!( (status, code.as_deref()), diff --git a/codex-rs/exec-server/src/remote/registration_retry_tests.rs b/codex-rs/exec-server/src/remote/registration_retry_tests.rs index 81f642f077..6db2608f1c 100644 --- a/codex-rs/exec-server/src/remote/registration_retry_tests.rs +++ b/codex-rs/exec-server/src/remote/registration_retry_tests.rs @@ -40,6 +40,7 @@ const SUCCESS_BODY: &str = r#"{"environment_id":"registration-retry-test","url": #[derive(Debug, Default)] struct RegistrationAuthProvider { calls: AtomicUsize, + fail_first: bool, } impl AuthProvider for RegistrationAuthProvider { @@ -48,11 +49,46 @@ impl AuthProvider for RegistrationAuthProvider { } fn apply_auth_headers(&self, headers: HeaderMap) -> codex_api::AuthHeadersFuture<'_> { - self.calls.fetch_add(1, Ordering::Relaxed); + if self.calls.fetch_add(1, Ordering::Relaxed) == 0 && self.fail_first { + return Box::pin(async { + Err(codex_api::AuthError::Transient( + "metadata unavailable".to_string(), + )) + }); + } Box::pin(async move { Ok(headers) }) } } +#[tokio::test] +async fn transient_credential_failure_retries_before_sending_registration() -> anyhow::Result<()> { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::any()) + .respond_with( + wiremock::ResponseTemplate::new(200).set_body_raw(SUCCESS_BODY, "application/json"), + ) + .expect(1) + .mount(&server) + .await; + let provider = Arc::new(RegistrationAuthProvider { + fail_first: true, + ..Default::default() + }); + let client = EnvironmentRegistryClient::new(server.uri(), provider.clone())?; + let response = timeout( + Duration::from_secs(10), + client.register_environment_with_retry( + ENVIRONMENT_ID, + &NoiseChannelIdentity::generate()?.public_key(), + ), + ) + .await??; + assert_eq!(response.environment_id, ENVIRONMENT_ID); + assert_eq!(provider.calls.load(Ordering::Relaxed), 2); + server.verify().await; + Ok(()) +} + struct RetryObserved(Arc); impl tracing_subscriber::Layer for RetryObserved { @@ -198,7 +234,7 @@ async fn registration_requires_a_confirmed_pre_write_failure_before_replay( assert_eq!(message, "registration unavailable"); } } - _ => anyhow::bail!("unexpected registration failure kind"), + error => anyhow::bail!("unexpected registration failure kind: {error:?}"), } Ok(()) } diff --git a/codex-rs/login/Cargo.toml b/codex-rs/login/Cargo.toml index 3eba4b495b..d9ade5cf5c 100644 --- a/codex-rs/login/Cargo.toml +++ b/codex-rs/login/Cargo.toml @@ -31,6 +31,7 @@ rand = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sha2 = { workspace = true } +tempfile = { workspace = true } thiserror = { workspace = true } tiny_http = { workspace = true } tokio = { workspace = true, features = [ @@ -45,6 +46,9 @@ url = { workspace = true } urlencoding = { workspace = true } webbrowser = { workspace = true } +[target.'cfg(windows)'.dependencies] +windows-sys = { workspace = true, features = ["Win32_Foundation", "Win32_Security", "Win32_Security_Authorization", "Win32_Storage_FileSystem"] } + [dev-dependencies] codex-utils-cargo-bin = { workspace = true } opentelemetry_sdk = { workspace = true, features = ["testing"] } @@ -54,7 +58,6 @@ keyring = { workspace = true } pretty_assertions = { workspace = true } regex-lite = { workspace = true } serial_test = { workspace = true } -tempfile = { workspace = true } tracing-subscriber = { workspace = true } wiremock = { workspace = true } diff --git a/codex-rs/login/src/auth/auth_tests.rs b/codex-rs/login/src/auth/auth_tests.rs index 48cbeb984a..d1c9206275 100644 --- a/codex-rs/login/src/auth/auth_tests.rs +++ b/codex-rs/login/src/auth/auth_tests.rs @@ -113,12 +113,10 @@ async fn refresh_without_id_token() -> anyhow::Result<()> { assert_eq!(tokens.refresh_token, "new-refresh-token"); assert_eq!(tokens.oauth_provenance, None); - // Preserve ordinary same-login refresh behavior when the stored tokens changed. - let mut expected = previous.clone(); - expected.access_token = "stale-access-token".into(); - expected.refresh_token = "stale-refresh-token".into(); - expected.oauth_provenance = Some(provenance.clone()); - assert_eq!( + // Reject this stale refresh response: newer tokens have already been saved + // for the same account. + let before = storage.load()?; + assert!( super::persist_tokens( &storage, &previous, @@ -126,10 +124,10 @@ async fn refresh_without_id_token() -> anyhow::Result<()> { /*id_token*/ None, Some("stale-access-token".into()), Some("stale-refresh-token".into()), - )? - .tokens, - Some(expected) + ) + .is_err() ); + assert_eq!(storage.load()?, before); Ok(()) } @@ -436,6 +434,8 @@ async fn stored_agent_identity_jwt_keeps_auth_json_unchanged() -> anyhow::Result AuthKeyringBackendKind::Direct, Some(&authapi_base_url), &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await? .expect("auth should load"); @@ -630,6 +630,8 @@ async fn chatgpt_auth_registers_agent_identity_when_enabled() -> anyhow::Result< AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await? .expect("auth should load"); @@ -719,6 +721,8 @@ async fn chatgpt_auth_registers_agent_identity_when_enabled() -> anyhow::Result< AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await? .expect("auth should reload"); @@ -762,6 +766,8 @@ async fn chatgpt_auth_retries_transient_agent_identity_registration() -> anyhow: AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await? .expect("auth should load"); @@ -829,6 +835,8 @@ async fn chatgpt_auth_registration_retry_exhaustion_is_fallback_eligible() -> an AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await? .expect("auth should load"); @@ -891,6 +899,8 @@ async fn chatgpt_auth_task_registration_retry_exhaustion_is_fallback_eligible() AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await? .expect("auth should load"); @@ -948,6 +958,8 @@ async fn chatgpt_auth_non_retryable_registration_error_is_hard_failure() -> anyh AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await? .expect("auth should load"); @@ -1094,6 +1106,8 @@ async fn pro_account_with_no_api_key_uses_chatgpt_auth() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .unwrap() @@ -1172,6 +1186,8 @@ async fn loads_api_key_from_auth_json() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .unwrap() @@ -1288,6 +1304,8 @@ async fn refresh_failure_is_scoped_to_the_matching_auth_snapshot() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("load auth") @@ -1309,6 +1327,7 @@ async fn refresh_failure_is_scoped_to_the_matching_auth_snapshot() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, ) .await .expect("updated auth should parse"); @@ -2093,6 +2112,8 @@ async fn load_auth_reads_access_token_from_env() { AuthKeyringBackendKind::Direct, Some(&authapi_base_url), &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("env auth should load") @@ -2142,6 +2163,8 @@ async fn load_auth_reads_personal_access_token_from_env() { AuthKeyringBackendKind::default(), /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("env auth should load") @@ -2316,6 +2339,8 @@ async fn load_auth_keeps_codex_api_key_env_precedence() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("env auth should load") @@ -3005,6 +3030,8 @@ async fn plan_type_maps_known_plan() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("load auth") @@ -3038,6 +3065,8 @@ async fn plan_type_maps_self_serve_business_usage_based_plan() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("load auth") @@ -3074,6 +3103,8 @@ async fn plan_type_maps_enterprise_cbp_usage_based_plan() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("load auth") @@ -3110,6 +3141,8 @@ async fn plan_type_maps_unknown_to_unknown() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("load auth") @@ -3143,6 +3176,8 @@ async fn missing_plan_type_maps_to_unknown() { AuthKeyringBackendKind::Direct, /*agent_identity_authapi_base_url*/ None, &crate::test_support::transport_default_auth_route_config(), + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await .expect("load auth") diff --git a/codex-rs/login/src/auth/manager.rs b/codex-rs/login/src/auth/manager.rs index 15864c40a8..97d42a309b 100644 --- a/codex-rs/login/src/auth/manager.rs +++ b/codex-rs/login/src/auth/manager.rs @@ -343,6 +343,7 @@ impl From for std::io::Error { } impl CodexAuth { + #[allow(clippy::too_many_arguments)] async fn from_auth_dot_json( codex_home: &Path, auth_dot_json: AuthDotJson, @@ -351,6 +352,7 @@ impl CodexAuth { keyring_backend_kind: AuthKeyringBackendKind, agent_identity_authapi_base_url: Option<&str>, auth_route_config: &AuthRouteConfig, + cached_auth: Option<&CodexAuth>, ) -> std::io::Result { let auth_mode = auth_dot_json.resolved_mode(); if auth_mode == AuthMode::ApiKey { @@ -399,8 +401,14 @@ impl CodexAuth { "personal access token auth is missing a personal access token.", )); }; - return Self::from_personal_access_token(personal_access_token, auth_route_config) - .await; + return Ok(Self::PersonalAccessToken( + PersonalAccessTokenAuth::load_with_cached( + personal_access_token, + auth_route_config, + cached_auth, + ) + .await?, + )); } if auth_mode == AuthMode::BedrockApiKey { let Some(auth) = auth_dot_json.bedrock_api_key else { @@ -429,8 +437,21 @@ impl CodexAuth { .tokens .as_ref() .and_then(|tokens| tokens.oauth_provenance.as_ref()); - let client = - create_default_auth_client(&refresh_token_endpoint(provenance), auth_route_config)?; + // Reuse only the transport, not the credential state or storage backend: the + // selected source can change even when its credential document is identical. + let cached_client = cached_auth + .filter(|auth| auth.get_current_auth_json().as_ref() == Some(&auth_dot_json)) + .and_then(|auth| match auth { + Self::Chatgpt(auth) => Some(auth.state.client.clone()), + Self::ChatgptAuthTokens(auth) => Some(auth.state.client.clone()), + _ => None, + }); + let client = match cached_client { + Some(client) => client, + None => { + create_default_auth_client(&refresh_token_endpoint(provenance), auth_route_config)? + } + }; let state = ChatgptAuthState { auth_dot_json: Arc::new(Mutex::new(Some(auth_dot_json))), client, @@ -480,6 +501,8 @@ impl CodexAuth { keyring_backend_kind, agent_identity_authapi_base_url.as_deref(), auth_route_config, + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await } @@ -1275,6 +1298,15 @@ impl AuthConfig { pub async fn load_auth( &self, enable_codex_api_key_env: bool, + ) -> std::io::Result> { + self.load_auth_with_modes(enable_codex_api_key_env, /*allowed_auth_modes*/ None) + .await + } + + async fn load_auth_with_modes( + &self, + enable_codex_api_key_env: bool, + allowed_auth_modes: Option<&[AuthMode]>, ) -> std::io::Result> { let allowed_login_methods = self.allowed_login_methods(); let workspaces = self.effective_chatgpt_workspaces(); @@ -1290,6 +1322,8 @@ impl AuthConfig { self.keyring_backend_kind, agent_identity_authapi_base_url.as_deref(), &self.auth_route_config, + /*cached_auth*/ None, + allowed_auth_modes, ) .await?; Ok(auth.filter(|auth| self.allows_auth(auth))) @@ -1389,6 +1423,8 @@ async fn enforce_login_restrictions_with_agent_identity_authapi_base_url( config.keyring_backend_kind, agent_identity_authapi_base_url, &config.auth_route_config, + /*cached_auth*/ None, + /*allowed_auth_modes*/ None, ) .await? else { @@ -1543,9 +1579,13 @@ async fn load_auth( keyring_backend_kind: AuthKeyringBackendKind, agent_identity_authapi_base_url: Option<&str>, auth_route_config: &AuthRouteConfig, + cached_auth: Option<&CodexAuth>, + allowed_auth_modes: Option<&[AuthMode]>, ) -> std::io::Result> { + let mode_allowed = |mode| allowed_auth_modes.is_none_or(|modes| modes.contains(&mode)); // API key via env var takes precedence over any other auth method. if enable_codex_api_key_env + && mode_allowed(AuthMode::ApiKey) && auth_mode_is_allowed(allowed_login_methods, AuthMode::ApiKey) && let Some(api_key) = read_codex_api_key_from_env() { @@ -1559,9 +1599,12 @@ async fn load_auth( AuthCredentialsStoreMode::Ephemeral, AuthKeyringBackendKind::default(), ); - if let Some(auth_dot_json) = ephemeral_storage.load()? + if let Some(auth_dot_json) = load_stored_auth(ephemeral_storage).await? && auth_mode_is_allowed(allowed_login_methods, auth_dot_json.resolved_mode()) { + if !mode_allowed(auth_dot_json.resolved_mode()) { + return Ok(None); + } if let Some(agent_identity) = auth_dot_json.agent_identity.as_ref() { ensure_agent_identity_workspace_allowed(forced_chatgpt_workspace_id, agent_identity)?; } @@ -1573,6 +1616,7 @@ async fn load_auth( keyring_backend_kind, agent_identity_authapi_base_url, auth_route_config, + cached_auth, ) .await?; if let CodexAuth::PersonalAccessToken(auth) = &auth { @@ -1586,11 +1630,22 @@ async fn load_auth( { return match classify_codex_access_token(&access_token) { CodexAccessToken::PersonalAccessToken(access_token) => { - let auth = PersonalAccessTokenAuth::load(access_token, auth_route_config).await?; + if !mode_allowed(AuthMode::PersonalAccessToken) { + return Ok(None); + } + let auth = PersonalAccessTokenAuth::load_with_cached( + access_token, + auth_route_config, + cached_auth, + ) + .await?; ensure_auth_workspace_allowed(forced_chatgpt_workspace_id, auth.account_id())?; Ok(Some(CodexAuth::PersonalAccessToken(auth))) } CodexAccessToken::AgentIdentityJwt(jwt) => { + if !mode_allowed(AuthMode::AgentIdentity) { + return Ok(None); + } let record = AgentIdentityAuthRecord::from_agent_identity_jwt(jwt)?; ensure_auth_workspace_allowed(forced_chatgpt_workspace_id, &record.account_id)?; CodexAuth::from_agent_identity_jwt_with_authapi_base_url( @@ -1616,11 +1671,13 @@ async fn load_auth( auth_credentials_store_mode, keyring_backend_kind, ); - let auth_dot_json = match storage.load()? { + let auth_dot_json = match load_stored_auth(storage).await? { Some(auth) => auth, None => return Ok(None), }; - if !auth_mode_is_allowed(allowed_login_methods, auth_dot_json.resolved_mode()) { + if !mode_allowed(auth_dot_json.resolved_mode()) + || !auth_mode_is_allowed(allowed_login_methods, auth_dot_json.resolved_mode()) + { return Ok(None); } if let Some(agent_identity) = auth_dot_json.agent_identity.as_ref() { @@ -1635,6 +1692,7 @@ async fn load_auth( keyring_backend_kind, agent_identity_authapi_base_url, auth_route_config, + cached_auth, ) .await?; if let CodexAuth::PersonalAccessToken(auth) = &auth { @@ -1643,6 +1701,16 @@ async fn load_auth( Ok(Some(auth)) } +// Blocking keyring access must not prevent an async caller's deadline from firing. +async fn load_stored_auth( + storage: Arc, +) -> std::io::Result> { + let originator = codex_otel::auth_storage::AuthStorageOriginator::current(); + tokio::task::spawn_blocking(move || originator.sync_scope(|| storage.load())) + .await + .map_err(|_| std::io::Error::other("credential storage read failed"))? +} + // Ordinary auth tolerates legacy metadata loss; EMA grant checks remain strict. fn same_login_for_persistence(expected: &TokenData, stored: &TokenData) -> bool { same_account_login(expected, stored) @@ -1666,16 +1734,17 @@ pub(super) fn persist_tokens( let _commit = storage.lock_for_write()?; let stored = storage.load()?; let mut auth_dot_json = stored.ok_or(std::io::Error::other("Token data is not available."))?; + let original = auth_dot_json.clone(); if auth_dot_json.resolved_mode() != AuthMode::Chatgpt { return Err(std::io::Error::other( "Account credentials changed during refresh.", )); } - if !auth_dot_json - .tokens - .as_ref() - .is_some_and(|tokens| same_login_for_persistence(expected, tokens)) - { + if !auth_dot_json.tokens.as_ref().is_some_and(|tokens| { + same_login_for_persistence(expected, tokens) + && tokens.refresh_token == expected.refresh_token + && tokens.access_token == expected.access_token + }) { return Err(std::io::Error::other( "Account credentials changed during refresh.", )); @@ -1696,7 +1765,7 @@ pub(super) fn persist_tokens( tokens.refresh_token = refresh_token; } auth_dot_json.last_refresh = Some(Utc::now()); - storage.save(&auth_dot_json)?; + storage.compare_and_save(&original, &auth_dot_json)?; Ok(auth_dot_json) } @@ -1715,7 +1784,7 @@ async fn request_chatgpt_token_refresh( url: endpoint, client_id, encoding: TokenEncoding::Json, - timeout: None, + timeout: Some(Duration::from_secs(45)), error_body_limit: ErrorBodyLimit::Unlimited, }, ); @@ -2150,6 +2219,7 @@ pub struct AuthManager { external_auth: RwLock>>, workload_identity_selected: bool, auth_route_config: AuthRouteConfig, + allowed_auth_modes: RwLock>>, } /// Configuration view required to construct a shared [`AuthManager`]. @@ -2261,13 +2331,18 @@ impl AuthManager { auth_route_config, }, enable_codex_api_key_env, + /*allowed_auth_modes*/ None, ) .await } - async fn new_from_auth_config(auth_config: AuthConfig, enable_codex_api_key_env: bool) -> Self { + async fn new_from_auth_config( + auth_config: AuthConfig, + enable_codex_api_key_env: bool, + allowed_auth_modes: Option>, + ) -> Self { let managed_auth = auth_config - .load_auth(enable_codex_api_key_env) + .load_auth_with_modes(enable_codex_api_key_env, allowed_auth_modes.as_deref()) .await .ok() .flatten(); @@ -2307,6 +2382,7 @@ impl AuthManager { external_auth: RwLock::new(None), workload_identity_selected: false, auth_route_config, + allowed_auth_modes: RwLock::new(allowed_auth_modes), } } @@ -2342,6 +2418,7 @@ impl AuthManager { agent_identity_bootstrap_cooldown: Mutex::default(), external_auth: RwLock::new(None), workload_identity_selected: false, + allowed_auth_modes: RwLock::new(None), auth_route_config: crate::test_support::transport_default_auth_route_config(), }) } @@ -2372,6 +2449,7 @@ impl AuthManager { agent_identity_bootstrap_cooldown: Mutex::default(), external_auth: RwLock::new(None), workload_identity_selected: false, + allowed_auth_modes: RwLock::new(None), auth_route_config: crate::test_support::transport_default_auth_route_config(), }) } @@ -2410,6 +2488,7 @@ impl AuthManager { agent_identity_bootstrap_cooldown: Mutex::default(), external_auth: RwLock::new(None), workload_identity_selected: false, + allowed_auth_modes: RwLock::new(None), auth_route_config: crate::test_support::transport_default_auth_route_config(), }) } @@ -2447,6 +2526,7 @@ impl AuthManager { agent_identity_bootstrap_cooldown: Mutex::default(), external_auth: RwLock::new(Some(Arc::new(BearerTokenRefresher::new(config, env)))), workload_identity_selected: false, + allowed_auth_modes: RwLock::new(None), // External bearer auth refreshes by running the provider's command and never makes // auth-owned HTTP requests, so this route is intentionally inert. auth_route_config: AuthRouteConfig::from_http_client_factory(HttpClientFactory::new( @@ -2719,7 +2799,7 @@ impl AuthManager { tracing::info!("Reloading auth"); loop { let generation = self.auth_change_state_tx.borrow().generation; - let new_auth = self.load_auth().await; + let new_auth = self.load_auth(/*cached_auth*/ None).await; if let Some(publication) = self.publish_reloaded_auth(generation, new_auth) { return publication.changed; } @@ -2729,6 +2809,38 @@ impl AuthManager { } } + /// Pins the stored credential kind before any subsequent load or hydration. + /// This prevents a running client from silently acquiring a different kind of secret. + pub fn restrict_auth_modes(&self, mut modes: Vec) -> std::io::Result<()> { + let mut allowed = self + .allowed_auth_modes + .write() + .map_err(|_| std::io::Error::other("credential policy is unavailable"))?; + if let Some(previous) = allowed.as_ref() { + modes.retain(|mode| previous.contains(mode)); + } + *allowed = Some(modes); + Ok(()) + } + + /// Reloads stored credentials without hiding storage or hydration failures. + /// Failed loads retain the last known owner so a temporary outage does not + /// invalidate application network policy as if the user had logged out. + /// Callers must serialize reload and refresh and reject cached headers on error. + pub async fn reload_with_cached_auth(&self) -> std::io::Result { + loop { + let generation = self.auth_change_state_tx.borrow().generation; + let cached = self.auth_cached(); + let new_auth = self.try_load_managed_auth(cached.as_ref()).await?; + if let Some(publication) = self.publish_reloaded_auth(generation, new_auth) { + return Ok(publication.changed); + } + if self.inner.is_poisoned() { + return Err(std::io::Error::other("credential cache is unavailable")); + } + } + } + async fn reload_if_account_id_matches( &self, expected_account_id: Option<&str>, @@ -2743,7 +2855,7 @@ impl AuthManager { loop { let generation = self.auth_change_state_tx.borrow().generation; - let new_auth = self.load_auth().await; + let new_auth = self.load_auth(/*cached_auth*/ None).await; let new_account_id = new_auth.as_ref().and_then(CodexAuth::get_account_id); if new_account_id.as_deref() != Some(expected_account_id) { @@ -2822,7 +2934,7 @@ impl AuthManager { } } - async fn load_auth(&self) -> Option { + async fn load_auth(&self, cached_auth: Option<&CodexAuth>) -> Option { if let Some(external_auth) = self.external_auth_provider() { let cached_auth = self.auth_cached(); if cached_auth @@ -2849,9 +2961,21 @@ impl AuthManager { }; } + self.try_load_managed_auth(cached_auth).await.ok().flatten() + } + + async fn try_load_managed_auth( + &self, + cached_auth: Option<&CodexAuth>, + ) -> std::io::Result> { + let allowed_auth_modes = self + .allowed_auth_modes + .read() + .map_err(|_| std::io::Error::other("credential policy is unavailable"))? + .clone(); let allowed_login_methods = self.allowed_login_methods(); let effective_chatgpt_workspaces = self.effective_chatgpt_workspaces(); - load_auth( + let auth = load_auth( &self.codex_home, self.enable_codex_api_key_env, self.auth_credentials_store_mode, @@ -2861,18 +2985,18 @@ impl AuthManager { self.keyring_backend_kind, self.agent_identity_authapi_base_url.as_deref(), &self.auth_route_config, + cached_auth, + allowed_auth_modes.as_deref(), ) - .await - .ok() - .flatten() - .filter(|auth| { + .await?; + Ok(auth.filter(|auth| { validate_auth_restrictions( Some(&allowed_login_methods), effective_chatgpt_workspaces.as_deref(), auth, ) .is_ok() - }) + })) } fn set_cached_auth(&self, new_auth: Option) -> bool { @@ -3083,9 +3207,52 @@ impl AuthManager { pub async fn shared_from_auth_config( auth_config: AuthConfig, enable_codex_api_key_env: bool, + ) -> Result, AuthManagerInitializationError> { + Self::shared_from_auth_config_with_modes( + auth_config, + enable_codex_api_key_env, + /*allowed_auth_modes*/ None, + ) + .await + } + + /// Restricts credential modes before loading or hydrating the initial credential. + pub async fn shared_from_config_with_auth_modes( + config: &impl AuthManagerConfig, + enable_codex_api_key_env: bool, + allowed_auth_modes: Vec, + ) -> Result, AuthManagerInitializationError> { + Self::shared_from_auth_config_with_auth_modes( + auth_config_from(config), + enable_codex_api_key_env, + allowed_auth_modes, + ) + .await + } + + /// Restricts credential modes during bootstrap, before loading runtime configuration. + pub async fn shared_from_auth_config_with_auth_modes( + auth_config: AuthConfig, + enable_codex_api_key_env: bool, + allowed_auth_modes: Vec, + ) -> Result, AuthManagerInitializationError> { + Self::shared_from_auth_config_with_modes( + auth_config, + enable_codex_api_key_env, + Some(allowed_auth_modes), + ) + .await + } + + async fn shared_from_auth_config_with_modes( + auth_config: AuthConfig, + enable_codex_api_key_env: bool, + allowed_auth_modes: Option>, ) -> Result, AuthManagerInitializationError> { let external_auth = WorkloadIdentityExternalAuth::from_process_config(&auth_config)?; - let mut manager = Self::new_from_auth_config(auth_config, enable_codex_api_key_env).await; + let mut manager = + Self::new_from_auth_config(auth_config, enable_codex_api_key_env, allowed_auth_modes) + .await; manager.workload_identity_selected = external_auth.is_some(); let manager = Arc::new(manager); if let Some(external_auth) = external_auth { @@ -3441,14 +3608,26 @@ impl AuthManager { endpoint == format!("{}/oauth/token", provenance.issuer.trim_end_matches('/')) }); - persist_tokens( - auth.storage(), - &tokens, - provenance, - refresh_response.id_token, - refresh_response.access_token, - refresh_response.refresh_token, - ) + let storage = Arc::clone(auth.storage()); + let originator = codex_otel::auth_storage::AuthStorageOriginator::current(); + tokio::task::spawn_blocking(move || { + originator.sync_scope(|| { + persist_tokens( + &storage, + &tokens, + provenance, + refresh_response.id_token, + refresh_response.access_token, + refresh_response.refresh_token, + ) + }) + }) + .await + .map_err(|_| { + RefreshTokenError::Transient(std::io::Error::other( + "credential persistence task failed", + )) + })? .map_err(RefreshTokenError::from)?; self.reload().await; diff --git a/codex-rs/login/src/auth/personal_access_token.rs b/codex-rs/login/src/auth/personal_access_token.rs index dba80c8f94..86fb6a87b2 100644 --- a/codex-rs/login/src/auth/personal_access_token.rs +++ b/codex-rs/login/src/auth/personal_access_token.rs @@ -37,6 +37,19 @@ impl fmt::Debug for PersonalAccessTokenAuth { } impl PersonalAccessTokenAuth { + pub(super) async fn load_with_cached( + access_token: &str, + auth_route_config: &AuthRouteConfig, + cached: Option<&super::CodexAuth>, + ) -> std::io::Result { + if let Some(super::CodexAuth::PersonalAccessToken(cached)) = cached + && cached.access_token() == access_token + { + return Ok(cached.clone()); + } + Self::load(access_token, auth_route_config).await + } + pub(super) async fn load( access_token: &str, auth_route_config: &AuthRouteConfig, diff --git a/codex-rs/login/src/auth/storage.rs b/codex-rs/login/src/auth/storage.rs index 60adea09ca..7b1e12a9ef 100644 --- a/codex-rs/login/src/auth/storage.rs +++ b/codex-rs/login/src/auth/storage.rs @@ -12,8 +12,6 @@ use std::fs::File; use std::fs::OpenOptions; use std::io::Read; use std::io::Write; -#[cfg(unix)] -use std::os::unix::fs::OpenOptionsExt; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; @@ -23,6 +21,12 @@ use tracing::warn; #[path = "storage_error.rs"] mod storage_error; +#[path = "storage_file.rs"] +mod storage_file; + +#[path = "storage_commit.rs"] +mod storage_commit; + #[path = "storage_telemetry.rs"] mod storage_telemetry; @@ -213,6 +217,17 @@ pub(super) trait AuthStorageBackend: Debug + Send + Sync { self.load() } fn save(&self, auth: &AuthDotJson) -> std::io::Result<()>; + /// Commits only while the selected credential document remains unchanged. + fn compare_and_save(&self, expected: &AuthDotJson, auth: &AuthDotJson) -> std::io::Result<()> { + // An unavailable keyring cannot prove that a fallback is still current. + // Do not overwrite a recovered secure-store generation we could not read. + if self.load_account_grant()?.as_ref() != Some(expected) { + return Err(std::io::Error::other( + "Credentials changed before refresh commit.", + )); + } + self.save(auth) + } fn delete(&self) -> std::io::Result; } @@ -254,21 +269,25 @@ impl AuthStorageBackend for FileAuthStorage { } fn save(&self, auth_dot_json: &AuthDotJson) -> std::io::Result<()> { - let auth_file = get_auth_file(&self.codex_home); - - if let Some(parent) = auth_file.parent() { - std::fs::create_dir_all(parent)?; - } + let auth_file = storage_file::resolve_auth_file(get_auth_file(&self.codex_home))?; + let parent = auth_file + .parent() + .ok_or_else(|| std::io::Error::other("auth file has no parent"))?; + std::fs::create_dir_all(parent)?; let json_data = serde_json::to_string_pretty(auth_dot_json)?; - let mut options = OpenOptions::new(); - options.truncate(true).write(true).create(true); - #[cfg(unix)] - { - options.mode(0o600); - } - let mut file = options.open(auth_file)?; - file.write_all(json_data.as_bytes())?; - file.flush()?; + storage_file::check_replaceable(&auth_file)?; + // tempfile creates a private (0600 on Unix) file in the same filesystem. + // Readers see the complete old or new document, never a truncated auth.json. + let mut replacement = storage_file::replacement(&auth_file)?; + replacement.write_all(json_data.as_bytes())?; + replacement.as_file().sync_all()?; + // Clear Windows' temporary attribute, then restore cleanup protection. + // std::fs::rename can replace a file held open by a reader on Windows; + // tempfile::persist only uses MoveFileExW, which rejects that case. + let (_file, path) = replacement.keep().map_err(|error| error.error)?; + let mut path = tempfile::TempPath::try_from_path(path)?; + std::fs::rename(&path, auth_file)?; + path.disable_cleanup(/*disable_cleanup*/ true); Ok(()) } @@ -659,6 +678,7 @@ fn create_auth_storage_with_store( keyring_store: Arc, keyring_backend_kind: AuthKeyringBackendKind, ) -> Arc { + let commit_home = codex_home.clone(); let storage: Arc = match mode { AuthCredentialsStoreMode::File => Arc::new(FileAuthStorage::new(codex_home)), AuthCredentialsStoreMode::Keyring => { @@ -671,7 +691,11 @@ fn create_auth_storage_with_store( )), AuthCredentialsStoreMode::Ephemeral => Arc::new(EphemeralAuthStorage::new(codex_home)), }; - storage_telemetry::observe(storage, mode, keyring_backend_kind) + storage_commit::wrap( + storage_telemetry::observe(storage, mode, keyring_backend_kind), + commit_home, + mode, + ) } fn create_keyring_auth_storage( diff --git a/codex-rs/login/src/auth/storage_commit.rs b/codex-rs/login/src/auth/storage_commit.rs new file mode 100644 index 0000000000..605d21638c --- /dev/null +++ b/codex-rs/login/src/auth/storage_commit.rs @@ -0,0 +1,99 @@ +//! Serializes cooperating processes' credential commits, including compare-and-save. +//! The stable sidecar is never renamed or removed; network requests never hold it. + +use super::*; +use std::fs::OpenOptions; +use std::fs::TryLockError; +use std::time::Duration; +use std::time::Instant; + +pub(super) fn wrap( + inner: Arc, + home: PathBuf, + mode: AuthCredentialsStoreMode, +) -> Arc { + if mode == AuthCredentialsStoreMode::Ephemeral { + return inner; + } + Arc::new(CommittedStorage { inner, home, mode }) +} + +#[derive(Debug)] +struct CommittedStorage { + inner: Arc, + home: PathBuf, + mode: AuthCredentialsStoreMode, +} + +impl CommittedStorage { + fn lock(&self) -> std::io::Result { + // File, keyring, and auto writers must coordinate: auto may use the file + // fallback. Resolve symlinks so aliases of that file share the same lock. + let path = + storage_file::resolve_auth_file(get_auth_file(&self.home))?.with_extension("json.lock"); + std::fs::create_dir_all( + path.parent() + .ok_or_else(|| std::io::Error::other("credential lock has no parent"))?, + )?; + let mut options = OpenOptions::new(); + options.read(true).write(true).create(true).truncate(false); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let file = options.open(path)?; + let deadline = Instant::now() + Duration::from_secs(5); + loop { + match file.try_lock() { + Ok(()) => return Ok(file), + Err(TryLockError::WouldBlock) if Instant::now() < deadline => { + std::thread::sleep(Duration::from_millis(10)) + } + Err(TryLockError::WouldBlock) => { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "credential commit lock timed out", + )); + } + Err(error) => return Err(error.into()), + } + } + } +} + +impl AuthStorageBackend for CommittedStorage { + fn persistent_home(&self) -> Option<&Path> { + self.inner.persistent_home() + } + + fn load(&self) -> std::io::Result> { + self.inner.load() + } + fn load_account_grant(&self) -> std::io::Result> { + self.inner.load_account_grant() + } + fn save(&self, auth: &AuthDotJson) -> std::io::Result<()> { + let _guard = self.lock()?; + self.inner.save(auth) + } + fn compare_and_save(&self, expected: &AuthDotJson, auth: &AuthDotJson) -> std::io::Result<()> { + let _guard = self.lock()?; + self.inner.compare_and_save(expected, auth) + } + fn delete(&self) -> std::io::Result { + // A missing file-store directory contains no credentials to delete. + // Linearize this no-op before any concurrent creator, without creating + // a directory just to leave a lock behind. Secure stores may still have + // credentials even when the directory is absent and must take the lock. + if self.mode == AuthCredentialsStoreMode::File && !self.home.try_exists()? { + return Ok(false); + } + let _guard = self.lock()?; + self.inner.delete() + } +} + +#[cfg(test)] +#[path = "storage_commit_tests.rs"] +mod tests; diff --git a/codex-rs/login/src/auth/storage_commit_tests.rs b/codex-rs/login/src/auth/storage_commit_tests.rs new file mode 100644 index 0000000000..3d3edf6d5d --- /dev/null +++ b/codex-rs/login/src/auth/storage_commit_tests.rs @@ -0,0 +1,99 @@ +#![cfg(test)] + +use super::*; +use pretty_assertions::assert_eq; +use std::io::BufRead; +use std::io::Read; +use std::process::Stdio; + +#[test] +fn deleting_absent_file_store_does_not_create_home() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let home = root.path().join("missing"); + let storage = create_auth_storage( + home.clone(), + AuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + ); + assert!(!storage.delete()?); + assert!(!home.try_exists()?); + Ok(()) +} + +#[test] +fn write_lock_coordinates_with_existing_auth_lock() -> anyhow::Result<()> { + let home = tempfile::tempdir()?; + let storage = create_auth_storage( + home.path().to_path_buf(), + AuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + ); + let _guard = storage.lock_for_write()?; + let other_writer = OpenOptions::new() + .read(true) + .write(true) + .open(home.path().join("auth.lock"))?; + assert!(matches!( + other_writer.try_lock(), + Err(TryLockError::WouldBlock) + )); + Ok(()) +} + +#[test] +fn commits_respect_another_process_lock() -> anyhow::Result<()> { + const CHILD_HOME: &str = "CODEX_AUTH_COMMIT_TEST_HOME"; + if let Some(home) = std::env::var_os(CHILD_HOME) { + let home = PathBuf::from(home); + let storage = CommittedStorage { + inner: Arc::new(FileAuthStorage::new(home.clone())), + home, + mode: AuthCredentialsStoreMode::File, + }; + let _guard = storage.lock()?; + println!("LOCK_HELD"); + let mut release = String::new(); + std::io::stdin().read_to_string(&mut release)?; + return Ok(()); + } + let home = tempfile::tempdir()?; + let storage = create_auth_storage( + home.path().to_path_buf(), + AuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + ); + let old: AuthDotJson = serde_json::from_value(serde_json::json!({"OPENAI_API_KEY":"old"}))?; + let new: AuthDotJson = serde_json::from_value(serde_json::json!({"OPENAI_API_KEY":"new"}))?; + storage.save(&old)?; + let mut child = codex_utils_process::background_command(std::env::current_exe()?) + .args([ + "--exact", + "auth::storage::storage_commit::tests::commits_respect_another_process_lock", + "--nocapture", + ]) + .env(CHILD_HOME, home.path()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn()?; + let input = child.stdin.take(); + let mut output = std::io::BufReader::new(child.stdout.take().unwrap()); + loop { + let mut line = String::new(); + anyhow::ensure!( + output.read_line(&mut line)? > 0, + "lock holder exited before acquiring the lock" + ); + if line.contains("LOCK_HELD") { + break; + } + } + let result = storage.compare_and_save(&old, &new); + drop(input); + let status = child.wait()?; + assert!(status.success()); + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::TimedOut); + assert_eq!(storage.load()?, Some(old.clone())); + storage.compare_and_save(&old, &new)?; + assert_eq!(storage.load()?, Some(new)); + Ok(()) +} diff --git a/codex-rs/login/src/auth/storage_file.rs b/codex-rs/login/src/auth/storage_file.rs new file mode 100644 index 0000000000..43dba97e90 --- /dev/null +++ b/codex-rs/login/src/auth/storage_file.rs @@ -0,0 +1,73 @@ +//! Atomic credential-file replacement without broadening existing file permissions. +//! Symlinks retain their target; hard links and read-only files fail rather than split identity. + +#[cfg(not(windows))] +use std::fs::File; +use std::io; +use std::path::Path; +use std::path::PathBuf; + +pub(super) fn resolve_auth_file(mut path: PathBuf) -> io::Result { + for _ in 0..40 { + match std::fs::symlink_metadata(&path) { + Ok(metadata) if metadata.file_type().is_symlink() => { + let target = std::fs::read_link(&path)?; + path = if target.is_absolute() { + target + } else { + path.parent() + .ok_or_else(|| io::Error::other("credential link has no parent"))? + .join(target) + }; + } + Ok(_) => return Ok(path), + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(path), + Err(error) => return Err(error), + } + } + Err(io::Error::new( + io::ErrorKind::InvalidInput, + "too many credential symlinks", + )) +} + +pub(super) fn check_replaceable(path: &Path) -> io::Result<()> { + let metadata = match std::fs::metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error), + }; + if metadata.permissions().readonly() { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "credential file is read-only", + )); + } + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt; + if metadata.nlink() > 1 && metadata.is_file() { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "atomic credential replacement does not support hard-linked files", + )); + } + } + // Check the destination's write permission, not only its parent directory's. + std::fs::OpenOptions::new().write(true).open(path)?; + Ok(()) +} + +#[cfg(not(windows))] +pub(super) fn replacement(path: &Path) -> io::Result> { + tempfile::Builder::new().prefix(".auth-").tempfile_in( + path.parent() + .ok_or_else(|| io::Error::other("credential file has no parent"))?, + ) +} + +#[cfg(windows)] +#[path = "storage_file_windows.rs"] +mod windows; +#[cfg(windows)] +pub(super) use windows::replacement; diff --git a/codex-rs/login/src/auth/storage_file_windows.rs b/codex-rs/login/src/auth/storage_file_windows.rs new file mode 100644 index 0000000000..efe620e8f2 --- /dev/null +++ b/codex-rs/login/src/auth/storage_file_windows.rs @@ -0,0 +1,128 @@ +//! Creates replacement files with the destination DACL before any secret is written. +//! Protect the captured DACL from broader parent inheritance during creation. + +use std::ffi::c_void; +use std::fs::File; +use std::fs::OpenOptions; +use std::io; +use std::os::windows::ffi::OsStrExt; +use std::os::windows::fs::OpenOptionsExt; +use std::os::windows::io::AsRawHandle; +use std::os::windows::io::FromRawHandle; +use std::path::Path; +use std::ptr; +use windows_sys::Win32::Foundation::ERROR_SUCCESS; +use windows_sys::Win32::Foundation::GENERIC_READ; +use windows_sys::Win32::Foundation::GENERIC_WRITE; +use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE; +use windows_sys::Win32::Foundation::LocalFree; +use windows_sys::Win32::Security::Authorization::GetSecurityInfo; +use windows_sys::Win32::Security::Authorization::SE_FILE_OBJECT; +use windows_sys::Win32::Security::DACL_SECURITY_INFORMATION; +use windows_sys::Win32::Security::SE_DACL_PROTECTED; +use windows_sys::Win32::Security::SECURITY_ATTRIBUTES; +use windows_sys::Win32::Security::SetSecurityDescriptorControl; +use windows_sys::Win32::Storage::FileSystem::CREATE_NEW; +use windows_sys::Win32::Storage::FileSystem::CreateFileW; +use windows_sys::Win32::Storage::FileSystem::FILE_ATTRIBUTE_NORMAL; +use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_DELETE; +use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ; +use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_WRITE; +use windows_sys::Win32::Storage::FileSystem::GetFileInformationByHandle; +use windows_sys::Win32::Storage::FileSystem::READ_CONTROL; + +struct Descriptor(*mut c_void); + +impl Drop for Descriptor { + fn drop(&mut self) { + // SAFETY: GetSecurityInfo allocates this descriptor with LocalAlloc. + unsafe { + LocalFree(self.0); + } + } +} + +pub(in crate::auth::storage) fn replacement( + path: &Path, +) -> io::Result> { + let descriptor = match OpenOptions::new().access_mode(READ_CONTROL).open(path) { + Ok(file) => { + // SAFETY: BY_HANDLE_FILE_INFORMATION is an all-integer output structure. + let mut info = unsafe { std::mem::zeroed() }; + // SAFETY: file is open and info is a valid output structure. + if unsafe { GetFileInformationByHandle(file.as_raw_handle(), &mut info) } == 0 { + return Err(io::Error::last_os_error()); + } + if info.nNumberOfLinks > 1 { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "atomic credential replacement does not support hard-linked files", + )); + } + let mut descriptor = ptr::null_mut(); + // SAFETY: the file handle and output pointer remain valid for this call. + let status = unsafe { + GetSecurityInfo( + file.as_raw_handle(), + SE_FILE_OBJECT, + DACL_SECURITY_INFORMATION, + ptr::null_mut(), + ptr::null_mut(), + ptr::null_mut(), + ptr::null_mut(), + &mut descriptor, + ) + }; + if status != ERROR_SUCCESS { + return Err(io::Error::from_raw_os_error(status as i32)); + } + let descriptor = Descriptor(descriptor); + // SAFETY: this is the owned descriptor returned by GetSecurityInfo. + if unsafe { + SetSecurityDescriptorControl(descriptor.0, SE_DACL_PROTECTED, SE_DACL_PROTECTED) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Some(descriptor) + } + Err(error) if error.kind() == io::ErrorKind::NotFound => None, + Err(error) => return Err(error), + }; + let parent = std::fs::canonicalize( + path.parent() + .ok_or_else(|| io::Error::other("credential file has no parent"))?, + )?; + tempfile::Builder::new() + .prefix(".auth-") + .make_in(parent, |temporary| { + let name: Vec = temporary.as_os_str().encode_wide().chain(Some(0)).collect(); + let attributes = SECURITY_ATTRIBUTES { + nLength: std::mem::size_of::() as u32, + lpSecurityDescriptor: descriptor.as_ref().map_or(ptr::null_mut(), |value| value.0), + bInheritHandle: 0, + }; + // SAFETY: name is NUL-terminated; the owned descriptor outlives creation. + // CREATE_NEW prevents opening an attacker-created temporary path. + let handle = unsafe { + CreateFileW( + name.as_ptr(), + GENERIC_READ | GENERIC_WRITE, + FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, + &attributes, + CREATE_NEW, + FILE_ATTRIBUTE_NORMAL, + ptr::null_mut(), + ) + }; + if handle == INVALID_HANDLE_VALUE { + return Err(io::Error::last_os_error()); + } + // SAFETY: this is a newly opened handle owned exclusively by this File. + Ok(unsafe { File::from_raw_handle(handle) }) + }) +} + +#[cfg(test)] +#[path = "storage_file_windows_tests.rs"] +mod tests; diff --git a/codex-rs/login/src/auth/storage_file_windows_tests.rs b/codex-rs/login/src/auth/storage_file_windows_tests.rs new file mode 100644 index 0000000000..3c61c1016f --- /dev/null +++ b/codex-rs/login/src/auth/storage_file_windows_tests.rs @@ -0,0 +1,100 @@ +#![cfg(test)] + +use super::*; +use pretty_assertions::assert_eq; +use std::io::Write; +use windows_sys::Win32::Security::Authorization::ConvertStringSecurityDescriptorToSecurityDescriptorW; +use windows_sys::Win32::Security::Authorization::SetNamedSecurityInfoW; +use windows_sys::Win32::Security::GetSecurityDescriptorDacl; +use windows_sys::Win32::Security::PROTECTED_DACL_SECURITY_INFORMATION; + +fn set_dacl(path: &Path, sddl: &str) -> io::Result<()> { + let text: Vec = sddl.encode_utf16().chain(Some(0)).collect(); + let mut descriptor = ptr::null_mut(); + // SAFETY: text is NUL-terminated, the result is freed by Descriptor. + if unsafe { + ConvertStringSecurityDescriptorToSecurityDescriptorW( + text.as_ptr(), + 1, + &mut descriptor, + ptr::null_mut(), + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + let descriptor = Descriptor(descriptor); + let mut present = 0; + let mut defaulted = 0; + let mut acl = ptr::null_mut(); + // SAFETY: all output pointers are valid; descriptor owns the returned ACL. + if unsafe { GetSecurityDescriptorDacl(descriptor.0, &mut present, &mut acl, &mut defaulted) } + == 0 + { + return Err(io::Error::last_os_error()); + } + assert_ne!(present, 0); + let path: Vec = path.as_os_str().encode_wide().chain(Some(0)).collect(); + // SAFETY: the path and ACL remain valid for this call. + let status = unsafe { + SetNamedSecurityInfoW( + path.as_ptr(), + SE_FILE_OBJECT, + DACL_SECURITY_INFORMATION | PROTECTED_DACL_SECURITY_INFORMATION, + ptr::null_mut(), + ptr::null_mut(), + acl, + ptr::null_mut(), + ) + }; + if status != ERROR_SUCCESS { + return Err(io::Error::from_raw_os_error(status as i32)); + } + Ok(()) +} + +fn dacl(path: &Path) -> io::Result> { + let file = OpenOptions::new().access_mode(READ_CONTROL).open(path)?; + let mut acl = ptr::null_mut(); + let mut descriptor = ptr::null_mut(); + // SAFETY: the file and output pointers are valid. + let status = unsafe { + GetSecurityInfo( + file.as_raw_handle(), + SE_FILE_OBJECT, + DACL_SECURITY_INFORMATION, + ptr::null_mut(), + ptr::null_mut(), + &mut acl, + ptr::null_mut(), + &mut descriptor, + ) + }; + if status != ERROR_SUCCESS { + return Err(io::Error::from_raw_os_error(status as i32)); + } + let _descriptor = Descriptor(descriptor); + assert!(!acl.is_null()); + // SAFETY: the descriptor owns an ACL with AclSize bytes for this scope. + Ok(unsafe { std::slice::from_raw_parts(acl.cast::(), (*acl).AclSize as usize) }.to_vec()) +} + +#[test] +fn replacement_keeps_restricted_dacl_before_secret_is_written() -> anyhow::Result<()> { + let home = tempfile::tempdir()?; + // Everyone can read child files by inheritance; the credential explicitly cannot. + set_dacl(home.path(), "D:P(A;OICI;FA;;;WD)")?; + let path = home.path().join("auth.json"); + std::fs::write(&path, "old")?; + // OWNER RIGHTS grants access only to the owner, not Everyone from the parent. + set_dacl(&path, "D:P(A;;FA;;;OW)")?; + let expected = dacl(&path)?; + let mut temporary = replacement(&path)?; + assert_eq!(dacl(temporary.path())?, expected); + temporary.write_all(b"fake-secret")?; + let (_file, temporary_path) = temporary.keep()?; + std::fs::rename(temporary_path, &path)?; + assert_eq!(dacl(&path)?, expected); + assert_eq!(std::fs::read(&path)?, b"fake-secret"); + Ok(()) +} diff --git a/codex-rs/login/src/auth/storage_policy_tests.rs b/codex-rs/login/src/auth/storage_policy_tests.rs index 17e4ab035a..1ac3c75871 100644 --- a/codex-rs/login/src/auth/storage_policy_tests.rs +++ b/codex-rs/login/src/auth/storage_policy_tests.rs @@ -302,10 +302,19 @@ fn auto_auth_storage_load_falls_back_when_keyring_errors() -> anyhow::Result<()> assert_eq!(loaded, Some(expected.clone())); mock_keyring.set_error(&key, KeyringError::Invalid("error".into(), "load".into())); assert!(storage.load_account_grant().is_err()); + assert_eq!(storage.load()?, Some(encrypted.clone())); + assert_eq!(storage.file_storage.load()?, Some(expected.clone())); + mock_keyring.set_error(&key, KeyringError::Invalid("error".into(), "load".into())); + assert!( + storage + .compare_and_save(&expected, &auth_with_prefix("must-not-save")) + .is_err() + ); assert_eq!(storage.load()?, Some(encrypted)); assert_eq!(storage.file_storage.load()?, Some(expected)); - // Preserve rotated tokens whether the keyring recovers before or during persistence. + // A recovered keyring may contain a newer generation than the fallback. + // Never overwrite it with a response obtained using the fallback token. let mut previous = auth_with_prefix("same-login"); previous.auth_mode = Some(AuthMode::Chatgpt); previous.openai_api_key = None; @@ -330,21 +339,40 @@ fn auto_auth_storage_load_falls_back_when_keyring_errors() -> anyhow::Result<()> /*id_token*/ None, Some("rotated-access".into()), Some("rotated-refresh".into()), - )?; - let tokens = refreshed - .tokens - .as_ref() - .context("refreshed account tokens")?; - assert_eq!(tokens.access_token, "rotated-access"); - assert_eq!(tokens.refresh_token, "rotated-refresh"); - assert_eq!(storage.load()?, Some(refreshed.clone())); - assert_keyring_saved_auth_and_removed_fallback( - &mock_keyring, - codex_home.path(), - &refreshed, - )?; + ); + if recover_before_persist { + assert!(refreshed.is_err()); + assert_eq!(storage.load()?, Some(previous.clone())); + assert_eq!(storage.file_storage.load()?, Some(cached)); + } else { + // Matching keyring and fallback generations can safely commit. + let refreshed = refreshed?; + assert_eq!(storage.load()?, Some(refreshed.clone())); + assert_keyring_saved_auth_and_removed_fallback( + &mock_keyring, + codex_home.path(), + &refreshed, + )?; + } } + // Once the selected source is stable, a fresh exchange can commit normally. + let refreshed = crate::auth::manager::persist_tokens( + &backend, + previous.tokens.as_ref().context("stored account tokens")?, + /*provenance*/ None, + /*id_token*/ None, + Some("rotated-access".into()), + Some("rotated-refresh".into()), + )?; + let tokens = refreshed + .tokens + .as_ref() + .context("refreshed account tokens")?; + assert_eq!(tokens.access_token, "rotated-access"); + assert_eq!(tokens.refresh_token, "rotated-refresh"); + assert_eq!(storage.load()?, Some(refreshed.clone())); + assert_keyring_saved_auth_and_removed_fallback(&mock_keyring, codex_home.path(), &refreshed)?; // An absent fallback must not select a plaintext destination for cached credentials. storage.file_storage.delete()?; mock_keyring.set_error(&key, KeyringError::Invalid("error".into(), "load".into())); diff --git a/codex-rs/login/src/auth/storage_tests.rs b/codex-rs/login/src/auth/storage_tests.rs index dc35215d8a..7da2c9c011 100644 --- a/codex-rs/login/src/auth/storage_tests.rs +++ b/codex-rs/login/src/auth/storage_tests.rs @@ -14,6 +14,83 @@ use tempfile::tempdir; use codex_keyring_store::tests::MockKeyringStore; use keyring::Error as KeyringError; +#[cfg(unix)] +#[test] +fn file_storage_creates_dangling_symlink_target_and_rejects_cycles() -> anyhow::Result<()> { + use std::os::unix::fs::symlink; + let home = tempdir()?; + let auth = serde_json::from_value(json!({"OPENAI_API_KEY":"fake-key"}))?; + let storage = FileAuthStorage::new(home.path().to_path_buf()); + symlink("intermediate", home.path().join("auth.json"))?; + symlink("new-target.json", home.path().join("intermediate"))?; + storage.save(&auth)?; + assert_eq!(storage.load()?, Some(auth.clone())); + assert!( + std::fs::symlink_metadata(home.path().join("auth.json"))? + .file_type() + .is_symlink() + ); + std::fs::remove_file(home.path().join("new-target.json"))?; + symlink("auth.json", home.path().join("new-target.json"))?; + assert_eq!( + storage.save(&auth).unwrap_err().kind(), + std::io::ErrorKind::InvalidInput + ); + Ok(()) +} + +#[cfg(unix)] +#[test] +fn file_storage_does_not_replace_read_only_or_hard_linked_credentials() -> anyhow::Result<()> { + use std::os::unix::fs::PermissionsExt; + let home = tempdir()?; + let path = home.path().join("auth.json"); + let storage = FileAuthStorage::new(home.path().to_path_buf()); + let first = serde_json::from_value(json!({"OPENAI_API_KEY":"first"}))?; + let second = serde_json::from_value(json!({"OPENAI_API_KEY":"second"}))?; + storage.save(&first)?; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o400))?; + assert_eq!( + storage.save(&second).unwrap_err().kind(), + std::io::ErrorKind::PermissionDenied + ); + assert_eq!(storage.load()?, Some(first.clone())); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))?; + let peer = home.path().join("peer.json"); + std::fs::hard_link(&path, &peer)?; + assert_eq!( + storage.save(&second).unwrap_err().kind(), + std::io::ErrorKind::Unsupported + ); + assert_eq!(std::fs::read(path)?, std::fs::read(peer)?); + assert_eq!(storage.load()?, Some(first)); + Ok(()) +} + +#[test] +fn refresh_commit_does_not_overwrite_another_writer() -> anyhow::Result<()> { + let home = tempdir()?; + let storage = create_auth_storage( + home.path().to_path_buf(), + AuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + ); + let first = serde_json::from_value(json!({"OPENAI_API_KEY":"first"}))?; + let second = serde_json::from_value(json!({"OPENAI_API_KEY":"second"}))?; + let stale = serde_json::from_value(json!({"OPENAI_API_KEY":"stale"}))?; + storage.save(&first)?; + // An independent storage instance publishes a later generation before commit. + let other = create_auth_storage( + home.path().to_path_buf(), + AuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + ); + other.save(&second)?; + assert!(storage.compare_and_save(&first, &stale).is_err()); + assert_eq!(storage.load()?, Some(second)); + Ok(()) +} + #[tokio::test] async fn file_storage_load_returns_auth_dot_json() -> anyhow::Result<()> { let codex_home = tempdir()?; @@ -62,6 +139,54 @@ async fn file_storage_save_persists_auth_dot_json() -> anyhow::Result<()> { .try_read_auth_json(&file) .context("failed to read auth file after save")?; assert_eq!(auth_dot_json, same_auth_dot_json); + + // An overlapping reader must keep seeing the complete old document. + let mut overlapping_reader = File::open(&file)?; + let mut replacement = auth_dot_json.clone(); + replacement.openai_api_key = Some("replacement-with-a-different-length".to_string()); + storage + .save(&replacement) + .context("atomically replace auth while the original reader remains open")?; + let mut original_contents = String::new(); + overlapping_reader.read_to_string(&mut original_contents)?; + assert_eq!( + serde_json::from_str::(&original_contents)?, + auth_dot_json + ); + assert_eq!(storage.load()?, Some(replacement)); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + std::fs::metadata(&file)?.permissions().mode() & 0o777, + 0o600 + ); + + let linked_home = tempdir()?; + let link = get_auth_file(linked_home.path()); + std::os::unix::fs::symlink(&file, &link)?; + FileAuthStorage::new(linked_home.path().to_path_buf()).save(&auth_dot_json)?; + assert!(std::fs::symlink_metadata(link)?.file_type().is_symlink()); + assert_eq!(storage.load()?, Some(auth_dot_json)); + } + Ok(()) +} + +#[test] +fn file_storage_failed_replacement_cleans_up_temporary_file() -> anyhow::Result<()> { + let home = tempdir()?; + let auth_path = get_auth_file(home.path()); + // A directory cannot be replaced by an auth file on any supported platform. + std::fs::create_dir(&auth_path)?; + let storage = FileAuthStorage::new(home.path().to_path_buf()); + let auth = serde_json::from_value(json!({"OPENAI_API_KEY": "test-key"}))?; + assert!(storage.save(&auth).is_err()); + let entries = std::fs::read_dir(home.path())? + .map(|entry| entry.map(|entry| entry.file_name())) + .collect::>>()?; + assert_eq!(entries, vec![std::ffi::OsString::from("auth.json")]); + assert!(auth_path.is_dir()); Ok(()) } diff --git a/codex-rs/rmcp-client/src/ema_http_client_tests.rs b/codex-rs/rmcp-client/src/ema_http_client_tests.rs index 82e8668d83..7540ae6618 100644 --- a/codex-rs/rmcp-client/src/ema_http_client_tests.rs +++ b/codex-rs/rmcp-client/src/ema_http_client_tests.rs @@ -376,16 +376,22 @@ async fn initial_bearer_aged_during_validation_is_renewed_before_dispatch() -> a .mount(&server) .await; let (client, source, params) = fixture(&format!("{}/mcp", server.uri())); + // Age the token by advancing time: backdating an Instant can underflow + // on a Windows runner that has only recently booted. + tokio::time::pause(); + let received_at = Instant::now(); + tokio::time::advance(Duration::from_secs(301)).await; let client = EmaAuthenticatedHttpClient::new( client.inner, Arc::clone(&source), EmaAccessToken { access_token: "aged-in-validation".to_string(), expires_in: Some(Duration::from_secs(300)), - received_at: Instant::now() - Duration::from_secs(301), + received_at, }, client.authority, ); + tokio::time::resume(); if streaming { client.http_request_stream(params).await?; } else {