openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

core-plugins: route requests through HTTP client factory

Browse Source
This commit is contained in:
Michael Bolin
2026-07-09 11:53:23 -07:00
parent bf21583df2
commit c50357588b
25 changed files with 627 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2 -1
View File
@@ -2786,6 +2786,7 @@ dependencies = [
"codex-exec-server",
"codex-git-utils",
"codex-hooks",
"codex-http-client",
"codex-login",
"codex-mcp",
"codex-model-provider",
@@ -2799,10 +2800,10 @@ dependencies = [
"codex-utils-plugins",
"dirs",
"flate2",
"http 1.4.0",
"libc",
"pretty_assertions",
"regex",
"reqwest 0.12.28",
"semver",
"serde",
"serde_json",
codex-rs/app-server/src/request_processors/plugins.rs
+22 -33
View File
@@ -127,6 +127,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<AbsolutePathBuf, String>,
@@ -581,9 +588,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(),
@@ -1030,9 +1035,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(),
@@ -1138,9 +1141,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,
@@ -1196,9 +1197,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(),
@@ -1249,9 +1248,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),
@@ -1292,9 +1289,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(),
@@ -1322,9 +1317,7 @@ impl PluginRequestProcessor {
_params: PluginShareListParams,
) -> Result<PluginShareListResponse, JSONRPCErrorError> {
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(),
@@ -1361,9 +1354,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(),
@@ -1394,9 +1385,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(),
@@ -1531,9 +1520,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,
@@ -1606,6 +1593,7 @@ impl PluginRequestProcessor {
})?;
let result = codex_core_plugins::remote_bundle::download_and_install_remote_plugin_bundle(
&remote_plugin_service_config,
config.codex_home.to_path_buf(),
validated_bundle,
)
@@ -2020,9 +2008,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(),
@@ -2296,6 +2282,7 @@ fn remote_plugin_catalog_error_type(err: &RemotePluginCatalogError) -> &'static
RemotePluginCatalogError::AuthRequired => "remote_catalog_auth_required",
RemotePluginCatalogError::UnsupportedAuthMode => "remote_catalog_unsupported_auth_mode",
RemotePluginCatalogError::AuthToken(_) => "remote_catalog_auth_token",
RemotePluginCatalogError::HttpClient { .. } => "remote_catalog_http_client",
RemotePluginCatalogError::Request { .. } => "remote_catalog_request",
RemotePluginCatalogError::UnexpectedStatus { .. } => "remote_catalog_unexpected_status",
RemotePluginCatalogError::Decode { .. } => "remote_catalog_decode",
@@ -2342,6 +2329,7 @@ fn remote_plugin_bundle_install_error_type(err: &RemotePluginBundleInstallError)
"remote_bundle_unsupported_download_url_scheme"
}
RemotePluginBundleInstallError::InvalidPluginId { .. } => "remote_bundle_invalid_plugin_id",
RemotePluginBundleInstallError::HttpClient { .. } => "remote_bundle_http_client",
RemotePluginBundleInstallError::DownloadRequest { .. } => "remote_bundle_download_request",
RemotePluginBundleInstallError::DownloadStatus { .. } => "remote_bundle_download_status",
RemotePluginBundleInstallError::DownloadBody { .. } => "remote_bundle_download_body",
@@ -2377,6 +2365,7 @@ fn remote_plugin_catalog_error_to_jsonrpc(
| RemotePluginCatalogError::ArchiveTooLarge { .. }
| RemotePluginCatalogError::UnknownMarketplace { .. } => invalid_request(message),
RemotePluginCatalogError::AuthToken(_)
| RemotePluginCatalogError::HttpClient { .. }
| RemotePluginCatalogError::Request { .. }
| RemotePluginCatalogError::UnexpectedStatus { .. }
| RemotePluginCatalogError::Decode { .. }
codex-rs/core-plugins/Cargo.toml
+2 -1
View File
@@ -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 }
@@ -36,7 +37,7 @@ codex-utils-plugins = { workspace = true }
chrono = { workspace = true }
dirs = { workspace = true }
flate2 = { workspace = true }
reqwest = { workspace = true }
http = { workspace = true }
regex = { workspace = true }
semver = { workspace = true }
serde = { workspace = true, features = ["derive"] }
codex-rs/core-plugins/src/discoverable_tests.rs
+4 -3
View File
@@ -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
codex-rs/core-plugins/src/http_client_selector.rs
+23
View File
@@ -0,0 +1,23 @@
use codex_http_client::HttpClient;
use codex_http_client::RouteAwareClientPool;
use codex_http_client::RouteAwareClientPoolError;
use std::fmt::Debug;
use std::future::Future;
use std::pin::Pin;
pub(crate) type HttpClientSelectionFuture<'a> =
Pin<Box<dyn Future<Output = Result<HttpClient, RouteAwareClientPoolError>> + Send + 'a>>;
/// Selects an HTTP client after resolving the outbound route for an exact request URL.
///
/// Implementations are expected to preserve request-specific route selection while reusing
/// transport clients when multiple URLs resolve to the same route.
pub(crate) trait HttpClientSelector: Debug + Send + Sync {
fn client_for_url<'a>(&'a self, request_url: &'a str) -> HttpClientSelectionFuture<'a>;
}
impl HttpClientSelector for RouteAwareClientPool {
fn client_for_url<'a>(&'a self, request_url: &'a str) -> HttpClientSelectionFuture<'a> {
Box::pin(RouteAwareClientPool::client_for_url(self, request_url))
}
}
codex-rs/core-plugins/src/lib.rs
+1
View File
@@ -1,5 +1,6 @@
mod app_mcp_routing;
mod discoverable;
mod http_client_selector;
pub mod installed_marketplaces;
pub mod loader;
mod manager;
codex-rs/core-plugins/src/manager.rs
+20 -9
View File
@@ -69,6 +69,7 @@ use codex_core_skills::SkillMetadata;
use codex_core_skills::config_rules::SkillConfigRules;
use codex_core_skills::config_rules::skill_config_rules_from_stack;
use codex_hooks::plugin_hook_declarations;
use codex_http_client::HttpClientFactory;
use codex_login::AuthManager;
use codex_login::CodexAuth;
use codex_plugin::AppConnectorId;
@@ -110,6 +111,7 @@ pub struct PluginsConfigInput {
pub plugins_enabled: bool,
pub remote_plugin_enabled: bool,
pub chatgpt_base_url: String,
http_client_factory: HttpClientFactory,
}
impl PluginsConfigInput {
@@ -118,14 +120,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(),
)
}
}
/// Inputs used to select endpoint-backed plugin install candidates.
@@ -223,9 +235,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(
@@ -1942,7 +1952,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() {
@@ -2298,7 +2308,7 @@ impl PluginsManager {
}
}
fn start_curated_repo_sync(self: &Arc<Self>) {
fn start_curated_repo_sync(self: &Arc<Self>, http_client_factory: HttpClientFactory) {
if CURATED_REPO_SYNC_STARTED.swap(true, Ordering::SeqCst) {
return;
}
@@ -2306,8 +2316,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());
@@ -2331,8 +2341,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}");
@@ -2718,6 +2728,7 @@ fn remote_plugin_mutation_error_type(err: &RemotePluginMutationError) -> &'stati
RemotePluginMutationError::AuthToken(_) => "remote_mutation_auth_token",
RemotePluginMutationError::InvalidBaseUrl(_) => "remote_mutation_invalid_base_url",
RemotePluginMutationError::InvalidBaseUrlPath => "remote_mutation_invalid_base_url_path",
RemotePluginMutationError::HttpClient { .. } => "remote_mutation_http_client",
RemotePluginMutationError::Request { .. } => "remote_mutation_request",
RemotePluginMutationError::UnexpectedStatus { .. } => "remote_mutation_unexpected_status",
RemotePluginMutationError::Decode { .. } => "remote_mutation_decode",
codex-rs/core-plugins/src/manager_tests.rs
+3
View File
@@ -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(),
)
}
@@ -2477,6 +2479,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());
codex-rs/core-plugins/src/remote.rs
+118 -42
View File
@@ -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;
@@ -9,8 +10,13 @@ use codex_app_server_protocol::PluginInstallPolicy;
use codex_app_server_protocol::PluginInstallPolicySource;
use codex_app_server_protocol::PluginInterface;
use codex_app_server_protocol::SkillInterface;
use codex_http_client::ClientRouteClass;
use codex_http_client::HttpClient;
use codex_http_client::HttpClientFactory;
use codex_http_client::RequestBuilder;
use codex_http_client::RouteAwareClientPool;
use codex_http_client::RouteAwareClientPoolError;
use codex_login::CodexAuth;
use codex_login::default_client::build_reqwest_client;
use codex_plugin::AppConnectorId;
use codex_plugin::AppDeclaration;
use codex_plugin::PluginCapabilitySummary;
@@ -18,7 +24,7 @@ use codex_plugin::PluginId;
use codex_plugin::app_connector_ids_from_declarations;
use codex_plugin::prompt_safe_plugin_description;
use codex_utils_absolute_path::AbsolutePathBuf;
use reqwest::RequestBuilder;
use http::StatusCode;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value as JsonValue;
@@ -29,6 +35,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;
@@ -118,11 +125,59 @@ 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_client_factory: HttpClientFactory,
pub(crate) http_clients: Arc<dyn HttpClientSelector>,
}
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(
http_client_factory.clone(),
ClientRouteClass::Api,
);
Self {
chatgpt_base_url,
http_client_factory,
http_clients: Arc::new(http_clients),
}
}
pub(crate) async fn http_client_for_url(
&self,
request_url: &str,
) -> Result<HttpClient, RouteAwareClientPoolError> {
self.http_clients.client_for_url(request_url).await
}
async fn catalog_client_for_url(
&self,
request_url: &str,
) -> Result<HttpClient, RemotePluginCatalogError> {
self.http_client_for_url(request_url)
.await
.map_err(|source| RemotePluginCatalogError::HttpClient {
url: request_url.to_string(),
source,
})
}
}
impl PartialEq for RemotePluginServiceConfig {
fn eq(&self, other: &Self) -> bool {
self.chatgpt_base_url == other.chatgpt_base_url
&& self.http_client_factory == other.http_client_factory
}
}
impl Eq for RemotePluginServiceConfig {}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemotePluginUninstallTarget {
pub plugin_id: PluginId,
@@ -305,17 +360,24 @@ pub enum RemotePluginCatalogError {
#[error("failed to read auth token for remote plugin catalog: {0}")]
AuthToken(#[source] std::io::Error),
#[error("failed to build HTTP client for remote plugin request to {url}: {source}")]
HttpClient {
url: String,
#[source]
source: RouteAwareClientPoolError,
},
#[error("failed to send remote plugin catalog request to {url}: {source}")]
Request {
url: String,
#[source]
source: reqwest::Error,
source: codex_http_client::HttpError,
},
#[error("remote plugin catalog request to {url} failed with status {status}: {body}")]
UnexpectedStatus {
url: String,
status: reqwest::StatusCode,
status: StatusCode,
body: String,
},
@@ -850,11 +912,13 @@ pub async fn fetch_recommended_plugins(
) -> Result<RecommendedPluginsMode, RemotePluginCatalogError> {
let auth = ensure_chatgpt_auth(auth)?;
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/ps/plugins/suggested");
let client = build_reqwest_client();
let request = authenticated_request(client.get(&url), auth)?
.timeout(RECOMMENDED_PLUGINS_TIMEOUT)
.query(&[("scope", "GLOBAL")]);
let mut url = Url::parse(&format!("{base_url}/ps/plugins/suggested"))
.map_err(RemotePluginCatalogError::InvalidBaseUrl)?;
url.query_pairs_mut().append_pair("scope", "GLOBAL");
let url = url.to_string();
let client = config.catalog_client_for_url(&url).await?;
let request =
authenticated_request(client.get(&url), auth)?.timeout(RECOMMENDED_PLUGINS_TIMEOUT);
let response: RecommendedPluginsResponse = send_and_decode(request, &url).await?;
Ok(recommended_plugins_mode(response))
}
@@ -1160,7 +1224,7 @@ pub async fn fetch_remote_plugin_skill_detail(
}
let url = remote_plugin_skill_detail_url(config, plugin_id, skill_name)?;
let client = build_reqwest_client();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.get(&url), auth)?;
let response: RemotePluginSkillDetailResponse = send_and_decode(request, &url).await?;
if response.plugin_id != plugin_id {
@@ -1315,14 +1379,13 @@ pub async fn install_remote_plugin(
// marketplace name is not validated before sending the install mutation.
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/ps/plugins/{plugin_id}/install");
let client = build_reqwest_client();
let request = authenticated_request(
client
.post(&url)
.query(&[("includeAppsNeedingAuth", "true")]),
auth,
)?;
let mut url = Url::parse(&format!("{base_url}/ps/plugins/{plugin_id}/install"))
.map_err(RemotePluginCatalogError::InvalidBaseUrl)?;
url.query_pairs_mut()
.append_pair("includeAppsNeedingAuth", "true");
let url = url.to_string();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.post(&url), auth)?;
let response: RemotePluginMutationResponse = send_and_decode(request, &url).await?;
if response.id != plugin_id {
return Err(RemotePluginCatalogError::UnexpectedPluginId {
@@ -1409,7 +1472,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 client = build_reqwest_client();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.post(&url), auth)?;
let response: RemotePluginMutationResponse = send_and_decode(request, &url).await?;
if response.id != remote_plugin_id {
@@ -1827,17 +1890,20 @@ async fn get_remote_plugin_list_page(
collection: Option<&str>,
) -> Result<RemotePluginListResponse, RemotePluginCatalogError> {
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/ps/plugins/list");
let client = build_reqwest_client();
let mut request = authenticated_request(client.get(&url), auth)?;
request = request.query(&[("scope", scope.api_value())]);
request = request.query(&[("limit", REMOTE_PLUGIN_LIST_PAGE_LIMIT)]);
let mut url = Url::parse(&format!("{base_url}/ps/plugins/list"))
.map_err(RemotePluginCatalogError::InvalidBaseUrl)?;
url.query_pairs_mut()
.append_pair("scope", scope.api_value())
.append_pair("limit", &REMOTE_PLUGIN_LIST_PAGE_LIMIT.to_string());
if let Some(collection) = collection {
request = request.query(&[("collection", collection)]);
url.query_pairs_mut().append_pair("collection", collection);
}
if let Some(page_token) = page_token {
request = request.query(&[("pageToken", page_token)]);
url.query_pairs_mut().append_pair("pageToken", page_token);
}
let url = url.to_string();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.get(&url), auth)?;
send_and_decode(request, &url).await
}
@@ -1847,13 +1913,16 @@ async fn get_remote_shared_workspace_plugins_page(
page_token: Option<&str>,
) -> Result<RemotePluginListResponse, RemotePluginCatalogError> {
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/ps/plugins/workspace/shared");
let client = build_reqwest_client();
let mut request = authenticated_request(client.get(&url), auth)?;
request = request.query(&[("limit", REMOTE_PLUGIN_LIST_PAGE_LIMIT)]);
let mut url = Url::parse(&format!("{base_url}/ps/plugins/workspace/shared"))
.map_err(RemotePluginCatalogError::InvalidBaseUrl)?;
url.query_pairs_mut()
.append_pair("limit", &REMOTE_PLUGIN_LIST_PAGE_LIMIT.to_string());
if let Some(page_token) = page_token {
request = request.query(&[("pageToken", page_token)]);
url.query_pairs_mut().append_pair("pageToken", page_token);
}
let url = url.to_string();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.get(&url), auth)?;
send_and_decode(request, &url).await
}
@@ -1865,16 +1934,20 @@ async fn get_remote_plugin_installed_page(
include_download_urls: bool,
) -> Result<RemotePluginInstalledResponse, RemotePluginCatalogError> {
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/ps/plugins/installed");
let client = build_reqwest_client();
let mut request = authenticated_request(client.get(&url), auth)?;
request = request.query(&[("scope", scope.api_value())]);
let mut url = Url::parse(&format!("{base_url}/ps/plugins/installed"))
.map_err(RemotePluginCatalogError::InvalidBaseUrl)?;
url.query_pairs_mut()
.append_pair("scope", scope.api_value());
if include_download_urls {
request = request.query(&[("includeDownloadUrls", true)]);
url.query_pairs_mut()
.append_pair("includeDownloadUrls", "true");
}
if let Some(page_token) = page_token {
request = request.query(&[("pageToken", page_token)]);
url.query_pairs_mut().append_pair("pageToken", page_token);
}
let url = url.to_string();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.get(&url), auth)?;
send_and_decode(request, &url).await
}
@@ -1885,12 +1958,15 @@ async fn fetch_plugin_detail(
include_download_urls: bool,
) -> Result<RemotePluginDirectoryItem, RemotePluginCatalogError> {
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/ps/plugins/{plugin_id}");
let client = build_reqwest_client();
let mut request = authenticated_request(client.get(&url), auth)?;
let mut url = Url::parse(&format!("{base_url}/ps/plugins/{plugin_id}"))
.map_err(RemotePluginCatalogError::InvalidBaseUrl)?;
if include_download_urls {
request = request.query(&[("includeDownloadUrls", true)]);
url.query_pairs_mut()
.append_pair("includeDownloadUrls", "true");
}
let url = url.to_string();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.get(&url), auth)?;
send_and_decode(request, &url).await
}
codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs
+5 -3
View File
@@ -248,6 +248,7 @@ pub async fn sync_remote_installed_plugin_bundles_once(
};
match crate::remote_bundle::download_and_install_remote_plugin_bundle(
config,
codex_home.clone(),
bundle,
)
@@ -542,9 +543,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(
codex-rs/core-plugins/src/remote/share.rs
+18 -14
View File
@@ -1,17 +1,17 @@
use super::*;
use crate::plugin_bundle_archive::PluginBundlePackError;
use crate::plugin_bundle_archive::pack_plugin_bundle_tar_gz;
use codex_http_client::RequestBuilder;
use codex_login::CodexAuth;
use codex_login::default_client::build_reqwest_client;
use codex_utils_absolute_path::AbsolutePathBuf;
use reqwest::RequestBuilder;
use reqwest::StatusCode;
use http::StatusCode;
use serde::Deserialize;
use serde::Serialize;
use std::collections::BTreeMap;
use std::io;
use std::path::Path;
use tracing::warn;
use url::Url;
mod checkout;
mod local_paths;
@@ -163,7 +163,7 @@ pub async fn save_remote_plugin_share(
let etag = upload
.etag
.ok_or(RemotePluginCatalogError::MissingUploadEtag)?;
put_workspace_plugin_upload(&upload.upload_url, archive_bytes).await?;
put_workspace_plugin_upload(config, &upload.upload_url, archive_bytes).await?;
let share_targets = access_policy.share_targets;
let share_targets =
ensure_unlisted_workspace_target(auth, access_policy.discoverability, share_targets)?;
@@ -279,7 +279,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 client = build_reqwest_client();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.delete(&url), auth)?;
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) {
@@ -312,7 +312,7 @@ 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 client = build_reqwest_client();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.put(&url), auth)?.json(
&RemotePluginShareUpdateTargetsRequest {
discoverability,
@@ -377,13 +377,16 @@ async fn get_created_workspace_plugins_page(
page_token: Option<&str>,
) -> Result<RemotePluginListResponse, RemotePluginCatalogError> {
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/ps/plugins/workspace/created");
let client = build_reqwest_client();
let mut request = authenticated_request(client.get(&url), auth)?;
request = request.query(&[("limit", REMOTE_PLUGIN_LIST_PAGE_LIMIT)]);
let mut url = Url::parse(&format!("{base_url}/ps/plugins/workspace/created"))
.map_err(RemotePluginCatalogError::InvalidBaseUrl)?;
url.query_pairs_mut()
.append_pair("limit", &REMOTE_PLUGIN_LIST_PAGE_LIMIT.to_string());
if let Some(page_token) = page_token {
request = request.query(&[("pageToken", page_token)]);
url.query_pairs_mut().append_pair("pageToken", page_token);
}
let url = url.to_string();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.get(&url), auth)?;
send_and_decode(request, &url).await
}
@@ -396,7 +399,7 @@ 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 client = build_reqwest_client();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.post(&url), auth)?.json(
&RemoteWorkspacePluginUploadUrlRequest {
filename,
@@ -409,10 +412,11 @@ async fn create_workspace_plugin_upload(
}
async fn put_workspace_plugin_upload(
config: &RemotePluginServiceConfig,
upload_url: &str,
archive_bytes: Vec<u8>,
) -> Result<(), RemotePluginCatalogError> {
let client = build_reqwest_client();
let client = config.catalog_client_for_url(upload_url).await?;
let request = client
.put(upload_url)
.timeout(REMOTE_PLUGIN_CATALOG_TIMEOUT)
@@ -450,7 +454,7 @@ async fn finalize_workspace_plugin_upload(
} else {
format!("{base_url}/public/plugins/workspace")
};
let client = build_reqwest_client();
let client = config.catalog_client_for_url(&url).await?;
let request = authenticated_request(client.post(&url), auth)?.json(&body);
send_and_decode(request, &url).await
}
codex-rs/core-plugins/src/remote/share/checkout.rs
+1
View File
@@ -93,6 +93,7 @@ pub async fn checkout_remote_plugin_share(
))
})?;
crate::remote_bundle::download_and_extract_remote_plugin_bundle_to_path(
config,
bundle,
local_plugin_path.clone(),
)
codex-rs/core-plugins/src/remote/share/tests.rs
+19 -4
View File
@@ -1,4 +1,6 @@
use super::*;
use crate::test_support::recorded_http_client_urls;
use crate::test_support::recording_remote_plugin_service_config;
use codex_app_server_protocol::PluginAuthPolicy;
use codex_app_server_protocol::PluginInstallPolicy;
use codex_app_server_protocol::PluginInterface;
@@ -23,9 +25,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 {
@@ -174,7 +177,8 @@ async fn save_remote_plugin_share_creates_workspace_plugin() {
.unwrap()
.len();
let server = MockServer::start().await;
let config = test_config(&server);
let (config, selected_urls) =
recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
let auth = test_auth();
Mock::given(method("POST"))
@@ -260,6 +264,17 @@ async fn save_remote_plugin_share_creates_workspace_plugin() {
local_paths::load_plugin_share_local_paths(codex_home.path()).unwrap(),
BTreeMap::from([("plugins_123".to_string(), plugin_path)])
);
assert_eq!(
recorded_http_client_urls(&selected_urls),
vec![
format!(
"{}/backend-api/public/plugins/workspace/upload-url",
server.uri()
),
format!("{}/upload/file_123", server.uri()),
format!("{}/backend-api/public/plugins/workspace", server.uri()),
]
);
let requests = server.received_requests().await.unwrap_or_default();
let upload_request = requests
codex-rs/core-plugins/src/remote_bundle.rs
+62 -7
View File
@@ -1,18 +1,19 @@
use crate::plugin_bundle_archive::PluginBundleUnpackError;
use crate::plugin_bundle_archive::unpack_plugin_bundle_tar_gz;
use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME;
use crate::remote::RemotePluginServiceConfig;
use crate::store::PluginInstallResult;
use crate::store::PluginStore;
use crate::store::PluginStoreError;
use crate::store::error_context_sub_error_type;
use crate::store::validate_plugin_version_segment;
use codex_login::default_client::build_reqwest_client;
use codex_http_client::HttpResponse;
use codex_http_client::RouteAwareClientPoolError;
use codex_plugin::PluginId;
use codex_plugin::PluginIdError;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_plugins::find_plugin_manifest_path;
use reqwest::Response;
use reqwest::StatusCode;
use http::StatusCode;
use serde_json::Value as JsonValue;
use std::fs;
use std::io;
@@ -83,11 +84,18 @@ pub enum RemotePluginBundleInstallError {
source: PluginIdError,
},
#[error("failed to build HTTP client for remote plugin bundle download from {url}: {source}")]
HttpClient {
url: String,
#[source]
source: RouteAwareClientPoolError,
},
#[error("failed to send remote plugin bundle download request to {url}: {source}")]
DownloadRequest {
url: String,
#[source]
source: reqwest::Error,
source: codex_http_client::HttpError,
},
#[error("remote plugin bundle download from {url} failed with status {status}: {body}")]
@@ -101,7 +109,7 @@ pub enum RemotePluginBundleInstallError {
DownloadBody {
url: String,
#[source]
source: reqwest::Error,
source: codex_http_client::HttpError,
},
#[error("remote plugin bundle download from {url} exceeded maximum size of {max_bytes} bytes")]
@@ -144,6 +152,7 @@ impl RemotePluginBundleInstallError {
| Self::InvalidBundleDownloadUrl { .. }
| Self::UnsupportedBundleDownloadUrlScheme { .. }
| Self::InvalidPluginId { .. }
| Self::HttpClient { .. }
| Self::DownloadRequest { .. }
| Self::DownloadStatus { .. }
| Self::DownloadBody { .. }
@@ -247,10 +256,12 @@ fn is_loopback_url(url: &Url) -> bool {
}
pub async fn download_and_install_remote_plugin_bundle(
config: &RemotePluginServiceConfig,
codex_home: PathBuf,
bundle: ValidatedRemotePluginBundle,
) -> Result<PluginInstallResult, RemotePluginBundleInstallError> {
let bundle_bytes = download_remote_plugin_bundle_with_limit(
config,
&bundle.bundle_download_url,
/*max_bytes*/ REMOTE_PLUGIN_BUNDLE_MAX_DOWNLOAD_BYTES,
)
@@ -267,10 +278,12 @@ pub async fn download_and_install_remote_plugin_bundle(
}
pub(crate) async fn download_and_extract_remote_plugin_bundle_to_path(
config: &RemotePluginServiceConfig,
bundle: ValidatedRemotePluginBundle,
destination: AbsolutePathBuf,
) -> Result<AbsolutePathBuf, RemotePluginBundleInstallError> {
let bundle_bytes = download_remote_plugin_bundle_with_limit(
config,
&bundle.bundle_download_url,
/*max_bytes*/ REMOTE_PLUGIN_BUNDLE_MAX_DOWNLOAD_BYTES,
)
@@ -287,10 +300,17 @@ pub(crate) async fn download_and_extract_remote_plugin_bundle_to_path(
}
async fn download_remote_plugin_bundle_with_limit(
config: &RemotePluginServiceConfig,
bundle_download_url: &str,
max_bytes: u64,
) -> Result<Vec<u8>, RemotePluginBundleInstallError> {
let client = build_reqwest_client();
let client = config
.http_client_for_url(bundle_download_url)
.await
.map_err(|source| RemotePluginBundleInstallError::HttpClient {
url: bundle_download_url.to_string(),
source,
})?;
let response = client
.get(bundle_download_url)
.timeout(REMOTE_PLUGIN_BUNDLE_DOWNLOAD_TIMEOUT)
@@ -354,7 +374,7 @@ async fn download_remote_plugin_bundle_with_limit(
}
async fn read_response_body_with_limit(
mut response: Response,
mut response: HttpResponse,
url: &str,
max_bytes: u64,
) -> Result<Vec<u8>, RemotePluginBundleInstallError> {
@@ -623,11 +643,18 @@ fn is_standard_plugin_root(path: &Path) -> bool {
#[cfg(test)]
mod tests {
use super::*;
use crate::test_support::recorded_http_client_urls;
use crate::test_support::recording_remote_plugin_service_config;
use flate2::Compression;
use flate2::write::GzEncoder;
use pretty_assertions::assert_eq;
use std::io::Write;
use tempfile::tempdir;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
const REMOTE_PLUGIN_ID: &str = "plugins~Plugin_00000000000000000000000000000000";
@@ -739,6 +766,34 @@ mod tests {
));
}
#[tokio::test]
async fn bundle_download_routes_the_backend_supplied_url() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/signed/plugin-bundle"))
.respond_with(ResponseTemplate::new(200).set_body_bytes(b"bundle"))
.expect(1)
.mount(&server)
.await;
let (config, selected_urls) =
recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
let download_url = format!("{}/signed/plugin-bundle?sig=signed-token", server.uri());
let err =
download_remote_plugin_bundle_with_limit(&config, &download_url, /*max_bytes*/ 64)
.await
.expect_err("plain HTTP final URL should remain unsupported");
assert!(matches!(
err,
RemotePluginBundleInstallError::UnsupportedBundleDownloadFinalUrl { .. }
));
assert_eq!(
recorded_http_client_urls(&selected_urls),
vec![download_url]
);
}
#[test]
fn install_rejects_invalid_tar_gz_bundle() {
let codex_home = tempdir().expect("tempdir");
codex-rs/core-plugins/src/remote_legacy.rs
+42 -12
View File
@@ -1,7 +1,8 @@
use crate::remote::RemotePluginServiceConfig;
use codex_http_client::RouteAwareClientPoolError;
use codex_login::CodexAuth;
use codex_login::default_client::build_reqwest_client;
use codex_protocol::protocol::Product;
use http::StatusCode;
use serde::Deserialize;
use std::time::Duration;
use url::Url;
@@ -35,17 +36,24 @@ pub enum RemotePluginMutationError {
#[error("chatgpt base url cannot be used for plugin mutation")]
InvalidBaseUrlPath,
#[error("failed to build HTTP client for remote plugin mutation request to {url}: {source}")]
HttpClient {
url: String,
#[source]
source: RouteAwareClientPoolError,
},
#[error("failed to send remote plugin mutation request to {url}: {source}")]
Request {
url: String,
#[source]
source: reqwest::Error,
source: codex_http_client::HttpError,
},
#[error("remote plugin mutation failed with status {status} from {url}: {body}")]
UnexpectedStatus {
url: String,
status: reqwest::StatusCode,
status: StatusCode,
body: String,
},
@@ -73,17 +81,27 @@ pub enum RemotePluginMutationError {
#[derive(Debug, thiserror::Error)]
pub enum RemotePluginFetchError {
#[error("invalid chatgpt base url for remote featured plugin request: {0}")]
InvalidBaseUrl(#[source] url::ParseError),
#[error("failed to build HTTP client for remote featured plugin request to {url}: {source}")]
HttpClient {
url: String,
#[source]
source: RouteAwareClientPoolError,
},
#[error("failed to send remote featured plugin request to {url}: {source}")]
Request {
url: String,
#[source]
source: reqwest::Error,
source: codex_http_client::HttpError,
},
#[error("remote featured plugin request to {url} failed with status {status}: {body}")]
UnexpectedStatus {
url: String,
status: reqwest::StatusCode,
status: StatusCode,
body: String,
},
@@ -101,14 +119,21 @@ pub async fn fetch_remote_featured_plugin_ids(
product: Option<Product>,
) -> Result<Vec<String>, RemotePluginFetchError> {
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/plugins/featured");
let client = build_reqwest_client();
let mut url = Url::parse(&format!("{base_url}/plugins/featured"))
.map_err(RemotePluginFetchError::InvalidBaseUrl)?;
url.query_pairs_mut().append_pair(
"platform",
product.unwrap_or(Product::Codex).to_app_platform(),
);
let url = url.to_string();
let client = config.http_client_for_url(&url).await.map_err(|source| {
RemotePluginFetchError::HttpClient {
url: url.clone(),
source,
}
})?;
let mut request = client
.get(&url)
.query(&[(
"platform",
product.unwrap_or(Product::Codex).to_app_platform(),
)])
.timeout(REMOTE_FEATURED_PLUGIN_FETCH_TIMEOUT);
if let Some(auth) = auth.filter(|auth| auth.uses_codex_backend()) {
@@ -173,7 +198,12 @@ async fn post_remote_plugin_mutation(
) -> Result<RemotePluginMutationResponse, RemotePluginMutationError> {
let auth = ensure_codex_backend_auth(auth)?;
let url = remote_plugin_mutation_url(config, plugin_id, action)?;
let client = build_reqwest_client();
let client = config.http_client_for_url(&url).await.map_err(|source| {
RemotePluginMutationError::HttpClient {
url: url.clone(),
source,
}
})?;
let request = client
.post(url.clone())
.timeout(REMOTE_PLUGIN_MUTATION_TIMEOUT)
codex-rs/core-plugins/src/remote_tests.rs
+42
View File
@@ -1,5 +1,47 @@
use super::*;
use crate::test_support::recorded_http_client_urls;
use crate::test_support::recording_remote_plugin_service_config;
use pretty_assertions::assert_eq;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
#[tokio::test]
async fn remote_plugin_list_routes_the_complete_query_url() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/backend-api/ps/plugins/list"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"plugins": [],
"pagination": {"next_page_token": null},
})))
.expect(1)
.mount(&server)
.await;
let (config, selected_urls) =
recording_remote_plugin_service_config(format!("{}/backend-api", server.uri()));
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
get_remote_plugin_list_page(
&config,
&auth,
RemotePluginScope::Global,
Some("next page/+"),
Some("vertical & special"),
)
.await
.expect("plugin list request should succeed");
assert_eq!(
recorded_http_client_urls(&selected_urls),
vec![format!(
"{}/backend-api/ps/plugins/list?scope=GLOBAL&limit=200&collection=vertical+%26+special&pageToken=next+page%2F%2B",
server.uri()
)]
);
}
#[test]
fn build_remote_marketplace_preserves_directory_order_and_appends_installed_only_plugins() {
codex-rs/core-plugins/src/startup_sync.rs
+87 -23
View File
@@ -6,16 +6,19 @@ use std::process::Output;
use std::process::Stdio;
use std::time::Duration;
use crate::http_client_selector::HttpClientSelector;
use codex_http_client::ClientRouteClass;
use codex_http_client::HttpClient;
use codex_http_client::HttpClientFactory;
use codex_http_client::RequestBuilder;
use codex_http_client::RouteAwareClientPool;
use codex_otel::CURATED_PLUGINS_STARTUP_SYNC_FINAL_METRIC;
use codex_otel::CURATED_PLUGINS_STARTUP_SYNC_METRIC;
use reqwest::Client;
use serde::Deserialize;
use tempfile::TempDir;
use tracing::warn;
use zip::ZipArchive;
use codex_login::default_client::build_reqwest_client;
const GITHUB_API_BASE_URL: &str = "https://api.github.com";
const GITHUB_API_ACCEPT_HEADER: &str = "application/vnd.github+json";
const GITHUB_API_VERSION_HEADER: &str = "2022-11-28";
@@ -92,7 +95,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<String, String> {
pub fn sync_openai_plugins_repo(
codex_home: &Path,
http_client_factory: HttpClientFactory,
) -> Result<String, String> {
#[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 +112,7 @@ pub fn sync_openai_plugins_repo(codex_home: &Path) -> Result<String, String> {
git_binary.as_deref(),
GITHUB_API_BASE_URL,
CURATED_PLUGINS_BACKUP_ARCHIVE_API_URL,
&http_client_factory,
)
}
@@ -114,6 +121,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<String, String> {
let _file_guard = lock_curated_plugins_startup_sync(codex_home)?;
@@ -134,7 +142,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 +170,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 +328,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<String, String> {
let repo_path = curated_plugins_repo_path(codex_home);
let sha_path = codex_home.join(CURATED_PLUGINS_SHA_FILE);
@@ -326,7 +336,12 @@ 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 = RouteAwareClientPool::with_chatgpt_cloudflare_cookies(
http_client_factory.clone(),
ClientRouteClass::Api,
);
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 +349,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 +364,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<String, String> {
let repo_path = curated_plugins_repo_path(codex_home);
let sha_path = curated_plugins_sha_path(codex_home);
@@ -353,7 +373,12 @@ 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 = RouteAwareClientPool::with_chatgpt_cloudflare_cookies(
http_client_factory.clone(),
ClientRouteClass::Api,
);
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 +788,14 @@ fn ensure_git_success(output: &Output, context: &str) -> Result<(), String> {
}
}
async fn fetch_curated_repo_remote_sha(api_base_url: &str) -> Result<String, String> {
async fn fetch_curated_repo_remote_sha(
http_clients: &dyn HttpClientSelector,
api_base_url: &str,
) -> Result<String, 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 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 +808,7 @@ async fn fetch_curated_repo_remote_sha(api_base_url: &str) -> Result<String, Str
let git_ref_url = format!("{repo_url}/git/ref/heads/{}", repo_summary.default_branch);
let git_ref_body =
fetch_github_text(&client, &git_ref_url, "get curated plugins HEAD ref").await?;
fetch_github_text(http_clients, &git_ref_url, "get curated plugins HEAD ref").await?;
let git_ref: GitHubGitRefSummary = serde_json::from_str(&git_ref_body).map_err(|err| {
format!("failed to parse curated plugins ref response from {git_ref_url}: {err}")
})?;
@@ -794,22 +822,27 @@ async fn fetch_curated_repo_remote_sha(api_base_url: &str) -> Result<String, Str
}
async fn fetch_curated_repo_zipball(
http_clients: &dyn HttpClientSelector,
api_base_url: &str,
remote_sha: &str,
) -> Result<Vec<u8>, 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: &dyn HttpClientSelector,
backup_archive_api_url: &str,
) -> Result<Vec<u8>, 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 +860,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 +957,13 @@ fn read_git_ref_sha(git_dir: &Path, reference: &str) -> Result<String, String> {
))
}
async fn fetch_github_text(client: &Client, url: &str, context: &str) -> Result<String, String> {
let response = github_request(client, url)
async fn fetch_github_text(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<String, String> {
let client = client_for_url(http_clients, url, context).await?;
let response = github_request(&client, url)
.send()
.await
.map_err(|err| format!("failed to {context} from {url}: {err}"))?;
@@ -939,8 +977,13 @@ 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<Vec<u8>, String> {
let response = github_request(client, url)
async fn fetch_github_bytes(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<Vec<u8>, String> {
let client = client_for_url(http_clients, url, context).await?;
let response = github_request(&client, url)
.send()
.await
.map_err(|err| format!("failed to {context} from {url}: {err}"))?;
@@ -958,7 +1001,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<String, String> {
async fn fetch_public_text(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<String, String> {
let client = client_for_url(http_clients, url, context).await?;
let response = client
.get(url)
.timeout(CURATED_PLUGINS_BACKUP_ARCHIVE_TIMEOUT)
@@ -975,7 +1023,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<Vec<u8>, String> {
async fn fetch_public_bytes(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<Vec<u8>, String> {
let client = client_for_url(http_clients, url, context).await?;
let response = client
.get(url)
.timeout(CURATED_PLUGINS_BACKUP_ARCHIVE_TIMEOUT)
@@ -996,7 +1049,18 @@ 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 {
async fn client_for_url(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<HttpClient, String> {
http_clients
.client_for_url(url)
.await
.map_err(|err| format!("failed to build HTTP client to {context} from {url}: {err}"))
}
fn github_request(client: &HttpClient, url: &str) -> RequestBuilder {
client
.get(url)
.timeout(CURATED_PLUGINS_HTTP_TIMEOUT)
codex-rs/core-plugins/src/startup_sync_tests.rs
+50 -1
View File
@@ -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;
@@ -17,6 +19,46 @@ use zip::write::SimpleFileOptions;
const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567";
#[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 body = fetch_curated_repo_backup_archive_zip(http_clients.as_ref(), &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"));
@@ -196,6 +238,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 +258,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 +271,11 @@ async fn run_http_sync(
) -> Result<String, String> {
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 +414,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);
codex-rs/core-plugins/src/test_support.rs
+69
View File
@@ -1,19 +1,87 @@
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::HttpClientSelectionFuture;
use crate::http_client_selector::HttpClientSelector;
use crate::remote::RemotePluginServiceConfig;
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_utils_absolute_path::AbsolutePathBuf;
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<Mutex<Vec<String>>>,
delegate: RouteAwareClientPool,
}
impl RecordingHttpClientSelector {
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(
test_http_client_factory(),
ClientRouteClass::Api,
);
(
Arc::new(Self {
selected_urls: Arc::clone(&selected_urls),
delegate,
}),
selected_urls,
)
}
}
impl HttpClientSelector for RecordingHttpClientSelector {
fn client_for_url<'a>(&'a self, request_url: &'a str) -> HttpClientSelectionFuture<'a> {
match self.selected_urls.lock() {
Ok(mut selected_urls) => selected_urls.push(request_url.to_string()),
Err(error) => panic!("selected URL recorder lock should not be poisoned: {error}"),
}
Box::pin(self.delegate.client_for_url(request_url))
}
}
pub(crate) fn recording_remote_plugin_service_config(
chatgpt_base_url: String,
) -> (RemotePluginServiceConfig, Arc<Mutex<Vec<String>>>) {
let http_client_factory = test_http_client_factory();
let (http_clients, selected_urls) = RecordingHttpClientSelector::new();
(
RemotePluginServiceConfig {
chatgpt_base_url,
http_client_factory,
http_clients,
},
selected_urls,
)
}
pub(crate) fn recorded_http_client_urls(selected_urls: &Mutex<Vec<String>>) -> Vec<String> {
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 +220,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(),
)
}
codex-rs/core/src/config/config_tests.rs
+9
View File
@@ -1594,6 +1594,15 @@ respect_system_proxy = true
config.http_client_factory().outbound_proxy_policy(),
codex_http_client::OutboundProxyPolicy::RespectSystemProxy
);
assert_eq!(
config.plugins_config_input().remote_plugin_service_config(),
codex_core_plugins::remote::RemotePluginServiceConfig::new(
config.chatgpt_base_url,
codex_http_client::HttpClientFactory::new(
codex_http_client::OutboundProxyPolicy::RespectSystemProxy,
),
)
);
Ok(())
}
codex-rs/core/src/config/mod.rs
+1
View File
@@ -1503,6 +1503,7 @@ impl Config {
self.features.enabled(Feature::Plugins),
self.features.enabled(Feature::RemotePlugin),
self.chatgpt_base_url.clone(),
self.http_client_factory(),
)
}
codex-rs/core/src/plugins/discoverable_tests.rs
+4 -3
View File
@@ -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
codex-rs/deny.toml
-1
View File
@@ -243,7 +243,6 @@ deny = [
"codex-api",
"codex-app-server",
"codex-core",
"codex-core-plugins",
"codex-exec-server",
"codex-lmstudio",
"codex-login",
codex-rs/http-client/src/default_client.rs
+21 -5
View File
@@ -1,4 +1,4 @@
use http::Error as HttpError;
use http::Error as HttpRequestBuildError;
use http::HeaderMap;
use http::HeaderName;
use http::HeaderValue;
@@ -6,13 +6,15 @@ use opentelemetry::global;
use opentelemetry::propagation::Injector;
use reqwest::IntoUrl;
use reqwest::Method;
use reqwest::Response;
use serde::Serialize;
use std::fmt::Display;
use std::time::Duration;
use tracing::Span;
use tracing_opentelemetry::OpenTelemetrySpanExt;
pub type HttpError = reqwest::Error;
pub type HttpResponse = reqwest::Response;
#[derive(Clone, Debug)]
pub struct HttpClient {
inner: reqwest::Client,
@@ -37,6 +39,20 @@ impl HttpClient {
self.request(Method::POST, url)
}
pub fn put<U>(&self, url: U) -> RequestBuilder
where
U: IntoUrl,
{
self.request(Method::PUT, url)
}
pub fn delete<U>(&self, url: U) -> RequestBuilder
where
U: IntoUrl,
{
self.request(Method::DELETE, url)
}
pub fn request<U>(&self, method: Method, url: U) -> RequestBuilder
where
U: IntoUrl,
@@ -78,9 +94,9 @@ impl RequestBuilder {
pub fn header<K, V>(self, key: K, value: V) -> Self
where
HeaderName: TryFrom<K>,
<HeaderName as TryFrom<K>>::Error: Into<HttpError>,
<HeaderName as TryFrom<K>>::Error: Into<HttpRequestBuildError>,
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: Into<HttpError>,
<HeaderValue as TryFrom<V>>::Error: Into<HttpRequestBuildError>,
{
self.map(|builder| builder.header(key, value))
}
@@ -110,7 +126,7 @@ impl RequestBuilder {
self.map(|builder| builder.body(body))
}
pub async fn send(self) -> Result<Response, reqwest::Error> {
pub async fn send(self) -> Result<HttpResponse, HttpError> {
let headers = trace_headers();
match self.builder.headers(headers).send().await {
codex-rs/http-client/src/lib.rs
+2
View File
@@ -22,6 +22,8 @@ pub use crate::custom_ca::build_reqwest_client_with_custom_ca;
pub use crate::custom_ca::build_rustls_client_config_with_custom_ca;
pub use crate::custom_ca::maybe_build_rustls_client_config_with_custom_ca;
pub use crate::default_client::HttpClient;
pub use crate::default_client::HttpError;
pub use crate::default_client::HttpResponse;
pub use crate::default_client::RequestBuilder;
pub use crate::error::StreamError;
pub use crate::error::TransportError;