diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index ff30cdf50f..911c873f5f 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2777,6 +2777,7 @@ dependencies = [ "codex-exec-server", "codex-git-utils", "codex-hooks", + "codex-http-client", "codex-login", "codex-mcp", "codex-model-provider", @@ -2791,6 +2792,7 @@ dependencies = [ "codex-utils-plugins", "dirs", "flate2", + "http 1.4.0", "libc", "pretty_assertions", "regex", diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 4566332102..b7c023eaa9 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -129,6 +129,13 @@ fn load_shared_plugin_ids_by_local_path( }) } +fn remote_plugin_service_config(config: &Config) -> RemotePluginServiceConfig { + RemotePluginServiceConfig::new( + config.chatgpt_base_url.clone(), + config.http_client_factory(), + ) +} + fn share_context_for_source( source: &MarketplacePluginSource, shared_plugin_ids_by_local_path: &std::collections::BTreeMap, @@ -569,9 +576,7 @@ impl PluginRequestProcessor { !explicit_marketplace_kinds && config.features.enabled(Feature::RemotePlugin); let use_remote_global_catalog = include_global_remote && auth_mode.is_some_and(DomainAuthMode::uses_codex_backend); - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let refresh_global_remote_catalog_cache = use_remote_global_catalog && codex_core_plugins::remote::has_cached_global_remote_plugin_catalog( config.codex_home.as_path(), @@ -1018,9 +1023,7 @@ impl PluginRequestProcessor { ); let share_context = match share_context { Some(context) => { - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); match codex_core_plugins::remote::fetch_remote_plugin_share_context( &remote_plugin_service_config, auth.as_ref(), @@ -1128,9 +1131,7 @@ impl PluginRequestProcessor { "remote plugin read is not enabled for marketplace {remote_marketplace_name}" ))); } - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); validate_remote_plugin_id(&plugin_name)?; let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail( &remote_plugin_service_config, @@ -1186,9 +1187,7 @@ impl PluginRequestProcessor { } let auth = self.auth_manager.auth().await; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let remote_skill_detail = codex_core_plugins::remote::fetch_remote_plugin_skill_detail( &remote_plugin_service_config, auth.as_ref(), @@ -1239,9 +1238,7 @@ impl PluginRequestProcessor { validate_client_plugin_share_targets(share_targets)?; } - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&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), @@ -1282,9 +1279,7 @@ impl PluginRequestProcessor { } validate_client_plugin_share_targets(&share_targets)?; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let result = codex_core_plugins::remote::update_remote_plugin_share_targets( &remote_plugin_service_config, auth.as_ref(), @@ -1312,9 +1307,7 @@ impl PluginRequestProcessor { _params: PluginShareListParams, ) -> Result { let (config, auth) = self.load_plugin_share_config_and_auth().await?; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let data = codex_core_plugins::remote::list_remote_plugin_shares( &remote_plugin_service_config, auth.as_ref(), @@ -1351,9 +1344,7 @@ impl PluginRequestProcessor { return Err(invalid_request("invalid remote plugin id")); } - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let result = codex_core_plugins::remote::checkout_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), @@ -1384,9 +1375,7 @@ impl PluginRequestProcessor { return Err(invalid_request("invalid remote plugin id")); } - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); codex_core_plugins::remote::delete_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), @@ -1521,9 +1510,7 @@ impl PluginRequestProcessor { validate_remote_plugin_id(&remote_plugin_id)?; let auth = self.auth_manager.auth().await; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail_with_download_urls( &remote_plugin_service_config, @@ -2011,9 +1998,7 @@ impl PluginRequestProcessor { validate_remote_plugin_id(&plugin_id)?; let auth = self.auth_manager.auth().await; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let uninstall_target = codex_core_plugins::remote::resolve_remote_plugin_uninstall_target( &remote_plugin_service_config, auth.as_ref(), diff --git a/codex-rs/core-plugins/Cargo.toml b/codex-rs/core-plugins/Cargo.toml index ebcd1d4832..5b8d0e6cce 100644 --- a/codex-rs/core-plugins/Cargo.toml +++ b/codex-rs/core-plugins/Cargo.toml @@ -22,6 +22,7 @@ codex-core-skills = { workspace = true } codex-exec-server = { workspace = true } codex-git-utils = { workspace = true } codex-hooks = { workspace = true } +codex-http-client = { workspace = true } codex-login = { workspace = true } codex-mcp = { workspace = true } codex-model-provider = { workspace = true } @@ -37,6 +38,7 @@ codex-utils-plugins = { workspace = true } chrono = { workspace = true } dirs = { workspace = true } flate2 = { workspace = true } +http = { workspace = true } reqwest = { workspace = true } regex = { workspace = true } semver = { workspace = true } diff --git a/codex-rs/core-plugins/src/discoverable_tests.rs b/codex-rs/core-plugins/src/discoverable_tests.rs index 48482ed430..1dd88969ac 100644 --- a/codex-rs/core-plugins/src/discoverable_tests.rs +++ b/codex-rs/core-plugins/src/discoverable_tests.rs @@ -851,9 +851,10 @@ plugins = true let plugins_manager = PluginsManager::new(codex_home.path().to_path_buf()); fetch_and_cache_global_remote_plugin_catalog( codex_home.path(), - &RemotePluginServiceConfig { - chatgpt_base_url: plugins.chatgpt_base_url.clone(), - }, + &RemotePluginServiceConfig::new( + plugins.chatgpt_base_url.clone(), + crate::test_support::test_http_client_factory(), + ), Some(&auth), ) .await diff --git a/codex-rs/core-plugins/src/http_client_selector.rs b/codex-rs/core-plugins/src/http_client_selector.rs new file mode 100644 index 0000000000..cd1515bde5 --- /dev/null +++ b/codex-rs/core-plugins/src/http_client_selector.rs @@ -0,0 +1,24 @@ +use codex_http_client::OutboundProxyPolicy; +use codex_http_client::RouteAwareClientPool; +use codex_http_client::RouteAwareRequestBuilder; +use http::Method; +use std::fmt::Debug; + +/// 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 request(&self, method: Method, url: &str) -> RouteAwareRequestBuilder; + fn outbound_proxy_policy(&self) -> OutboundProxyPolicy; +} + +impl HttpClientSelector for RouteAwareClientPool { + fn request(&self, method: Method, url: &str) -> RouteAwareRequestBuilder { + RouteAwareClientPool::request(self, method, url) + } + + fn outbound_proxy_policy(&self) -> OutboundProxyPolicy { + RouteAwareClientPool::outbound_proxy_policy(self) + } +} diff --git a/codex-rs/core-plugins/src/lib.rs b/codex-rs/core-plugins/src/lib.rs index eee1939ce5..ce02d0c7ba 100644 --- a/codex-rs/core-plugins/src/lib.rs +++ b/codex-rs/core-plugins/src/lib.rs @@ -1,6 +1,7 @@ mod app_mcp_routing; mod command_migration; mod discoverable; +mod http_client_selector; pub mod installed_marketplaces; pub mod loader; mod manager; diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index d64d8fabdf..8f9a470258 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -70,6 +70,7 @@ use codex_core_skills::PluginSkillSnapshots; use codex_core_skills::config_rules::skill_config_rules_from_stack; use codex_core_skills::loader::MAX_CONCURRENT_ROOT_SCANS; use codex_hooks::plugin_hook_declarations; +use codex_http_client::HttpClientFactory; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_plugin::AppConnectorId; @@ -115,6 +116,7 @@ pub struct PluginsConfigInput { pub plugins_enabled: bool, pub remote_plugin_enabled: bool, pub chatgpt_base_url: String, + http_client_factory: HttpClientFactory, } impl PluginsConfigInput { @@ -123,14 +125,24 @@ impl PluginsConfigInput { plugins_enabled: bool, remote_plugin_enabled: bool, chatgpt_base_url: String, + http_client_factory: HttpClientFactory, ) -> Self { Self { config_layer_stack, plugins_enabled, remote_plugin_enabled, chatgpt_base_url, + http_client_factory, } } + + /// Builds route-aware service state for remote plugin requests. + pub fn remote_plugin_service_config(&self) -> RemotePluginServiceConfig { + RemotePluginServiceConfig::new( + self.chatgpt_base_url.clone(), + self.http_client_factory.clone(), + ) + } } /// Effective-plugin changes that downstream composition layers may act on. @@ -236,9 +248,7 @@ struct ConfiguredMarketplaceUpgradeState { } fn remote_plugin_service_config(config: &PluginsConfigInput) -> RemotePluginServiceConfig { - RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - } + config.remote_plugin_service_config() } fn featured_plugin_ids_cache_key( @@ -1975,7 +1985,7 @@ impl PluginsManager { let use_remote_global_catalog = config.remote_plugin_enabled && auth_manager.current_auth_uses_codex_backend(); if !use_remote_global_catalog { - self.start_curated_repo_sync(); + self.start_curated_repo_sync(config.http_client_factory.clone()); } let should_spawn_marketplace_auto_upgrade = { let mut state = match self.configured_marketplace_upgrade_state.write() { @@ -2356,7 +2366,7 @@ impl PluginsManager { } } - fn start_curated_repo_sync(self: &Arc) { + fn start_curated_repo_sync(self: &Arc, http_client_factory: HttpClientFactory) { if CURATED_REPO_SYNC_STARTED.swap(true, Ordering::SeqCst) { return; } @@ -2364,8 +2374,8 @@ impl PluginsManager { let codex_home = self.codex_home.clone(); if let Err(err) = std::thread::Builder::new() .name("plugins-curated-repo-sync".to_string()) - .spawn( - move || match sync_openai_plugins_repo(codex_home.as_path()) { + .spawn(move || { + match sync_openai_plugins_repo(codex_home.as_path(), http_client_factory) { Ok(curated_plugin_version) => { let configured_curated_plugin_ids = configured_curated_plugin_ids_from_codex_home(codex_home.as_path()); @@ -2389,8 +2399,8 @@ impl PluginsManager { CURATED_REPO_SYNC_STARTED.store(false, Ordering::SeqCst); warn!("failed to sync curated plugins repo: {err}"); } - }, - ) + } + }) { CURATED_REPO_SYNC_STARTED.store(false, Ordering::SeqCst); warn!("failed to start curated plugins repo sync task: {err}"); diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index c6fdc26db8..41cbd4b0bc 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -20,6 +20,7 @@ use crate::startup_sync::curated_plugins_repo_path; use crate::test_support::TEST_CURATED_PLUGIN_CACHE_VERSION; use crate::test_support::TEST_CURATED_PLUGIN_SHA; use crate::test_support::load_plugins_config as load_plugins_config_input; +use crate::test_support::test_http_client_factory; use crate::test_support::write_curated_plugin; use crate::test_support::write_curated_plugin_sha_with as write_curated_plugin_sha; use crate::test_support::write_file; @@ -110,6 +111,7 @@ fn plugins_config_input_with_requirements( /*plugins_enabled*/ true, /*remote_plugin_enabled*/ false, String::new(), + test_http_client_factory(), ) } @@ -2626,6 +2628,7 @@ async fn plugin_cache_ignores_unrelated_session_overrides() { /*plugins_enabled*/ true, /*remote_plugin_enabled*/ false, "https://chatgpt.com".to_string(), + test_http_client_factory(), ) }; let manager = PluginsManager::new(codex_home.path().to_path_buf()); @@ -6071,9 +6074,10 @@ fn remote_installed_plugins_cache_refresh_coalesces_materializations() { }; let request = |change, on_effective_plugins_changed| RemoteInstalledPluginsCacheRefreshRequest { - service_config: RemotePluginServiceConfig { - chatgpt_base_url: "https://example.com".to_string(), - }, + service_config: RemotePluginServiceConfig::new( + "https://example.com".to_string(), + test_http_client_factory(), + ), auth: None, notify: RemoteInstalledPluginsCacheRefreshNotify::IfCacheChanged, on_effective_plugins_changed: Some(on_effective_plugins_changed), diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index 3a907afa22..c8595b6174 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -1,4 +1,5 @@ use crate::app_mcp_routing::apply_app_mcp_routing_policy; +use crate::http_client_selector::HttpClientSelector; use crate::loader::plugin_app_declarations_from_value; use crate::store::PLUGINS_CACHE_DIR; use crate::store::PluginStore; @@ -10,6 +11,9 @@ 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_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::RouteAwareClientPool; use codex_login::CodexAuth; use codex_login::default_client::build_reqwest_client; use codex_plugin::AppConnectorId; @@ -30,6 +34,7 @@ use std::collections::HashSet; use std::fs; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; use std::time::Duration; use tracing::instrument; use url::Url; @@ -120,11 +125,40 @@ const REMOTE_INSTALLED_MARKETPLACE_DISPLAY_ORDER: [(&str, &str); 6] = [ ), ]; -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] pub struct RemotePluginServiceConfig { pub chatgpt_base_url: String, + pub(crate) http_clients: Arc, } +impl RemotePluginServiceConfig { + /// 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. + pub fn new(chatgpt_base_url: String, http_client_factory: HttpClientFactory) -> Self { + let http_clients = + RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging( + http_client_factory, + ClientRouteClass::Api, + ); + Self { + chatgpt_base_url, + http_clients: Arc::new(http_clients), + } + } +} + +impl PartialEq for RemotePluginServiceConfig { + fn eq(&self, other: &Self) -> bool { + self.chatgpt_base_url == other.chatgpt_base_url + && self.http_clients.outbound_proxy_policy() + == other.http_clients.outbound_proxy_policy() + } +} + +impl Eq for RemotePluginServiceConfig {} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct RemotePluginUninstallTarget { pub plugin_id: PluginId, diff --git a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs index e168698722..e2a665beee 100644 --- a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs +++ b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs @@ -570,9 +570,10 @@ mod tests { .expect(1) .mount(&server) .await; - let config = RemotePluginServiceConfig { - chatgpt_base_url: format!("{}/backend-api", server.uri()), - }; + let config = RemotePluginServiceConfig::new( + format!("{}/backend-api", server.uri()), + crate::test_support::test_http_client_factory(), + ); let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); let outcome = sync_remote_installed_plugin_bundles_once( diff --git a/codex-rs/core-plugins/src/remote/share/tests.rs b/codex-rs/core-plugins/src/remote/share/tests.rs index d9b036889a..baaffb335a 100644 --- a/codex-rs/core-plugins/src/remote/share/tests.rs +++ b/codex-rs/core-plugins/src/remote/share/tests.rs @@ -23,9 +23,10 @@ use wiremock::matchers::query_param; use wiremock::matchers::query_param_is_missing; fn test_config(server: &MockServer) -> RemotePluginServiceConfig { - RemotePluginServiceConfig { - chatgpt_base_url: format!("{}/backend-api", server.uri()), - } + RemotePluginServiceConfig::new( + format!("{}/backend-api", server.uri()), + crate::test_support::test_http_client_factory(), + ) } fn test_auth() -> CodexAuth { diff --git a/codex-rs/core-plugins/src/startup_sync.rs b/codex-rs/core-plugins/src/startup_sync.rs index 0dd437af8a..69fe045464 100644 --- a/codex-rs/core-plugins/src/startup_sync.rs +++ b/codex-rs/core-plugins/src/startup_sync.rs @@ -6,15 +6,19 @@ use std::process::Output; use std::process::Stdio; use std::time::Duration; +use self::http_client::StartupSyncHttpClient; +use self::http_client::StartupSyncRequestBuilder; +use codex_http_client::HttpClientFactory; +use codex_login::default_client::default_headers; use codex_otel::CURATED_PLUGINS_STARTUP_SYNC_FINAL_METRIC; use codex_otel::CURATED_PLUGINS_STARTUP_SYNC_METRIC; -use reqwest::Client; +use http::Method; use serde::Deserialize; use tempfile::TempDir; use tracing::warn; use zip::ZipArchive; -use codex_login::default_client::build_reqwest_client; +mod http_client; const GITHUB_API_BASE_URL: &str = "https://api.github.com"; const GITHUB_API_ACCEPT_HEADER: &str = "application/vnd.github+json"; @@ -92,7 +96,10 @@ fn curated_plugins_sha_path(codex_home: &Path) -> PathBuf { codex_home.join(CURATED_PLUGINS_SHA_FILE) } -pub fn sync_openai_plugins_repo(codex_home: &Path) -> Result { +pub fn sync_openai_plugins_repo( + codex_home: &Path, + http_client_factory: HttpClientFactory, +) -> Result { #[cfg(target_os = "macos")] let git_binary = match which::which("git") { Ok(git_path) => macos_git_binary_from_path(git_path, apple_developer_tools_available()), @@ -106,6 +113,7 @@ pub fn sync_openai_plugins_repo(codex_home: &Path) -> Result { git_binary.as_deref(), GITHUB_API_BASE_URL, CURATED_PLUGINS_BACKUP_ARCHIVE_API_URL, + &http_client_factory, ) } @@ -114,6 +122,7 @@ fn sync_openai_plugins_repo_with_transport_overrides( git_binary: Option<&Path>, api_base_url: &str, backup_archive_api_url: &str, + http_client_factory: &HttpClientFactory, ) -> Result { let _file_guard = lock_curated_plugins_startup_sync(codex_home)?; @@ -134,7 +143,7 @@ fn sync_openai_plugins_repo_with_transport_overrides( error = %err, "git sync failed for curated plugin sync; falling back to GitHub HTTP" ); - match sync_openai_plugins_repo_via_http(codex_home, api_base_url) { + match sync_openai_plugins_repo_via_http(codex_home, api_base_url, http_client_factory) { Ok(remote_sha) => { emit_curated_plugins_startup_sync_metric("http", "success"); emit_curated_plugins_startup_sync_final_metric("http", "success"); @@ -162,6 +171,7 @@ fn sync_openai_plugins_repo_with_transport_overrides( let result = sync_openai_plugins_repo_via_backup_archive( codex_home, backup_archive_api_url, + http_client_factory, ); let status = if result.is_ok() { "success" } else { "failure" }; emit_curated_plugins_startup_sync_metric("export_archive", status); @@ -319,6 +329,7 @@ fn run_git_in_repo( fn sync_openai_plugins_repo_via_http( codex_home: &Path, api_base_url: &str, + http_client_factory: &HttpClientFactory, ) -> Result { let repo_path = curated_plugins_repo_path(codex_home); let sha_path = codex_home.join(CURATED_PLUGINS_SHA_FILE); @@ -326,7 +337,9 @@ fn sync_openai_plugins_repo_via_http( .enable_all() .build() .map_err(|err| format!("failed to create curated plugins sync runtime: {err}"))?; - let remote_sha = runtime.block_on(fetch_curated_repo_remote_sha(api_base_url))?; + let http_clients = StartupSyncHttpClient::new(http_client_factory); + let remote_sha = + runtime.block_on(fetch_curated_repo_remote_sha(&http_clients, api_base_url))?; let local_sha = read_sha_file(&sha_path); if local_sha.as_deref() == Some(remote_sha.as_str()) && repo_path.is_dir() { @@ -334,7 +347,11 @@ fn sync_openai_plugins_repo_via_http( } let staged_repo_dir = prepare_curated_repo_parent_and_temp_dir(&repo_path)?; - let zipball_bytes = runtime.block_on(fetch_curated_repo_zipball(api_base_url, &remote_sha))?; + let zipball_bytes = runtime.block_on(fetch_curated_repo_zipball( + &http_clients, + api_base_url, + &remote_sha, + ))?; extract_zipball_to_dir(&zipball_bytes, staged_repo_dir.path())?; ensure_marketplace_manifest_exists(staged_repo_dir.path())?; activate_curated_repo(&repo_path, staged_repo_dir)?; @@ -345,6 +362,7 @@ fn sync_openai_plugins_repo_via_http( fn sync_openai_plugins_repo_via_backup_archive( codex_home: &Path, backup_archive_api_url: &str, + http_client_factory: &HttpClientFactory, ) -> Result { let repo_path = curated_plugins_repo_path(codex_home); let sha_path = curated_plugins_sha_path(codex_home); @@ -353,7 +371,9 @@ fn sync_openai_plugins_repo_via_backup_archive( .build() .map_err(|err| format!("failed to create curated plugins sync runtime: {err}"))?; let staged_repo_dir = prepare_curated_repo_parent_and_temp_dir(&repo_path)?; + let http_clients = StartupSyncHttpClient::new(http_client_factory); let zipball_bytes = runtime.block_on(fetch_curated_repo_backup_archive_zip( + &http_clients, backup_archive_api_url, ))?; extract_zipball_to_dir(&zipball_bytes, staged_repo_dir.path())?; @@ -763,11 +783,14 @@ fn ensure_git_success(output: &Output, context: &str) -> Result<(), String> { } } -async fn fetch_curated_repo_remote_sha(api_base_url: &str) -> Result { +async fn fetch_curated_repo_remote_sha( + http_clients: &StartupSyncHttpClient, + api_base_url: &str, +) -> Result { let api_base_url = api_base_url.trim_end_matches('/'); let repo_url = format!("{api_base_url}/repos/{OPENAI_PLUGINS_OWNER}/{OPENAI_PLUGINS_REPO}"); - let client = build_reqwest_client(); - let repo_body = fetch_github_text(&client, &repo_url, "get curated plugins repository").await?; + let repo_body = + fetch_github_text(http_clients, &repo_url, "get curated plugins repository").await?; let repo_summary: GitHubRepositorySummary = serde_json::from_str(&repo_body).map_err(|err| { format!("failed to parse curated plugins repository response from {repo_url}: {err}") @@ -780,7 +803,7 @@ async fn fetch_curated_repo_remote_sha(api_base_url: &str) -> Result Result Result, String> { let api_base_url = api_base_url.trim_end_matches('/'); let repo_url = format!("{api_base_url}/repos/{OPENAI_PLUGINS_OWNER}/{OPENAI_PLUGINS_REPO}"); let zipball_url = format!("{repo_url}/zipball/{remote_sha}"); - let client = build_reqwest_client(); - fetch_github_bytes(&client, &zipball_url, "download curated plugins archive").await + fetch_github_bytes( + http_clients, + &zipball_url, + "download curated plugins archive", + ) + .await } async fn fetch_curated_repo_backup_archive_zip( + http_clients: &StartupSyncHttpClient, backup_archive_api_url: &str, ) -> Result, String> { - let client = build_reqwest_client(); let export_body = fetch_public_text( - &client, + http_clients, backup_archive_api_url, "get curated plugins export archive metadata", ) @@ -827,7 +855,7 @@ async fn fetch_curated_repo_backup_archive_zip( } fetch_public_bytes( - &client, + http_clients, &export_response.download_url, "download curated plugins export archive", ) @@ -924,8 +952,12 @@ fn read_git_ref_sha(git_dir: &Path, reference: &str) -> Result { )) } -async fn fetch_github_text(client: &Client, url: &str, context: &str) -> Result { - let response = github_request(client, url) +async fn fetch_github_text( + http_clients: &StartupSyncHttpClient, + url: &str, + context: &str, +) -> Result { + let response = github_request(http_clients, url) .send() .await .map_err(|err| format!("failed to {context} from {url}: {err}"))?; @@ -939,8 +971,12 @@ async fn fetch_github_text(client: &Client, url: &str, context: &str) -> Result< Ok(body) } -async fn fetch_github_bytes(client: &Client, url: &str, context: &str) -> Result, String> { - let response = github_request(client, url) +async fn fetch_github_bytes( + http_clients: &StartupSyncHttpClient, + url: &str, + context: &str, +) -> Result, String> { + let response = github_request(http_clients, url) .send() .await .map_err(|err| format!("failed to {context} from {url}: {err}"))?; @@ -958,9 +994,12 @@ async fn fetch_github_bytes(client: &Client, url: &str, context: &str) -> Result Ok(body.to_vec()) } -async fn fetch_public_text(client: &Client, url: &str, context: &str) -> Result { - let response = client - .get(url) +async fn fetch_public_text( + http_clients: &StartupSyncHttpClient, + url: &str, + context: &str, +) -> Result { + let response = startup_sync_request(http_clients, url) .timeout(CURATED_PLUGINS_BACKUP_ARCHIVE_TIMEOUT) .send() .await @@ -975,9 +1014,12 @@ async fn fetch_public_text(client: &Client, url: &str, context: &str) -> Result< Ok(body) } -async fn fetch_public_bytes(client: &Client, url: &str, context: &str) -> Result, String> { - let response = client - .get(url) +async fn fetch_public_bytes( + http_clients: &StartupSyncHttpClient, + url: &str, + context: &str, +) -> Result, String> { + let response = startup_sync_request(http_clients, url) .timeout(CURATED_PLUGINS_BACKUP_ARCHIVE_TIMEOUT) .send() .await @@ -996,14 +1038,22 @@ async fn fetch_public_bytes(client: &Client, url: &str, context: &str) -> Result Ok(body.to_vec()) } -fn github_request(client: &Client, url: &str) -> reqwest::RequestBuilder { - client - .get(url) +fn github_request(http_clients: &StartupSyncHttpClient, url: &str) -> StartupSyncRequestBuilder { + startup_sync_request(http_clients, url) .timeout(CURATED_PLUGINS_HTTP_TIMEOUT) .header("accept", GITHUB_API_ACCEPT_HEADER) .header("x-github-api-version", GITHUB_API_VERSION_HEADER) } +fn startup_sync_request( + http_clients: &StartupSyncHttpClient, + url: &str, +) -> StartupSyncRequestBuilder { + http_clients + .request(Method::GET, url) + .headers(default_headers()) +} + fn read_sha_file(sha_path: &Path) -> Option { std::fs::read_to_string(sha_path) .ok() diff --git a/codex-rs/core-plugins/src/startup_sync/http_client.rs b/codex-rs/core-plugins/src/startup_sync/http_client.rs new file mode 100644 index 0000000000..15fcc452a8 --- /dev/null +++ b/codex-rs/core-plugins/src/startup_sync/http_client.rs @@ -0,0 +1,101 @@ +//! Startup-sync-specific HTTP transport selection. +//! +//! Curated plugin startup sync normally uses git, so its HTTP path is also a recovery path for +//! machines where git is unavailable or fails. Under `ReqwestDefault`, that recovery path must +//! preserve the legacy `codex_login::default_client::build_reqwest_client()` behavior: invalid +//! custom-CA configuration is logged and falls back to a normal reqwest client instead of making +//! HTTP sync fail as well. +//! +//! When `RespectSystemProxy` is enabled, however, every concrete request URL—including download +//! URLs returned by another endpoint—must be routed through `RouteAwareClientPool` so PAC and +//! operating-system proxy settings are respected. The route-aware pool intentionally surfaces +//! client-construction errors and therefore cannot provide the legacy custom-CA fallback for free. +//! +//! `StartupSyncHttpClient` keeps those two policies behind one request API without making lenient +//! custom-CA handling a global HTTP-client behavior. This module selects the transport only; +//! startup-sync request helpers remain responsible for applying the standard Codex headers. + +use std::sync::Arc; +use std::time::Duration; + +use crate::http_client_selector::HttpClientSelector; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use codex_http_client::RouteAwareClientPool; +use codex_http_client::RouteAwareRequestBuilder; +use codex_login::default_client::build_reqwest_client; +use http::HeaderMap; +use http::Method; + +pub(super) enum StartupSyncHttpClient { + Default(reqwest::Client), + RouteAware(Arc), +} + +impl StartupSyncHttpClient { + pub(super) fn new(http_client_factory: &HttpClientFactory) -> Self { + match http_client_factory.outbound_proxy_policy() { + OutboundProxyPolicy::ReqwestDefault => Self::Default(build_reqwest_client()), + OutboundProxyPolicy::RespectSystemProxy => { + let http_clients = + RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging( + http_client_factory.clone(), + ClientRouteClass::Api, + ); + Self::RouteAware(Arc::new(http_clients)) + } + } + } + + #[cfg(test)] + pub(super) fn route_aware(http_clients: Arc) -> Self { + Self::RouteAware(http_clients) + } + + pub(super) fn request(&self, method: Method, url: &str) -> StartupSyncRequestBuilder { + match self { + Self::Default(client) => { + StartupSyncRequestBuilder::Default(client.request(method, url)) + } + Self::RouteAware(http_clients) => { + StartupSyncRequestBuilder::RouteAware(http_clients.request(method, url)) + } + } + } +} + +pub(super) enum StartupSyncRequestBuilder { + Default(reqwest::RequestBuilder), + RouteAware(RouteAwareRequestBuilder), +} + +impl StartupSyncRequestBuilder { + pub(super) fn timeout(self, timeout: Duration) -> Self { + match self { + Self::Default(request) => Self::Default(request.timeout(timeout)), + Self::RouteAware(request) => Self::RouteAware(request.timeout(timeout)), + } + } + + pub(super) fn header(self, key: &'static str, value: &'static str) -> Self { + match self { + Self::Default(request) => Self::Default(request.header(key, value)), + Self::RouteAware(request) => Self::RouteAware(request.header(key, value)), + } + } + + pub(super) fn headers(self, headers: HeaderMap) -> Self { + match self { + Self::Default(request) => Self::Default(request.headers(headers)), + Self::RouteAware(request) => Self::RouteAware(request.headers(headers)), + } + } + + pub(super) async fn send(self) -> Result { + match self { + Self::Default(request) => request.send().await.map_err(|err| err.to_string()), + Self::RouteAware(request) => request.send().await.map_err(|err| err.to_string()), + } + } +} diff --git a/codex-rs/core-plugins/src/startup_sync_tests.rs b/codex-rs/core-plugins/src/startup_sync_tests.rs index a4ea778443..a760eab66f 100644 --- a/codex-rs/core-plugins/src/startup_sync_tests.rs +++ b/codex-rs/core-plugins/src/startup_sync_tests.rs @@ -1,4 +1,6 @@ use super::*; +use crate::test_support::RecordingHttpClientSelector; +use crate::test_support::recorded_http_client_urls; use pretty_assertions::assert_eq; use std::ffi::OsStr; use std::io::Write; @@ -10,6 +12,7 @@ use tempfile::tempdir; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; +use wiremock::matchers::header_exists; use wiremock::matchers::method; use wiremock::matchers::path; use zip::ZipWriter; @@ -17,6 +20,76 @@ use zip::write::SimpleFileOptions; const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567"; +#[tokio::test] +async fn github_http_routes_repository_ref_and_zipball_urls() { + let server = MockServer::start().await; + let sha = TEST_CURATED_PLUGIN_SHA; + let zipball = b"archive".to_vec(); + mount_github_repo_and_ref(&server, sha).await; + mount_github_zipball(&server, sha, zipball.clone()).await; + let api_base_url = server.uri(); + let repo_url = format!("{api_base_url}/repos/openai/plugins"); + let ref_url = format!("{repo_url}/git/ref/heads/main"); + let zipball_url = format!("{repo_url}/zipball/{sha}"); + let (http_clients, selected_urls) = RecordingHttpClientSelector::new(); + let http_clients = StartupSyncHttpClient::route_aware(http_clients); + + let remote_sha = fetch_curated_repo_remote_sha(&http_clients, &api_base_url) + .await + .expect("remote SHA request should succeed"); + let downloaded_zipball = fetch_curated_repo_zipball(&http_clients, &api_base_url, &remote_sha) + .await + .expect("zipball request should succeed"); + + assert_eq!(remote_sha, sha); + assert_eq!(downloaded_zipball, zipball); + assert_eq!( + recorded_http_client_urls(&selected_urls), + vec![repo_url, ref_url, zipball_url] + ); +} + +#[tokio::test] +async fn backup_archive_routes_metadata_and_backend_supplied_download_urls() { + let metadata_server = MockServer::start().await; + let download_server = MockServer::start().await; + let download_url = format!( + "{}/files/curated-plugins.zip?sig=signed", + download_server.uri() + ); + Mock::given(method("GET")) + .and(path("/backend-api/plugins/export/curated")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(serde_json::json!({"download_url": download_url.clone()})), + ) + .expect(1) + .mount(&metadata_server) + .await; + Mock::given(method("GET")) + .and(path("/files/curated-plugins.zip")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"archive".to_vec())) + .expect(1) + .mount(&download_server) + .await; + let metadata_url = format!( + "{}/backend-api/plugins/export/curated", + metadata_server.uri() + ); + let (http_clients, selected_urls) = RecordingHttpClientSelector::new(); + let http_clients = StartupSyncHttpClient::route_aware(http_clients); + + let body = fetch_curated_repo_backup_archive_zip(&http_clients, &metadata_url) + .await + .expect("backup archive download should succeed"); + + assert_eq!(body, b"archive"); + assert_eq!( + recorded_http_client_urls(&selected_urls), + vec![metadata_url, download_url] + ); +} + #[test] fn git_command_sanitizes_ambient_repository_environment() { let command = git_command(Path::new("git")); @@ -133,11 +206,15 @@ fn run_git(repo: &Path, args: &[&str]) -> std::process::Output { async fn mount_github_repo_and_ref(server: &MockServer, sha: &str) { Mock::given(method("GET")) .and(path("/repos/openai/plugins")) + .and(header_exists("user-agent")) + .and(header_exists("originator")) .respond_with(ResponseTemplate::new(200).set_body_string(r#"{"default_branch":"main"}"#)) .mount(server) .await; Mock::given(method("GET")) .and(path("/repos/openai/plugins/git/ref/heads/main")) + .and(header_exists("user-agent")) + .and(header_exists("originator")) .respond_with( ResponseTemplate::new(200) .set_body_string(format!(r#"{{"object":{{"sha":"{sha}"}}}}"#)), @@ -149,6 +226,8 @@ async fn mount_github_repo_and_ref(server: &MockServer, sha: &str) { async fn mount_github_zipball(server: &MockServer, sha: &str, bytes: Vec) { Mock::given(method("GET")) .and(path(format!("/repos/openai/plugins/zipball/{sha}"))) + .and(header_exists("user-agent")) + .and(header_exists("originator")) .respond_with( ResponseTemplate::new(200) .insert_header("content-type", "application/zip") @@ -162,6 +241,8 @@ async fn mount_export_archive(server: &MockServer, bytes: Vec) -> String { let export_api_url = format!("{}/backend-api/plugins/export/curated", server.uri()); Mock::given(method("GET")) .and(path("/backend-api/plugins/export/curated")) + .and(header_exists("user-agent")) + .and(header_exists("originator")) .respond_with(ResponseTemplate::new(200).set_body_string(format!( r#"{{"download_url":"{}/files/curated-plugins.zip"}}"#, server.uri() @@ -170,6 +251,8 @@ async fn mount_export_archive(server: &MockServer, bytes: Vec) -> String { .await; Mock::given(method("GET")) .and(path("/files/curated-plugins.zip")) + .and(header_exists("user-agent")) + .and(header_exists("originator")) .respond_with( ResponseTemplate::new(200) .insert_header("content-type", "application/zip") @@ -196,6 +279,7 @@ async fn run_sync_with_transport_overrides( Some(git_binary.as_path()), &api_base_url, &backup_archive_api_url, + &crate::test_support::test_http_client_factory(), ) }) .await @@ -215,6 +299,7 @@ async fn run_sync_without_git( /*git_binary*/ None, &api_base_url, &backup_archive_api_url, + &crate::test_support::test_http_client_factory(), ) }) .await @@ -227,7 +312,11 @@ async fn run_http_sync( ) -> Result { let api_base_url = api_base_url.into(); tokio::task::spawn_blocking(move || { - sync_openai_plugins_repo_via_http(codex_home.as_path(), &api_base_url) + sync_openai_plugins_repo_via_http( + codex_home.as_path(), + &api_base_url, + &crate::test_support::test_http_client_factory(), + ) }) .await .expect("sync task should join") @@ -366,6 +455,7 @@ exit 1 Some(git_path.as_path()), "http://127.0.0.1:9", "http://127.0.0.1:9/backend-api/plugins/export/curated", + &crate::test_support::test_http_client_factory(), ) }; let first = scope.spawn(run_sync); diff --git a/codex-rs/core-plugins/src/test_support.rs b/codex-rs/core-plugins/src/test_support.rs index 9e06de9d16..32bf1f92b4 100644 --- a/codex-rs/core-plugins/src/test_support.rs +++ b/codex-rs/core-plugins/src/test_support.rs @@ -1,19 +1,76 @@ use std::fs; use std::path::Path; +use std::sync::Arc; +use std::sync::Mutex; use crate::OPENAI_API_CURATED_MARKETPLACE_NAME; use crate::OPENAI_CURATED_MARKETPLACE_NAME; use crate::PluginsConfigInput; +use crate::http_client_selector::HttpClientSelector; use codex_config::LoaderOverrides; use codex_config::NoopThreadConfigLoader; use codex_config::loader::load_config_layers_state; use codex_exec_server::LOCAL_FS; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use codex_http_client::RouteAwareClientPool; +use codex_http_client::RouteAwareRequestBuilder; use codex_utils_absolute_path::AbsolutePathBuf; +use http::Method; use toml::Value; pub(crate) const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567"; pub(crate) const TEST_CURATED_PLUGIN_CACHE_VERSION: &str = "01234567"; +pub(crate) fn test_http_client_factory() -> HttpClientFactory { + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) +} + +#[derive(Debug)] +pub(crate) struct RecordingHttpClientSelector { + selected_urls: Arc>>, + delegate: RouteAwareClientPool, +} + +impl RecordingHttpClientSelector { + pub(crate) fn new() -> (Arc, Arc>>) { + let selected_urls = Arc::new(Mutex::new(Vec::new())); + let delegate = RouteAwareClientPool::with_chatgpt_cloudflare_cookies( + test_http_client_factory(), + ClientRouteClass::Api, + ); + ( + Arc::new(Self { + selected_urls: Arc::clone(&selected_urls), + delegate, + }), + selected_urls, + ) + } +} + +impl HttpClientSelector for RecordingHttpClientSelector { + fn request(&self, method: Method, url: &str) -> RouteAwareRequestBuilder { + 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.delegate.request(method, url) + } + + fn outbound_proxy_policy(&self) -> OutboundProxyPolicy { + self.delegate.outbound_proxy_policy() + } +} + +pub(crate) fn recorded_http_client_urls(selected_urls: &Mutex>) -> Vec { + match selected_urls.lock() { + Ok(selected_urls) => selected_urls.clone(), + Err(error) => panic!("selected URL recorder lock should not be poisoned: {error}"), + } +} + pub(crate) fn write_file(path: &Path, contents: &str) { fs::create_dir_all(path.parent().expect("file should have a parent")).unwrap(); fs::write(path, contents).unwrap(); @@ -152,6 +209,7 @@ pub(crate) async fn load_plugins_config(codex_home: &Path, cwd: &Path) -> Plugin /*default_enabled*/ true, ), "https://chatgpt.com/backend-api/".to_string(), + test_http_client_factory(), ) } diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 1fe1b80558..689126b4ef 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1535,6 +1535,7 @@ impl Config { self.features.enabled(Feature::Plugins), self.features.enabled(Feature::RemotePlugin), self.chatgpt_base_url.clone(), + self.http_client_factory(), ) } diff --git a/codex-rs/core/src/plugins/discoverable_tests.rs b/codex-rs/core/src/plugins/discoverable_tests.rs index 53d010d190..a5472f18db 100644 --- a/codex-rs/core/src/plugins/discoverable_tests.rs +++ b/codex-rs/core/src/plugins/discoverable_tests.rs @@ -221,9 +221,10 @@ plugins = true let plugins_manager = PluginsManager::new(config.codex_home.to_path_buf()); fetch_and_cache_global_remote_plugin_catalog( codex_home.path(), - &RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }, + &RemotePluginServiceConfig::new( + config.chatgpt_base_url.clone(), + config.http_client_factory(), + ), Some(&auth), ) .await