diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 2e2ec60434..347cf9d64b 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3160,14 +3160,17 @@ name = "codex-core-plugins" version = "0.0.0" dependencies = [ "anyhow", + "base64 0.22.1", "chrono", "codex-analytics", + "codex-api", "codex-app-server-protocol", "codex-config", "codex-connectors", "codex-core-plugin-common", "codex-exec-server", "codex-exec-server-test-support", + "codex-features", "codex-git-utils", "codex-hooks", "codex-http-client", @@ -3177,6 +3180,7 @@ dependencies = [ "codex-otel", "codex-plugin", "codex-protocol", + "codex-runtime-agent-instance", "codex-shell-command", "codex-skills", "codex-tools", @@ -6012,22 +6016,26 @@ dependencies = [ "codex-login", "codex-model-provider-info", "codex-models-manager", + "codex-network-proxy", "codex-protocol", "codex-thread-store", "codex-utils-absolute-path", "codex-utils-cargo-bin", "codex-utils-path-uri", + "codex-utils-rustls-provider", "ctor 0.6.3", "futures", "notify", "opentelemetry", "opentelemetry_sdk", "pretty_assertions", + "rcgen", "regex-lite", "serde_json", "shlex", "tempfile", "tokio", + "tokio-rustls", "tokio-tungstenite", "tracing", "tracing-opentelemetry", diff --git a/codex-rs/app-server/src/request_processors/account_processor.rs b/codex-rs/app-server/src/request_processors/account_processor.rs index 80f88efb7f..46841fe415 100644 --- a/codex-rs/app-server/src/request_processors/account_processor.rs +++ b/codex-rs/app-server/src/request_processors/account_processor.rs @@ -17,6 +17,7 @@ use codex_model_provider::is_supported_amazon_bedrock_region; use codex_otel::auth_storage::AuthStorageOriginator; use codex_rmcp_client::EnterpriseOAuthCredentialGuard; +mod backend_client; mod bedrock_gov_cloud; mod bedrock_setup; mod enterprise_login; @@ -290,6 +291,7 @@ impl AccountRequestProcessor { .await { Ok(config) => { + let runtime_auth = thread_manager.runtime_auth_for_config(&config); Self::spawn_effective_plugins_changed_task( Arc::clone(thread_manager), config_manager.clone(), @@ -311,6 +313,10 @@ impl AccountRequestProcessor { &plugins_config, Some(Arc::clone(&on_effective_plugins_changed)), ); + if let Err(err) = runtime_auth { + warn!("failed to initialize runtime auth for remote plugin refresh: {err}"); + return; + } thread_manager .plugins_manager() .maybe_start_remote_plugin_caches_refresh( @@ -1186,11 +1192,7 @@ impl AccountRequestProcessor { )); } - let client = BackendClient::from_auth( - self.config.chatgpt_base_url.clone(), - &auth, - http_client_factory, - ); + let client = self.backend_client(&auth, http_client_factory).await?; let usage_request = async { if params.supports_luna_reserve @@ -1303,11 +1305,7 @@ impl AccountRequestProcessor { )); } - let client = BackendClient::from_auth( - self.config.chatgpt_base_url.clone(), - &auth, - http_client_factory, - ); + let client = self.backend_client(&auth, http_client_factory).await?; if let Some(thread_id) = thread_id { let thread_id = thread_id.to_string(); let usage = tokio::time::timeout( @@ -1391,11 +1389,7 @@ impl AccountRequestProcessor { )); } - let client = BackendClient::from_auth( - self.config.chatgpt_base_url.clone(), - &auth, - http_client_factory, - ); + let client = self.backend_client(&auth, http_client_factory).await?; let messages = tokio::time::timeout( ACCOUNT_WORKSPACE_MESSAGES_FETCH_TIMEOUT, client.list_workspace_messages(), @@ -1485,11 +1479,7 @@ impl AccountRequestProcessor { )); } - let client = BackendClient::from_auth( - self.config.chatgpt_base_url.clone(), - &auth, - http_client_factory, - ); + let client = self.backend_client(&auth, http_client_factory).await?; match client .send_add_credits_nudge_email(Self::backend_credit_type(params.credit_type)) diff --git a/codex-rs/app-server/src/request_processors/account_processor/backend_client.rs b/codex-rs/app-server/src/request_processors/account_processor/backend_client.rs new file mode 100644 index 0000000000..9e3967ac05 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/account_processor/backend_client.rs @@ -0,0 +1,31 @@ +//! Build account API clients from the latest config and captured account authority. + +use super::*; +use futures::future::BoxFuture; + +impl AccountRequestProcessor { + // Keep config loading and runtime bootstrap out of request-handler future types. + #[inline(never)] + pub(super) fn backend_client<'a>( + &'a self, + auth: &'a CodexAuth, + http_client_factory: codex_http_client::HttpClientFactory, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + let config = self.load_latest_config().await; + let runtime = self + .thread_manager + .runtime_auth_for_config(&config) + .map_err(|error| internal_error(error.to_string()))?; + BackendClient::from_auth_with_runtime( + config.chatgpt_base_url.clone(), + auth, + http_client_factory, + runtime.as_ref(), + /*headers*/ &Default::default(), + ) + .await + .map_err(|error| internal_error(error.to_string())) + }) + } +} diff --git a/codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs b/codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs index c923429f56..5a1ee4c0e6 100644 --- a/codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs +++ b/codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs @@ -111,11 +111,7 @@ impl AccountRequestProcessor { )); } - Ok(BackendClient::from_auth( - self.config.chatgpt_base_url.clone(), - &auth, - http_client_factory, - )) + self.backend_client(&auth, http_client_factory).await } } diff --git a/codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs b/codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs index aef1593c8e..e768c8cec4 100644 --- a/codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs +++ b/codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs @@ -399,15 +399,23 @@ impl AccountRequestProcessor { let routing = if let Some(cached) = cached { cached } else { + let runtime = self + .thread_manager + .runtime_auth_for_config(&config) + .map_err(|_| WorkspaceRoutingError::DiscoveryFailed)?; let mut recovery = self.auth_manager.unauthorized_recovery(); let mut discovery_auth = auth.clone(); let mut discovery_generation = auth_state.generation; let response = loop { - let client = BackendClient::from_auth( + let client = BackendClient::from_auth_with_runtime( &config.chatgpt_base_url, &discovery_auth, config.http_client_factory(), - ); + runtime.as_ref(), + /*headers*/ &Default::default(), + ) + .await + .map_err(|_| WorkspaceRoutingError::DiscoveryFailed)?; match client.get_accounts_check().await { Ok(response) => break response, Err(error) => { diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 159641f0bd..875006ce10 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -159,8 +159,16 @@ fn load_shared_plugin_ids_by_local_path( }) } -fn remote_plugin_service_config(config: &Config) -> RemotePluginServiceConfig { - config.plugins_config_input().remote_plugin_service_config() +fn remote_plugin_service_config( + thread_manager: &ThreadManager, + config: &Config, +) -> Result { + let _ = thread_manager + .runtime_auth_for_config(config) + .map_err(|error| internal_error(error.to_string()))?; + Ok(thread_manager + .plugins_manager() + .remote_plugin_service_config(&config.plugins_config_input())) } fn share_context_for_source( @@ -599,7 +607,8 @@ impl PluginRequestProcessor { let include_global_remote = context.remote_plugins_enabled() && !explicit_marketplace_kinds; let use_remote_global_catalog = include_global_remote && auth_mode.is_some_and(DomainAuthMode::uses_codex_backend); - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; let plugin_policy = &config.plugins; let remote_catalog_cache_mode = if force_refetch { RemotePluginCatalogCacheMode::ForceRefetch @@ -1056,7 +1065,8 @@ impl PluginRequestProcessor { ); let share_context = match share_context { Some(context) => { - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; match codex_core_plugins::remote::fetch_remote_plugin_share_context( &remote_plugin_service_config, auth.as_ref(), @@ -1184,7 +1194,8 @@ impl PluginRequestProcessor { "remote plugin read is not enabled for marketplace {remote_marketplace_name}" ))); } - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; validate_remote_plugin_id(&plugin_name)?; let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail( &remote_plugin_service_config, @@ -1246,7 +1257,8 @@ impl PluginRequestProcessor { } let auth = self.auth_manager.auth().await; - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; let remote_skill_detail = codex_core_plugins::remote::fetch_remote_plugin_skill_detail( &remote_plugin_service_config, auth.as_ref(), @@ -1297,7 +1309,8 @@ impl PluginRequestProcessor { validate_client_plugin_share_targets(share_targets)?; } - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; let access_policy = codex_core_plugins::remote::RemotePluginShareAccessPolicy { discoverability: discoverability.map(remote_plugin_share_discoverability), share_targets: share_targets.map(remote_plugin_share_targets), @@ -1345,7 +1358,8 @@ impl PluginRequestProcessor { } validate_client_plugin_share_targets(&share_targets)?; - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; let result = codex_core_plugins::remote::update_remote_plugin_share_targets( &remote_plugin_service_config, auth.as_ref(), @@ -1379,7 +1393,8 @@ impl PluginRequestProcessor { _params: PluginShareListParams, ) -> Result { let (config, auth) = self.load_plugin_share_config_and_auth().await?; - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; let plugin_policy = &config.plugins; let data = codex_core_plugins::remote::list_remote_plugin_shares( &remote_plugin_service_config, @@ -1417,7 +1432,8 @@ impl PluginRequestProcessor { return Err(invalid_request("invalid remote plugin id")); } - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; let result = codex_core_plugins::remote::checkout_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), @@ -1448,7 +1464,8 @@ impl PluginRequestProcessor { return Err(invalid_request("invalid remote plugin id")); } - let remote_plugin_service_config = remote_plugin_service_config(&config); + let remote_plugin_service_config = + remote_plugin_service_config(&self.thread_manager, &config)?; codex_core_plugins::remote::delete_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), @@ -1599,6 +1616,10 @@ impl PluginRequestProcessor { install_attempt_id: Option, ) -> Result { let mut config = self.load_latest_config(/*fallback_cwd*/ None).await?; + let _ = self + .thread_manager + .runtime_auth_for_config(&config) + .map_err(|error| internal_error(error.to_string()))?; let auth = match self.auth_manager.auth_with_http_client_factory().await { Some((auth, factory)) => { config.application_network_policy = factory.network_policy().clone(); @@ -2105,6 +2126,10 @@ impl PluginRequestProcessor { plugin_id: String, ) -> Result { let config = self.load_latest_config(/*fallback_cwd*/ None).await?; + let _ = self + .thread_manager + .runtime_auth_for_config(&config) + .map_err(|error| internal_error(error.to_string()))?; let auth = self.auth_manager.auth().await; let outcome = self .thread_manager diff --git a/codex-rs/app-server/src/request_processors/plugins/search.rs b/codex-rs/app-server/src/request_processors/plugins/search.rs index b3d3de93c7..f259544ac2 100644 --- a/codex-rs/app-server/src/request_processors/plugins/search.rs +++ b/codex-rs/app-server/src/request_processors/plugins/search.rs @@ -78,7 +78,7 @@ impl PluginRequestProcessor { && let Some(remote_scope) = remote_scope { let page = search_remote_plugins( - &remote_plugin_service_config(&config), + &remote_plugin_service_config(&self.thread_manager, &config)?, auth.as_ref(), RemotePluginSearchRequest { query: search_term, diff --git a/codex-rs/app-server/tests/suite/v2/plugin_install.rs b/codex-rs/app-server/tests/suite/v2/plugin_install.rs index 19a4327b82..0fafd9cce4 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_install.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_install.rs @@ -77,6 +77,9 @@ use wiremock::matchers::method; use wiremock::matchers::path; use wiremock::matchers::query_param; +#[path = "plugin_runtime_auth.rs"] +mod runtime_auth_tests; + // Plugin install tests wait on connector discovery after the install response path // starts, which is noticeably slower on Windows CI. const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60); diff --git a/codex-rs/app-server/tests/suite/v2/plugin_runtime_auth.rs b/codex-rs/app-server/tests/suite/v2/plugin_runtime_auth.rs new file mode 100644 index 0000000000..0732ffb615 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/plugin_runtime_auth.rs @@ -0,0 +1,194 @@ +//! Cover direct plugin mutations when runtime authentication is enabled after startup. + +use super::*; +use anyhow::Context; +use base64::Engine; +use codex_app_server_protocol::PluginUninstallParams; +use codex_app_server_protocol::PluginUninstallResponse; +use core_test_support::runtime_auth::PROBE_SERVER; +use core_test_support::runtime_auth::run_harness_through_proxy; +use pretty_assertions::assert_eq; +use test_case::test_case; + +#[test_case("dual_header", "install"; "install_dual_header")] +#[test_case("assertion_only", "install"; "install_assertion_only")] +#[test_case("dual_header", "uninstall"; "uninstall_dual_header")] +#[test_case("assertion_only", "uninstall"; "uninstall_assertion_only")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn direct_mutation_initializes_runtime_after_config_reload( + mode: &str, + operation: &str, +) -> Result<()> { + core_test_support::skip_if_no_network!(Ok(())); + let token = format!( + "e30.{}.c2ln", + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode( + json!({ + "exp": 4_102_444_800_u64, + "session_id": "session-a", + "iss": "https://auth.openai.com", + "https://api.openai.com/auth": { + "chatgpt_user_id": "user-123", "chatgpt_account_id": "account-123", + }, + }) + .to_string() + ) + ); + if std::env::var(PROBE_SERVER).is_ok() { + let home = TempDir::new()?; + write_chatgpt_auth( + home.path(), + ChatGptAuthFixture::new(&token) + .account_id("account-123") + .chatgpt_user_id("user-123") + .chatgpt_account_id("account-123"), + AuthCredentialsStoreMode::File, + )?; + write_remote_plugin_catalog_config(home.path(), "https://chatgpt.com/backend-api")?; + let config_path = home.path().join("config.toml"); + let initial = std::fs::read_to_string(&config_path)?; + std::fs::write( + &config_path, + format!( + "{initial}\n[features.runtime_agent_instances]\nenabled = false\nmode = {mode:?}\n" + ), + )?; + let mut app = TestAppServer::builder() + .with_codex_home(home.path()) + .without_auto_env() + .with_env_overrides(&[ + ("OPENAI_API_KEY", None), + ("CODEX_ACCESS_TOKEN", None), + (TEST_ALLOW_HTTP_REMOTE_PLUGIN_BUNDLE_DOWNLOADS, Some("1")), + ]) + .build_initialized_with_timeout(DEFAULT_TIMEOUT) + .await?; + std::fs::write( + &config_path, + format!( + "{initial}\n[features.runtime_agent_instances]\nenabled = true\nmode = {mode:?}\n" + ), + )?; + let cache = home + .path() + .join("plugins/cache/openai-curated-remote/linear"); + match operation { + "install" => { + let id = send_remote_plugin_install_request(&mut app, REMOTE_PLUGIN_ID).await?; + let response: PluginInstallResponse = + timeout(DEFAULT_TIMEOUT, app.read_response(id)).await??; + assert_eq!( + response, + PluginInstallResponse { + auth_policy: PluginAuthPolicy::OnUse, + apps_needing_auth: Vec::new(), + } + ); + assert!(cache.join("1.2.3/.codex-plugin/plugin.json").is_file()); + } + "uninstall" => { + std::fs::create_dir_all(cache.join("1.2.3/.codex-plugin"))?; + std::fs::write( + cache.join("1.2.3/.codex-plugin/plugin.json"), + r#"{"name":"linear","version":"1.2.3"}"#, + )?; + let id = app + .send_plugin_uninstall_request(PluginUninstallParams { + plugin_id: REMOTE_PLUGIN_ID.into(), + }) + .await?; + let response: PluginUninstallResponse = + timeout(DEFAULT_TIMEOUT, app.read_response(id)).await??; + assert_eq!(response, PluginUninstallResponse {}); + assert!(!cache.exists()); + } + _ => unreachable!("test operation"), + } + return Ok(()); + } + + let server = MockServer::start().await; + let bundle = mount_remote_plugin_bundle( + &server, + /*status_code*/ 200, + remote_plugin_bundle_tar_gz_bytes("linear")?, + ) + .await; + Mock::given(method("POST")) + .and(path("/agent/register")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "runtime_agent_instance_id": "runtime-a", "agent_host_id": "host-a", + "actor": {"kind": "chatgpt", "user_id": "user-123", "account_id": "account-123"}, + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path(format!("/backend-api/ps/plugins/{REMOTE_PLUGIN_ID}"))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "id": REMOTE_PLUGIN_ID, "name": "linear", "scope": "GLOBAL", + "installation_policy": "AVAILABLE", "authentication_policy": "ON_USE", "status": "ENABLED", + "release": { + "version": "1.2.3", "bundle_download_url": bundle, "display_name": "Linear", + "description": "Track work in Linear.", + "app_ids": [], "interface": {}, "skills": [], + }, + }))) + .expect(1) + .mount(&server).await; + Mock::given(method("POST")) + .and(path(format!( + "/backend-api/ps/plugins/{REMOTE_PLUGIN_ID}/{operation}" + ))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "id": REMOTE_PLUGIN_ID, "enabled": operation == "install", + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/installed")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "plugins": [], "pagination": {"limit": 50, "next_page_token": null}, + }))) + .mount(&server) + .await; + run_harness_through_proxy( + &server, + &format!("suite::v2::plugin_install::runtime_auth_tests::direct_mutation_initializes_runtime_after_config_reload::{operation}_{mode}"), + /*realtime_target*/ None, + ).await?; + let requests = server + .received_requests() + .await + .context("recorded requests")?; + let plugin_requests = requests + .iter() + .filter(|request| { + request + .url + .path() + .starts_with(&format!("/backend-api/ps/plugins/{REMOTE_PLUGIN_ID}")) + }) + .collect::>(); + assert_eq!(plugin_requests.len(), 2); + for request in plugin_requests { + let proof = if mode == "dual_header" { + assert_eq!(request.headers["authorization"], format!("Bearer {token}")); + request.headers["x-openai-agent-assertion"].to_str()? + } else { + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + request.headers["authorization"] + .to_str()? + .strip_prefix("AgentAssertion ") + .context("assertion auth")? + }; + let claims: serde_json::Value = serde_json::from_slice( + &base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(proof.split('.').nth(2).context("assertion payload")?)?, + )?; + assert_eq!(claims["runtime_agent_instance"], "runtime-a"); + } + server.verify().await; + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/rate_limits.rs b/codex-rs/app-server/tests/suite/v2/rate_limits.rs index 76057f8ea1..9f3b2154ef 100644 --- a/codex-rs/app-server/tests/suite/v2/rate_limits.rs +++ b/codex-rs/app-server/tests/suite/v2/rate_limits.rs @@ -120,7 +120,8 @@ async fn get_account_rate_limits_returns_snapshot( let server = MockServer::start().await; let server_url = server.uri(); - write_chatgpt_base_url(codex_home.path(), &server_url)?; + let startup_server = MockServer::start().await; + write_chatgpt_base_url(codex_home.path(), &startup_server.uri())?; let primary_reset_timestamp = chrono::DateTime::parse_from_rfc3339("2025-01-01T00:02:00Z") .expect("parse primary reset timestamp") @@ -246,6 +247,14 @@ async fn get_account_rate_limits_returns_snapshot( .build_initialized_with_timeout(DEFAULT_READ_TIMEOUT) .await?; + // Account RPCs reload the effective config, but custom endpoints keep their original auth. + std::fs::write( + codex_home.path().join("config.toml"), + format!( + "chatgpt_base_url = {server_url:?}\nruntime_auth_mode = \"assertion_only\"\n[features]\nruntime_agent_instances = true\n", + ), + )?; + let request_id = mcp.send_get_account_rate_limits_request().await?; let received: GetAccountRateLimitsResponse = @@ -361,6 +370,9 @@ async fn get_account_rate_limits_returns_snapshot( }; assert_eq!(received, expected); + for request in server.received_requests().await.expect("recorded requests") { + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + } Ok(()) } diff --git a/codex-rs/cli/src/marketplace_cmd.rs b/codex-rs/cli/src/marketplace_cmd.rs index d942de63e0..6c616c4f11 100644 --- a/codex-rs/cli/src/marketplace_cmd.rs +++ b/codex-rs/cli/src/marketplace_cmd.rs @@ -5,7 +5,7 @@ use clap::Parser; use codex_core::config::Config; use codex_core::config::ConfigOverrides; use codex_core::config::LoaderOverrides; -use codex_core::plugins_manager_for_config; +use codex_core::plugins_manager_for_config_with_runtime; use codex_core_plugins::PluginMarketplaceUpgradeOutcome; use codex_core_plugins::PluginsConfigInput; use codex_core_plugins::installed_marketplaces::marketplace_install_root; @@ -211,7 +211,9 @@ impl JsonMarketplaceAddOutput { } async fn run_list(config: Config, args: ListMarketplaceArgs) -> Result<()> { - let manager = plugins_manager_for_config(&config, load_cli_auth_manager(&config).await?); + let manager = + plugins_manager_for_config_with_runtime(&config, load_cli_auth_manager(&config).await?) + .await?; let plugins_input = config.plugins_config_input(); let marketplace_listing = manager .discover_marketplaces_for_config(&plugins_input, &[]) @@ -373,7 +375,9 @@ async fn run_upgrade( marketplace_name, json, } = args; - let manager = plugins_manager_for_config(&config, load_cli_auth_manager(&config).await?); + let manager = + plugins_manager_for_config_with_runtime(&config, load_cli_auth_manager(&config).await?) + .await?; let plugins_input = config.plugins_config_input(); let runtime = tokio::runtime::Handle::current(); let reload_config: codex_core_plugins::ConfigLayerReload = std::sync::Arc::new(move || { diff --git a/codex-rs/cli/src/mcp_cmd.rs b/codex-rs/cli/src/mcp_cmd.rs index 1915cb2abf..cf65a9bef5 100644 --- a/codex-rs/cli/src/mcp_cmd.rs +++ b/codex-rs/cli/src/mcp_cmd.rs @@ -19,7 +19,7 @@ use codex_core::config::LoaderOverrides; use codex_core::config::edit::ConfigEditsBuilder; use codex_core::config::find_codex_home; use codex_core::config::load_global_mcp_servers; -use codex_core::plugins_manager_for_config; +use codex_core::plugins_manager_for_config_with_runtime; use codex_exec_server::EnvironmentManager; use codex_exec_server::HttpClient; use codex_exec_server::RouteAwareHttpClient; @@ -533,10 +533,10 @@ async fn run_remove(config_overrides: &CliConfigOverrides, remove_args: RemoveAr } async fn load_mcp_manager(config: &Config) -> Result { - let plugins_manager = Arc::new(plugins_manager_for_config( - config, - load_cli_auth_manager(config).await?, - )); + let plugins_manager = Arc::new( + plugins_manager_for_config_with_runtime(config, load_cli_auth_manager(config).await?) + .await?, + ); Ok(McpManager::new(plugins_manager)) } diff --git a/codex-rs/cli/src/plugin_cmd.rs b/codex-rs/cli/src/plugin_cmd.rs index e32c1d8d19..e82274267a 100644 --- a/codex-rs/cli/src/plugin_cmd.rs +++ b/codex-rs/cli/src/plugin_cmd.rs @@ -7,7 +7,7 @@ use codex_app_server_protocol::PluginAuthPolicy; use codex_app_server_protocol::PluginInstallPolicy; use codex_core::config::Config; use codex_core::config::find_codex_home; -use codex_core::plugins_manager_for_config; +use codex_core::plugins_manager_for_config_with_runtime; use codex_core_plugins::ConfiguredMarketplace; use codex_core_plugins::OPENAI_BUNDLED_MARKETPLACE_NAME; use codex_core_plugins::PluginInstallOutcome; @@ -773,10 +773,9 @@ async fn load_plugin_command_context( .context("failed to load configuration")?; let plugins_input = config.plugins_config_input(); let auth_manager = load_cli_auth_manager(&config).await?; - let manager = Arc::new(plugins_manager_for_config( - &config, - Arc::clone(&auth_manager), - )); + let manager = Arc::new( + plugins_manager_for_config_with_runtime(&config, Arc::clone(&auth_manager)).await?, + ); Ok(PluginCommandContext { codex_home: codex_home.to_path_buf(), plugins_input, @@ -861,7 +860,9 @@ async fn fetch_remote_marketplaces( ); return Ok(RemoteMarketplaceListing::default()); } - let service = context.plugins_input.remote_plugin_service_config(); + let service = context + .manager + .remote_plugin_service_config(&context.plugins_input); let result = if context.plugins_input.remote_plugin_enabled { remote::fetch_remote_marketplaces( &service, diff --git a/codex-rs/codex-api/src/files.rs b/codex-rs/codex-api/src/files.rs index 3d2376bb52..b7fcdb6e56 100644 --- a/codex-rs/codex-api/src/files.rs +++ b/codex-rs/codex-api/src/files.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::future::Future; use std::time::Duration; @@ -106,6 +107,7 @@ pub enum OpenAiFileError { struct CreateFileResponse { file_id: String, upload_url: String, + upload_headers: Option>, #[serde(default)] pdf_c2pa_reservation: bool, } @@ -237,12 +239,18 @@ where if remaining.is_zero() { return Err(request_error(RouteAwareRequestError::Timeout)); } - client_pool + let mut request = client_pool .put(&create_payload.upload_url) .timeout(remaining) .header("x-ms-blob-type", "BlockBlob") .header("x-ms-client-request-id", &azure_client_request_id) - .header(CONTENT_LENGTH, file_size_bytes) + .header(CONTENT_LENGTH, file_size_bytes); + if let Some(headers) = &create_payload.upload_headers { + for (name, value) in headers { + request = request.header(name.as_str(), value.as_str()); + } + } + request .body_stream(contents) .send() .await @@ -538,20 +546,36 @@ mod tests { Mock::given(method("POST")) .and(path("/backend-api/files")) .and(header("chatgpt-account-id", "account_id")) + .and(|request: &Request| !request.headers.contains_key("x-amz-tagging")) .and(body_json(serde_json::json!({ "file_name": "hello.txt", "file_size": 5, "use_case": "codex", }))) - .respond_with( - ResponseTemplate::new(200) - .set_body_json(serde_json::json!({"file_id": "file_123", "upload_url": format!("{}/upload/file_123", server.uri())})), - ) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "file_id": "file_123", + "upload_url": format!("{}/upload/file_123", server.uri()), + "upload_headers": { + "Content-Type": "text/plain", + "Content-Disposition": "attachment; filename=hello.txt", + "x-amz-tagging": "purpose=codex", + }, + }))) .mount(&server) .await; Mock::given(method("PUT")) .and(path("/upload/file_123")) .and(header("content-length", "5")) + .and(header("content-type", "text/plain")) + .and(header( + "content-disposition", + "attachment; filename=hello.txt", + )) + .and(header("x-amz-tagging", "purpose=codex")) + .and(|request: &Request| { + !request.headers.contains_key("authorization") + && !request.headers.contains_key("chatgpt-account-id") + }) .and(header_regex("x-ms-client-request-id", "^[0-9a-f-]{36}$")) .respond_with(ResponseTemplate::new(200)) .mount(&server) @@ -561,6 +585,7 @@ mod tests { let download_url = format!("{}/download/file_123", server.uri()); Mock::given(method("POST")) .and(path("/backend-api/files/file_123/uploaded")) + .and(|request: &Request| !request.headers.contains_key("x-amz-tagging")) .respond_with(move |_request: &Request| { if finalize_attempts_responder.fetch_add(1, Ordering::SeqCst) == 0 { return ResponseTemplate::new(200).set_body_json(serde_json::json!({ @@ -622,6 +647,7 @@ mod tests { .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "file_id": "file_pdf", "upload_url": format!("{}/upload/file_pdf", server.uri()), + "upload_headers": null, "pdf_c2pa_reservation": true, }))) .expect(1) diff --git a/codex-rs/core-plugins/Cargo.toml b/codex-rs/core-plugins/Cargo.toml index 3025c3a62b..2c190c07bf 100644 --- a/codex-rs/core-plugins/Cargo.toml +++ b/codex-rs/core-plugins/Cargo.toml @@ -15,12 +15,15 @@ workspace = true [dependencies] codex-utils-process = { workspace = true } anyhow = { workspace = true } +chrono = { workspace = true } codex-analytics = { workspace = true } +codex-api = { workspace = true } codex-app-server-protocol = { workspace = true } codex-config = { workspace = true } codex-connectors = { workspace = true } codex-core-plugin-common = { workspace = true } codex-exec-server = { workspace = true } +codex-features = { workspace = true } codex-git-utils = { workspace = true } codex-hooks = { workspace = true } codex-http-client = { workspace = true } @@ -30,14 +33,14 @@ codex-model-provider = { workspace = true } codex-otel = { workspace = true } codex-plugin = { workspace = true } codex-protocol = { workspace = true } -codex-skills = { workspace = true } +codex-runtime-agent-instance = { workspace = true } codex-shell-command = { workspace = true } +codex-skills = { workspace = true } codex-tools = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-path = { workspace = true } codex-utils-path-uri = { workspace = true } codex-utils-plugins = { workspace = true } -chrono = { workspace = true } dirs = { workspace = true } flate2 = { workspace = true } futures = { workspace = true } @@ -59,6 +62,7 @@ uuid = { workspace = true, features = ["v4"] } zip = { workspace = true } [dev-dependencies] +base64 = { workspace = true } codex-exec-server-test-support = { workspace = true } codex-utils-cargo-bin = { workspace = true } libc = { workspace = true } diff --git a/codex-rs/core-plugins/src/http_client_selector.rs b/codex-rs/core-plugins/src/http_client_selector.rs index f42abea4cc..0650f2a55a 100644 --- a/codex-rs/core-plugins/src/http_client_selector.rs +++ b/codex-rs/core-plugins/src/http_client_selector.rs @@ -1,23 +1,45 @@ +use codex_http_client::NetworkPolicy; use codex_http_client::OutboundProxyPolicy; use codex_http_client::RequestBuilder; use codex_http_client::RouteAwareClientPool; use http::Method; use std::fmt::Debug; +use std::sync::Arc; /// Builds requests whose URL is also used to resolve their outbound route. /// /// Implementations must keep route selection coupled to the request URL. Returning a transport /// client would let callers send a different URL than the one used for route selection. pub(crate) trait HttpClientSelector: Debug + Send + Sync { + fn for_current_account(&self) -> Arc; fn request(&self, method: Method, url: &str) -> RequestBuilder; + fn request_for_account( + &self, + method: Method, + url: &str, + binding: &NetworkPolicy, + ) -> Option; fn outbound_proxy_policy(&self) -> OutboundProxyPolicy; } impl HttpClientSelector for RouteAwareClientPool { + fn for_current_account(&self) -> Arc { + Arc::new(RouteAwareClientPool::for_current_account(self)) + } + fn request(&self, method: Method, url: &str) -> RequestBuilder { RouteAwareClientPool::request(self, method, url) } + fn request_for_account( + &self, + method: Method, + url: &str, + binding: &NetworkPolicy, + ) -> Option { + Some(self.with_account_binding(binding)?.request(method, url)) + } + fn outbound_proxy_policy(&self) -> OutboundProxyPolicy { RouteAwareClientPool::outbound_proxy_policy(self) } diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 3fcc3fdf49..8558837809 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -108,6 +108,7 @@ use codex_config::types::ToolSuggestDisabledTool; use codex_config::types::ToolSuggestDiscoverableType; use codex_connectors::ConnectorSnapshot; use codex_connectors::PluginConnectorSource; +use codex_features::RuntimeAuthMode; use codex_hooks::plugin_hook_declarations; use codex_http_client::ClientRouteClass; use codex_http_client::HttpClientFactory; @@ -124,6 +125,7 @@ use codex_plugin::prompt_safe_plugin_description; use codex_protocol::auth::AuthMode; use codex_protocol::protocol::HookEventName; use codex_protocol::protocol::Product; +use codex_runtime_agent_instance::RuntimeAuthSlot; use codex_skills::SkillMetadata; use codex_skills::SkillRootLoader; use codex_skills::SkillRootSnapshots; @@ -175,6 +177,8 @@ pub struct PluginsConfigInput { pub product_sku: Option, http_client_factory: HttpClientFactory, remote_http_clients: Arc>>, + runtime_http_clients: Arc>>, + runtime_auth_mode: Option, } impl PluginsConfigInput { @@ -200,16 +204,40 @@ impl PluginsConfigInput { http_client_factory, product_sku, remote_http_clients: Arc::new(OnceLock::new()), + runtime_http_clients: Arc::new(OnceLock::new()), + runtime_auth_mode: None, } } - /// Shares a lazy route-aware connection pool across this input and its clones. - /// Request metadata remains current; new config inputs own a new pool. - pub fn remote_plugin_service_config(&self) -> RemotePluginServiceConfig { - let http_clients = self.remote_http_clients.get_or_init(|| { + pub fn with_runtime_auth_mode(mut self, mode: Option) -> Self { + self.runtime_auth_mode = mode; + self + } + + /// Shares lazy route-aware connection pools across this input and its clones. + /// Runtime mode strips sensitive headers on cross-origin redirects. + pub fn remote_plugin_service_config( + &self, + runtime_auth: Option<(RuntimeAuthSlot, RuntimeAuthMode)>, + ) -> RemotePluginServiceConfig { + let runtime_auth = runtime_auth.filter(|_| { + codex_http_client::classify_openai_api_url(&self.chatgpt_base_url) + != codex_http_client::OpenAiDestination::Other + }); + let (http_clients, factory) = if runtime_auth.is_some() { + ( + &self.runtime_http_clients, + self.http_client_factory + .clone() + .with_sensitive_header_protection(), + ) + } else { + (&self.remote_http_clients, self.http_client_factory.clone()) + }; + let http_clients = http_clients.get_or_init(|| { Arc::new( RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging( - self.http_client_factory.clone(), + factory, ClientRouteClass::Api, ), ) @@ -221,6 +249,7 @@ impl PluginsConfigInput { .clone() .unwrap_or_else(|| CODEX_PRODUCT_SKU.to_string()), http_clients: Arc::clone(http_clients), + runtime_auth, } } } @@ -424,10 +453,6 @@ struct ConfiguredMarketplaceUpgradeState { in_flight: bool, } -fn remote_plugin_service_config(config: &PluginsConfigInput) -> RemotePluginServiceConfig { - config.remote_plugin_service_config() -} - fn featured_plugin_ids_cache_key( config: &PluginsConfigInput, auth: Option<&CodexAuth>, @@ -583,6 +608,7 @@ pub struct PluginsManager { remote_catalog_cache_refresh_state: RwLock, restriction_product: Option, auth_manager: Arc, + runtime_auth: RuntimeAuthSlot, analytics_events_client: RwLock>, plugin_install_source: PluginInstallSource, } @@ -692,6 +718,34 @@ fn target_curated_marketplace(auth_mode: Option) -> TargetCuratedMarke } impl PluginsManager { + /// Build remote service state from the request's effective mode and this harness's runtime. + pub fn remote_plugin_service_config( + &self, + config: &PluginsConfigInput, + ) -> RemotePluginServiceConfig { + config.remote_plugin_service_config( + config + .runtime_auth_mode + .map(|mode| (Arc::clone(&self.runtime_auth), mode)), + ) + } + + /// Reuse host configuration without sharing another actor's remote caches or credentials. + pub fn for_auth_context( + &self, + auth_manager: Arc, + runtime_auth: RuntimeAuthSlot, + ) -> Self { + Self::new_with_options( + self.codex_home.clone(), + self.restriction_product, + auth_manager, + runtime_auth, + Arc::clone(&self.skill_root_loader), + ) + .with_plugin_install_source(self.plugin_install_source) + } + pub fn new( codex_home: PathBuf, auth_manager: Arc, @@ -701,6 +755,7 @@ impl PluginsManager { codex_home, Some(Product::Codex), auth_manager, + RuntimeAuthSlot::default(), skill_root_loader, ) } @@ -709,6 +764,7 @@ impl PluginsManager { codex_home: PathBuf, restriction_product: Option, auth_manager: Arc, + runtime_auth: RuntimeAuthSlot, skill_root_loader: Arc>, ) -> Self { // Product restrictions are enforced at marketplace admission time for a given CODEX_HOME: @@ -752,6 +808,7 @@ impl PluginsManager { ), restriction_product, auth_manager, + runtime_auth, analytics_events_client: RwLock::new(None), plugin_install_source: PluginInstallSource::Manual, } @@ -1333,7 +1390,7 @@ impl PluginsManager { crate::remote::cached_global_remote_discoverable_plugins( self.codex_home.as_path(), - &remote_plugin_service_config(config), + &self.remote_plugin_service_config(config), auth, ) } @@ -1368,7 +1425,7 @@ impl PluginsManager { .prepare_remote_installed_plugins_cache_generation(auth) .ok_or(RemotePluginCatalogError::AuthRequired)?; let plugins = crate::remote::fetch_remote_installed_plugins_with_display_assets( - &remote_plugin_service_config(config), + &self.remote_plugin_service_config(config), auth, display_assets, ) @@ -1659,7 +1716,7 @@ impl PluginsManager { self.schedule_remote_installed_plugins_cache_refresh( RemoteInstalledPluginsCacheRefreshRequest { generation, - service_config: remote_plugin_service_config(config), + service_config: self.remote_plugin_service_config(config), auth, notify, on_effective_plugins_changed, @@ -1796,7 +1853,7 @@ impl PluginsManager { .collect::>(); let result = crate::remote::sync_remote_installed_plugin_bundles_once_with_snapshot( self.codex_home.clone(), - &remote_plugin_service_config(config), + &self.remote_plugin_service_config(config), auth, &previous_plugin_ids, ) @@ -1866,7 +1923,7 @@ impl PluginsManager { } self.schedule_remote_catalog_cache_refresh(RemoteCatalogCacheRefreshRequest { - service_config: remote_plugin_service_config(config), + service_config: self.remote_plugin_service_config(config), auth, scopes, mode, @@ -1962,7 +2019,7 @@ impl PluginsManager { return Ok(featured_plugin_ids); } let featured_plugin_ids = crate::remote_legacy::fetch_remote_featured_plugin_ids( - &remote_plugin_service_config(config), + &self.remote_plugin_service_config(config), auth, self.restriction_product, ) @@ -2013,7 +2070,7 @@ impl PluginsManager { let mode = refresh .get_or_init(|| async { match crate::remote::fetch_recommended_plugins( - &remote_plugin_service_config(config), + &self.remote_plugin_service_config(config), auth, ) .await @@ -2208,7 +2265,7 @@ impl PluginsManager { let plugin_id = resolved.plugin_id.as_key(); // This only forwards the backend mutation before the local install flow. if let Err(err) = crate::remote_legacy::enable_remote_plugin( - &remote_plugin_service_config(config), + &self.remote_plugin_service_config(config), auth, &plugin_id, ) @@ -2416,7 +2473,7 @@ impl PluginsManager { let plugin_key = plugin_id.as_key(); // This only forwards the backend mutation before the local uninstall flow. crate::remote_legacy::uninstall_remote_plugin( - &remote_plugin_service_config(config), + &self.remote_plugin_service_config(config), auth, &plugin_key, ) @@ -3004,7 +3061,7 @@ impl PluginsManager { }; let mut scopes = crate::remote::cached_remote_plugin_catalog_scopes( manager.codex_home.as_path(), - &remote_plugin_service_config(&config_for_remote_sync), + &manager.remote_plugin_service_config(&config_for_remote_sync), auth.as_ref(), scopes_to_probe, ); diff --git a/codex-rs/core-plugins/src/plugin_measurement_references.rs b/codex-rs/core-plugins/src/plugin_measurement_references.rs index 9c04db968f..4d3a8dce52 100644 --- a/codex-rs/core-plugins/src/plugin_measurement_references.rs +++ b/codex-rs/core-plugins/src/plugin_measurement_references.rs @@ -7,8 +7,10 @@ use crate::remote::fetch_measurement_reference_bundle; use crate::remote_bundle::download_remote_plugin_bundle; use crate::remote_bundle::install_remote_plugin_bundle; use crate::script_attribution::PluginMeasurementTarget; +use codex_features::RuntimeAuthMode; use codex_login::CodexAuth; use codex_plugin::PluginId; +use codex_runtime_agent_instance::RuntimeAuthSlot; use codex_utils_path_uri::PathConvention; use std::sync::Arc; use tempfile::TempDir; @@ -84,6 +86,7 @@ impl RemotePluginMeasurementCache { config: &PluginsConfigInput, auth: &CodexAuth, target: &PluginMeasurementTarget, + runtime_auth: Option<(RuntimeAuthSlot, RuntimeAuthMode)>, ) -> anyhow::Result>> { if !config.plugins_enabled || !config.remote_plugin_enabled || !auth.uses_codex_backend() { return Ok(None); @@ -92,7 +95,9 @@ impl RemotePluginMeasurementCache { // authoritative for enabled state, including default installations. // GLOBAL plugins have already passed server-side admission; local // marketplace product policy and skill capability loading are separate. - let service = config.remote_plugin_service_config(); + let service = config + .remote_plugin_service_config(runtime_auth) + .for_current_account(); let Some((remote_id, bundle)) = fetch_measurement_reference_bundle(&service, auth, target) .await .map_err(|_| anyhow::anyhow!("plugin measurement catalog lookup failed"))? diff --git a/codex-rs/core-plugins/src/plugin_measurement_references_tests.rs b/codex-rs/core-plugins/src/plugin_measurement_references_tests.rs index 3966c494e7..6902fcf7f2 100644 --- a/codex-rs/core-plugins/src/plugin_measurement_references_tests.rs +++ b/codex-rs/core-plugins/src/plugin_measurement_references_tests.rs @@ -174,7 +174,9 @@ impl Fixture { &self, target: &PluginMeasurementTarget, ) -> Result>> { - self.cache.prepare(&self.config, &self.auth, target).await + self.cache + .prepare(&self.config, &self.auth, target, /*runtime_auth*/ None) + .await } async fn assert_request_count(&self, expected: usize) { @@ -254,7 +256,7 @@ async fn narrow_preparation_reuses_versions_and_keeps_old_leases_alive() -> Resu // A newer release cannot supply an older version's declaration on a cold cache. assert!( RemotePluginMeasurementCache::new() - .prepare(&f.config, &f.auth, &first) + .prepare(&f.config, &f.auth, &first, /*runtime_auth*/ None) .await? .is_none() ); diff --git a/codex-rs/core-plugins/src/plugins_config_input_tests.rs b/codex-rs/core-plugins/src/plugins_config_input_tests.rs index 69cadf4510..73a1840833 100644 --- a/codex-rs/core-plugins/src/plugins_config_input_tests.rs +++ b/codex-rs/core-plugins/src/plugins_config_input_tests.rs @@ -28,12 +28,14 @@ fn repeated_service_configs_and_early_clones_share_one_lazy_pool() { ); let clone = input.clone(); assert!(input.remote_http_clients.get().is_none()); - let first = input.remote_plugin_service_config(); + let first = input.remote_plugin_service_config(/*runtime_auth*/ None); let reuse: Vec<_> = (0..220) .map(|_| { Arc::ptr_eq( &first.http_clients, - &clone.remote_plugin_service_config().http_clients, + &clone + .remote_plugin_service_config(/*runtime_auth*/ None) + .http_clients, ) }) .collect(); @@ -52,7 +54,9 @@ fn repeated_service_configs_and_early_clones_share_one_lazy_pool() { reuse, Arc::ptr_eq( &first.http_clients, - &new_input.remote_plugin_service_config().http_clients, + &new_input + .remote_plugin_service_config(/*runtime_auth*/ None) + .http_clients, ), ), (vec![true; 220], false) @@ -95,7 +99,7 @@ async fn reused_pool_uses_current_endpoint_product_and_authentication() { test_http_client_factory(), /*product_sku*/ None, ); - let first = input.remote_plugin_service_config(); + let first = input.remote_plugin_service_config(/*runtime_auth*/ None); let first_auth = CodexAuth::from_external_chatgpt_tokens( "header.e30.first", "first-account", @@ -108,7 +112,7 @@ async fn reused_pool_uses_current_endpoint_product_and_authentication() { input.chatgpt_base_url = format!("{}/backend-api", second_server.uri()); input.product_sku = Some("updated".to_string()); - let second = input.remote_plugin_service_config(); + let second = input.remote_plugin_service_config(/*runtime_auth*/ None); let second_auth = CodexAuth::from_external_chatgpt_tokens( "header.e30.second", "second-account", diff --git a/codex-rs/core-plugins/src/recommended_plugin_install.rs b/codex-rs/core-plugins/src/recommended_plugin_install.rs index b63846d3f3..16ab4a46fb 100644 --- a/codex-rs/core-plugins/src/recommended_plugin_install.rs +++ b/codex-rs/core-plugins/src/recommended_plugin_install.rs @@ -1,10 +1,10 @@ -use crate::PluginsConfigInput; use crate::remote::RemotePluginCatalogError; +use crate::remote::RemotePluginServiceConfig; use codex_login::CodexAuth; use codex_tools::DiscoverableTool; pub async fn hydrate_selected_recommended_plugin_install_metadata( - config: &PluginsConfigInput, + config: &RemotePluginServiceConfig, auth: Option<&CodexAuth>, mut tool: DiscoverableTool, ) -> Result, RemotePluginCatalogError> { @@ -17,12 +17,9 @@ pub async fn hydrate_selected_recommended_plugin_install_metadata( plugin.id )) })?; - let Some(app_connector_ids) = crate::remote::fetch_recommended_plugin_install_metadata( - &config.remote_plugin_service_config(), - auth, - &remote_plugin_id, - ) - .await? + let Some(app_connector_ids) = + crate::remote::fetch_recommended_plugin_install_metadata(config, auth, &remote_plugin_id) + .await? else { return Ok(None); }; diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index d51d40a122..6d06b3b503 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -19,6 +19,7 @@ use codex_app_server_protocol::PluginInstallPolicySource; use codex_app_server_protocol::PluginInterface; use codex_app_server_protocol::ScheduledTaskSummary; use codex_app_server_protocol::SkillInterface; +use codex_features::RuntimeAuthMode; use codex_http_client::ClientRouteClass; use codex_http_client::HttpClientFactory; use codex_http_client::RequestBuilder; @@ -32,7 +33,9 @@ use codex_plugin::PluginCapabilitySummary; use codex_plugin::PluginId; use codex_plugin::app_connector_ids_from_declarations; use codex_plugin::prompt_safe_plugin_description; +use codex_runtime_agent_instance::RuntimeAuthSlot; use codex_utils_absolute_path::AbsolutePathBuf; +use http::HeaderMap; use http::Method; use http::StatusCode; use serde::Deserialize; @@ -153,13 +156,20 @@ pub struct RemotePluginServiceConfig { pub chatgpt_base_url: String, pub(crate) product_sku: String, pub(crate) http_clients: Arc, + pub(crate) runtime_auth: Option<(RuntimeAuthSlot, RuntimeAuthMode)>, } impl RemotePluginServiceConfig { + /// Keeps metadata and subsequent storage requests on the account that starts the operation. + pub(crate) fn for_current_account(&self) -> Self { + Self { + http_clients: self.http_clients.for_current_account(), + ..self.clone() + } + } + /// Creates remote plugin service state from the effective application HTTP configuration. - /// - /// Keeping the factory mandatory ensures every catalog, mutation, upload, and bundle request - /// follows the same outbound proxy policy. + /// Every catalog, mutation, upload, and bundle request retains the configured proxy policy. pub fn new( chatgpt_base_url: String, http_client_factory: HttpClientFactory, @@ -170,13 +180,109 @@ impl RemotePluginServiceConfig { http_client_factory, ClientRouteClass::Api, ); + let http_clients: Arc = Arc::new(http_clients); Self { chatgpt_base_url, product_sku: product_sku.unwrap_or_else(|| CODEX_PRODUCT_SKU.to_string()), - http_clients: Arc::new(http_clients), + http_clients, + runtime_auth: None, } } + #[cfg(test)] + pub(crate) fn new_with_runtime_auth( + chatgpt_base_url: String, + http_client_factory: HttpClientFactory, + product_sku: Option, + runtime_auth: (RuntimeAuthSlot, RuntimeAuthMode), + ) -> Self { + if codex_http_client::classify_openai_api_url(&chatgpt_base_url) + == codex_http_client::OpenAiDestination::Other + { + return Self::new(chatgpt_base_url, http_client_factory, product_sku); + } + Self::new_with_selected_runtime_auth( + chatgpt_base_url, + http_client_factory, + product_sku, + runtime_auth, + ) + } + + #[cfg(test)] + pub(crate) fn new_with_selected_runtime_auth( + chatgpt_base_url: String, + http_client_factory: HttpClientFactory, + product_sku: Option, + runtime_auth: (RuntimeAuthSlot, RuntimeAuthMode), + ) -> Self { + let mut config = Self::new( + chatgpt_base_url, + http_client_factory.with_sensitive_header_protection(), + product_sku, + ); + config.runtime_auth = Some(runtime_auth); + config + } + + // Keep runtime registration out of callers' async stack frames. + pub(crate) fn authenticated_request<'a>( + &'a self, + method: Method, + url: &'a str, + auth: &'a CodexAuth, + ) -> std::pin::Pin< + Box< + dyn std::future::Future> + + Send + + 'a, + >, + > { + Box::pin(async move { + if self.runtime_auth.is_some() { + codex_model_provider::validate_runtime_destination(Some(&self.chatgpt_base_url)) + .map_err(|error| codex_api::AuthError::Build(error.to_string()))?; + } + let runtime = self + .runtime_auth + .as_ref() + .map(|(slot, mode)| { + slot.get() + .map(|manager| (Arc::clone(manager), *mode)) + .ok_or_else(|| { + codex_api::AuthError::Build( + "runtime authentication has not been initialized".to_string(), + ) + }) + }) + .transpose()?; + let resolved = codex_model_provider::auth_provider_for_request( + auth, + runtime.as_ref(), + &HeaderMap::new(), + ) + .await?; + let request = match &resolved.runtime { + Some(runtime) => self + .http_clients + .request_for_account( + method, + url, + runtime.http_client_factory().network_policy(), + ) + .ok_or_else(|| { + codex_api::AuthError::Build( + "plugin credentials and transport must share the application network policy" + .to_string(), + ) + })?, + None => self.http_clients.request(method, url), + }; + let headers = resolved.auth.apply_auth_headers(HeaderMap::new()).await?; + Ok(request.headers(default_headers()).headers(headers)) + }) + } + pub(crate) fn http_request(&self, method: Method, url: &str) -> RequestBuilder { self.http_clients .request(method, url) @@ -188,6 +294,13 @@ impl PartialEq for RemotePluginServiceConfig { fn eq(&self, other: &Self) -> bool { self.chatgpt_base_url == other.chatgpt_base_url && self.product_sku == other.product_sku + && match (&self.runtime_auth, &other.runtime_auth) { + (Some((left, left_mode)), Some((right, right_mode))) => { + Arc::ptr_eq(left, right) && left_mode == right_mode + } + (None, None) => true, + _ => false, + } && self.http_clients.outbound_proxy_policy() == other.http_clients.outbound_proxy_policy() } @@ -1045,12 +1158,9 @@ pub async fn fetch_recommended_plugins( .map_err(RemotePluginCatalogError::InvalidBaseUrl)?; url.query_pairs_mut().append_pair("scope", "GLOBAL"); let url = url.to_string(); - let request = authenticated_request( - config.http_request(Method::GET, &url), - auth, - &config.product_sku, - )? - .timeout(RECOMMENDED_PLUGINS_TIMEOUT); + let request = authenticated_request(config, Method::GET, &url, auth) + .await? + .timeout(RECOMMENDED_PLUGINS_TIMEOUT); let response: RecommendedPluginsResponse = send_and_decode(request, &url).await?; Ok(recommended_plugins_mode(response)) } @@ -1445,11 +1555,7 @@ pub async fn fetch_remote_plugin_skill_detail( } let url = remote_plugin_skill_detail_url(config, plugin_id, skill_name)?; - let request = authenticated_request( - config.http_request(Method::GET, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::GET, &url, auth).await?; let response: RemotePluginSkillDetailResponse = send_and_decode(request, &url).await?; if response.plugin_id != plugin_id { return Err(RemotePluginCatalogError::UnexpectedPluginId { @@ -1656,11 +1762,7 @@ async fn install_remote_plugin_inner( url.query_pairs_mut() .append_pair("includeAppsNeedingAuth", "true"); let url = url.to_string(); - let request = authenticated_request( - config.http_request(Method::POST, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::POST, &url, auth).await?; let request = if let Some(install_attempt_id) = install_attempt_id { request.json(&RemotePluginInstallRequest { install_attempt_id }) } else { @@ -1753,11 +1855,7 @@ pub async fn uninstall_remote_plugin( let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/ps/plugins/{remote_plugin_id}/uninstall"); - let request = authenticated_request( - config.http_request(Method::POST, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::POST, &url, auth).await?; let response: RemotePluginMutationResponse = send_and_decode(request, &url).await?; if response.id != remote_plugin_id { return Err(RemotePluginCatalogError::UnexpectedPluginId { @@ -2265,11 +2363,7 @@ async fn get_remote_plugin_list_page( url.query_pairs_mut().append_pair("pageToken", page_token); } let url = url.to_string(); - let request = authenticated_request( - config.http_request(Method::GET, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::GET, &url, auth).await?; send_and_decode(request, &url).await } @@ -2287,11 +2381,7 @@ async fn get_remote_shared_workspace_plugins_page( url.query_pairs_mut().append_pair("pageToken", page_token); } let url = url.to_string(); - let request = authenticated_request( - config.http_request(Method::GET, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::GET, &url, auth).await?; send_and_decode(request, &url).await } @@ -2324,11 +2414,7 @@ async fn get_remote_plugin_installed_page( url.query_pairs_mut().append_pair("pageToken", page_token); } let url = url.to_string(); - let request = authenticated_request( - config.http_request(Method::GET, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::GET, &url, auth).await?; send_and_decode(request, &url).await } @@ -2363,12 +2449,9 @@ async fn fetch_plugin_detail_with_timeout( .append_pair("includeDownloadUrls", "true"); } let url = url.to_string(); - let request = authenticated_request( - config.http_request(Method::GET, &url), - auth, - &config.product_sku, - )? - .timeout(timeout); + let request = authenticated_request(config, Method::GET, &url, auth) + .await? + .timeout(timeout); send_and_decode(request, &url).await } @@ -2403,20 +2486,19 @@ fn ensure_chatgpt_auth(auth: Option<&CodexAuth>) -> Result<&CodexAuth, RemotePlu Ok(auth) } -fn authenticated_request( - request: RequestBuilder, +async fn authenticated_request( + config: &RemotePluginServiceConfig, + method: Method, + url: &str, auth: &CodexAuth, - product_sku: &str, ) -> Result { - Ok(request + Ok(config + .authenticated_request(method, url, auth) + .await + .map_err(std::io::Error::other) + .map_err(RemotePluginCatalogError::AuthToken)? .timeout(REMOTE_PLUGIN_CATALOG_TIMEOUT) - .headers( - codex_model_provider::auth_provider_from_auth(auth) - .to_auth_headers() - .map_err(std::io::Error::other) - .map_err(RemotePluginCatalogError::AuthToken)?, - ) - .header(OAI_PRODUCT_SKU_HEADER, product_sku)) + .header(OAI_PRODUCT_SKU_HEADER, &config.product_sku)) } async fn send_and_decode Deserialize<'de>>( @@ -2451,3 +2533,7 @@ async fn send_and_decode Deserialize<'de>>( source, }) } + +#[cfg(test)] +#[path = "remote_runtime_auth_tests.rs"] +mod runtime_auth_tests; diff --git a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs index 89f5681d29..00748af827 100644 --- a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs +++ b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs @@ -136,6 +136,7 @@ pub(crate) async fn sync_remote_installed_plugin_bundles_once_with_snapshot( previous_plugin_ids: &[PluginId], ) -> Result { let auth = ensure_chatgpt_auth(auth)?; + let config = &config.for_current_account(); let authenticated_account_id = auth.get_account_id(); let fetched_installed_plugins = fetch_installed_plugins( config, diff --git a/codex-rs/core-plugins/src/remote/search.rs b/codex-rs/core-plugins/src/remote/search.rs index 71a56f9425..2d9c518897 100644 --- a/codex-rs/core-plugins/src/remote/search.rs +++ b/codex-rs/core-plugins/src/remote/search.rs @@ -60,11 +60,7 @@ pub async fn search_remote_plugins( } let url = url.to_string(); - let request = authenticated_request( - config.http_request(Method::GET, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::GET, &url, auth).await?; let response: RemotePluginListResponse = send_and_decode(request, &url_for_error) .await .map_err(|error| match error { diff --git a/codex-rs/core-plugins/src/remote/share.rs b/codex-rs/core-plugins/src/remote/share.rs index f5aa54937a..98ee12aef0 100644 --- a/codex-rs/core-plugins/src/remote/share.rs +++ b/codex-rs/core-plugins/src/remote/share.rs @@ -149,6 +149,7 @@ pub async fn save_remote_plugin_share( access_policy: RemotePluginShareAccessPolicy, ) -> Result { let auth = ensure_chatgpt_auth(auth)?; + let config = &config.for_current_account(); let plugin_path_for_archive = plugin_path.as_path().to_path_buf(); let (filename, archive_bytes) = tokio::task::spawn_blocking(move || { let filename = archive_filename(&plugin_path_for_archive)?; @@ -285,11 +286,7 @@ pub async fn delete_remote_plugin_share( let auth = ensure_chatgpt_auth(auth)?; let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/public/plugins/workspace/{remote_plugin_id}"); - let request = authenticated_request( - config.http_request(Method::DELETE, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::DELETE, &url, auth).await?; send_and_expect_status(request, &url, &[StatusCode::NO_CONTENT]).await?; if let Err(err) = local_paths::remove_plugin_share_local_path(codex_home, remote_plugin_id) { warn!( @@ -322,15 +319,12 @@ pub async fn update_remote_plugin_share_targets( .unwrap_or_default(); let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/ps/plugins/{remote_plugin_id}/shares"); - let request = authenticated_request( - config.http_request(Method::PUT, &url), - auth, - &config.product_sku, - )? - .json(&RemotePluginShareUpdateTargetsRequest { - discoverability, - targets, - }); + let request = authenticated_request(config, Method::PUT, &url, auth) + .await? + .json(&RemotePluginShareUpdateTargetsRequest { + discoverability, + targets, + }); let response: RemotePluginShareUpdateTargetsResponse = send_and_decode(request, &url).await?; Ok(RemotePluginShareUpdateTargetsResult { principals: response.principals, @@ -397,11 +391,7 @@ async fn get_created_workspace_plugins_page( url.query_pairs_mut().append_pair("pageToken", page_token); } let url = url.to_string(); - let request = authenticated_request( - config.http_request(Method::GET, &url), - auth, - &config.product_sku, - )?; + let request = authenticated_request(config, Method::GET, &url, auth).await?; send_and_decode(request, &url).await } @@ -414,17 +404,14 @@ async fn create_workspace_plugin_upload( ) -> Result { let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/public/plugins/workspace/upload-url"); - let request = authenticated_request( - config.http_request(Method::POST, &url), - auth, - &config.product_sku, - )? - .json(&RemoteWorkspacePluginUploadUrlRequest { - filename, - mime_type: "application/gzip", - size_bytes, - plugin_id: remote_plugin_id, - }); + let request = authenticated_request(config, Method::POST, &url, auth) + .await? + .json(&RemoteWorkspacePluginUploadUrlRequest { + filename, + mime_type: "application/gzip", + size_bytes, + plugin_id: remote_plugin_id, + }); send_and_decode(request, &url).await } @@ -476,12 +463,9 @@ async fn finalize_workspace_plugin_upload( } else { format!("{base_url}/public/plugins/workspace") }; - let request = authenticated_request( - config.http_request(Method::POST, &url), - auth, - &config.product_sku, - )? - .json(&body); + let request = authenticated_request(config, Method::POST, &url, auth) + .await? + .json(&body); send_and_decode(request, &url).await } diff --git a/codex-rs/core-plugins/src/remote/share/checkout.rs b/codex-rs/core-plugins/src/remote/share/checkout.rs index f846a9b90e..223c667b9c 100644 --- a/codex-rs/core-plugins/src/remote/share/checkout.rs +++ b/codex-rs/core-plugins/src/remote/share/checkout.rs @@ -40,6 +40,7 @@ pub async fn checkout_remote_plugin_share( codex_home: &Path, remote_plugin_id: &str, ) -> Result { + let config = &config.for_current_account(); let detail = super::super::fetch_remote_plugin_detail_with_download_urls( config, auth, diff --git a/codex-rs/core-plugins/src/remote_bundle.rs b/codex-rs/core-plugins/src/remote_bundle.rs index cb884aa40f..9292e2c2c5 100644 --- a/codex-rs/core-plugins/src/remote_bundle.rs +++ b/codex-rs/core-plugins/src/remote_bundle.rs @@ -1148,3 +1148,7 @@ mod tests { } } } + +#[cfg(test)] +#[path = "remote_bundle_runtime_auth_tests.rs"] +mod runtime_auth_tests; diff --git a/codex-rs/core-plugins/src/remote_bundle_runtime_auth_tests.rs b/codex-rs/core-plugins/src/remote_bundle_runtime_auth_tests.rs new file mode 100644 index 0000000000..2ca0934e0f --- /dev/null +++ b/codex-rs/core-plugins/src/remote_bundle_runtime_auth_tests.rs @@ -0,0 +1,55 @@ +//! Runtime-authenticated catalogs keep presigned bundle transfers unauthenticated. + +use super::*; +use codex_features::RuntimeAuthMode; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use pretty_assertions::assert_eq; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::path; + +#[tokio::test] +async fn runtime_bundle_download_follows_redirect_without_actor_auth() { + let origin = MockServer::start().await; + let destination = MockServer::start().await; + let final_url = format!("{}/bundle", destination.uri()); + Mock::given(path("/bundle")) + .respond_with(ResponseTemplate::new(307).insert_header("location", &final_url)) + .expect(1) + .mount(&origin) + .await; + Mock::given(path("/bundle")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"bundle")) + .expect(1) + .mount(&destination) + .await; + let config = RemotePluginServiceConfig::new_with_selected_runtime_auth( + origin.uri(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + /*product_sku*/ None, + (Default::default(), RuntimeAuthMode::AssertionOnly), + ); + let error = download_remote_plugin_bundle_with_limit( + &config, + &format!("{}/bundle", origin.uri()), + /*max_bytes*/ 64, + ) + .await + .expect_err("HTTP final URLs remain unsupported after following redirects"); + assert!( + matches!( + &error, + RemotePluginBundleInstallError::UnsupportedBundleDownloadFinalUrl { .. } + ), + "expected rejection of the HTTP redirect destination: {error}" + ); + let mut requests = origin.received_requests().await.unwrap(); + requests.extend(destination.received_requests().await.unwrap()); + assert_eq!(requests.len(), 2); + for request in requests { + assert!(!request.headers.contains_key("authorization")); + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + } +} diff --git a/codex-rs/core-plugins/src/remote_legacy.rs b/codex-rs/core-plugins/src/remote_legacy.rs index 846634358b..3d5e94afc2 100644 --- a/codex-rs/core-plugins/src/remote_legacy.rs +++ b/codex-rs/core-plugins/src/remote_legacy.rs @@ -135,18 +135,15 @@ pub async fn fetch_remote_featured_plugin_ids( product.unwrap_or(Product::Codex).to_app_platform(), ); let url = url.to_string(); - let mut request = config - .http_request(Method::GET, &url) - .timeout(REMOTE_FEATURED_PLUGIN_FETCH_TIMEOUT); - - if let Some(auth) = auth.filter(|auth| auth.uses_codex_backend()) { - request = request.headers( - codex_model_provider::auth_provider_from_auth(auth) - .to_auth_headers() - .map_err(std::io::Error::other) - .map_err(RemotePluginFetchError::AuthToken)?, - ); + let request = match auth.filter(|auth| auth.uses_codex_backend()) { + Some(auth) => config + .authenticated_request(Method::GET, &url, auth) + .await + .map_err(std::io::Error::other) + .map_err(RemotePluginFetchError::AuthToken)?, + None => config.http_request(Method::GET, &url), } + .timeout(REMOTE_FEATURED_PLUGIN_FETCH_TIMEOUT); let response = request .send() @@ -206,14 +203,11 @@ async fn post_remote_plugin_mutation( let auth = ensure_codex_backend_auth(auth)?; let url = remote_plugin_mutation_url(config, plugin_id, action)?; let request = config - .http_request(Method::POST, &url) - .timeout(REMOTE_PLUGIN_MUTATION_TIMEOUT) - .headers( - codex_model_provider::auth_provider_from_auth(auth) - .to_auth_headers() - .map_err(std::io::Error::other) - .map_err(RemotePluginMutationError::AuthToken)?, - ); + .authenticated_request(Method::POST, &url, auth) + .await + .map_err(std::io::Error::other) + .map_err(RemotePluginMutationError::AuthToken)? + .timeout(REMOTE_PLUGIN_MUTATION_TIMEOUT); let response = request .send() diff --git a/codex-rs/core-plugins/src/remote_mutations.rs b/codex-rs/core-plugins/src/remote_mutations.rs index 5e07d6f8a7..0ec012d28c 100644 --- a/codex-rs/core-plugins/src/remote_mutations.rs +++ b/codex-rs/core-plugins/src/remote_mutations.rs @@ -106,7 +106,9 @@ impl PluginsManager { } remote::validate_remote_plugin_id(&remote_plugin_id) .map_err(|err| unresolved(Error::InvalidRequest(err.message)))?; - let service = config.remote_plugin_service_config(); + let service = self + .remote_plugin_service_config(config) + .for_current_account(); let detail = remote::fetch_remote_plugin_detail_with_download_urls( &service, auth, @@ -232,7 +234,7 @@ impl PluginsManager { } remote::validate_remote_plugin_id(remote_plugin_id) .map_err(|err| unresolved(Error::InvalidRequest(err.message)))?; - let service = config.remote_plugin_service_config(); + let service = self.remote_plugin_service_config(config); let target = remote::resolve_remote_plugin_uninstall_target(&service, auth, remote_plugin_id) .await diff --git a/codex-rs/core-plugins/src/remote_mutations_tests.rs b/codex-rs/core-plugins/src/remote_mutations_tests.rs index 727001d73b..f82d3bc4f1 100644 --- a/codex-rs/core-plugins/src/remote_mutations_tests.rs +++ b/codex-rs/core-plugins/src/remote_mutations_tests.rs @@ -1,4 +1,4 @@ -//! Regression coverage for the shared mutation gate, refresh, and backend failure handling. +//! Regression coverage for remote mutations and account-bound storage transfers. use super::*; use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME; @@ -7,6 +7,7 @@ use crate::test_support::test_auth_manager; use crate::test_support::test_plugins_manager_with_auth_manager; use crate::test_support::write_file; use codex_protocol::auth::AuthMode; +use codex_utils_absolute_path::AbsolutePathBuf; use pretty_assertions::assert_eq; use serde_json::json; use std::sync::atomic::AtomicBool; @@ -109,3 +110,364 @@ async fn uninstall_serializes_backend_mutations_and_preserves_cache_on_failure() } } } + +/// Switch only after metadata has been read, immediately before constructing the storage request. +/// Metadata-body cancellation would not exercise the separate request these tests protect. +#[derive(Clone)] +struct TransferHttpClients { + delegate: codex_http_client::RouteAwareClientPool, + controller: codex_http_client::NetworkPolicyController, + switch_before_storage: Arc, + api_origin: String, + storage_origin: String, +} + +impl std::fmt::Debug for TransferHttpClients { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TransferHttpClients") + .finish_non_exhaustive() + } +} + +impl crate::http_client_selector::HttpClientSelector for TransferHttpClients { + fn for_current_account(&self) -> Arc { + Arc::new(Self { + delegate: self.delegate.for_current_account(), + ..self.clone() + }) + } + + fn request(&self, method: http::Method, url: &str) -> codex_http_client::RequestBuilder { + if url.starts_with("https://storage.example/") + && self.switch_before_storage.swap(false, Ordering::SeqCst) + { + let policy = self.controller.policy(); + policy.invalidate(); + assert!(self.controller.publish( + policy.revision(), + codex_http_client::DestinationPolicy::Unrestricted, + )); + } + self.delegate.request( + method, + url.replace("https://storage.example", &self.storage_origin) + .replace("https://chatgpt.com", &self.api_origin), + ) + } + + fn request_for_account( + &self, + method: http::Method, + url: &str, + binding: &codex_http_client::NetworkPolicy, + ) -> Option { + Some( + self.delegate + .with_account_binding(binding)? + .request(method, url.replace("https://chatgpt.com", &self.api_origin)), + ) + } + + fn outbound_proxy_policy(&self) -> codex_http_client::OutboundProxyPolicy { + self.delegate.outbound_proxy_policy() + } +} + +struct PluginTransferTest { + api: MockServer, + storage: MockServer, + config: PluginsConfigInput, + manager: Arc, + auth: CodexAuth, + switch_before_storage: Arc, +} + +impl PluginTransferTest { + async fn new( + home: &std::path::Path, + mode: codex_features::RuntimeAuthMode, + switch: bool, + ) -> Self { + use base64::Engine; + use codex_http_client::ClientRouteClass; + use codex_http_client::DestinationPolicy; + use codex_http_client::HttpClientFactory; + use codex_http_client::NetworkPolicyController; + use codex_http_client::OutboundProxyPolicy; + use codex_http_client::RouteAwareClientPool; + use codex_runtime_agent_instance::AgentHostAuthorization; + use codex_runtime_agent_instance::RuntimeAuthManager; + use codex_runtime_agent_instance::RuntimeAuthSlot; + use codex_runtime_agent_instance::RuntimeRegistrationMetadata; + use codex_runtime_agent_instance::RuntimeRole; + + let api = MockServer::start().await; + let storage = MockServer::start().await; + Mock::given(path("/agent/register")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "runtime_agent_instance_id": "runtime-plugins", "agent_host_id": "host-a", + "actor": {"kind": "chatgpt", "user_id": "user-a", "account_id": "account-a"}, + }))) + .expect(1) + .mount(&api) + .await; + let claims = json!({ + "iss": "https://auth.openai.com/", "session_id": "login-a", + "https://api.openai.com/auth": {"user_id": "user-a", "chatgpt_account_id": "account-a"}, + }); + let token = format!( + "header.{}.signature", + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(claims.to_string()), + ); + let auth = CodexAuth::from_external_chatgpt_tokens( + &token, + "account-a", + /*chatgpt_plan_type*/ None, + ) + .unwrap(); + let credentials = codex_login::AuthManager::from_auth_for_testing(auth.clone()); + let slot = RuntimeAuthSlot::default(); + slot.set(Arc::new(RuntimeAuthManager::new( + Arc::clone(&credentials), + AgentHostAuthorization::InstallationId { + installation_id: uuid::Uuid::nil(), + }, + RuntimeRegistrationMetadata { + harness: Some("codex".into()), + harness_version: None, + role: RuntimeRole::Orchestrator, + }, + format!("{}/agent/register", api.uri()).parse().unwrap(), + ))) + .unwrap(); + let controller = NetworkPolicyController::default(); + assert!(controller.publish( + controller.policy().revision(), + DestinationPolicy::Unrestricted + )); + let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) + .with_network_policy(controller.policy()) + .with_sensitive_header_protection(); + let switch_before_storage = Arc::new(AtomicBool::new(switch)); + let clients = Arc::new(TransferHttpClients { + delegate: RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging( + factory.clone(), + ClientRouteClass::Api, + ), + controller, + switch_before_storage: Arc::clone(&switch_before_storage), + api_origin: api.uri(), + storage_origin: storage.uri(), + }); + let config = PluginsConfigInput::new( + Default::default(), + Default::default(), + "openai".into(), + /*plugins_enabled*/ true, + /*remote_plugin_enabled*/ true, + "https://chatgpt.com/backend-api".into(), + factory, + /*product_sku*/ None, + ) + .with_runtime_auth_mode(Some(mode)); + config.runtime_http_clients.set(clients).unwrap(); + let manager = Arc::new(crate::PluginsManager::new_with_options( + home.to_path_buf(), + /*restriction_product*/ None, + credentials, + slot, + crate::test_support::test_skill_root_loader(), + )); + Self { + api, + storage, + config, + manager, + auth, + switch_before_storage, + } + } + + async fn verify(&self, account_changed: bool, mode: codex_features::RuntimeAuthMode) { + assert!(!self.switch_before_storage.load(Ordering::SeqCst)); + self.api.verify().await; + let storage_requests = self.storage.received_requests().await.unwrap(); + assert_eq!(storage_requests.len(), usize::from(!account_changed)); + for request in storage_requests { + for name in [ + "authorization", + "x-openai-agent-assertion", + "chatgpt-account-id", + ] { + assert!(!request.headers.contains_key(name)); + } + } + for request in self.api.received_requests().await.unwrap() { + if request.url.path() == "/agent/register" { + continue; + } + let authorization = request.headers["authorization"].to_str().unwrap(); + match mode { + codex_features::RuntimeAuthMode::DualHeader => { + assert!(authorization.starts_with("Bearer ")); + assert!(request.headers.contains_key("x-openai-agent-assertion")); + } + codex_features::RuntimeAuthMode::AssertionOnly => { + assert!(authorization.starts_with("AgentAssertion ")); + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + } + } + } + } +} + +#[tokio::test] +async fn plugin_share_keeps_account_binding_through_storage_upload() { + use codex_features::RuntimeAuthMode; + use codex_http_client::NetworkPolicyDenied; + use codex_http_client::RouteAwareRequestError; + + for mode in [RuntimeAuthMode::DualHeader, RuntimeAuthMode::AssertionOnly] { + for account_changed in [false, true] { + let home = TempDir::new().unwrap(); + let fixture = PluginTransferTest::new(home.path(), mode, account_changed).await; + let plugin_path = home.path().join("sample"); + write_file( + &plugin_path.join(".codex-plugin/plugin.json"), + r#"{"name":"sample"}"#, + ); + Mock::given(method("POST")) + .and(path("/backend-api/public/plugins/workspace/upload-url")) + .respond_with(ResponseTemplate::new(201).set_body_json(json!({ + "file_id": "file_123", "upload_url": "https://storage.example/upload", + "etag": "\"metadata-etag\"", + }))) + .expect(1) + .mount(&fixture.api) + .await; + Mock::given(method("PUT")) + .and(path("/upload")) + .respond_with(ResponseTemplate::new(201)) + .mount(&fixture.storage) + .await; + Mock::given(method("POST")) + .and(path("/backend-api/public/plugins/workspace")) + .respond_with(ResponseTemplate::new(201).set_body_json(json!({ + "plugin_id": "plugins_123", "share_url": null, + }))) + .expect(u64::from(!account_changed)) + .mount(&fixture.api) + .await; + let result = remote::save_remote_plugin_share( + &fixture + .manager + .remote_plugin_service_config(&fixture.config), + Some(&fixture.auth), + home.path(), + &AbsolutePathBuf::try_from(plugin_path).unwrap(), + /*remote_plugin_id*/ None, + remote::RemotePluginShareAccessPolicy::default(), + ) + .await; + if account_changed { + assert!( + matches!( + result, + Err(RemotePluginCatalogError::Request { + source: RouteAwareRequestError::Policy(NetworkPolicyDenied::Revoked), + .. + }) + ), + "{result:?}" + ); + } else { + assert_eq!(result.unwrap().remote_plugin_id, "plugins_123"); + } + fixture.verify(account_changed, mode).await; + } + } +} + +#[tokio::test] +async fn plugin_install_keeps_account_binding_through_storage_download() { + use codex_features::RuntimeAuthMode; + use codex_http_client::NetworkPolicyDenied; + use codex_http_client::RouteAwareRequestError; + + for mode in [RuntimeAuthMode::DualHeader, RuntimeAuthMode::AssertionOnly] { + for account_changed in [false, true] { + let home = TempDir::new().unwrap(); + let fixture = PluginTransferTest::new(home.path(), mode, account_changed).await; + let remote_id = "b1234567-89ab-4cde-8f01-234567890abc"; + Mock::given(method("GET")) + .and(path(format!("/backend-api/ps/plugins/{remote_id}"))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "id": remote_id, "name": "sample", "scope": "GLOBAL", + "installation_policy": "AVAILABLE", "authentication_policy": "ON_USE", + "release": { + "version": "1.0.0", "display_name": "Sample", + "description": "Sample plugin", "interface": {}, + "bundle_download_url": "https://storage.example/bundle", + } + }))) + .expect(1) + .mount(&fixture.api) + .await; + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/installed")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "plugins": [], "pagination": {"next_page_token": null}, + }))) + .expect(1) + .mount(&fixture.api) + .await; + Mock::given(method("GET")) + .and(path("/bundle")) + .respond_with(ResponseTemplate::new(200).set_body_string("bundle")) + .mount(&fixture.storage) + .await; + let result = fixture + .manager + .install_remote_plugin( + &fixture.config, + Some(&fixture.auth), + RemotePluginInstallRequest { + marketplace_name: REMOTE_GLOBAL_MARKETPLACE_NAME.to_string(), + remote_plugin_id: remote_id.to_string(), + install_attempt_id: None, + }, + /*on_effective_plugins_changed*/ None, + ) + .await; + let error = result.err().expect("test transfer should fail"); + if account_changed { + assert!( + matches!( + *error.kind, + RemotePluginOperationErrorKind::Bundle( + RemotePluginBundleInstallError::DownloadRequest { + source: RouteAwareRequestError::Policy( + NetworkPolicyDenied::Revoked + ), + .. + } + ) + ), + "{error:?}" + ); + } else { + // The local HTTP control reaches storage, then fails the final-URL HTTPS check. + assert!( + matches!( + *error.kind, + RemotePluginOperationErrorKind::Bundle( + RemotePluginBundleInstallError::UnsupportedBundleDownloadFinalUrl { .. } + ) + ), + "{error:?}" + ); + } + fixture.verify(account_changed, mode).await; + } + } +} diff --git a/codex-rs/core-plugins/src/remote_runtime_auth_tests.rs b/codex-rs/core-plugins/src/remote_runtime_auth_tests.rs new file mode 100644 index 0000000000..b9bdadd3eb --- /dev/null +++ b/codex-rs/core-plugins/src/remote_runtime_auth_tests.rs @@ -0,0 +1,307 @@ +//! Exercise modern and legacy plugin requests with a shared, late-initialized runtime. + +use super::*; +use base64::Engine; +use codex_features::RuntimeAuthMode; +use codex_http_client::DestinationPolicy; +use codex_http_client::NetworkPolicyController; +use codex_http_client::NetworkPolicyDenied; +use codex_http_client::OutboundProxyPolicy; +use codex_login::AuthCredentialsStoreMode; +use codex_login::AuthKeyringBackendKind; +use codex_login::AuthManager; +use codex_login::AuthRouteConfig; +use codex_runtime_agent_instance::AgentHostAuthorization; +use codex_runtime_agent_instance::RuntimeAuthManager; +use codex_runtime_agent_instance::RuntimeAuthSlot; +use codex_runtime_agent_instance::RuntimeRegistrationMetadata; +use codex_runtime_agent_instance::RuntimeRole; +use pretty_assertions::assert_eq; +use serde_json::json; +use uuid::Uuid; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::path; + +#[tokio::test] +async fn plugin_runtime_auth_shares_registration_and_strips_cross_origin_auth() { + for (mode, pooled) in [ + (RuntimeAuthMode::DualHeader, false), + (RuntimeAuthMode::AssertionOnly, false), + (RuntimeAuthMode::DualHeader, true), + (RuntimeAuthMode::AssertionOnly, true), + ] { + let server = MockServer::start().await; + let destination = MockServer::start().await; + Mock::given(path("/")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "enabled": true, "plugins": [], + }))) + .expect(1) + .mount(&destination) + .await; + Mock::given(path("/agent/register")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "runtime_agent_instance_id": "runtime-plugins", "agent_host_id": "host-a", + "actor": {"kind": "chatgpt", "user_id": "user-a", "account_id": "account-a"}, + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(path("/backend-api/ps/plugins/list")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "plugins": [], "pagination": {"next_page_token": null}, + }))) + .expect(2) + .mount(&server) + .await; + Mock::given(path("/backend-api/plugins/featured")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!([]))) + .expect(1) + .mount(&server) + .await; + Mock::given(path("/backend-api/ps/plugins/suggested/codex")) + .respond_with(ResponseTemplate::new(307).insert_header("location", destination.uri())) + .expect(1) + .mount(&server) + .await; + let claims = json!({ + "iss": "https://auth.openai.com/", "session_id": "login-a", + "https://api.openai.com/auth": {"user_id": "user-a", "chatgpt_account_id": "account-a"}, + }); + let token = format!( + "header.{}.signature", + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(claims.to_string()) + ); + let auth = CodexAuth::from_external_chatgpt_tokens( + &token, + "account-a", + /*chatgpt_plan_type*/ None, + ) + .unwrap(); + let slot = RuntimeAuthSlot::default(); + let controller = NetworkPolicyController::default(); + let policy = controller.policy(); + let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) + .with_network_policy(policy.clone()); + let home = tempfile::tempdir().unwrap(); + codex_login::auth::login_with_chatgpt_auth_tokens( + home.path(), + &token, + "account-a", + /*chatgpt_plan_type*/ None, + ) + .unwrap(); + let credentials = Arc::new( + AuthManager::new( + home.path().to_path_buf(), + /*enable_codex_api_key_env*/ false, + AuthCredentialsStoreMode::Ephemeral, + /*forced_chatgpt_workspace_id*/ None, + /*chatgpt_base_url*/ None, + AuthKeyringBackendKind::Direct, + AuthRouteConfig::from_http_client_factory(factory.clone()), + ) + .await, + ); + assert!(controller.publish(policy.revision(), DestinationPolicy::Unrestricted)); + let mut config = if pooled { + crate::manager::PluginsConfigInput::new( + Default::default(), + Default::default(), + "openai".into(), + /*plugins_enabled*/ true, + /*remote_plugin_enabled*/ true, + "https://chatgpt.com/backend-api".into(), + factory, + /*product_sku*/ None, + ) + .remote_plugin_service_config(Some((Arc::clone(&slot), mode))) + } else { + RemotePluginServiceConfig::new_with_selected_runtime_auth( + "https://chatgpt.com/backend-api".into(), + factory, + /*product_sku*/ None, + (Arc::clone(&slot), mode), + ) + }; + config.chatgpt_base_url = format!("{}/backend-api", server.uri()); + // Enabled requests cannot silently fall back before the harness initializes the slot. + assert!( + fetch_recommended_plugins(&config, Some(&auth)) + .await + .is_err() + ); + assert!(server.received_requests().await.unwrap().is_empty()); + slot.set(Arc::new(RuntimeAuthManager::new( + credentials, + AgentHostAuthorization::InstallationId { + installation_id: Uuid::nil(), + }, + RuntimeRegistrationMetadata { + harness: Some("codex".into()), + harness_version: None, + role: RuntimeRole::Orchestrator, + }, + format!("{}/agent/register", server.uri()).parse().unwrap(), + ))) + .unwrap(); + get_remote_plugin_list_page( + &config, + &auth, + RemotePluginScope::Global, + /*page_token*/ None, + /*collection*/ None, + ) + .await + .unwrap(); + crate::remote_legacy::fetch_remote_featured_plugin_ids( + &config, + Some(&auth), + /*product*/ None, + ) + .await + .unwrap(); + fetch_recommended_plugins(&config, Some(&auth)) + .await + .unwrap(); + let redirected = destination.received_requests().await.unwrap(); + assert_eq!(redirected.len(), 1); + assert!(!redirected[0].headers.contains_key("authorization")); + assert!( + !redirected[0] + .headers + .contains_key("x-openai-agent-assertion") + ); + destination.verify().await; + let url = format!("{}/backend-api/ps/plugins/list", server.uri()); + let pending = config + .authenticated_request(Method::GET, &url, &auth) + .await + .unwrap(); + policy.invalidate(); + assert!(controller.publish(policy.revision(), DestinationPolicy::Unrestricted)); + assert!(matches!( + pending.send().await, + Err(RouteAwareRequestError::Policy(NetworkPolicyDenied::Revoked)) + )); + let response = config + .authenticated_request(Method::GET, &url, &auth) + .await + .unwrap() + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let requests = server.received_requests().await.unwrap(); + let mut proof_ids = HashSet::new(); + for request in requests + .iter() + .filter(|r| r.url.path() != "/agent/register") + { + let authorization = request.headers["authorization"].to_str().unwrap(); + let proof = match mode { + RuntimeAuthMode::DualHeader => { + assert_eq!(authorization, format!("Bearer {token}")); + request.headers["x-openai-agent-assertion"] + .to_str() + .unwrap() + } + RuntimeAuthMode::AssertionOnly => { + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + authorization.strip_prefix("AgentAssertion ").unwrap() + } + }; + assert_eq!(request.headers["chatgpt-account-id"], "account-a"); + let payload: JsonValue = serde_json::from_slice( + &base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(proof.split('.').nth(2).unwrap()) + .unwrap(), + ) + .unwrap(); + assert_eq!(payload["runtime_agent_instance"], "runtime-plugins"); + assert!(proof_ids.insert(payload["id"].as_str().unwrap().to_string())); + } + assert_eq!(proof_ids.len(), 4); + server.verify().await; + } +} + +#[tokio::test] +async fn custom_plugin_backend_keeps_legacy_auth_with_runtime_enabled() { + for mode in [RuntimeAuthMode::DualHeader, RuntimeAuthMode::AssertionOnly] { + let server = MockServer::start().await; + Mock::given(path("/backend-api/plugins/featured")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!([]))) + .expect(1) + .mount(&server) + .await; + let config = RemotePluginServiceConfig::new_with_runtime_auth( + format!("{}/backend-api", server.uri()), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + /*product_sku*/ None, + (RuntimeAuthSlot::default(), mode), + ); + let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); + crate::remote_legacy::fetch_remote_featured_plugin_ids( + &config, + Some(&auth), + /*product*/ None, + ) + .await + .unwrap(); + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 1); + assert!( + requests[0].headers["authorization"] + .to_str() + .unwrap() + .starts_with("Bearer ") + ); + assert!(!requests[0].headers.contains_key("x-openai-agent-assertion")); + server.verify().await; + } +} + +#[tokio::test] +async fn unsupported_openai_plugin_backend_fails_before_registration() { + for mode in [RuntimeAuthMode::DualHeader, RuntimeAuthMode::AssertionOnly] { + let server = MockServer::start().await; + let auth = CodexAuth::from_api_key("key-a"); + let runtime = ( + Arc::new(std::sync::OnceLock::from(Arc::new( + RuntimeAuthManager::new( + AuthManager::from_auth_for_testing(auth.clone()), + AgentHostAuthorization::InstallationId { + installation_id: Uuid::nil(), + }, + RuntimeRegistrationMetadata { + harness: Some("codex".into()), + harness_version: None, + role: RuntimeRole::Orchestrator, + }, + format!("{}/register", server.uri()).parse().unwrap(), + ), + ))), + mode, + ); + let config = RemotePluginServiceConfig::new_with_runtime_auth( + "https://future.openai.com/backend-api".to_string(), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + /*product_sku*/ None, + runtime, + ); + assert!( + config + .authenticated_request( + Method::GET, + "https://future.openai.com/backend-api/plugins/featured", + &auth, + ) + .await + .is_err() + ); + assert!(server.received_requests().await.unwrap().is_empty()); + } +} diff --git a/codex-rs/core-plugins/src/test_support.rs b/codex-rs/core-plugins/src/test_support.rs index 39e5ea0b48..60dcdc2971 100644 --- a/codex-rs/core-plugins/src/test_support.rs +++ b/codex-rs/core-plugins/src/test_support.rs @@ -33,6 +33,7 @@ use codex_login::test_support::auth_manager_from_optional_auth; use codex_protocol::auth::AuthMode; use codex_protocol::protocol::Product; use codex_protocol::protocol::SkillScope; +use codex_runtime_agent_instance::RuntimeAuthSlot; use codex_skills::LoadedSkillRoot; use codex_skills::LoadedSkills; use codex_skills::SkillError; @@ -81,6 +82,7 @@ pub(crate) fn test_plugins_manager_with_options( codex_home, restriction_product, test_auth_manager(auth_mode), + RuntimeAuthSlot::default(), test_skill_root_loader(), ) } @@ -94,6 +96,7 @@ pub(crate) fn test_plugins_manager_with_auth_manager( codex_home, restriction_product, auth_manager, + RuntimeAuthSlot::default(), test_skill_root_loader(), ) } @@ -327,6 +330,13 @@ pub(crate) struct RecordingHttpClientSelector { } impl RecordingHttpClientSelector { + fn record_url(&self, url: &str) { + match self.selected_urls.lock() { + Ok(mut selected_urls) => selected_urls.push(url.to_string()), + Err(error) => panic!("selected URL recorder lock should not be poisoned: {error}"), + } + } + pub(crate) fn new() -> (Arc, Arc>>) { let selected_urls = Arc::new(Mutex::new(Vec::new())); let delegate = RouteAwareClientPool::with_chatgpt_cloudflare_cookies( @@ -344,13 +354,27 @@ impl RecordingHttpClientSelector { } impl HttpClientSelector for RecordingHttpClientSelector { + fn for_current_account(&self) -> Arc { + Arc::new(Self { + selected_urls: Arc::clone(&self.selected_urls), + delegate: self.delegate.for_current_account(), + }) + } + fn request(&self, method: Method, url: &str) -> RequestBuilder { - match self.selected_urls.lock() { - Ok(mut selected_urls) => selected_urls.push(url.to_string()), - Err(error) => panic!("selected URL recorder lock should not be poisoned: {error}"), - } + self.record_url(url); self.delegate.request(method, url) } + + fn request_for_account( + &self, + method: Method, + url: &str, + binding: &codex_http_client::NetworkPolicy, + ) -> Option { + self.record_url(url); + self.delegate.request_for_account(method, url, binding) + } fn outbound_proxy_policy(&self) -> OutboundProxyPolicy { self.delegate.outbound_proxy_policy() } @@ -362,6 +386,7 @@ pub(crate) fn recording_remote_plugin_service_config( let (http_clients, selected_urls) = RecordingHttpClientSelector::new(); ( RemotePluginServiceConfig { + runtime_auth: None, chatgpt_base_url, product_sku: crate::remote::CODEX_PRODUCT_SKU.to_string(), http_clients, diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 3970fac710..6f93cd216f 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -2203,7 +2203,9 @@ respect_system_proxy = true codex_http_client::OutboundProxyPolicy::RespectSystemProxy ); assert_eq!( - config.plugins_config_input().remote_plugin_service_config(), + config + .plugins_config_input() + .remote_plugin_service_config(/*runtime_auth*/ None), codex_core_plugins::remote::RemotePluginServiceConfig::new( config.chatgpt_base_url, codex_http_client::HttpClientFactory::new( diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 7f934808be..f11bb4ab10 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1861,6 +1861,11 @@ impl Config { self.http_client_factory(), self.apps_mcp_product_sku.clone(), ) + .with_runtime_auth_mode( + (self.features.enabled(Feature::RuntimeAgentInstances) + && codex_runtime_agent_instance::is_openai_api_url(&self.chatgpt_base_url)) + .then_some(self.runtime_auth_mode), + ) } /// Applies managed MCP requirements to servers supplied by one plugin. diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 59b661f1db..7ec46339cc 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -22,7 +22,7 @@ use tracing::warn; use crate::config::Config; use crate::mcp::McpManager; use crate::plugins::list_tool_suggest_discoverable_plugins; -use crate::plugins::plugins_manager_for_config; +use crate::plugins::plugins_manager_for_config_with_runtime; use crate::session::INITIAL_SUBMIT_ID; use codex_config::types::ApprovalsReviewer; use codex_config::types::ToolSuggestDiscoverableType; @@ -187,10 +187,8 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_environment_manager( ) -> anyhow::Result { let auth_manager = AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await?; - let plugins_manager = Arc::new(plugins_manager_for_config( - config, - Arc::clone(&auth_manager), - )); + let plugins_manager = + Arc::new(plugins_manager_for_config_with_runtime(config, Arc::clone(&auth_manager)).await?); let mcp_manager = Arc::new(McpManager::new(plugins_manager)); list_accessible_connectors_from_mcp_tools_with_mcp_manager( config, diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 553689953f..c27ec56297 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -127,6 +127,7 @@ pub use mention_syntax::TOOL_MENTION_SIGIL; pub use utils::path_utils; pub(crate) mod plugins; pub use plugins::plugins_manager_for_config; +pub use plugins::plugins_manager_for_config_with_runtime; #[doc(hidden)] pub(crate) mod prompt_debug; #[doc(hidden)] diff --git a/codex-rs/core/src/mcp.rs b/codex-rs/core/src/mcp.rs index 3bfa3b412c..eba1bd977c 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -146,6 +146,16 @@ impl McpManager { } } + pub(crate) fn for_plugins_manager(&self, plugins_manager: Arc) -> Self { + let mut manager = Self::new_with_extensions( + plugins_manager, + Arc::clone(&self.extensions), + ConnectorRuntimeManager::default(), + ); + manager.connection_pool = self.connection_pool.clone(); + manager + } + pub fn codex_apps_tools_cache(&self) -> ConnectorRuntimeManager { self.codex_apps_tools_cache.clone() } diff --git a/codex-rs/core/src/mcp_openai_file.rs b/codex-rs/core/src/mcp_openai_file.rs index f9f9156f96..1a7608d7ab 100644 --- a/codex-rs/core/src/mcp_openai_file.rs +++ b/codex-rs/core/src/mcp_openai_file.rs @@ -236,11 +236,38 @@ async fn build_uploaded_argument_value( } } }; - let upload_auth = codex_model_provider::auth_provider_from_auth(auth); + let runtime = if turn_context + .config + .features + .enabled(codex_features::Feature::RuntimeAgentInstances) + && codex_http_client::classify_openai_api_url(&turn_context.config.chatgpt_base_url) + != codex_http_client::OpenAiDestination::Other + { + codex_model_provider::validate_runtime_destination(Some( + &turn_context.config.chatgpt_base_url, + )) + .map_err(|error| contextualize_error(error.to_string()))?; + let manager = sess.services.runtime_auth.get().ok_or_else(|| { + contextualize_error("runtime authentication has not been initialized".to_string()) + })?; + Some((Arc::clone(manager), turn_context.config.runtime_auth_mode)) + } else { + None + }; + let upload_auth = codex_model_provider::auth_provider_for_request( + auth, + runtime.as_ref(), + &http::HeaderMap::new(), + ) + .await + .map_err(|error| contextualize_error(error.to_string()))?; + let upload_clients = upload_auth + .bind_http_client_pool(&sess.services.openai_file_upload_client_pool) + .map_err(|error| contextualize_error(error.to_string()))?; let uploaded = upload_openai_file( turn_context.config.chatgpt_base_url.trim_end_matches('/'), - upload_auth.as_ref(), - &sess.services.openai_file_upload_client_pool, + upload_auth.auth.as_ref(), + &upload_clients, file_name, metadata.size, open_contents, diff --git a/codex-rs/core/src/plugins/mod.rs b/codex-rs/core/src/plugins/mod.rs index e47c8e28e6..b2753e587f 100644 --- a/codex-rs/core/src/plugins/mod.rs +++ b/codex-rs/core/src/plugins/mod.rs @@ -44,3 +44,39 @@ pub fn plugins_manager_for_config( skill_root_loader, ) } + +/// Initialize a standalone command's runtime before it can make first-party plugin requests. +/// Callers that own a ThreadManager should reuse that manager's PluginsManager instead. +pub async fn plugins_manager_for_config_with_runtime( + config: &Config, + auth_manager: Arc, +) -> anyhow::Result { + if !config + .features + .enabled(codex_features::Feature::RuntimeAgentInstances) + { + return Ok(plugins_manager_for_config(config, auth_manager)); + } + codex_model_provider::validate_runtime_destination(Some(&config.chatgpt_base_url))?; + if !codex_runtime_agent_instance::is_openai_api_url(&config.chatgpt_base_url) { + return Ok(plugins_manager_for_config(config, auth_manager)); + } + let installation_id = + crate::installation_id::resolve_installation_id(&config.codex_home).await?; + let runtime = crate::runtime_auth::build_runtime_auth_manager( + config, + Arc::clone(&auth_manager), + &installation_id, + ); + let slot = Arc::new(runtime.map(std::sync::OnceLock::from).unwrap_or_default()); + Ok(PluginsManager::new_with_options( + config.codex_home.to_path_buf(), + Some(codex_protocol::protocol::Product::Codex), + auth_manager, + slot, + Arc::new(HostSkillsService::new( + config.codex_home.clone(), + /*bundled_skills_enabled*/ false, + )), + )) +} diff --git a/codex-rs/core/src/plugins/skill_snapshot_tests.rs b/codex-rs/core/src/plugins/skill_snapshot_tests.rs index 175c2b8a8e..06e9dce356 100644 --- a/codex-rs/core/src/plugins/skill_snapshot_tests.rs +++ b/codex-rs/core/src/plugins/skill_snapshot_tests.rs @@ -57,6 +57,7 @@ enabled = false codex_home.path().to_path_buf(), Some(Product::Codex), AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()), + codex_runtime_agent_instance::RuntimeAuthSlot::default(), skills_service.clone(), ); let plugin_outcome = plugins_manager.plugins_for_config(&plugins_input).await; diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index dde89e55fa..3cc3efbd39 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1925,11 +1925,24 @@ impl Session { exec_policy, auth_manager: Arc::clone(&auth_manager), runtime_auth, - openai_file_upload_client_pool: RouteAwareClientPool::new_without_request_logging( - config.http_client_factory(), - ClientRouteClass::Api, - ) - .with_legacy_custom_ca_fallback(), + openai_file_upload_client_pool: if config + .features + .enabled(Feature::RuntimeAgentInstances) + && codex_runtime_agent_instance::is_openai_api_url(&config.chatgpt_base_url) + { + RouteAwareClientPool::new_without_request_logging( + config + .http_client_factory() + .with_sensitive_header_protection(), + ClientRouteClass::Api, + ) + } else { + RouteAwareClientPool::new_without_request_logging( + config.http_client_factory(), + ClientRouteClass::Api, + ) + .with_legacy_custom_ca_fallback() + }, session_telemetry, models_manager: Arc::clone(&models_manager), git_root_discovery, diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index c11fda446e..12562fa3a0 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -99,6 +99,7 @@ use codex_protocol::protocol::TurnEnvironmentRequest; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::W3cTraceContext; use codex_rollout::state_db::StateDbHandle; +use codex_runtime_agent_instance::RuntimeAuthManager; use codex_skills_extension::HostSkillsService; use codex_thread_store::LoadThreadHistoryParams; use codex_thread_store::LocalThreadStore; @@ -599,6 +600,7 @@ impl ThreadManager { codex_home.to_path_buf(), restriction_product, Arc::clone(&auth_manager), + Arc::clone(&runtime_auth), skills_service.clone(), )); let mcp_manager = Arc::new(McpManager::new_with_extensions( @@ -809,10 +811,12 @@ impl ThreadManager { /*bundled_skills_enabled*/ true, restriction_product, )); + let runtime_auth = crate::runtime_auth::RuntimeAuthSlot::default(); let plugins_manager = Arc::new(PluginsManager::new_with_options( codex_home.clone(), restriction_product, Arc::clone(&auth_manager), + Arc::clone(&runtime_auth), skills_service.clone(), )); let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); @@ -829,7 +833,7 @@ impl ThreadManager { let agent_graph_store = local_agent_graph_store_from_state_db(state_db.as_ref()); Self { state: Arc::new(ThreadManagerState { - runtime_auth: Default::default(), + runtime_auth, threads: Arc::new(RwLock::new(HashMap::new())), shared_thread_instructions: Default::default(), thread_created_tx, @@ -882,6 +886,30 @@ impl ThreadManager { self.state.auth_manager.clone() } + /// Share the harness runtime with an API client using this effective ChatGPT configuration. + pub fn runtime_auth_for_config( + &self, + config: &Config, + ) -> std::io::Result, codex_features::RuntimeAuthMode)>> { + if config + .features + .enabled(codex_features::Feature::RuntimeAgentInstances) + { + codex_model_provider::validate_runtime_destination(Some(&config.chatgpt_base_url)) + .map_err(std::io::Error::other)?; + } + if !codex_runtime_agent_instance::is_openai_api_url(&config.chatgpt_base_url) { + return Ok(None); + } + Ok(crate::runtime_auth::runtime_for_config( + &self.state.runtime_auth, + config, + Arc::clone(&self.state.auth_manager), + &self.state.installation_id, + ) + .map(|runtime| (runtime, config.runtime_auth_mode))) + } + pub fn skills_service(&self) -> Arc { self.state.skills_service.clone() } @@ -2577,20 +2605,37 @@ impl ThreadManagerState { } else { codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile }; - // Resume may supply a different credential owner. Keep its runtime separate from the - // harness default, and let sessions using that parent's credentials inherit its owner. - let runtime_auth = if let Some(runtime_auth) = inherited_runtime_auth { - runtime_auth - } else if Arc::ptr_eq(&auth_manager, &self.auth_manager) { - Arc::clone(&self.runtime_auth) + // A resumed actor and its descendants share an owner, not the harness default's credentials. + let (runtime_auth, plugins_manager) = if Arc::ptr_eq(&auth_manager, &self.auth_manager) { + ( + inherited_runtime_auth.unwrap_or_else(|| Arc::clone(&self.runtime_auth)), + Arc::clone(&self.plugins_manager), + ) } else { let threads = self.threads.read().await; - parent_thread_id + if let Some(parent) = parent_thread_id .or(forked_from_thread_id) .and_then(|id| threads.get(&id)) .filter(|parent| Arc::ptr_eq(&parent.session.services.auth_manager, &auth_manager)) - .map(|parent| Arc::clone(&parent.session.services.runtime_auth)) - .unwrap_or_default() + { + ( + inherited_runtime_auth + .unwrap_or_else(|| Arc::clone(&parent.session.services.runtime_auth)), + Arc::clone(&parent.session.services.plugins_manager), + ) + } else { + let runtime_auth = inherited_runtime_auth.unwrap_or_default(); + let plugins = Arc::new( + self.plugins_manager + .for_auth_context(Arc::clone(&auth_manager), Arc::clone(&runtime_auth)), + ); + (runtime_auth, plugins) + } + }; + let mcp_manager = if Arc::ptr_eq(&plugins_manager, &self.plugins_manager) { + mcp_manager + } else { + Arc::new(mcp_manager.for_plugins_manager(Arc::clone(&plugins_manager))) }; let runtime = crate::runtime_auth::runtime_for_config( &runtime_auth, @@ -2621,7 +2666,7 @@ impl ThreadManagerState { git_root_discovery: Arc::clone(&self.git_root_discovery), environment_manager: Arc::clone(&self.environment_manager), skills_service: Arc::clone(&self.skills_service), - plugins_manager: Arc::clone(&self.plugins_manager), + plugins_manager, mcp_manager, code_mode_session_provider: Arc::clone(&self.code_mode_session_provider), extensions, diff --git a/codex-rs/core/src/tools/handlers/request_plugin_install.rs b/codex-rs/core/src/tools/handlers/request_plugin_install.rs index 281f938b5c..335584ab4c 100644 --- a/codex-rs/core/src/tools/handlers/request_plugin_install.rs +++ b/codex-rs/core/src/tools/handlers/request_plugin_install.rs @@ -191,7 +191,10 @@ impl RequestPluginInstallHandler { let auth = session.services.auth_manager.auth().await; let plugins_config = turn.config.plugins_config_input(); match codex_core_plugins::hydrate_selected_recommended_plugin_install_metadata( - &plugins_config, + &session + .services + .plugins_manager + .remote_plugin_service_config(&plugins_config), auth.as_ref(), tool, ) diff --git a/codex-rs/core/tests/common/Cargo.toml b/codex-rs/core/tests/common/Cargo.toml index 89abae915a..0837095d38 100644 --- a/codex-rs/core/tests/common/Cargo.toml +++ b/codex-rs/core/tests/common/Cargo.toml @@ -31,20 +31,24 @@ codex-hooks = { workspace = true } codex-login = { workspace = true } codex-model-provider-info = { workspace = true } codex-models-manager = { workspace = true } +codex-network-proxy = { workspace = true } codex-protocol = { workspace = true } codex-thread-store = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-cargo-bin = { workspace = true } codex-utils-path-uri = { workspace = true } +codex-utils-rustls-provider = { workspace = true } ctor = { workspace = true } futures = { workspace = true } notify = { workspace = true } opentelemetry = { workspace = true } opentelemetry_sdk = { workspace = true } +rcgen = { workspace = true } regex-lite = { workspace = true } serde_json = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["net", "test-util", "time"] } +tokio-rustls = { workspace = true } tokio-tungstenite = { workspace = true } tracing = { workspace = true } tracing-opentelemetry = { workspace = true } diff --git a/codex-rs/core/tests/common/lib.rs b/codex-rs/core/tests/common/lib.rs index 91848e8b9a..add2d3a932 100644 --- a/codex-rs/core/tests/common/lib.rs +++ b/codex-rs/core/tests/common/lib.rs @@ -32,6 +32,7 @@ pub mod exec_server; pub mod hooks; pub mod process; pub mod responses; +pub mod runtime_auth; pub mod startup; pub mod streaming_sse; pub mod test_codex; diff --git a/codex-rs/core/tests/common/runtime_auth.rs b/codex-rs/core/tests/common/runtime_auth.rs new file mode 100644 index 0000000000..85e4f9280b --- /dev/null +++ b/codex-rs/core/tests/common/runtime_auth.rs @@ -0,0 +1,132 @@ +//! Route isolated first-party auth tests to loopback without relaxing destination checks. + +use anyhow::Result; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpListener; +use tokio::net::TcpStream; +use tokio::process::Command; +use tokio::task::JoinSet; +use tokio_rustls::TlsAcceptor; +use tokio_rustls::rustls; +use wiremock::MockServer; + +pub const PROBE_SERVER: &str = "CODEX_RUNTIME_AUTH_TEST_SERVER"; + +/// Run one test in an isolated subprocess with first-party HTTP routed to the mock server. +/// An optional Realtime target receives chatgpt.com Realtime WebSocket upgrades. +pub async fn run_harness_through_proxy( + server: &MockServer, + test_name: &str, + realtime_target: Option, +) -> Result<()> { + // The production destination check stays intact. A test-only CA and CONNECT proxy route + // the real host names to wiremock, only inside this isolated test subprocess. + codex_utils_rustls_provider::ensure_rustls_crypto_provider(); + let rcgen::CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![ + "api.openai.com".into(), + "auth.openai.com".into(), + "chatgpt.com".into(), + "us.chatgpt.com".into(), + ])?; + let ca_dir = tempfile::tempdir()?; + let ca_path = ca_dir.path().join("ca.pem"); + std::fs::write(&ca_path, cert.pem())?; + let acceptor = TlsAcceptor::from(Arc::new( + rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![cert.der().clone()], signing_key.into())?, + )); + let proxy = TcpListener::bind("127.0.0.1:0").await?; + let proxy_url = format!("http://{}", proxy.local_addr()?); + let target = *server.address(); + let proxy_task = tokio::spawn(async move { + let mut connections = JoinSet::new(); + loop { + let (mut stream, _) = proxy.accept().await?; + let acceptor = acceptor.clone(); + connections.spawn(async move { + let mut headers = Vec::new(); + while !headers.ends_with(b"\r\n\r\n") && headers.len() < 8192 { + headers.push(stream.read_u8().await?); + } + let headers = std::str::from_utf8(&headers)?; + anyhow::ensure!( + matches!( + headers.lines().next(), + Some( + "CONNECT api.openai.com:443 HTTP/1.1" + | "CONNECT auth.openai.com:443 HTTP/1.1" + | "CONNECT chatgpt.com:443 HTTP/1.1" + | "CONNECT us.chatgpt.com:443 HTTP/1.1" + ) + ), + "unexpected proxy destination" + ); + stream + .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n") + .await?; + let realtime_host = headers.starts_with("CONNECT chatgpt.com:443 "); + let mut stream = acceptor.accept(stream).await?; + let mut prefix = Vec::new(); + let mut destination = target; + if let Some(realtime_target) = realtime_target.filter(|_| realtime_host) { + // Background HTTP requests share this host with Realtime. Only upgrades + // belong on the WebSocket fixture; ordinary HTTP still goes to wiremock. + while !prefix.ends_with(b"\r\n\r\n") && prefix.len() < 65536 { + prefix.push(stream.read_u8().await?); + } + anyhow::ensure!( + prefix.ends_with(b"\r\n\r\n"), + "oversized test request headers" + ); + let request_headers = std::str::from_utf8(&prefix)?; + let realtime_path = request_headers.lines().next().is_some_and(|line| { + line.starts_with("GET /v1/realtime") || line.starts_with("GET /v1/live") + }); + if realtime_path + && request_headers.lines().any(|line| { + line.split_once(':').is_some_and(|(name, value)| { + name.eq_ignore_ascii_case("upgrade") + && value.trim().eq_ignore_ascii_case("websocket") + }) + }) + { + destination = realtime_target; + } + } + let mut target = TcpStream::connect(destination).await?; + target.write_all(&prefix).await?; + tokio::io::copy_bidirectional(&mut stream, &mut target).await?; + Ok::<_, anyhow::Error>(()) + }); + } + #[allow(unreachable_code)] + Ok::<_, anyhow::Error>(()) + }); + let mut command = Command::new(std::env::current_exe()?); + command + .args(["--exact", test_name, "--nocapture"]) + .kill_on_drop(true); + for &key in codex_network_proxy::PROXY_ENV_KEYS { + command.env_remove(key); + } + command + .env(PROBE_SERVER, server.uri()) + .env("HTTPS_PROXY", proxy_url) + .env("NO_PROXY", codex_network_proxy::DEFAULT_NO_PROXY_VALUE) + .env("CODEX_CA_CERTIFICATE", ca_path); + let output = tokio::time::timeout(Duration::from_secs(90), command.output()).await; + proxy_task.abort(); + let output = output??; + assert!( + output.status.success(), + "harness subprocess failed\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + Ok(()) +} diff --git a/codex-rs/core/tests/suite/request_plugin_install.rs b/codex-rs/core/tests/suite/request_plugin_install.rs index 38cb627ed5..834e338eff 100644 --- a/codex-rs/core/tests/suite/request_plugin_install.rs +++ b/codex-rs/core/tests/suite/request_plugin_install.rs @@ -75,6 +75,9 @@ use wiremock::matchers::query_param; use super::rmcp_client::remote_aware_environment_id; use super::rmcp_client::remote_aware_stdio_server_bin; +#[path = "request_plugin_install_runtime_auth.rs"] +mod runtime_auth_tests; + const TOOL_SEARCH_TOOL_NAME: &str = "tool_search"; const LIST_AVAILABLE_PLUGINS_TO_INSTALL_TOOL_NAME: &str = "list_available_plugins_to_install"; const REQUEST_PLUGIN_INSTALL_TOOL_NAME: &str = "request_plugin_install"; diff --git a/codex-rs/core/tests/suite/request_plugin_install_runtime_auth.rs b/codex-rs/core/tests/suite/request_plugin_install_runtime_auth.rs new file mode 100644 index 0000000000..eb0e51ece2 --- /dev/null +++ b/codex-rs/core/tests/suite/request_plugin_install_runtime_auth.rs @@ -0,0 +1,248 @@ +//! Exercise recommended-plugin metadata after resuming with another account. + +use super::*; +use base64::Engine; +use codex_features::RuntimeAuthMode; +use codex_history::InitialHistory; +use codex_history::ResumedHistory; +use codex_protocol::mcp::ClientMcpExtensions; +use core_test_support::runtime_auth::PROBE_SERVER; +use core_test_support::runtime_auth::run_harness_through_proxy; +use pretty_assertions::assert_eq; +use test_case::test_case; + +fn actor_token(actor: &str) -> String { + let claims = json!({ + "exp": 4_102_444_800_u64, + "session_id": format!("session-{actor}"), + "iss": "https://auth.openai.com", + "https://api.openai.com/auth": { + "chatgpt_user_id": format!("user-{actor}"), + "chatgpt_account_id": format!("account-{actor}"), + }, + }); + format!( + "e30.{}.c2ln", + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(claims.to_string()) + ) +} + +#[test_case(RuntimeAuthMode::DualHeader; "dual_header")] +#[test_case(RuntimeAuthMode::AssertionOnly; "assertion_only")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn resumed_plugin_install_uses_the_threads_account(mode: RuntimeAuthMode) -> Result<()> { + skip_if_no_network!(Ok(())); + const PLUGIN_ID: &str = "plugin_connector_github"; + if std::env::var(PROBE_SERVER).is_ok() { + let server = start_mock_server().await; + let requests = mount_sse_sequence( + &server, + vec![ + sse(vec![ev_completed("original")]), + sse(vec![ + ev_function_call( + "install-github", + REQUEST_PLUGIN_INSTALL_TOOL_NAME, + &json!({ + "plugin_id": "github@openai-curated-remote", + "suggest_reason": "Use GitHub for this request", + }) + .to_string(), + ), + ev_completed("install"), + ]), + sse(vec![ev_completed("finished")]), + ], + ) + .await; + let mut builder = test_codex() + .with_auth(CodexAuth::from_external_chatgpt_tokens( + &actor_token("a"), + "account-a", + Some("plus"), + )?) + .with_config(move |config| { + config + .permissions + .set_permission_profile(PermissionProfile::Disabled) + .expect("test config should allow disabled permissions"); + configure_apps_without_search_tool(config, "https://chatgpt.com"); + config + .features + .enable(Feature::RuntimeAgentInstances) + .unwrap(); + config.runtime_auth_mode = mode; + }); + let test = builder.build_with_auto_env(&server).await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Original account".into(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.codex.ensure_rollout_materialized().await; + test.codex.flush_rollout().await?; + let rollout = test.codex.rollout_path().context("original rollout")?; + test.codex.shutdown_and_wait().await?; + let _ = test + .thread_manager + .remove_thread(&test.session_configured.thread_id) + .await; + let history = test + .thread_store + .load_latest_model_context(codex_thread_store::LoadThreadHistoryParams { + thread_id: test.session_configured.thread_id, + include_archived: false, + }) + .await?; + let replacement = codex_core::test_support::auth_manager_from_auth_with_home( + CodexAuth::from_external_chatgpt_tokens(&actor_token("b"), "account-b", Some("plus"))?, + test.codex_home_path().to_path_buf(), + ); + let resumed = test + .thread_manager + .resume_thread_with_history( + test.config.clone(), + InitialHistory::Resumed(ResumedHistory { + history_revision: None, + conversation_id: history.thread_id, + history: Arc::new(history.items), + rollout_path: Some(rollout), + }), + replacement, + /*parent_trace*/ None, + ClientMcpExtensions::default(), + ) + .await?; + resumed + .thread + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Use GitHub".into(), + text_elements: Vec::new(), + }])) + .await?; + let elicitation = wait_for_event_match(&resumed.thread, |event| match event { + EventMsg::ElicitationRequest(request) => Some(Ok(request.clone())), + EventMsg::TurnComplete(_) => Some(Err(anyhow::anyhow!( + "the resumed account must reach the plugin installation prompt" + ))), + _ => None, + }) + .await?; + let ElicitationRequest::Form { + meta: Some(meta), .. + } = &elicitation.request + else { + anyhow::bail!("expected plugin installation metadata"); + }; + assert_eq!(meta["remote_plugin_id"], PLUGIN_ID); + resumed + .thread + .submit(Op::ResolveElicitation { + server_name: elicitation.server_name, + request_id: elicitation.id, + decision: ElicitationAction::Decline, + content: None, + meta: None, + }) + .await?; + wait_for_event(&resumed.thread, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + assert_eq!(requests.requests().len(), 3); + resumed.thread.shutdown_and_wait().await?; + return Ok(()); + } + + let server = start_mock_server().await; + let _apps = AppsTestServer::mount(&server).await?; + mount_recommendations( + &server, + ResponseTemplate::new(200).set_body_json(json!({ + "enabled": true, + "plugins": [{"id": PLUGIN_ID, "name": "github", "display_name": "GitHub"}], + })), + ) + .await; + Mock::given(method("POST")) + .and(path("/agent/register")) + .respond_with(|request: &wiremock::Request| { + let body: Value = request.body_json().unwrap(); + let actor = if body["actor_auth"]["credential"] == actor_token("a") { + "a" + } else { + "b" + }; + assert_eq!(body["actor_auth"]["credential"], actor_token(actor)); + ResponseTemplate::new(200).set_body_json(json!({ + "runtime_agent_instance_id": format!("runtime-{actor}"), + "agent_host_id": "host-a", + "actor": { + "kind": "chatgpt", + "user_id": format!("user-{actor}"), + "account_id": format!("account-{actor}"), + }, + })) + }) + .expect(2) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path(format!("/ps/plugins/{PLUGIN_ID}"))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "id": PLUGIN_ID, "name": "github", "scope": "GLOBAL", "status": "ENABLED", + "installation_policy": "AVAILABLE", "authentication_policy": "ON_USE", + "release": { + "display_name": "GitHub", "description": "Work with GitHub repositories.", + "app_ids": [], "interface": {}, + }, + }))) + .expect(1) + .mount(&server) + .await; + let case = match mode { + RuntimeAuthMode::DualHeader => "dual_header", + RuntimeAuthMode::AssertionOnly => "assertion_only", + }; + run_harness_through_proxy( + &server, + &format!("suite::request_plugin_install::runtime_auth_tests::resumed_plugin_install_uses_the_threads_account::{case}"), + /*realtime_target*/ None, + ).await?; + let requests = server + .received_requests() + .await + .context("recorded requests")?; + let metadata = requests + .iter() + .find(|request| request.url.path() == format!("/ps/plugins/{PLUGIN_ID}")) + .context("plugin metadata request")?; + let proof = match mode { + RuntimeAuthMode::DualHeader => { + assert_eq!( + metadata.headers["authorization"], + format!("Bearer {}", actor_token("b")) + ); + metadata.headers["x-openai-agent-assertion"].to_str()? + } + RuntimeAuthMode::AssertionOnly => { + assert!(!metadata.headers.contains_key("x-openai-agent-assertion")); + metadata.headers["authorization"] + .to_str()? + .strip_prefix("AgentAssertion ") + .context("assertion auth")? + } + }; + let claims: Value = serde_json::from_slice( + &base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(proof.split('.').nth(2).context("assertion payload")?)?, + )?; + assert_eq!(claims["runtime_agent_instance"], "runtime-b"); + server.verify().await; + Ok(()) +} diff --git a/codex-rs/core/tests/suite/runtime_auth.rs b/codex-rs/core/tests/suite/runtime_auth.rs index 65a831ca7c..51f417aff8 100644 --- a/codex-rs/core/tests/suite/runtime_auth.rs +++ b/codex-rs/core/tests/suite/runtime_auth.rs @@ -87,6 +87,27 @@ async fn harness_shares_runtime_and_honors_thread_auth_config() -> Result<()> { .mount(&server) .await; } + Mock::given(path("/backend-api/files")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "file_id": "file-a", "upload_url": format!("{}/blob", server.uri()), + }))) + .expect(2) + .mount(&server) + .await; + Mock::given(path("/backend-api/files/file-a/uploaded")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "status": "success", "download_url": format!("{}/blob", server.uri()), + "file_name": "empty.txt", "file_size_bytes": 0, + }))) + .expect(2) + .mount(&server) + .await; + Mock::given(path("/blob")) + .and(method("PUT")) + .respond_with(ResponseTemplate::new(200)) + .expect(2) + .mount(&server) + .await; let responses = responses::mount_response_sequence( &server, (0..14) @@ -131,11 +152,13 @@ async fn harness_shares_runtime_and_honors_thread_auth_config() -> Result<()> { .await; // The production destination check stays intact. A test-only CA and CONNECT proxy route - // the two real host names to wiremock, only inside this isolated test subprocess. + // the real host names to wiremock, only inside this isolated test subprocess. codex_utils_rustls_provider::ensure_rustls_crypto_provider(); let rcgen::CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![ "api.openai.com".into(), "auth.openai.com".into(), + "chatgpt.com".into(), + "us.chatgpt.com".into(), ])?; let ca_dir = tempfile::tempdir()?; let ca_path = ca_dir.path().join("ca.pem"); @@ -165,6 +188,8 @@ async fn harness_shares_runtime_and_honors_thread_auth_config() -> Result<()> { Some( "CONNECT api.openai.com:443 HTTP/1.1" | "CONNECT auth.openai.com:443 HTTP/1.1" + | "CONNECT chatgpt.com:443 HTTP/1.1" + | "CONNECT us.chatgpt.com:443 HTTP/1.1" ) ), "unexpected proxy destination" @@ -337,6 +362,29 @@ async fn harness_shares_runtime_and_honors_thread_auth_config() -> Result<()> { assert_eq!(payload["runtime_agent_instance"], "runtime-shared"); assert!(proof_ids.insert(payload["id"].as_str().unwrap().to_owned())); } + let files: Vec<_> = requests + .iter() + .filter(|r| r.url.path().starts_with("/backend-api/files")) + .collect(); + assert_eq!(files.len(), 4); + for (index, request) in files.iter().enumerate() { + let authorization = request.headers["authorization"].to_str()?; + let proof = if index < 2 { + assert_eq!(authorization, "Bearer key-a"); + request.headers["x-openai-agent-assertion"].to_str()? + } else { + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + authorization + .strip_prefix("AgentAssertion ") + .context("assertion-only file auth")? + }; + let payload: Value = serde_json::from_slice( + &base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(proof.split('.').nth(2).unwrap())?, + )?; + assert_eq!(payload["runtime_agent_instance"], "runtime-shared"); + assert!(proof_ids.insert(payload["id"].as_str().unwrap().to_owned())); + } server.verify().await; Ok(()) } @@ -420,6 +468,7 @@ async fn exercise_harness(server_uri: String) -> Result<()> { .await?; turn(&parent.thread, "parent-dual").await?; exercise_provider_destinations(&parent.thread, &enabled).await?; + upload_with_harness(&test.thread_manager, &enabled).await?; let mut child_config = enabled.clone(); child_config.runtime_auth_mode = RuntimeAuthMode::AssertionOnly; parent.thread.ensure_rollout_materialized().await; @@ -442,13 +491,14 @@ async fn exercise_harness(server_uri: String) -> Result<()> { agent_nickname: None, agent_role: None, })), - ..StartThreadOptions::new(child_config) + ..StartThreadOptions::new(child_config.clone()) }, prepared, ) .await?; turn(&child.thread, "child-assertion").await?; exercise_provider_destinations(&child.thread, &enabled).await?; + upload_with_harness(&test.thread_manager, &child_config).await?; parent .thread .submit(Op::Review { @@ -648,3 +698,33 @@ async fn turn(thread: &Arc, label: &str) -> Result<()> { wait_for_event(thread, |e| matches!(e, EventMsg::TurnComplete(_))).await; Ok(()) } + +async fn upload_with_harness( + manager: &codex_core::ThreadManager, + config: &codex_core::config::Config, +) -> Result<()> { + let (runtime, mode) = manager + .runtime_auth_for_config(config)? + .context("enabled harness runtime")?; + let selected = runtime.auth().await?; + let clients = + codex_http_client::RouteAwareClientPool::new_without_redirects_or_request_logging( + config.http_client_factory(), + codex_http_client::ClientRouteClass::Api, + ) + .with_account_binding(selected.http_client_factory().network_policy()) + .context("upload auth and transport share the application policy")?; + let auth = selected.into_auth_provider(mode); + let uploaded = codex_api::upload_openai_file( + &config.chatgpt_base_url, + auth.as_ref(), + &clients, + "empty.txt".into(), + /*file_size_bytes*/ 0, + || async { Ok(futures::stream::empty()) }, + /*hosted_upload*/ None, + ) + .await?; + assert_eq!(uploaded.uri, "sediment://file-a"); + Ok(()) +} diff --git a/codex-rs/external-agent-migration/src/plugins.rs b/codex-rs/external-agent-migration/src/plugins.rs index 26932c4278..1382d1a83c 100644 --- a/codex-rs/external-agent-migration/src/plugins.rs +++ b/codex-rs/external-agent-migration/src/plugins.rs @@ -1,6 +1,6 @@ use codex_analytics::PluginInstallSource; use codex_core::config::ConfigBuilder; -use codex_core::plugins_manager_for_config; +use codex_core::plugins_manager_for_config_with_runtime; use codex_core_plugins::PluginInstallError; use codex_core_plugins::PluginInstallRequest; use codex_core_plugins::marketplace::MarketplaceError; @@ -106,8 +106,11 @@ impl ExternalAgentConfigService { .map_err(|err| io::Error::other(format!("failed to load config: {err}")))?; let requirements = config.config_layer_stack.requirements().clone(); let mut outcome = PluginImportOutcome::default(); - let plugins_manager = plugins_manager_for_config(&config, Arc::clone(&self.auth_manager)) - .with_plugin_install_source(PluginInstallSource::ExternalAgentMigration); + let plugins_manager = + plugins_manager_for_config_with_runtime(&config, Arc::clone(&self.auth_manager)) + .await + .map_err(io::Error::other)? + .with_plugin_install_source(PluginInstallSource::ExternalAgentMigration); if let Some(analytics_events_client) = self.analytics_events_client.clone() { plugins_manager.set_analytics_events_client(analytics_events_client); } diff --git a/codex-rs/http-client/src/route_aware_client_pool.rs b/codex-rs/http-client/src/route_aware_client_pool.rs index fe39417e69..796a4a2659 100644 --- a/codex-rs/http-client/src/route_aware_client_pool.rs +++ b/codex-rs/http-client/src/route_aware_client_pool.rs @@ -18,6 +18,7 @@ use crate::ClientRouteClass; use crate::HttpClient; use crate::HttpClientBuilder; use crate::HttpClientFactory; +use crate::NetworkPolicy; use crate::NetworkPolicyDenied; use crate::OutboundProxyPolicy; use crate::OutboundProxyRoute; @@ -328,6 +329,19 @@ impl RouteAwareClientPool { pool } + /// Bind an operation to the account that supplied its credentials while retaining the + /// caller's restrictions and the existing transport cache. + pub fn with_account_binding(&self, binding: &NetworkPolicy) -> Option { + let policy = self + .http_client_factory + .network_policy() + .clone() + .with_account_binding(binding)?; + let mut pool = self.clone(); + pool.http_client_factory = pool.http_client_factory.with_network_policy(policy); + Some(pool) + } + /// Retries recognized TLS protocol-negotiation failures once using rustls. /// /// Successful fallback is remembered for each HTTPS origin and resolved outbound route. diff --git a/codex-rs/http-client/src/route_aware_policy_tests.rs b/codex-rs/http-client/src/route_aware_policy_tests.rs index ec88589bef..f35256b6eb 100644 --- a/codex-rs/http-client/src/route_aware_policy_tests.rs +++ b/codex-rs/http-client/src/route_aware_policy_tests.rs @@ -1,6 +1,7 @@ //! Unavailable policy rejects requests before a socket is opened. use super::*; +use http::StatusCode; use pretty_assertions::assert_eq; #[tokio::test] @@ -93,3 +94,39 @@ async fn managed_redirect_failure_is_tracked_before_proxy_retry() { assert!(redirected.load(std::sync::atomic::Ordering::Relaxed)); assert_eq!(server.join().unwrap().len(), 1); } + +#[tokio::test] +async fn credential_binding_rejects_old_operations_after_account_replacement() { + let (address, server) = super::tests::spawn_response_server(vec![ + "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".into(), + "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".into(), + ]); + let url = format!("http://{address}/account"); + let controller = crate::NetworkPolicyController::default(); + let policy = controller.policy(); + assert!(controller.publish(policy.revision(), crate::DestinationPolicy::Unrestricted)); + let pool = RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) + .with_network_policy(policy.clone()), + ClientRouteClass::Api, + ); + let old = pool + .with_account_binding(&policy.clone().for_current_account()) + .unwrap(); + assert_eq!(old.get(&url).send().await.unwrap().status(), StatusCode::OK); + + policy.invalidate(); + assert!(controller.publish(policy.revision(), crate::DestinationPolicy::Unrestricted)); + assert!(matches!( + old.get(&url).send().await, + Err(RouteAwareRequestError::Policy(NetworkPolicyDenied::Revoked)) + )); + let fresh = pool + .with_account_binding(&policy.for_current_account()) + .unwrap(); + assert_eq!( + fresh.get(&url).send().await.unwrap().status(), + StatusCode::OK + ); + assert_eq!(server.join().unwrap().len(), 2); +} diff --git a/codex-rs/model-provider/src/auth.rs b/codex-rs/model-provider/src/auth.rs index 84a6baa913..95dca9a74f 100644 --- a/codex-rs/model-provider/src/auth.rs +++ b/codex-rs/model-provider/src/auth.rs @@ -13,6 +13,7 @@ use codex_api::SharedAuthProvider; use codex_features::RuntimeAuthMode; use codex_http_client::CHATGPT_ACCOUNT_ID_HEADER; use codex_http_client::HttpClientFactory; +use codex_http_client::RouteAwareClientPool; use codex_login::AuthHeaders; use codex_login::AuthManager; use codex_login::CodexAuth; @@ -112,6 +113,24 @@ impl ResolvedProviderAuth { } } + /// Preserve the caller's connection pool and bind it to the selected login. + pub fn bind_http_client_pool( + &self, + pool: &RouteAwareClientPool, + ) -> Result { + match &self.runtime { + Some(runtime) => pool + .with_account_binding(runtime.http_client_factory().network_policy()) + .ok_or_else(|| { + AuthError::Build( + "runtime credentials and transport must share the application network policy" + .to_string(), + ) + }), + None => Ok(pool.clone()), + } + } + /// Reject account selectors that disagree with the selected runtime. pub fn validate_headers(&self, headers: &HeaderMap) -> codex_protocol::error::Result<()> { if let Some(runtime) = &self.runtime { diff --git a/codex-rs/tui/src/app/background_requests.rs b/codex-rs/tui/src/app/background_requests.rs index 6894b98915..21216fec23 100644 --- a/codex-rs/tui/src/app/background_requests.rs +++ b/codex-rs/tui/src/app/background_requests.rs @@ -1386,6 +1386,11 @@ mod tests { .respond_with(wiremock::ResponseTemplate::new(/*s*/ 200).set_body_string("{}")) .mount(&server) .await; + std::fs::write( + app.config.codex_home.join("config.toml"), + format!("chatgpt_base_url = {:?}\n", server.uri()), + ) + .expect("persist endpoint for account config reloads"); write_chatgpt_auth( app.config.codex_home.as_path(), ChatGptAuthFixture::new("chatgpt-token").account_id("account-123"),