openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Extend runtime auth to account, plugin, and file requests (#59564)

Browse Source 
## What changed

- Use the shared runtime and effective configuration for account API clients and remote plugin requests, including standalone plugin commands.
- Support dual-header and assertion-only authentication for remote plugins and MCP file uploads when runtime agent instances are enabled.
- Keep plugin credentials and caches scoped to the thread's account when resuming with different credentials. Bind request pools and multi-step plugin storage operations to the selected account so account replacement revokes stale operations.
- Protect runtime credentials on cross-origin redirects while preserving configured routing policies and legacy authentication for custom backends.
- Apply server-provided `upload_headers` to file storage uploads.

## Testing

Add coverage for shared runtime registration, both authentication modes, configuration reloads, resumed-thread account selection, account replacement during plugin uploads and downloads, and credential stripping on redirects. Extend file upload tests to verify runtime assertions and storage-only upload headers.

GitOrigin-RevId: 68aef81051a6f9bf919fffda52f372b108fce805
This commit is contained in:
cooper-oai
2026-11-05 03:49:13 +00:00
committed by copyberry
parent c90354172c
commit 5c0b1ac71f
55 changed files with 2145 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+8
View File
@@ -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",
codex-rs/app-server/src/request_processors/account_processor.rs
+10 -20
View File
@@ -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))
codex-rs/app-server/src/request_processors/account_processor/backend_client.rs
+31
View File
@@ -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<BackendClient, JSONRPCErrorError>> {
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()))
})
}
}
codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs
+1 -5
View File
@@ -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
}
}
codex-rs/app-server/src/request_processors/account_processor/workspace_routing.rs
+10 -2
View File
@@ -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) => {
codex-rs/app-server/src/request_processors/plugins.rs
+36 -11
View File
@@ -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<RemotePluginServiceConfig, JSONRPCErrorError> {
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<PluginShareListResponse, JSONRPCErrorError> {
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<String>,
) -> Result<PluginInstallResponse, JSONRPCErrorError> {
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<PluginUninstallResponse, JSONRPCErrorError> {
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
codex-rs/app-server/src/request_processors/plugins/search.rs
+1 -1
View File
@@ -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,
codex-rs/app-server/tests/suite/v2/plugin_install.rs
+3
View File
@@ -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);
codex-rs/app-server/tests/suite/v2/plugin_runtime_auth.rs
+194
View File
@@ -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::<Vec<_>>();
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(())
}
codex-rs/app-server/tests/suite/v2/rate_limits.rs
+13 -1
View File
@@ -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(())
}
codex-rs/cli/src/marketplace_cmd.rs
+7 -3
View File
@@ -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 || {
codex-rs/cli/src/mcp_cmd.rs
+5 -5
View File
@@ -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<McpManager> {
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))
}
codex-rs/cli/src/plugin_cmd.rs
+7 -6
View File
@@ -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,
codex-rs/codex-api/src/files.rs
+32 -6
View File
@@ -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<HashMap<String, String>>,
#[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)
codex-rs/core-plugins/Cargo.toml
+6 -2
View File
@@ -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 }
codex-rs/core-plugins/src/http_client_selector.rs
+22
View File
@@ -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<dyn HttpClientSelector>;
fn request(&self, method: Method, url: &str) -> RequestBuilder;
fn request_for_account(
&self,
method: Method,
url: &str,
binding: &NetworkPolicy,
) -> Option<RequestBuilder>;
fn outbound_proxy_policy(&self) -> OutboundProxyPolicy;
}
impl HttpClientSelector for RouteAwareClientPool {
fn for_current_account(&self) -> Arc<dyn HttpClientSelector> {
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<RequestBuilder> {
Some(self.with_account_binding(binding)?.request(method, url))
}
fn outbound_proxy_policy(&self) -> OutboundProxyPolicy {
RouteAwareClientPool::outbound_proxy_policy(self)
}
codex-rs/core-plugins/src/manager.rs
+76 -19
View File
@@ -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<String>,
http_client_factory: HttpClientFactory,
remote_http_clients: Arc<OnceLock<Arc<dyn HttpClientSelector>>>,
runtime_http_clients: Arc<OnceLock<Arc<dyn HttpClientSelector>>>,
runtime_auth_mode: Option<RuntimeAuthMode>,
}
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<RuntimeAuthMode>) -> 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<RemoteCatalogCacheRefreshState>,
restriction_product: Option<Product>,
auth_manager: Arc<AuthManager>,
runtime_auth: RuntimeAuthSlot,
analytics_events_client: RwLock<Option<AnalyticsEventsClient>>,
plugin_install_source: PluginInstallSource,
}
@@ -692,6 +718,34 @@ fn target_curated_marketplace(auth_mode: Option<AuthMode>) -> 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<AuthManager>,
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<AuthManager>,
@@ -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<Product>,
auth_manager: Arc<AuthManager>,
runtime_auth: RuntimeAuthSlot,
skill_root_loader: Arc<dyn SkillRootLoader<PluginSkillRoot>>,
) -> 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::<Vec<_>>();
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,
);
codex-rs/core-plugins/src/plugin_measurement_references.rs
+6 -1
View File
@@ -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<Option<Arc<PluginMeasurementReference>>> {
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"))?
codex-rs/core-plugins/src/plugin_measurement_references_tests.rs
+4 -2
View File
@@ -174,7 +174,9 @@ impl Fixture {
&self,
target: &PluginMeasurementTarget,
) -> Result<Option<Arc<PluginMeasurementReference>>> {
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()
);
codex-rs/core-plugins/src/plugins_config_input_tests.rs
+9 -5
View File
@@ -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",
codex-rs/core-plugins/src/recommended_plugin_install.rs
+5 -8
View File
@@ -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<Option<DiscoverableTool>, 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);
};
codex-rs/core-plugins/src/remote.rs
+143 -57
View File
@@ -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<dyn HttpClientSelector>,
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<dyn HttpClientSelector> = 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<String>,
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<String>,
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<Output = Result<RequestBuilder, codex_api::AuthError>>
+ 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<RequestBuilder, RemotePluginCatalogError> {
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<T: for<'de> Deserialize<'de>>(
@@ -2451,3 +2533,7 @@ async fn send_and_decode<T: for<'de> Deserialize<'de>>(
source,
})
}
#[cfg(test)]
#[path = "remote_runtime_auth_tests.rs"]
mod runtime_auth_tests;
codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs
+1
View File
@@ -136,6 +136,7 @@ pub(crate) async fn sync_remote_installed_plugin_bundles_once_with_snapshot(
previous_plugin_ids: &[PluginId],
) -> Result<RemoteInstalledPluginBundleSyncResult, RemoteInstalledPluginBundleSyncError> {
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,
codex-rs/core-plugins/src/remote/search.rs
+1 -5
View File
@@ -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 {
codex-rs/core-plugins/src/remote/share.rs
+20 -36
View File
@@ -149,6 +149,7 @@ pub async fn save_remote_plugin_share(
access_policy: RemotePluginShareAccessPolicy,
) -> Result<RemotePluginShareSaveResult, RemotePluginCatalogError> {
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<RemoteWorkspacePluginUploadUrlResponse, RemotePluginCatalogError> {
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
}
codex-rs/core-plugins/src/remote/share/checkout.rs
+1
View File
@@ -40,6 +40,7 @@ pub async fn checkout_remote_plugin_share(
codex_home: &Path,
remote_plugin_id: &str,
) -> Result<RemotePluginShareCheckoutResult, RemotePluginCatalogError> {
let config = &config.for_current_account();
let detail = super::super::fetch_remote_plugin_detail_with_download_urls(
config,
auth,
codex-rs/core-plugins/src/remote_bundle.rs
+4
View File
@@ -1148,3 +1148,7 @@ mod tests {
}
}
}
#[cfg(test)]
#[path = "remote_bundle_runtime_auth_tests.rs"]
mod runtime_auth_tests;
codex-rs/core-plugins/src/remote_bundle_runtime_auth_tests.rs
+55
View File
@@ -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"));
}
}
codex-rs/core-plugins/src/remote_legacy.rs
+13 -19
View File
@@ -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()
codex-rs/core-plugins/src/remote_mutations.rs
+4 -2
View File
@@ -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
codex-rs/core-plugins/src/remote_mutations_tests.rs
+363 -1
View File
@@ -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<AtomicBool>,
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<dyn crate::http_client_selector::HttpClientSelector> {
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<codex_http_client::RequestBuilder> {
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<PluginsManager>,
auth: CodexAuth,
switch_before_storage: Arc<AtomicBool>,
}
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;
}
}
}
codex-rs/core-plugins/src/remote_runtime_auth_tests.rs
+307
View File
@@ -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());
}
}
codex-rs/core-plugins/src/test_support.rs
+29 -4
View File
@@ -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<Self>, Arc<Mutex<Vec<String>>>) {
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<dyn HttpClientSelector> {
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<RequestBuilder> {
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,
codex-rs/core/src/config/config_tests.rs
+3 -1
View File
@@ -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(
codex-rs/core/src/config/mod.rs
+5
View File
@@ -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.
codex-rs/core/src/connectors.rs
+3 -5
View File
@@ -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<AccessibleConnectorsStatus> {
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,
codex-rs/core/src/lib.rs
+1
View File
@@ -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)]
codex-rs/core/src/mcp.rs
+10
View File
@@ -146,6 +146,16 @@ impl McpManager {
}
}
pub(crate) fn for_plugins_manager(&self, plugins_manager: Arc<PluginsManager>) -> 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<ToolInfo> {
self.codex_apps_tools_cache.clone()
}
codex-rs/core/src/mcp_openai_file.rs
+30 -3
View File
@@ -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,
codex-rs/core/src/plugins/mod.rs
+36
View File
@@ -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<AuthManager>,
) -> anyhow::Result<PluginsManager> {
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,
)),
))
}
codex-rs/core/src/plugins/skill_snapshot_tests.rs
+1
View File
@@ -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;
codex-rs/core/src/session/session.rs
+18 -5
View File
@@ -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,
codex-rs/core/src/thread_manager.rs
+56 -11
View File
@@ -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<Option<(Arc<RuntimeAuthManager>, 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<HostSkillsService> {
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,
codex-rs/core/src/tools/handlers/request_plugin_install.rs
+4 -1
View File
@@ -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,
)
codex-rs/core/tests/common/Cargo.toml
+4
View File
@@ -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 }
codex-rs/core/tests/common/lib.rs
+1
View File
@@ -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;
codex-rs/core/tests/common/runtime_auth.rs
+132
View File
@@ -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<std::net::SocketAddr>,
) -> 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(())
}
codex-rs/core/tests/suite/request_plugin_install.rs
+3
View File
@@ -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";
codex-rs/core/tests/suite/request_plugin_install_runtime_auth.rs
+248
View File
@@ -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(())
}
codex-rs/core/tests/suite/runtime_auth.rs
+82 -2
View File
@@ -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<CodexThread>, 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(())
}
codex-rs/external-agent-migration/src/plugins.rs
+6 -3
View File
@@ -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);
}
codex-rs/http-client/src/route_aware_client_pool.rs
+14
View File
@@ -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<Self> {
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.
codex-rs/http-client/src/route_aware_policy_tests.rs
+37
View File
@@ -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);
}
codex-rs/model-provider/src/auth.rs
+19
View File
@@ -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<RouteAwareClientPool, AuthError> {
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 {
codex-rs/tui/src/app/background_requests.rs
+5
View File
@@ -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"),