Use runtime authentication for OpenAI MCP and app requests (#59749)

## What changed

- Select fresh authentication for each HTTP operation on eligible OpenAI MCP connections while preserving the connection's credential source and account network policy. Stop redirects when requests carry runtime assertions.
- Pass the owning thread's authentication and runtime context through app discovery, connector metadata, and trusted-access requests. Include the runtime authentication mode in MCP connection and tool-cache identities.
- Give resumed actors analytics clients bound to their own authentication context, and serialize stored-credential reloads with token refresh while preserving host-provided credentials.

## Testing

Add coverage for fresh request assertions, revoked account policies, redirect handling, resumed-actor isolation, and MCP uploads in dual-header and assertion-only modes.

GitOrigin-RevId: f06900a6db61cd6cf53c3aa8075afeec4d5706d9
This commit is contained in:
cooper-oai
2026-11-05 22:03:29 +00:00
committed by copyberry
parent e934edd68c
commit 219ab4043e
51 changed files with 1901 additions and 250 deletions
@@ -484,7 +484,6 @@ impl MessageProcessor {
Arc::clone(&model_catalog),
);
let apps_processor = AppsRequestProcessor::new(
auth_manager.clone(),
Arc::clone(&thread_manager),
outgoing.clone(),
config_manager.clone(),
@@ -8,7 +8,6 @@ mod read;
pub(super) use read::APP_READ_MAX_IDS;
pub(crate) struct AppsRequestProcessor {
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
config_manager: ConfigManager,
@@ -18,7 +17,6 @@ pub(crate) struct AppsRequestProcessor {
impl AppsRequestProcessor {
pub(crate) fn new(
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
config_manager: ConfigManager,
@@ -26,7 +24,6 @@ impl AppsRequestProcessor {
) -> Self {
let shutdown_drop_guard = shutdown_token.clone().drop_guard();
Self {
auth_manager,
thread_manager,
outgoing,
config_manager,
@@ -64,13 +61,27 @@ impl AppsRequestProcessor {
};
let mut config = self.load_latest_config(fallback_cwd).await?;
if let Some(thread) = thread {
if let Some(thread) = thread.as_ref() {
let _ = config
.features
.set_enabled(Feature::Apps, thread.enabled(Feature::Apps));
}
let auth = self.auth_manager.auth().await;
let mcp_manager = thread
.as_ref()
.map(|thread| thread.mcp_manager())
.unwrap_or_else(|| self.thread_manager.mcp_manager());
self.thread_manager
.runtime_auth_for_mcp_manager(&config, &mcp_manager)
.map_err(|error| internal_error(error.to_string()))?;
let auth_manager = mcp_manager.auth_manager();
if params.force_refetch && !auth_manager.has_external_auth() {
auth_manager
.reload_with_cached_auth()
.await
.map_err(|error| internal_error(error.to_string()))?;
}
let auth = auth_manager.auth().await;
if !config
.features
.apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend))
@@ -86,8 +97,7 @@ impl AppsRequestProcessor {
let request = request_id.clone();
let outgoing = Arc::clone(&self.outgoing);
let environment_manager = self.thread_manager.environment_manager();
let mcp_manager = self.thread_manager.mcp_manager();
let plugins_manager = self.thread_manager.plugins_manager();
let plugins_manager = mcp_manager.plugins_manager();
let shutdown_token = self.shutdown_token.child_token();
tokio::spawn(async move {
tokio::select! {
@@ -195,10 +205,33 @@ impl AppsRequestProcessor {
loaded_plugins.capability_summaries(),
);
let plugin_apps = connector_snapshot.connector_ids().to_vec();
let (mut accessible_connectors, mut all_connectors) = tokio::join!(
connectors::list_cached_accessible_connectors_from_mcp_tools(&config),
connectors::list_cached_all_connectors(&config, &plugin_apps)
);
let mut config = config;
let auth = match mcp_manager
.auth_manager()
.auth_with_http_client_factory()
.await
{
Some((auth, factory)) => {
config.application_network_policy = factory.network_policy().clone();
Some(auth)
}
None => None,
};
let runtime = mcp_manager
.runtime_auth_for_config(&config)
.map(|(slot, mode)| {
slot.get()
.cloned()
.map(|runtime| (runtime, mode))
.ok_or_else(|| {
internal_error("runtime authentication is not initialized".to_string())
})
})
.transpose()?;
let mut accessible_connectors =
connectors::list_cached_accessible_connectors_from_mcp_tools(&config, auth.as_ref());
let mut all_connectors =
connectors::list_cached_all_connectors(&config, auth.as_ref(), &plugin_apps);
let cached_all_connectors = all_connectors.clone();
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
@@ -222,6 +255,8 @@ impl AppsRequestProcessor {
tokio::spawn(async move {
let result = connectors::list_all_connectors_with_options(
&all_config,
auth.as_ref(),
runtime.as_ref(),
force_refetch,
&all_plugin_apps,
)
@@ -354,19 +389,32 @@ impl AppsRequestProcessor {
.map_err(|err| internal_error(format!("failed to reload config: {err}")))
}
async fn load_apps_config(&self, thread_id: Option<&str>) -> Result<Config, JSONRPCErrorError> {
async fn load_apps_config(
&self,
thread_id: Option<&str>,
) -> Result<(Config, Arc<McpManager>), JSONRPCErrorError> {
let Some(thread_id) = thread_id else {
return self.load_latest_config(/*fallback_cwd*/ None).await;
let config = self.load_latest_config(/*fallback_cwd*/ None).await?;
self.thread_manager
.runtime_auth_for_config(&config)
.map_err(|error| internal_error(error.to_string()))?;
return Ok((config, self.thread_manager.mcp_manager()));
};
let (_, thread) = self.load_thread(thread_id).await?;
let thread_config = thread.config().await;
self.config_manager
let config = self
.config_manager
.load_latest_config_with_session_layers(
&thread_config.config_layer_stack,
&thread_config.cwd,
)
.await
.map_err(|err| internal_error(format!("failed to reload config: {err}")))
.map_err(|err| internal_error(format!("failed to reload config: {err}")))?;
let mcp_manager = thread.mcp_manager();
self.thread_manager
.runtime_auth_for_mcp_manager(&config, &mcp_manager)
.map_err(|error| internal_error(error.to_string()))?;
Ok((config, mcp_manager))
}
}
@@ -58,12 +58,14 @@ impl AppsRequestProcessor {
None,
),
};
let auth = self.auth_manager.auth().await;
let mcp_manager = thread.as_ref().map(|thread| thread.mcp_manager())
.unwrap_or_else(|| self.thread_manager.mcp_manager());
let auth_manager = mcp_manager.auth_manager();
let auth = auth_manager.auth().await;
let runtime_enabled = config
.features
.apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend));
let mcp_manager = self.thread_manager.mcp_manager();
let cache_key = connector_runtime_context_key(auth.as_ref());
let previous_snapshot = mcp_manager
.codex_apps_tools_cache()
@@ -96,9 +98,11 @@ impl AppsRequestProcessor {
let cancellation_token = CancellationToken::new();
let codex_apps_auth_manager =
host_owned_codex_apps_enabled(&mcp_config, auth.as_ref())
.then(|| Arc::clone(&self.auth_manager));
.then(|| Arc::clone(&auth_manager));
self.thread_manager.runtime_auth_for_mcp_manager(&config, &mcp_manager)?;
let runtime = McpRuntime::new(McpRuntimeInput {
network_policy_metadata: None,
runtime_auth: mcp_manager.runtime_auth_for_config(&config),
startup_policy: McpStartupPolicy::Eager,
config: Arc::clone(&mcp_config),
plugins_available: false,
@@ -26,14 +26,22 @@ impl AppsRequestProcessor {
.into_iter()
.filter(|app_id| seen_app_ids.insert(app_id.clone()))
.collect::<Vec<_>>();
let mut config = self.load_apps_config(thread_id.as_deref()).await?;
let auth = match self.auth_manager.auth_with_http_client_factory().await {
let (mut config, mcp_manager) = self.load_apps_config(thread_id.as_deref()).await?;
let auth = match mcp_manager
.auth_manager()
.auth_with_http_client_factory()
.await
{
Some((auth, factory)) => {
config.application_network_policy = factory.network_policy().clone();
Some(auth)
}
None => None,
};
let runtime = self
.thread_manager
.runtime_auth_for_mcp_manager(&config, &mcp_manager)
.map_err(|error| internal_error(error.to_string()))?;
if !config
.features
.apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend))
@@ -52,11 +60,16 @@ impl AppsRequestProcessor {
let connectors::ConnectorMetadataReadResult {
apps,
missing_app_ids,
} = connectors::read_connector_metadata(&config, auth, &app_ids, include_tools)
.await
.map_err(|err| internal_error(format!("failed to read app metadata: {err}")))?;
let loaded_plugins = self
.thread_manager
} = connectors::read_connector_metadata(
&config,
auth,
runtime.as_ref(),
&app_ids,
include_tools,
)
.await
.map_err(|err| internal_error(format!("failed to read app metadata: {err}")))?;
let loaded_plugins = mcp_manager
.plugins_manager()
.plugins_for_config(&config.plugins_config_input())
.await;
@@ -428,8 +428,15 @@ impl McpRequestProcessor {
}
None => self.load_global_config().await?,
};
let mcp_manager = self.thread_manager.mcp_manager();
let auth = self.auth_manager.auth().await;
let mcp_manager = thread
.as_ref()
.map(|thread| thread.mcp_manager())
.unwrap_or_else(|| self.thread_manager.mcp_manager());
self.thread_manager
.runtime_auth_for_mcp_manager(&config, &mcp_manager)
.map_err(|error| internal_error(error.to_string()))?;
let auth = mcp_manager.auth_manager().auth().await;
let runtime_auth = mcp_manager.runtime_auth_for_config(&config);
let (mcp_config, runtime_context) = match thread.as_ref() {
Some(thread) => thread.runtime_mcp_config_and_context(&config).await,
None => (
@@ -444,6 +451,7 @@ impl McpRequestProcessor {
&mcp_config,
auth.as_ref(),
Some(&self.auth_manager),
runtime_auth,
runtime_context,
mcp_manager.codex_apps_tools_cache(),
mcp_manager.tool_catalog_cache(),
@@ -700,7 +708,11 @@ impl McpRequestProcessor {
let mcp_config = mcp_manager.runtime_config(&config).await;
let codex_apps_tools_cache = mcp_manager.codex_apps_tools_cache();
let tool_catalog_cache = mcp_manager.tool_catalog_cache();
let auth = self.auth_manager.auth().await;
self.thread_manager
.runtime_auth_for_mcp_manager(&config, &mcp_manager)
.map_err(|error| internal_error(error.to_string()))?;
let runtime_auth = mcp_manager.runtime_auth_for_config(&config);
let auth = mcp_manager.auth_manager().auth().await;
let environment_manager = self.thread_manager.environment_manager();
// This threadless resource-read path has no turn cwd or turn-selected
// environment. Use config cwd only as the local stdio fallback; named
@@ -713,6 +725,7 @@ impl McpRequestProcessor {
let result = read_mcp_resource_without_thread(
&mcp_config,
auth.as_ref(),
runtime_auth,
runtime_context,
codex_apps_tools_cache,
tool_catalog_cache,
@@ -1113,6 +1113,7 @@ impl PluginRequestProcessor {
None => None,
};
let app_summaries = load_plugin_app_summaries(
&self.thread_manager,
&config,
auth.as_ref(),
&outcome.plugin.apps,
@@ -1219,6 +1220,7 @@ impl PluginRequestProcessor {
.map(plugin_app_category_by_id_from_value)
.unwrap_or_default();
let app_summaries = load_plugin_app_summaries(
&self.thread_manager,
&config,
auth.as_ref(),
&plugin_apps,
@@ -1738,37 +1740,42 @@ impl PluginRequestProcessor {
}
let is_chatgpt_auth = auth.as_ref().is_some_and(CodexAuth::is_chatgpt_auth);
let apps_needing_auth = if let Some(app_ids_needing_auth) =
installation.app_ids_needing_auth
{
if app_ids_needing_auth.is_empty()
|| !config.features.apps_enabled_for_auth(is_chatgpt_auth)
{
Vec::new()
} else {
let plugin_apps = app_ids_needing_auth
.into_iter()
.map(codex_plugin::AppConnectorId)
.collect::<Vec<_>>();
let app_category_by_id = remote_detail
.app_manifest
.as_ref()
.map(plugin_app_category_by_id_from_value)
.unwrap_or_default();
load_plugin_app_summaries(&config, auth.as_ref(), &plugin_apps, &app_category_by_id)
let apps_needing_auth =
if let Some(app_ids_needing_auth) = installation.app_ids_needing_auth {
if app_ids_needing_auth.is_empty()
|| !config.features.apps_enabled_for_auth(is_chatgpt_auth)
{
Vec::new()
} else {
let plugin_apps = app_ids_needing_auth
.into_iter()
.map(codex_plugin::AppConnectorId)
.collect::<Vec<_>>();
let app_category_by_id = remote_detail
.app_manifest
.as_ref()
.map(plugin_app_category_by_id_from_value)
.unwrap_or_default();
load_plugin_app_summaries(
&self.thread_manager,
&config,
auth.as_ref(),
&plugin_apps,
&app_category_by_id,
)
.await
}
} else {
let plugin_app_declarations =
load_plugin_apps(result.installed_path.as_path(), &result.plugin_id).await;
self.plugin_apps_needing_auth_for_install(
&config,
auth.as_ref(),
&result.plugin_id.as_key(),
&plugin_app_declarations,
)
.await
};
}
} else {
let plugin_app_declarations =
load_plugin_apps(result.installed_path.as_path(), &result.plugin_id).await;
self.plugin_apps_needing_auth_for_install(
&config,
auth.as_ref(),
&result.plugin_id.as_key(),
&plugin_app_declarations,
)
.await
};
Ok(PluginInstallResponse {
auth_policy: remote_detail.summary.auth_policy,
@@ -1839,7 +1846,13 @@ impl PluginRequestProcessor {
.collect();
let environment_manager = self.thread_manager.environment_manager();
let (app_summaries, accessible_connectors_result) = tokio::join!(
load_plugin_app_summaries(config, auth, &plugin_apps, &app_category_by_id),
load_plugin_app_summaries(
&self.thread_manager,
config,
auth,
&plugin_apps,
&app_category_by_id
),
connectors::list_accessible_connectors_from_mcp_tools_with_mcp_manager(
config,
/*force_refetch*/ true,
@@ -1856,8 +1869,7 @@ impl PluginRequestProcessor {
"failed to load accessible apps after plugin install: {err:#}"
);
(
connectors::list_cached_accessible_connectors_from_mcp_tools(config)
.await
connectors::list_cached_accessible_connectors_from_mcp_tools(config, auth)
.unwrap_or_default(),
false,
)
@@ -2159,6 +2171,7 @@ impl PluginRequestProcessor {
}
async fn load_plugin_app_summaries(
thread_manager: &ThreadManager,
config: &Config,
auth: Option<&CodexAuth>,
plugin_apps: &[codex_plugin::AppConnectorId],
@@ -2185,9 +2198,17 @@ async fn load_plugin_app_summaries(
)
.fresh_records(&app_ids, /*include_tools*/ false),
);
let runtime = thread_manager.runtime_auth_for_config(config);
for app_ids in app_ids.chunks(APP_READ_MAX_IDS) {
let runtime = match &runtime {
Ok(runtime) => runtime.as_ref(),
Err(error) => {
warn!("failed to initialize app metadata authentication: {error}");
break;
}
};
match connectors::read_connector_metadata(
config, auth, app_ids, /*include_tools*/ false,
config, auth, runtime, app_ids, /*include_tools*/ false,
)
.await
{
@@ -27,6 +27,8 @@ use codex_app_server_protocol::AppReview;
use codex_app_server_protocol::AppScreenshot;
use codex_app_server_protocol::AppsListParams;
use codex_app_server_protocol::AppsListResponse;
use codex_app_server_protocol::GetAccountParams;
use codex_app_server_protocol::GetAccountResponse;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::LoginAccountResponse;
use codex_app_server_protocol::RequestId;
@@ -1122,6 +1124,40 @@ async fn list_apps_force_refetch_preserves_previous_cache_on_failure() -> Result
assert_eq!(initial_data.len(), 1);
assert!(initial_data.iter().all(|app| app.is_accessible));
let account_request = mcp
.send_get_account_request(GetAccountParams {
refresh_token: false,
})
.await?;
let initial_account: GetAccountResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(account_request)).await??;
assert!(initial_account.account.is_some());
let auth_path = codex_home.path().join("auth.json");
std::fs::write(&auth_path, "")?;
let refetch_request = mcp
.send_apps_list_request(AppsListParams {
limit: None,
cursor: None,
thread_id: None,
force_refetch: true,
})
.await?;
let _: JSONRPCError = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(refetch_request)),
)
.await??;
let account_request = mcp
.send_get_account_request(GetAccountParams {
refresh_token: false,
})
.await?;
let account: GetAccountResponse =
timeout(DEFAULT_TIMEOUT, mcp.read_response(account_request)).await??;
assert_eq!(account, initial_account);
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("chatgpt-token-invalid")
@@ -1586,6 +1622,18 @@ async fn start_apps_server_with_delays_and_control_inner(
);
let router = Router::new()
.route(
"/api/codex/accounts/check",
get(|State(state): State<Arc<AppsServerState>>| async move {
Json(json!({
"accounts": [{
"id": state.expected_account_id,
"workspace_backend_origin": "https://chatgpt.com",
"account_routing_override": "NO_CONSTRAINT",
}],
}))
}),
)
.route("/connectors/directory/list", get(list_directory_connectors))
.route(
"/connectors/directory/list_workspace",
+16 -3
View File
@@ -115,14 +115,19 @@ async fn connector_requests_preserve_account_policy() -> Result<()> {
let (auth, factory) = manager.auth_with_http_client_factory().await.unwrap();
config.application_network_policy = factory.network_policy().clone();
let app_ids = vec!["alpha".to_string()];
connectors::read_connector_metadata(&config, &auth, &app_ids, /*include_tools*/ false).await?;
connectors::read_connector_metadata(
&config, &auth, /*runtime*/ None, &app_ids, /*include_tools*/ false,
)
.await?;
assert_eq!(state.requests().len(), 1);
policy.invalidate();
controller.publish(policy.revision(), DestinationPolicy::Unrestricted);
let error = timeout(
DEFAULT_TIMEOUT,
connectors::read_connector_metadata(&config, &auth, &app_ids, /*include_tools*/ false),
connectors::read_connector_metadata(
&config, &auth, /*runtime*/ None, &app_ids, /*include_tools*/ false,
),
)
.await?
.err()
@@ -139,9 +144,17 @@ async fn connector_requests_preserve_account_policy() -> Result<()> {
allowed_hosts: Default::default(),
},
);
let (auth, factory) = manager.auth_with_http_client_factory().await.unwrap();
config.application_network_policy = factory.network_policy().clone();
let error = timeout(
DEFAULT_TIMEOUT,
connectors::list_all_connectors_with_options(&config, /*force_refetch*/ true, &[]),
connectors::list_all_connectors_with_options(
&config,
Some(&auth),
/*runtime*/ None,
/*force_refetch*/ true,
&[],
),
)
.await?
.unwrap_err();