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

refactor: keep authorization headers in provider

Browse Source
This commit is contained in:
Edward Frazer
2026-04-22 10:31:01 -07:00
parent b9bf66f1d5
commit 27d0647682
15 changed files with 281 additions and 716 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+50 -527
View File
@@ -32,6 +32,7 @@ use codex_app_server_protocol::AccountUpdatedNotification;
use codex_app_server_protocol::AddCreditsNudgeCreditType;
use codex_app_server_protocol::AddCreditsNudgeEmailStatus;
use codex_app_server_protocol::AppInfo;
use codex_app_server_protocol::AppSummary;
use codex_app_server_protocol::AppsListParams;
use codex_app_server_protocol::AppsListResponse;
use codex_app_server_protocol::AskForApproval;
@@ -260,6 +261,12 @@ use codex_core_plugins::loader::load_plugin_mcp_servers;
use codex_core_plugins::manifest::PluginManifestInterface;
use codex_core_plugins::marketplace::MarketplaceError;
use codex_core_plugins::marketplace::MarketplacePluginSource;
use codex_core_plugins::remote::RemoteMarketplace;
use codex_core_plugins::remote::RemotePluginCatalogError;
use codex_core_plugins::remote::RemotePluginDetail as RemoteCatalogPluginDetail;
use codex_core_plugins::remote::RemotePluginServiceConfig;
use codex_core_plugins::remote::RemotePluginSummary as RemoteCatalogPluginSummary;
use codex_exec_server::EnvironmentManager;
use codex_exec_server::LOCAL_FS;
use codex_features::FEATURES;
use codex_features::Feature;
@@ -317,6 +324,7 @@ use codex_protocol::protocol::ReviewTarget as CoreReviewTarget;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionConfiguredEvent;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use codex_protocol::protocol::W3cTraceContext;
use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS;
@@ -372,6 +380,7 @@ use codex_app_server_protocol::ServerRequest;
mod apps_list_helpers;
mod plugin_app_helpers;
mod plugin_mcp_oauth;
mod plugins;
mod token_usage_replay;
use crate::filters::compute_source_filters;
@@ -5681,27 +5690,17 @@ impl CodexMessageProcessor {
.to_mcp_config(self.thread_manager.plugins_manager().as_ref())
.await;
let auth = self.auth_manager.auth().await;
let runtime_environment = match self.thread_manager.environment_manager().current().await {
Ok(Some(environment)) => {
let environment_manager = self.thread_manager.environment_manager();
let runtime_environment = match environment_manager.default_environment() {
Some(environment) => {
// Status listing has no turn cwd. This fallback is used only
// by executor-backed stdio MCPs whose config omits `cwd`.
McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf())
}
Ok(None) => McpRuntimeEnvironment::new(
Arc::new(codex_exec_server::Environment::default()),
None => McpRuntimeEnvironment::new(
environment_manager.local_environment(),
config.cwd.to_path_buf(),
),
Err(err) => {
// TODO(aibrahim): Investigate degrading MCP status listing when
// executor environment creation fails.
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to create environment: {err}"),
data: None,
};
self.outgoing.send_error(request, error).await;
return;
}
};
tokio::spawn(async move {
@@ -5860,25 +5859,14 @@ impl CodexMessageProcessor {
.to_mcp_config(self.thread_manager.plugins_manager().as_ref())
.await;
let auth = self.auth_manager.auth().await;
let runtime_environment = match self.thread_manager.environment_manager().current().await {
Ok(Some(environment)) => {
// Resource reads without a thread have no turn cwd. This fallback
// is used only by executor-backed stdio MCPs whose config omits `cwd`.
McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf())
}
Ok(None) => McpRuntimeEnvironment::new(
Arc::new(codex_exec_server::Environment::default()),
config.cwd.to_path_buf(),
),
Err(err) => {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to create environment: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
let runtime_environment = {
let environment_manager = self.thread_manager.environment_manager();
let environment = environment_manager
.default_environment()
.unwrap_or_else(|| environment_manager.local_environment());
// Resource reads without a thread have no turn cwd. This fallback
// is used only by executor-backed stdio MCPs whose config omits `cwd`.
McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf())
};
tokio::spawn(async move {
@@ -6226,8 +6214,9 @@ impl CodexMessageProcessor {
let request = request_id.clone();
let outgoing = Arc::clone(&self.outgoing);
let environment_manager = self.thread_manager.environment_manager();
tokio::spawn(async move {
Self::apps_list_task(outgoing, request, params, config).await;
Self::apps_list_task(outgoing, request, params, config, environment_manager).await;
});
}
@@ -6236,6 +6225,7 @@ impl CodexMessageProcessor {
request_id: ConnectionRequestId,
params: AppsListParams,
config: Config,
environment_manager: Arc<EnvironmentManager>,
) {
let AppsListParams {
cursor,
@@ -6270,12 +6260,15 @@ impl CodexMessageProcessor {
let accessible_config = config.clone();
let accessible_tx = tx.clone();
tokio::spawn(async move {
let result = connectors::list_accessible_connectors_from_mcp_tools_with_options(
&accessible_config,
force_refetch,
)
.await
.map_err(|err| format!("failed to load accessible apps: {err}"));
let result =
connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager(
&accessible_config,
force_refetch,
&environment_manager,
)
.await
.map(|status| status.connectors)
.map_err(|err| format!("failed to load accessible apps: {err}"));
let _ = accessible_tx.send(AppListLoadResult::Accessible(result));
});
@@ -6465,23 +6458,11 @@ impl CodexMessageProcessor {
};
let skills_manager = self.thread_manager.skills_manager();
let plugins_manager = self.thread_manager.plugins_manager();
let fs = match self.thread_manager.environment_manager().current().await {
Ok(Some(environment)) => Some(environment.get_filesystem()),
Ok(None) => None,
Err(err) => {
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to create environment: {err}"),
data: None,
},
)
.await;
return;
}
};
let fs = self
.thread_manager
.environment_manager()
.default_environment()
.map(|environment| environment.get_filesystem());
let mut data = Vec::new();
for cwd in cwds {
let extra_roots = extra_roots_by_cwd
@@ -6553,7 +6534,6 @@ impl CodexMessageProcessor {
.send_response(request_id, SkillsListResponse { data })
.await;
}
async fn marketplace_remove(
&self,
request_id: ConnectionRequestId,
@@ -6587,120 +6567,6 @@ impl CodexMessageProcessor {
}
}
}
async fn plugin_list(&self, request_id: ConnectionRequestId, params: PluginListParams) {
let plugins_manager = self.thread_manager.plugins_manager();
let PluginListParams { cwds } = params;
let roots = cwds.unwrap_or_default();
plugins_manager.maybe_start_non_curated_plugin_cache_refresh(&roots);
let config = match self.load_latest_config(/*fallback_cwd*/ None).await {
Ok(config) => config,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
return;
}
};
let auth = self.auth_manager.auth().await;
let config_for_marketplace_listing = config.clone();
let plugins_manager_for_marketplace_listing = plugins_manager.clone();
let (data, marketplace_load_errors) = match tokio::task::spawn_blocking(move || {
let outcome = plugins_manager_for_marketplace_listing
.list_marketplaces_for_config(&config_for_marketplace_listing, &roots)?;
Ok::<
(
Vec<PluginMarketplaceEntry>,
Vec<codex_app_server_protocol::MarketplaceLoadErrorInfo>,
),
MarketplaceError,
>((
outcome
.marketplaces
.into_iter()
.map(|marketplace| PluginMarketplaceEntry {
name: marketplace.name,
path: Some(marketplace.path),
interface: marketplace.interface.map(|interface| MarketplaceInterface {
display_name: interface.display_name,
}),
plugins: marketplace
.plugins
.into_iter()
.map(|plugin| PluginSummary {
id: plugin.id,
installed: plugin.installed,
enabled: plugin.enabled,
name: plugin.name,
source: marketplace_plugin_source_to_info(plugin.source),
install_policy: plugin.policy.installation.into(),
auth_policy: plugin.policy.authentication.into(),
interface: plugin.interface.map(local_plugin_interface_to_info),
})
.collect(),
})
.collect(),
outcome
.errors
.into_iter()
.map(|err| codex_app_server_protocol::MarketplaceLoadErrorInfo {
marketplace_path: err.path,
message: err.message,
})
.collect(),
))
})
.await
{
Ok(Ok(outcome)) => outcome,
Ok(Err(err)) => {
self.send_marketplace_error(request_id, err, "list marketplace plugins")
.await;
return;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to list marketplace plugins: {err}"),
)
.await;
return;
}
};
let featured_plugin_ids = if data
.iter()
.any(|marketplace| marketplace.name == OPENAI_CURATED_MARKETPLACE_NAME)
{
match plugins_manager
.featured_plugin_ids_for_config(&config, auth.as_ref())
.await
{
Ok(featured_plugin_ids) => featured_plugin_ids,
Err(err) => {
warn!(
error = %err,
"plugin/list featured plugin fetch failed; returning empty featured ids"
);
Vec::new()
}
}
} else {
Vec::new()
};
self.outgoing
.send_response(
request_id,
PluginListResponse {
marketplaces: data,
marketplace_load_errors,
featured_plugin_ids,
},
)
.await;
}
async fn marketplace_add(&self, request_id: ConnectionRequestId, params: MarketplaceAddParams) {
let result = add_marketplace_to_codex_home(
self.config.codex_home.to_path_buf(),
@@ -6734,106 +6600,6 @@ impl CodexMessageProcessor {
}
}
async fn plugin_read(&self, request_id: ConnectionRequestId, params: PluginReadParams) {
let plugins_manager = self.thread_manager.plugins_manager();
let PluginReadParams {
marketplace_path,
remote_marketplace_name,
plugin_name,
} = params;
let marketplace_path = match (marketplace_path, remote_marketplace_name) {
(Some(marketplace_path), None) => marketplace_path,
(None, Some(remote_marketplace_name)) => {
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!(
"remote plugin read is not supported yet for marketplace {remote_marketplace_name}"
),
data: None,
},
)
.await;
return;
}
(Some(_), Some(_)) | (None, None) => {
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "plugin/read requires exactly one of marketplacePath or remoteMarketplaceName".to_string(),
data: None,
},
)
.await;
return;
}
};
let config_cwd = marketplace_path.as_path().parent().map(Path::to_path_buf);
let config = match self.load_latest_config(config_cwd).await {
Ok(config) => config,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
return;
}
};
let request = PluginReadRequest {
plugin_name,
marketplace_path,
};
let outcome = match plugins_manager
.read_plugin_for_config(&config, &request)
.await
{
Ok(outcome) => outcome,
Err(err) => {
self.send_marketplace_error(request_id, err, "read plugin details")
.await;
return;
}
};
let app_summaries =
plugin_app_helpers::load_plugin_app_summaries(&config, &outcome.plugin.apps).await;
let visible_skills = outcome
.plugin
.skills
.iter()
.filter(|skill| {
skill.matches_product_restriction_for_product(
self.thread_manager.session_source().restriction_product(),
)
})
.cloned()
.collect::<Vec<_>>();
let plugin = PluginDetail {
marketplace_name: outcome.marketplace_name,
marketplace_path: outcome.marketplace_path,
summary: PluginSummary {
id: outcome.plugin.id,
name: outcome.plugin.name,
source: marketplace_plugin_source_to_info(outcome.plugin.source),
installed: outcome.plugin.installed,
enabled: outcome.plugin.enabled,
install_policy: outcome.plugin.policy.installation.into(),
auth_policy: outcome.plugin.policy.authentication.into(),
interface: outcome.plugin.interface.map(local_plugin_interface_to_info),
},
description: outcome.plugin.description,
skills: plugin_skills_to_info(&visible_skills, &outcome.plugin.disabled_skill_paths),
apps: app_summaries,
mcp_servers: outcome.plugin.mcp_server_names,
};
self.outgoing
.send_response(request_id, PluginReadResponse { plugin })
.await;
}
async fn skills_config_write(
&self,
request_id: ConnectionRequestId,
@@ -6892,259 +6658,6 @@ impl CodexMessageProcessor {
}
}
async fn plugin_install(&self, request_id: ConnectionRequestId, params: PluginInstallParams) {
let PluginInstallParams {
marketplace_path,
remote_marketplace_name,
plugin_name,
} = params;
let marketplace_path = match (marketplace_path, remote_marketplace_name) {
(Some(marketplace_path), None) => marketplace_path,
(None, Some(remote_marketplace_name)) => {
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!(
"remote plugin install is not supported yet for marketplace {remote_marketplace_name}"
),
data: None,
},
)
.await;
return;
}
(Some(_), Some(_)) | (None, None) => {
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "plugin/install requires exactly one of marketplacePath or remoteMarketplaceName".to_string(),
data: None,
},
)
.await;
return;
}
};
let config_cwd = marketplace_path.as_path().parent().map(Path::to_path_buf);
let plugins_manager = self.thread_manager.plugins_manager();
let request = PluginInstallRequest {
plugin_name,
marketplace_path,
};
let install_result = plugins_manager.install_plugin(request).await;
match install_result {
Ok(result) => {
let config = match self.load_latest_config(config_cwd).await {
Ok(config) => config,
Err(err) => {
warn!(
"failed to reload config after plugin install, using current config: {err:?}"
);
self.config.as_ref().clone()
}
};
self.clear_plugin_related_caches();
let plugin_mcp_servers =
load_plugin_mcp_servers(result.installed_path.as_path()).await;
if !plugin_mcp_servers.is_empty() {
if let Err(err) = self.queue_mcp_server_refresh_for_config(&config).await {
warn!(
plugin = result.plugin_id.as_key(),
"failed to queue MCP refresh after plugin install: {err:?}"
);
}
self.start_plugin_mcp_oauth_logins(&config, plugin_mcp_servers)
.await;
}
let plugin_apps = load_plugin_apps(result.installed_path.as_path()).await;
let auth = self.auth_manager.auth().await;
let apps_needing_auth = if plugin_apps.is_empty()
|| !config.features.apps_enabled_for_auth(
auth.as_ref().is_some_and(CodexAuth::uses_codex_backend),
) {
Vec::new()
} else {
let (all_connectors_result, accessible_connectors_result) = tokio::join!(
connectors::list_all_connectors_with_options(&config, /*force_refetch*/ true),
connectors::list_accessible_connectors_from_mcp_tools_with_options_and_status(
&config, /*force_refetch*/ true
),
);
let all_connectors = match all_connectors_result {
Ok(connectors) => connectors,
Err(err) => {
warn!(
plugin = result.plugin_id.as_key(),
"failed to load app metadata after plugin install: {err:#}"
);
connectors::list_cached_all_connectors(&config)
.await
.unwrap_or_default()
}
};
let all_connectors =
connectors::connectors_for_plugin_apps(all_connectors, &plugin_apps);
let (accessible_connectors, codex_apps_ready) =
match accessible_connectors_result {
Ok(status) => (status.connectors, status.codex_apps_ready),
Err(err) => {
warn!(
plugin = result.plugin_id.as_key(),
"failed to load accessible apps after plugin install: {err:#}"
);
(
connectors::list_cached_accessible_connectors_from_mcp_tools(
&config,
)
.await
.unwrap_or_default(),
false,
)
}
};
if !codex_apps_ready {
warn!(
plugin = result.plugin_id.as_key(),
"codex_apps MCP not ready after plugin install; skipping appsNeedingAuth check"
);
}
plugin_app_helpers::plugin_apps_needing_auth(
&all_connectors,
&accessible_connectors,
&plugin_apps,
codex_apps_ready,
)
};
self.outgoing
.send_response(
request_id,
PluginInstallResponse {
auth_policy: result.auth_policy.into(),
apps_needing_auth,
},
)
.await;
}
Err(err) => {
if err.is_invalid_request() {
self.send_invalid_request_error(request_id, err.to_string())
.await;
return;
}
match err {
CorePluginInstallError::Marketplace(err) => {
self.send_marketplace_error(request_id, err, "install plugin")
.await;
}
CorePluginInstallError::Config(err) => {
self.send_internal_error(
request_id,
format!("failed to persist installed plugin config: {err}"),
)
.await;
}
CorePluginInstallError::Remote(err) => {
self.send_internal_error(
request_id,
format!("failed to enable remote plugin: {err}"),
)
.await;
}
CorePluginInstallError::Join(err) => {
self.send_internal_error(
request_id,
format!("failed to install plugin: {err}"),
)
.await;
}
CorePluginInstallError::Store(err) => {
self.send_internal_error(
request_id,
format!("failed to install plugin: {err}"),
)
.await;
}
}
}
}
}
async fn plugin_uninstall(
&self,
request_id: ConnectionRequestId,
params: PluginUninstallParams,
) {
let PluginUninstallParams { plugin_id } = params;
let plugins_manager = self.thread_manager.plugins_manager();
let uninstall_result = plugins_manager.uninstall_plugin(plugin_id).await;
match uninstall_result {
Ok(()) => {
self.clear_plugin_related_caches();
self.outgoing
.send_response(request_id, PluginUninstallResponse {})
.await;
}
Err(err) => {
if err.is_invalid_request() {
self.send_invalid_request_error(request_id, err.to_string())
.await;
return;
}
match err {
CorePluginUninstallError::Config(err) => {
self.send_internal_error(
request_id,
format!("failed to clear plugin config: {err}"),
)
.await;
}
CorePluginUninstallError::Remote(err) => {
self.send_internal_error(
request_id,
format!("failed to uninstall remote plugin: {err}"),
)
.await;
}
CorePluginUninstallError::Join(err) => {
self.send_internal_error(
request_id,
format!("failed to uninstall plugin: {err}"),
)
.await;
}
CorePluginUninstallError::Store(err) => {
self.send_internal_error(
request_id,
format!("failed to uninstall plugin: {err}"),
)
.await;
}
CorePluginUninstallError::InvalidPluginId(_) => {
unreachable!("invalid plugin ids are handled above");
}
}
}
}
}
async fn turn_start(
&self,
request_id: ConnectionRequestId,
@@ -7187,6 +6700,15 @@ impl CodexMessageProcessor {
let collaboration_mode = params.collaboration_mode.map(|mode| {
self.normalize_turn_start_collaboration_mode(mode, collaboration_modes_config)
});
let environments = params.environments.map(|environments| {
environments
.into_iter()
.map(|environment| TurnEnvironmentSelection {
environment_id: environment.environment_id,
cwd: environment.cwd,
})
.collect()
});
// Map v2 input items to core input items.
let mapped_items: Vec<CoreInputItem> = params
@@ -7238,6 +6760,7 @@ impl CodexMessageProcessor {
thread.as_ref(),
Op::UserInput {
items: mapped_items,
environments,
final_output_json_schema: params.output_schema,
responsesapi_client_metadata: params.responsesapi_client_metadata,
},
@@ -9259,7 +8782,7 @@ fn plugin_skills_to_info(
default_prompt: interface.default_prompt,
}
}),
path: skill.path_to_skills_md.clone(),
path: Some(skill.path_to_skills_md.clone()),
enabled: !disabled_skill_paths.contains(&skill.path_to_skills_md),
})
.collect()
codex-rs/backend-client/src/client.rs
+2
View File
@@ -821,6 +821,7 @@ mod tests {
base_url: "https://example.test".to_string(),
http: reqwest::Client::new(),
bearer_token: None,
auth_headers: HeaderMap::new(),
user_agent: None,
chatgpt_account_id: None,
chatgpt_account_is_fedramp: false,
@@ -835,6 +836,7 @@ mod tests {
base_url: "https://chatgpt.com/backend-api".to_string(),
http: reqwest::Client::new(),
bearer_token: None,
auth_headers: HeaderMap::new(),
user_agent: None,
chatgpt_account_id: None,
chatgpt_account_is_fedramp: false,
codex-rs/chatgpt/src/connectors.rs
+17 -16
View File
@@ -32,6 +32,21 @@ async fn apps_enabled(config: &Config) -> bool {
.features
.apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend))
}
async fn connector_auth(config: &Config) -> anyhow::Result<CodexAuth> {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
let auth = auth_manager
.auth()
.await
.ok_or_else(|| anyhow::anyhow!("ChatGPT auth not available"))?;
anyhow::ensure!(
auth.uses_codex_backend(),
"ChatGPT connectors require Codex backend auth"
);
Ok(auth)
}
pub async fn list_connectors(config: &Config) -> anyhow::Result<Vec<AppInfo>> {
if !apps_enabled(config).await {
return Ok(Vec::new());
@@ -59,12 +74,7 @@ pub async fn list_cached_all_connectors(config: &Config) -> Option<Vec<AppInfo>>
return Some(Vec::new());
}
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
let auth = auth_manager.auth().await?;
if !auth.uses_codex_backend() {
return Some(Vec::new());
}
let auth = connector_auth(config).await.ok()?;
let cache_key = all_connectors_cache_key(config, &auth);
let connectors = codex_connectors::cached_all_connectors(&cache_key)?;
let connectors = merge_plugin_connectors(
@@ -87,16 +97,7 @@ pub async fn list_all_connectors_with_options(
if !apps_enabled(config).await {
return Ok(Vec::new());
}
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
let auth = auth_manager
.auth()
.await
.ok_or_else(|| anyhow::anyhow!("ChatGPT auth not available"))?;
anyhow::ensure!(
auth.uses_codex_backend(),
"ChatGPT connectors require Codex backend auth"
);
let auth = connector_auth(config).await?;
let cache_key = all_connectors_cache_key(config, &auth);
let connectors = codex_connectors::list_all_connectors_with_options(
cache_key,
codex-rs/codex-api/src/auth.rs
+7
View File
@@ -34,6 +34,13 @@ pub trait AuthProvider: Send + Sync {
/// used by telemetry and non-HTTP request paths.
fn add_auth_headers(&self, headers: &mut HeaderMap);
/// Returns any auth headers that are available without request body access.
fn to_auth_headers(&self) -> HeaderMap {
let mut headers = HeaderMap::new();
self.add_auth_headers(&mut headers);
headers
}
/// Applies auth to a complete outbound request and returns the request to send.
///
/// The input `request` is moved into this method. Implementations may mutate
codex-rs/core-plugins/src/remote_legacy.rs
+13 -28
View File
@@ -123,23 +123,17 @@ pub async fn fetch_remote_plugin_status(
let Some(auth) = auth else {
return Err(RemotePluginFetchError::AuthRequired);
};
if !auth.is_chatgpt_auth() {
if !auth.uses_codex_backend() {
return Err(RemotePluginFetchError::UnsupportedAuthMode);
}
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/plugins/list");
let client = build_reqwest_client();
let token = auth
.get_token()
.map_err(RemotePluginFetchError::AuthToken)?;
let mut request = client
let request = client
.get(&url)
.timeout(REMOTE_PLUGIN_FETCH_TIMEOUT)
.bearer_auth(token);
if let Some(account_id) = auth.get_account_id() {
request = request.header("chatgpt-account-id", account_id);
}
.headers(codex_model_provider::auth_provider_from_auth(auth).to_auth_headers());
let response = request
.send()
@@ -176,14 +170,9 @@ pub async fn fetch_remote_featured_plugin_ids(
)])
.timeout(REMOTE_FEATURED_PLUGIN_FETCH_TIMEOUT);
if let Some(auth) = auth.filter(|auth| auth.is_chatgpt_auth()) {
let token = auth
.get_token()
.map_err(RemotePluginFetchError::AuthToken)?;
request = request.bearer_auth(token);
if let Some(account_id) = auth.get_account_id() {
request = request.header("chatgpt-account-id", account_id);
}
if let Some(auth) = auth.filter(|auth| auth.uses_codex_backend()) {
request =
request.headers(codex_model_provider::auth_provider_from_auth(auth).to_auth_headers());
}
let response = request
@@ -223,11 +212,13 @@ pub async fn uninstall_remote_plugin(
Ok(())
}
fn ensure_chatgpt_auth(auth: Option<&CodexAuth>) -> Result<&CodexAuth, RemotePluginMutationError> {
fn ensure_codex_backend_auth(
auth: Option<&CodexAuth>,
) -> Result<&CodexAuth, RemotePluginMutationError> {
let Some(auth) = auth else {
return Err(RemotePluginMutationError::AuthRequired);
};
if !auth.is_chatgpt_auth() {
if !auth.uses_codex_backend() {
return Err(RemotePluginMutationError::UnsupportedAuthMode);
}
Ok(auth)
@@ -243,19 +234,13 @@ async fn post_remote_plugin_mutation(
plugin_id: &str,
action: &str,
) -> Result<RemotePluginMutationResponse, RemotePluginMutationError> {
let auth = ensure_chatgpt_auth(auth)?;
let auth = ensure_codex_backend_auth(auth)?;
let url = remote_plugin_mutation_url(config, plugin_id, action)?;
let client = build_reqwest_client();
let token = auth
.get_token()
.map_err(RemotePluginMutationError::AuthToken)?;
let mut request = client
let request = client
.post(url.clone())
.timeout(REMOTE_PLUGIN_MUTATION_TIMEOUT)
.bearer_auth(token);
if let Some(account_id) = auth.get_account_id() {
request = request.header("chatgpt-account-id", account_id);
}
.headers(codex_model_provider::auth_provider_from_auth(auth).to_auth_headers());
let response = request
.send()
codex-rs/core/src/connectors.rs
+19 -33
View File
@@ -16,7 +16,6 @@ use codex_connectors::DirectoryListResponse;
use codex_exec_server::EnvironmentManager;
use codex_exec_server::EnvironmentManagerArgs;
use codex_exec_server::ExecServerRuntimePaths;
use codex_login::token_data::TokenData;
use codex_protocol::protocol::SandboxPolicy;
use codex_tools::DiscoverableTool;
use rmcp::model::ToolAnnotations;
@@ -351,16 +350,9 @@ fn accessible_connectors_cache_key(
config: &Config,
auth: Option<&CodexAuth>,
) -> AccessibleConnectorsCacheKey {
let token_data: Option<TokenData> = auth.and_then(|auth| auth.get_token_data().ok());
let account_id = token_data
.as_ref()
.and_then(|token_data| token_data.account_id.clone());
let chatgpt_user_id = token_data
.as_ref()
.and_then(|token_data| token_data.id_token.chatgpt_user_id.clone());
let is_workspace_account = token_data
.as_ref()
.is_some_and(|token_data| token_data.id_token.is_workspace_account());
let account_id = auth.and_then(CodexAuth::get_account_id);
let chatgpt_user_id = auth.and_then(CodexAuth::get_chatgpt_user_id);
let is_workspace_account = auth.is_some_and(CodexAuth::is_workspace_account);
AccessibleConnectorsCacheKey {
chatgpt_base_url: config.chatgpt_base_url.clone(),
account_id,
@@ -431,31 +423,29 @@ async fn list_directory_connectors_for_tool_suggest_with_auth(
return Ok(Vec::new());
}
let token_data = if let Some(auth) = auth {
auth.get_token_data().ok()
let loaded_auth;
let auth = if let Some(auth) = auth {
Some(auth)
} else {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
auth_manager
.auth()
.await
.and_then(|auth| auth.get_token_data().ok())
loaded_auth = auth_manager.auth().await;
loaded_auth.as_ref()
};
let Some(token_data) = token_data else {
let Some(auth) = auth.filter(|auth| auth.uses_codex_backend()) else {
return Ok(Vec::new());
};
let account_id = match token_data.account_id.as_deref() {
let account_id = match auth.get_account_id() {
Some(account_id) if !account_id.is_empty() => account_id,
_ => return Ok(Vec::new()),
};
let access_token = token_data.access_token.clone();
let account_id = account_id.to_string();
let is_workspace_account = token_data.id_token.is_workspace_account();
let auth_headers = codex_model_provider::auth_provider_from_auth(auth).to_auth_headers();
let is_workspace_account = auth.is_workspace_account();
let cache_key = AllConnectorsCacheKey::new(
config.chatgpt_base_url.clone(),
Some(account_id.clone()),
token_data.id_token.chatgpt_user_id.clone(),
auth.get_chatgpt_user_id(),
is_workspace_account,
);
@@ -464,14 +454,12 @@ async fn list_directory_connectors_for_tool_suggest_with_auth(
is_workspace_account,
/*force_refetch*/ false,
|path| {
let access_token = access_token.clone();
let account_id = account_id.clone();
let auth_headers = auth_headers.clone();
async move {
chatgpt_get_request_with_token::<DirectoryListResponse>(
chatgpt_get_request_with_auth_headers::<DirectoryListResponse>(
config,
path,
access_token.as_str(),
account_id.as_str(),
auth_headers,
)
.await
}
@@ -480,18 +468,16 @@ async fn list_directory_connectors_for_tool_suggest_with_auth(
.await
}
async fn chatgpt_get_request_with_token<T: DeserializeOwned>(
async fn chatgpt_get_request_with_auth_headers<T: DeserializeOwned>(
config: &Config,
path: String,
access_token: &str,
account_id: &str,
auth_headers: http::HeaderMap,
) -> anyhow::Result<T> {
let client = create_client();
let url = format!("{}{}", config.chatgpt_base_url, path);
let response = client
.get(&url)
.bearer_auth(access_token)
.header("chatgpt-account-id", account_id)
.headers(auth_headers)
.header("Content-Type", "application/json")
.timeout(DIRECTORY_CONNECTORS_TIMEOUT)
.send()
codex-rs/core/src/session/turn_context.rs
+2 -7
View File
@@ -5,10 +5,7 @@ use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_sandboxing::policy_transforms::merge_permission_profiles;
pub(super) fn image_generation_tool_auth_allowed(auth_manager: Option<&AuthManager>) -> bool {
auth_manager
.and_then(AuthManager::auth_cached)
.as_ref()
.is_some_and(CodexAuth::uses_codex_backend)
auth_manager.is_some_and(AuthManager::current_auth_uses_codex_backend)
}
#[derive(Clone, Debug)]
@@ -104,9 +101,7 @@ impl TurnContext {
let uses_codex_backend = self
.auth_manager
.as_deref()
.and_then(AuthManager::auth_cached)
.as_ref()
.is_some_and(CodexAuth::uses_codex_backend);
.is_some_and(AuthManager::current_auth_uses_codex_backend);
self.features.apps_enabled_for_auth(uses_codex_backend)
}
codex-rs/login/src/auth/agent_identity.rs
+1 -1
View File
@@ -15,7 +15,7 @@ const DEFAULT_CHATGPT_BACKEND_BASE_URL: &str = "https://chatgpt.com/backend-api"
#[derive(Debug)]
pub struct AgentIdentityAuth {
record: AgentIdentityAuthRecord,
pub(super) process_task_id: Arc<OnceCell<String>>,
process_task_id: Arc<OnceCell<String>>,
}
impl Clone for AgentIdentityAuth {
codex-rs/login/src/auth/manager.rs
+9 -11
View File
@@ -398,17 +398,8 @@ impl CodexAuth {
}
pub fn is_workspace_account(&self) -> bool {
matches!(
self.account_plan_type(),
Some(
AccountPlanType::Team
| AccountPlanType::SelfServeBusinessUsageBased
| AccountPlanType::Business
| AccountPlanType::EnterpriseCbpUsageBased
| AccountPlanType::Enterprise
| AccountPlanType::Edu
)
)
self.account_plan_type()
.is_some_and(AccountPlanType::is_workspace_account)
}
/// Returns `None` if token-backed ChatGPT auth is unavailable.
@@ -1723,6 +1714,13 @@ impl AuthManager {
self.auth_cached().as_ref().map(CodexAuth::auth_mode)
}
pub fn current_auth_uses_codex_backend(&self) -> bool {
matches!(
self.auth_mode(),
Some(AuthMode::Chatgpt | AuthMode::ChatgptAuthTokens | AuthMode::AgentIdentity)
)
}
fn is_stale_for_proactive_refresh(auth: &CodexAuth) -> bool {
let chatgpt_auth = match auth {
CodexAuth::Chatgpt(chatgpt_auth) => chatgpt_auth,
codex-rs/model-provider/src/auth.rs
+88 -54
View File
@@ -14,40 +14,30 @@ use http::HeaderValue;
use crate::bearer_auth_provider::BearerAuthProvider;
#[derive(Clone, Debug)]
struct CodexAuthProvider {
auth: CodexAuth,
struct AgentIdentityAuthProvider {
auth: codex_login::auth::AgentIdentityAuth,
}
impl AuthProvider for CodexAuthProvider {
impl AuthProvider for AgentIdentityAuthProvider {
fn add_auth_headers(&self, headers: &mut HeaderMap) {
let header_value = match &self.auth {
CodexAuth::AgentIdentity(auth) => {
let record = auth.record();
let process_task_id = auth.process_task_id().ok_or_else(|| {
std::io::Error::other("agent identity process task is not initialized")
});
process_task_id.and_then(|task_id| {
authorization_header_for_agent_task(
AgentIdentityKey {
agent_runtime_id: &record.agent_runtime_id,
private_key_pkcs8_base64: &record.agent_private_key,
},
AgentTaskAuthorizationTarget {
agent_runtime_id: &record.agent_runtime_id,
task_id,
},
)
.map_err(std::io::Error::other)
})
}
CodexAuth::ApiKey(_) => self.auth.api_key().map_or_else(
|| Err(std::io::Error::other("API key auth missing API key")),
|api_key| Ok(format!("Bearer {api_key}")),
),
CodexAuth::Chatgpt(_) | CodexAuth::ChatgptAuthTokens(_) => {
self.auth.get_token().map(|token| format!("Bearer {token}"))
}
};
let record = self.auth.record();
let header_value = self
.auth
.process_task_id()
.ok_or_else(|| std::io::Error::other("agent identity process task is not initialized"))
.and_then(|task_id| {
authorization_header_for_agent_task(
AgentIdentityKey {
agent_runtime_id: &record.agent_runtime_id,
private_key_pkcs8_base64: &record.agent_private_key,
},
AgentTaskAuthorizationTarget {
agent_runtime_id: &record.agent_runtime_id,
task_id,
},
)
.map_err(std::io::Error::other)
});
if let Ok(header_value) = header_value
&& let Ok(header) = HeaderValue::from_str(&header_value)
@@ -55,9 +45,7 @@ impl AuthProvider for CodexAuthProvider {
let _ = headers.insert(http::header::AUTHORIZATION, header);
}
if let Some(account_id) = self.auth.get_account_id()
&& let Ok(header) = HeaderValue::from_str(&account_id)
{
if let Ok(header) = HeaderValue::from_str(self.auth.account_id()) {
let _ = headers.insert("ChatGPT-Account-ID", header);
}
@@ -67,6 +55,31 @@ impl AuthProvider for CodexAuthProvider {
}
}
#[derive(Clone, Debug)]
struct CodexBearerAuthProvider {
auth: CodexAuth,
}
impl AuthProvider for CodexBearerAuthProvider {
fn add_auth_headers(&self, headers: &mut HeaderMap) {
BearerAuthProvider {
token: self.auth.get_token().ok(),
account_id: self.auth.get_account_id(),
is_fedramp_account: self.auth.is_fedramp_account(),
}
.add_auth_headers(headers);
}
}
// Some providers are meant to send no auth headers. Examples include local OSS
// providers and custom test providers with `requires_openai_auth = false`.
#[derive(Clone, Debug)]
struct NoAuthProvider;
impl AuthProvider for NoAuthProvider {
fn add_auth_headers(&self, _headers: &mut HeaderMap) {}
}
/// Returns the provider-scoped auth manager when this provider uses command-backed auth.
///
/// Providers without custom auth continue using the caller-supplied base manager, when present.
@@ -84,34 +97,55 @@ pub(crate) fn resolve_provider_auth(
auth: Option<&CodexAuth>,
provider: &ModelProviderInfo,
) -> codex_protocol::error::Result<SharedAuthProvider> {
if let Some(auth) = bearer_auth_for_provider(provider)? {
return Ok(Arc::new(auth));
}
Ok(match auth {
Some(auth) => auth_provider_from_auth(auth),
None => Arc::new(NoAuthProvider),
})
}
fn bearer_auth_for_provider(
provider: &ModelProviderInfo,
) -> codex_protocol::error::Result<Option<BearerAuthProvider>> {
if let Some(api_key) = provider.api_key()? {
return Ok(Arc::new(BearerAuthProvider {
token: Some(api_key),
account_id: None,
is_fedramp_account: false,
}));
return Ok(Some(BearerAuthProvider::new(api_key)));
}
if let Some(token) = provider.experimental_bearer_token.clone() {
return Ok(Arc::new(BearerAuthProvider {
token: Some(token),
account_id: None,
is_fedramp_account: false,
}));
return Ok(Some(BearerAuthProvider::new(token)));
}
let Some(auth) = auth else {
return Ok(Arc::new(BearerAuthProvider {
token: None,
account_id: None,
is_fedramp_account: false,
}));
};
Ok(auth_provider_from_auth(auth))
Ok(None)
}
/// Builds request-header auth for a first-party Codex auth snapshot.
pub fn auth_provider_from_auth(auth: &CodexAuth) -> SharedAuthProvider {
Arc::new(CodexAuthProvider { auth: auth.clone() })
match auth {
CodexAuth::AgentIdentity(auth) => {
Arc::new(AgentIdentityAuthProvider { auth: auth.clone() })
}
CodexAuth::ApiKey(_) | CodexAuth::Chatgpt(_) | CodexAuth::ChatgptAuthTokens(_) => {
Arc::new(CodexBearerAuthProvider { auth: auth.clone() })
}
}
}
#[cfg(test)]
mod tests {
use codex_model_provider_info::WireApi;
use codex_model_provider_info::create_oss_provider_with_base_url;
use super::*;
#[test]
fn no_auth_provider_adds_no_headers() {
let provider =
create_oss_provider_with_base_url("http://localhost:11434/v1", WireApi::Responses);
let auth = resolve_provider_auth(/*auth*/ None, &provider).expect("auth should resolve");
assert!(auth.to_auth_headers().is_empty());
}
}
codex-rs/model-provider/src/bearer_auth_provider.rs
+8
View File
@@ -11,6 +11,14 @@ pub struct BearerAuthProvider {
}
impl BearerAuthProvider {
pub fn new(token: String) -> Self {
Self {
token: Some(token),
account_id: None,
is_fedramp_account: false,
}
}
pub fn for_test(token: Option<&str>, account_id: Option<&str>) -> Self {
Self {
token: token.map(str::to_string),
codex-rs/models-manager/src/manager.rs
+2 -2
View File
@@ -540,8 +540,8 @@ impl ModelsManager {
let uses_codex_backend = self
.provider
.auth_manager()
.and_then(|auth_manager| auth_manager.auth_cached())
.is_some_and(|auth| auth.uses_codex_backend());
.as_deref()
.is_some_and(AuthManager::current_auth_uses_codex_backend);
presets = ModelPreset::filter_by_auth(presets, uses_codex_backend);
ModelPreset::mark_default_by_picker_visibility(&mut presets);
codex-rs/protocol/src/account.rs
+20
View File
@@ -35,6 +35,12 @@ impl PlanType {
pub fn is_business_like(self) -> bool {
matches!(self, Self::Business | Self::EnterpriseCbpUsageBased)
}
pub fn is_workspace_account(self) -> bool {
self.is_team_like()
|| self.is_business_like()
|| matches!(self, Self::Enterprise | Self::Edu)
}
}
#[cfg(test)]
@@ -83,5 +89,19 @@ mod tests {
assert_eq!(PlanType::Business.is_business_like(), true);
assert_eq!(PlanType::EnterpriseCbpUsageBased.is_business_like(), true);
assert_eq!(PlanType::Team.is_business_like(), false);
assert_eq!(PlanType::Team.is_workspace_account(), true);
assert_eq!(
PlanType::SelfServeBusinessUsageBased.is_workspace_account(),
true
);
assert_eq!(PlanType::Business.is_workspace_account(), true);
assert_eq!(
PlanType::EnterpriseCbpUsageBased.is_workspace_account(),
true
);
assert_eq!(PlanType::Enterprise.is_workspace_account(), true);
assert_eq!(PlanType::Edu.is_workspace_account(), true);
assert_eq!(PlanType::Pro.is_workspace_account(), false);
}
}
codex-rs/protocol/src/auth.rs
+11 -9
View File
@@ -50,6 +50,14 @@ pub enum KnownPlan {
}
impl KnownPlan {
pub fn is_team_like(self) -> bool {
matches!(self, Self::Team | Self::SelfServeBusinessUsageBased)
}
pub fn is_business_like(self) -> bool {
matches!(self, Self::Business | Self::EnterpriseCbpUsageBased)
}
pub fn display_name(self) -> &'static str {
match self {
Self::Free => "Free",
@@ -83,15 +91,9 @@ impl KnownPlan {
}
pub fn is_workspace_account(self) -> bool {
matches!(
self,
Self::Team
| Self::SelfServeBusinessUsageBased
| Self::Business
| Self::EnterpriseCbpUsageBased
| Self::Enterprise
| Self::Edu
)
self.is_team_like()
|| self.is_business_like()
|| matches!(self, Self::Enterprise | Self::Edu)
}
}
codex-rs/protocol/src/error.rs
+32 -28
View File
@@ -472,40 +472,44 @@ impl std::fmt::Display for UsageLimitReachedError {
}
let message = match self.plan_type.as_ref() {
Some(PlanType::Known(KnownPlan::Plus)) => format!(
"You've hit your usage limit. Upgrade to Pro (https://chatgpt.com/explore/pro), visit https://chatgpt.com/codex/settings/usage to purchase more credits{}",
retry_suffix_after_or(self.resets_at.as_ref())
),
Some(PlanType::Known(
Some(PlanType::Known(plan)) => match plan {
KnownPlan::Plus => format!(
"You've hit your usage limit. Upgrade to Pro (https://chatgpt.com/explore/pro), visit https://chatgpt.com/codex/settings/usage to purchase more credits{}",
retry_suffix_after_or(self.resets_at.as_ref())
),
plan if plan.is_team_like() || plan.is_business_like() => {
format!(
"You've hit your usage limit. To get more access now, send a request to your admin{}",
retry_suffix_after_or(self.resets_at.as_ref())
)
}
KnownPlan::Free | KnownPlan::Go => {
format!(
"You've hit your usage limit. Upgrade to Plus to continue using Codex (https://chatgpt.com/explore/plus),{}",
retry_suffix_after_or(self.resets_at.as_ref())
)
}
KnownPlan::Pro | KnownPlan::ProLite => format!(
"You've hit your usage limit. Visit https://chatgpt.com/codex/settings/usage to purchase more credits{}",
retry_suffix_after_or(self.resets_at.as_ref())
),
KnownPlan::Enterprise | KnownPlan::Edu => format!(
"You've hit your usage limit.{}",
retry_suffix(self.resets_at.as_ref())
),
KnownPlan::Team
| KnownPlan::SelfServeBusinessUsageBased
| KnownPlan::Business
| KnownPlan::EnterpriseCbpUsageBased,
)) => {
| KnownPlan::EnterpriseCbpUsageBased => {
unreachable!("team-like and business-like plans are handled above")
}
},
Some(PlanType::Unknown(_)) | None => {
format!(
"You've hit your usage limit. To get more access now, send a request to your admin{}",
retry_suffix_after_or(self.resets_at.as_ref())
"You've hit your usage limit.{}",
retry_suffix(self.resets_at.as_ref())
)
}
Some(PlanType::Known(KnownPlan::Free)) | Some(PlanType::Known(KnownPlan::Go)) => {
format!(
"You've hit your usage limit. Upgrade to Plus to continue using Codex (https://chatgpt.com/explore/plus),{}",
retry_suffix_after_or(self.resets_at.as_ref())
)
}
Some(PlanType::Known(KnownPlan::Pro | KnownPlan::ProLite)) => format!(
"You've hit your usage limit. Visit https://chatgpt.com/codex/settings/usage to purchase more credits{}",
retry_suffix_after_or(self.resets_at.as_ref())
),
Some(PlanType::Known(KnownPlan::Enterprise))
| Some(PlanType::Known(KnownPlan::Edu)) => format!(
"You've hit your usage limit.{}",
retry_suffix(self.resets_at.as_ref())
),
Some(PlanType::Unknown(_)) | None => format!(
"You've hit your usage limit.{}",
retry_suffix(self.resets_at.as_ref())
),
};
write!(f, "{message}")