diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 9eba750ee0..edd0687601 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 63d861ae44..d93bfefb7d 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/src/export.rs b/codex-rs/app-server-protocol/src/export.rs index 4123b3f189..db5ff718b6 100644 --- a/codex-rs/app-server-protocol/src/export.rs +++ b/codex-rs/app-server-protocol/src/export.rs @@ -39,6 +39,9 @@ use std::path::PathBuf; use std::process::Command; use std::thread; +#[path = "export_user_verification.rs"] +mod user_verification; + pub(crate) const GENERATED_TS_HEADER: &str = "// GENERATED CODE! DO NOT MODIFY BY HAND!\n\n"; const IGNORED_DEFINITIONS: &[&str] = &["Option<()>"]; const JSON_V1_ALLOWLIST: &[&str] = &["InitializeParams", "InitializeResponse"]; @@ -271,6 +274,11 @@ fn filter_experimental_ts(out_dir: &Path) -> Result<()> { filter_request_ts(out_dir, "ServerRequest.ts", EXPERIMENTAL_SERVER_METHODS)?; filter_experimental_type_fields_ts(out_dir, ®istered_fields)?; remove_generated_type_files(out_dir, &experimental_method_types, "ts")?; + let elicitation_path = out_dir.join("v2/McpServerElicitationRequestParams.ts"); + if elicitation_path.exists() { + let content = fs::read_to_string(&elicitation_path)?; + fs::write(elicitation_path, user_verification::filter_ts(&content))?; + } Ok(()) } @@ -295,6 +303,11 @@ pub(crate) fn filter_experimental_ts_tree(tree: &mut BTreeMap) } for (path, content) in tree.iter_mut() { + if path.file_stem().and_then(|stem| stem.to_str()) + == Some("McpServerElicitationRequestParams") + { + *content = user_verification::filter_ts(content); + } let Some(type_name) = path.file_stem().and_then(|stem| stem.to_str()) else { continue; }; @@ -339,7 +352,7 @@ fn filter_request_ts_contents(mut content: String, experimental_methods: &[&str] let filtered_arms: Vec = arms .into_iter() .filter(|arm| { - extract_method_from_arm(arm) + extract_discriminator_from_arm(arm, "method") .is_none_or(|method| !experimental_methods.contains(method.as_str())) }) .collect(); @@ -425,6 +438,7 @@ fn filter_experimental_schema(bundle: &mut Value) -> Result<()> { prune_experimental_methods(bundle, EXPERIMENTAL_CLIENT_METHODS); prune_experimental_methods(bundle, EXPERIMENTAL_SERVER_METHODS); remove_experimental_method_type_definitions(bundle); + user_verification::filter_json(bundle); Ok(()) } @@ -804,14 +818,14 @@ fn split_top_level_multi(input: &str, delimiters: &[char]) -> Vec { parts } -fn extract_method_from_arm(arm: &str) -> Option { +fn extract_discriminator_from_arm(arm: &str, discriminator: &str) -> Option { let (open, close) = find_top_level_brace_span(arm)?; let inner = &arm[open + 1..close]; for field in split_top_level(inner, ',') { let Some((name, value)) = parse_property(field.as_str()) else { continue; }; - if name != "method" { + if name != discriminator { continue; } let value = value.trim_start(); diff --git a/codex-rs/app-server-protocol/src/export_user_verification.rs b/codex-rs/app-server-protocol/src/export_user_verification.rs new file mode 100644 index 0000000000..b0102ecafc --- /dev/null +++ b/codex-rs/app-server-protocol/src/export_user_verification.rs @@ -0,0 +1,57 @@ +//! Removes the opt-in verification mode from the flattened stable elicitation union. +//! +//! `ExperimentalApi` enum annotations currently drive runtime checks only. Keep this new mode +//! out of stable exports without changing the export behavior of existing gated enum variants. + +use super::extract_discriminator_from_arm; +use super::split_top_level; +use super::split_type_alias; +use serde_json::Value; + +const MODE: &str = "openai/userVerification"; + +pub(super) fn filter_ts(content: &str) -> String { + let Some((prefix, body, suffix)) = split_type_alias(content) else { + return content.to_string(); + }; + let parts = split_top_level(&body, '&') + .into_iter() + .map(|part| { + let Some(union) = part + .strip_prefix('(') + .and_then(|part| part.strip_suffix(')')) + else { + return part; + }; + let arms = split_top_level(union, '|') + .into_iter() + .filter(|arm| extract_discriminator_from_arm(arm, "mode").as_deref() != Some(MODE)) + .collect::>(); + format!("({})", arms.join(" | ")) + }) + .collect::>(); + format!("{prefix} {}{suffix}", parts.join(" & ")) +} + +pub(super) fn filter_json(value: &mut Value) { + match value { + Value::Array(items) => { + items.retain(|item| { + item.pointer("/properties/mode/enum") != Some(&serde_json::json!([MODE])) + }); + for item in items { + filter_json(item); + } + } + Value::Object(map) => { + for entry in map.values_mut() { + filter_json(entry); + } + } + Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => {} + } +} + +#[cfg(test)] +#[path = "export_user_verification_tests.rs"] +mod tests; diff --git a/codex-rs/app-server-protocol/src/export_user_verification_tests.rs b/codex-rs/app-server-protocol/src/export_user_verification_tests.rs new file mode 100644 index 0000000000..4a5d2ef3f6 --- /dev/null +++ b/codex-rs/app-server-protocol/src/export_user_verification_tests.rs @@ -0,0 +1,39 @@ +//! Checks that filtering the new mode preserves the stable elicitation contracts. + +use super::*; +use crate::McpServerElicitationRequestParams; +use crate::TS; +use anyhow::Result; +use pretty_assertions::assert_eq; + +#[test] +fn verification_mode_is_removed_from_flattened_typescript_only_for_stable_exports() -> Result<()> { + let generated = McpServerElicitationRequestParams::export_to_string()?; + assert!(generated.contains(MODE)); + let filtered = filter_ts(&generated); + assert!(!filtered.contains(MODE)); + assert_eq!(filter_ts(&filtered), filtered); + for mode in ["form", "openai/form", "openaiForm", "url"] { + assert!(filtered.contains(&format!("\"mode\": \"{mode}\""))); + } + assert!(filtered.contains("threadId: string")); + assert!(filtered.contains("requestedSchema: McpElicitationSchema")); + Ok(()) +} + +#[test] +fn verification_mode_is_removed_from_inline_json_without_changing_other_modes() -> Result<()> { + let generated = serde_json::to_value(schemars::schema_for!(McpServerElicitationRequestParams))?; + assert!(generated.to_string().contains(MODE)); + let mut expected = generated.clone(); + let modes = expected["oneOf"].as_array_mut().expect("flattened modes"); + let verification = modes.remove(0); + assert_eq!( + verification["properties"]["mode"]["enum"], + serde_json::json!([MODE]) + ); + let mut filtered = generated; + filter_json(&mut filtered); + assert_eq!(filtered, expected); + Ok(()) +} diff --git a/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs b/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs index c1ec63c55f..25a61167b3 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs @@ -1,6 +1,7 @@ use super::shared::v2_enum_from_core; use crate::JsonSchema; use crate::TS; +use codex_experimental_api_macros::ExperimentalApi; use codex_protocol::approvals::ElicitationRequest as CoreElicitationRequest; use codex_protocol::items::McpToolCallError as CoreMcpToolCallError; use codex_protocol::mcp::CallToolResult as CoreMcpCallToolResult; @@ -379,7 +380,7 @@ impl From for McpServerElicitationAction { } } -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct McpServerElicitationRequestParams { @@ -393,6 +394,7 @@ pub struct McpServerElicitationRequestParams { pub turn_id: Option, pub server_name: String, #[serde(flatten)] + #[experimental(nested)] pub request: McpServerElicitationRequest, // TODO: When core can correlate an elicitation with an MCP tool call, expose the associated // McpToolCall item id here as an optional field. The current core event does not carry that @@ -744,11 +746,20 @@ pub struct McpElicitationConstOption { pub title: String, } -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] #[serde(tag = "mode", rename_all = "camelCase")] #[ts(tag = "mode")] #[ts(export_to = "v2/")] pub enum McpServerElicitationRequest { + /// A device-authenticated approval; accepted responses contain the proof in `content`. + #[experimental("mcpServer/elicitation/request.userVerification")] + #[serde(rename = "openai/userVerification", rename_all = "camelCase")] + #[ts(rename = "openai/userVerification", rename_all = "camelCase")] + UserVerification { + title: String, + description: String, + challenge: String, + }, #[serde(rename_all = "camelCase")] #[ts(rename_all = "camelCase")] Form { @@ -794,6 +805,15 @@ impl TryFrom for McpServerElicitationRequest { fn try_from(value: CoreElicitationRequest) -> Result { match value { + CoreElicitationRequest::UserVerification { + title, + description, + challenge, + } => Ok(Self::UserVerification { + title, + description, + challenge, + }), CoreElicitationRequest::Form { meta, message, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 90f4317de7..fcfdc15d5a 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -2061,12 +2061,13 @@ Order of messages: - an OpenAI form request: `{ "mode": "openaiForm", "message": "...", "requestedSchema": { ... } }` - a legacy OpenAI extended form request: `{ "mode": "openai/form", "message": "...", "requestedSchema": { ... } }` - a URL request: `{ "mode": "url", "message": "...", "url": "...", "elicitationId": "..." }` + - a user-verification request: `{ "mode": "openai/userVerification", "title": "...", "description": "...", "challenge": "AQID" }` 2. Client response — `{ "action": "accept", "content": ... }`, `{ "action": "decline", "content": null }`, or `{ "action": "cancel", "content": null }`. 3. `serverRequest/resolved` — `{ threadId, requestId }` confirms the pending request has been resolved or cleared, including lifecycle cleanup on turn start/complete/interrupt. `turnId` is best-effort. When the elicitation is correlated with an active turn, the request includes that turn id; otherwise it is `null`. -MCP `openai/elicitation/create` requests must explicitly specify `mode: "form"`. +MCP form requests to `openai/elicitation/create` must explicitly specify `mode: "form"`. App-server forwards them as `mode: "openaiForm"`, preserving `requestedSchema` as opaque JSON, including `x-openai-*` annotations. The legacy `openai/form` route remains independent and also preserves its schema. @@ -2075,9 +2076,27 @@ The client owns validation and rendering. Graphical clients show an unsupported state for unknown semantic inputs, never a partial form or generic approval, and wait for the user to decline or cancel instead of returning a JSON-RPC error. The TUI automatically declines OpenAI forms, including requests replayed -from another client. Capability advertisement describes the session's initial +from another client. Form capability advertisement describes the session's initial client, not all clients that may later attach. +User verification uses the same MCP method with `mode: "openai/userVerification"`. +App-server forwards the title, description, and challenge to one connected app +enabled by trusted host activation. +The app returns `{ "action": "accept", "content": { "credentialId": "...", "signature": "..." } }`. +TUI/Desktop connected to local app-server can obtain that proof through `userVerification/verify`; a mobile app +can sign on its own device and answer directly. Neither path returns the proof in +metadata or requires a previous `verify` call to resolve the elicitation. + +Verification bypasses automated approval and review. Only the owning connection +can respond; disconnect, turn interruption, or a cached authentication change +cancels the request. Authentication revisions are checked even when an account +switches away and back before its change notification is processed. +Malformed acceptance without a proof becomes cancellation. The MCP boundary +validates bounded, unpadded base64url credential IDs and signatures before forwarding. +The Codex MCP layer will advertise support when the complete device/UI integration +is enabled; this transport groundwork does not enable advertisement or add a UI +capability field. + For MCP tool approval elicitations, form request `meta` includes `codex_approval_kind: "mcp_tool_call"` and may include `persist: "session"`, `persist: "always"`, or `persist: ["session", "always"]` to advertise whether diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 92c595b7e1..7c86ecb84e 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -869,6 +869,10 @@ pub(crate) async fn apply_bespoke_event_handling( }); } EventMsg::ElicitationRequest(request) => { + let user_verification = matches!( + &request.request, + codex_protocol::approvals::ElicitationRequest::UserVerification { .. } + ); let permission_guard = thread_watch_manager .note_permission_requested(&conversation_id.to_string()) .await; @@ -918,7 +922,10 @@ pub(crate) async fn apply_bespoke_event_handling( request.server_name, request.id, pending_request_id, - rx, + PendingMcpElicitationResponse { + receiver: rx, + user_verification, + }, conversation, thread_state, permission_guard, @@ -1811,19 +1818,28 @@ async fn on_request_user_input_response( } } +struct PendingMcpElicitationResponse { + receiver: oneshot::Receiver, + user_verification: bool, +} + async fn on_mcp_server_elicitation_response( server_name: String, request_id: codex_protocol::mcp::RequestId, pending_request_id: RequestId, - receiver: oneshot::Receiver, + pending: PendingMcpElicitationResponse, conversation: Arc, thread_state: Arc>, permission_guard: ThreadWatchActiveGuard, ) { - let response = receiver.await; + let response = pending.receiver.await; resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await; drop(permission_guard); - let response = mcp_server_elicitation_response_from_client_result(response); + let response = if pending.user_verification { + crate::user_verification_response::from_client_result(response) + } else { + mcp_server_elicitation_response_from_client_result(response) + }; if let Err(err) = conversation .submit(Op::ResolveElicitation { diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index d8613e90c5..bc012ce7c9 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -630,12 +630,12 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult { outgoing_message_sender - .notify_client_response(request_id, result) + .notify_client_response(IN_PROCESS_CONNECTION_ID, request_id, result) .await; } Some(InProcessClientMessage::ServerRequestError { request_id, error }) => { outgoing_message_sender - .notify_client_error(request_id, error) + .notify_client_error(IN_PROCESS_CONNECTION_ID, request_id, error) .await; } Some(InProcessClientMessage::Shutdown { done_tx }) => { @@ -704,7 +704,7 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult unreachable!("we just sent a ServerRequest variant"), }; outgoing_message_sender - .notify_client_error(request_id, error) + .notify_client_error(IN_PROCESS_CONNECTION_ID, request_id, error) .await; } } diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 82c51803c4..c4ec80c9bf 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -134,6 +134,7 @@ mod thread_status; mod transport; mod turn_cost_worker; mod user_verification; +mod user_verification_response; pub use crate::code_mode_host::AppServerCodeModeHostArgs; pub use crate::code_mode_host::CodeModeHostTransport; @@ -1135,7 +1136,7 @@ pub async fn run_main_with_transport_options( warn!("dropping response from unknown connection: {connection_id:?}"); continue; } - processor.process_response(response).await; + processor.process_response(connection_id, response).await; } JSONRPCMessage::Notification(notification) => { if !connections.contains_key(&connection_id) { @@ -1149,7 +1150,7 @@ pub async fn run_main_with_transport_options( warn!("dropping error from unknown connection: {connection_id:?}"); continue; } - processor.process_error(err).await; + processor.process_error(connection_id, err).await; } } } diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 3be4c9770e..da84311484 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -288,6 +288,7 @@ impl MessageProcessor { plugin_startup_tasks, } = args; let thread_state_manager = ThreadStateManager::new(); + outgoing.watch_user_verification_auth(Arc::clone(&auth_manager)); // The thread store is intentionally process-scoped. Config reloads can // affect per-thread behavior, but they must not move newly started, // resumed, or forked threads to a different persistence backend/root. @@ -804,6 +805,9 @@ impl MessageProcessor { session_state: &ConnectionSessionState, ) { session_state.rpc_gate.close().await; + self.outgoing + .disconnect_user_verification_connection(connection_id) + .await; session_state.mcp_event_streams.clear().await; if timeout( CONNECTION_RPC_DRAIN_TIMEOUT, @@ -835,14 +839,22 @@ impl MessageProcessor { } /// Handle a standalone JSON-RPC response originating from the peer. - pub(crate) async fn process_response(&self, response: JSONRPCResponse) { + pub(crate) async fn process_response( + &self, + connection_id: ConnectionId, + response: JSONRPCResponse, + ) { let JSONRPCResponse { id, result, .. } = response; - self.outgoing.notify_client_response(id, result).await + self.outgoing + .notify_client_response(connection_id, id, result) + .await } /// Handle an error object received from the peer. - pub(crate) async fn process_error(&self, err: JSONRPCError) { - self.outgoing.notify_client_error(err.id, err.error).await; + pub(crate) async fn process_error(&self, connection_id: ConnectionId, err: JSONRPCError) { + self.outgoing + .notify_client_error(connection_id, err.id, err.error) + .await; } async fn handle_client_request( diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 6904cfb389..e6e0ae87a4 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -1,5 +1,7 @@ use std::collections::HashMap; +use std::collections::HashSet; use std::sync::Arc; +use std::sync::OnceLock; use std::sync::atomic::AtomicI64; use std::sync::atomic::Ordering; use std::time::SystemTime; @@ -44,6 +46,9 @@ pub(crate) type ClientRequestResult = std::result::Result>, + verification_connections: Mutex>, next_server_request_id: AtomicI64, sender: mpsc::Sender, request_id_to_callback: Mutex>, @@ -120,6 +127,9 @@ pub(crate) struct ThreadScopedOutgoingMessageSender { } struct PendingCallbackEntry { + verification_owner: Option, + verification_auth_revision: Option, + verification_identity: Option, callback: oneshot::Sender, thread_id: Option, request: ServerRequest, @@ -221,6 +231,8 @@ impl OutgoingMessageSender { analytics_events_client: AnalyticsEventsClient, ) -> Self { Self { + verification_auth: OnceLock::new(), + verification_connections: Mutex::new(HashSet::new()), next_server_request_id: AtomicI64::new(0), sender, request_id_to_callback: Mutex::new(HashMap::new()), @@ -240,6 +252,8 @@ impl OutgoingMessageSender { } pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) { + self.disconnect_user_verification_connection(connection_id) + .await; let mut request_contexts = self.request_contexts.lock().await; request_contexts.retain(|request_id, _| request_id.connection_id != connection_id); } @@ -301,13 +315,50 @@ impl OutgoingMessageSender { let id = self.next_request_id(); let outgoing_message_id = id.clone(); let request = request.request_with_id(outgoing_message_id.clone()); - + let user_verification = matches!( + &request, + ServerRequest::McpServerElicitationRequest { params, .. } + if matches!(¶ms.request, codex_app_server_protocol::McpServerElicitationRequest::UserVerification { .. }) + ); + // Snapshot before waiting on eligibility or callback locks. A request cannot inherit + // whichever account happens to be current after an unrelated operation releases a lock. + let verification_auth_revision = user_verification + .then(|| self.verification_auth_revision()) + .flatten(); + let verification_identity = user_verification + .then(|| self.verification_identity()) + .flatten(); + let auth_changed = || { + user_verification + && (verification_auth_revision != self.verification_auth_revision() + || verification_identity != self.verification_identity()) + }; let (tx_approve, rx_approve) = oneshot::channel(); + // One app owns this ceremony. Reconnect and other subscribers cannot answer it. + let verification_owner = if user_verification { + let eligible = self.verification_connections.lock().await; + connection_ids + .and_then(|ids| ids.iter().find(|id| eligible.contains(*id))) + .copied() + } else { + None + }; + if user_verification && (verification_owner.is_none() || auth_changed()) { + return (outgoing_message_id, rx_approve); + } + let connection_ids = if user_verification { + verification_owner.as_ref().map(std::slice::from_ref) + } else { + connection_ids + }; { let mut request_id_to_callback = self.request_id_to_callback.lock().await; request_id_to_callback.insert( id, PendingCallbackEntry { + verification_owner, + verification_auth_revision, + verification_identity: verification_identity.clone(), callback: tx_approve, thread_id, request: request.clone(), @@ -315,6 +366,18 @@ impl OutgoingMessageSender { }, ); } + // Disconnect may finish its callback cleanup before registration acquires the lock. + // Recheck afterward so that ordering cannot leave an orphaned verification callback. + if let Some(owner) = verification_owner { + let eligible = self.verification_connections.lock().await.contains(&owner); + if !eligible || auth_changed() { + self.request_id_to_callback + .lock() + .await + .remove(&outgoing_message_id); + return (outgoing_message_id, rx_approve); + } + } let outgoing_message = OutgoingMessage::Request(request.clone()); let send_result = match connection_ids { @@ -380,13 +443,20 @@ impl OutgoingMessageSender { } } - pub(crate) async fn notify_client_response(&self, id: RequestId, result: Result) { - let entry = self.take_request_callback(&id).await; + pub(crate) async fn notify_client_response( + &self, + connection_id: ConnectionId, + id: RequestId, + result: Result, + ) { + let entry = self.take_connection_callback(connection_id, &id).await; match entry { Some((id, entry)) => { let completed_at_ms = now_unix_timestamp_ms(); - if let Ok(response) = entry.request.response_from_result(result.clone()) { + if entry.verification_owner.is_none() + && let Ok(response) = entry.request.response_from_result(result.clone()) + { tracing::info!("<- response: {response:?}"); if !matches!(response, ServerResponse::PermissionsRequestApproval { .. }) { self.analytics_events_client @@ -403,8 +473,13 @@ impl OutgoingMessageSender { } } - pub(crate) async fn notify_client_error(&self, id: RequestId, error: JSONRPCErrorError) { - let entry = self.take_request_callback(&id).await; + pub(crate) async fn notify_client_error( + &self, + connection_id: ConnectionId, + id: RequestId, + error: JSONRPCErrorError, + ) { + let entry = self.take_connection_callback(connection_id, &id).await; match entry { Some((id, entry)) => { @@ -462,6 +537,27 @@ impl OutgoingMessageSender { request_id_to_callback.remove_entry(id) } + async fn take_connection_callback( + &self, + connection_id: ConnectionId, + id: &RequestId, + ) -> Option<(RequestId, PendingCallbackEntry)> { + let mut callbacks = self.request_id_to_callback.lock().await; + let entry = callbacks.get(id)?; + if let Some(owner) = entry.verification_owner { + if owner != connection_id { + return None; + } + if entry.verification_identity != self.verification_identity() + || entry.verification_auth_revision != self.verification_auth_revision() + { + callbacks.remove(id); + return None; + } + } + callbacks.remove_entry(id) + } + pub(crate) async fn pending_requests_for_thread( &self, thread_id: ThreadId, @@ -470,7 +566,8 @@ impl OutgoingMessageSender { let mut requests = request_id_to_callback .values() .filter_map(|entry| { - (entry.thread_id == Some(thread_id)).then_some(entry.request.clone()) + (entry.thread_id == Some(thread_id) && entry.verification_owner.is_none()) + .then_some(entry.request.clone()) }) .collect::>(); requests.sort_by(|left, right| left.id().cmp(right.id())); @@ -738,6 +835,10 @@ fn timestamped_server_notification(notification: ServerNotification) -> Outgoing }) } +#[cfg(test)] +#[path = "user_verification_ownership_tests.rs"] +mod user_verification_ownership_tests; + #[cfg(test)] mod tests { use std::time::Duration; @@ -1316,7 +1417,7 @@ mod tests { let error = internal_error("refresh failed"); outgoing - .notify_client_error(request_id, error.clone()) + .notify_client_error(ConnectionId(1), request_id, error.clone()) .await; let result = timeout(Duration::from_secs(1), wait_for_result) diff --git a/codex-rs/app-server/src/user_verification_auth.rs b/codex-rs/app-server/src/user_verification_auth.rs new file mode 100644 index 0000000000..9a27d26a3d --- /dev/null +++ b/codex-rs/app-server/src/user_verification_auth.rs @@ -0,0 +1,73 @@ +//! Cancels device ceremonies when cached authentication changes. + +use codex_login::AuthManager; + +use super::*; + +#[derive(Clone, PartialEq, Eq)] +pub(super) struct Identity { + user_id: String, + account_id: String, +} + +impl OutgoingMessageSender { + /// Trusted host activation enables a UI integration after its signing path is available. + #[allow(dead_code)] // Called by the subsequent host activation change. + pub(crate) async fn enable_user_verification_connection(&self, connection_id: ConnectionId) { + self.verification_connections + .lock() + .await + .insert(connection_id); + } + + /// Stop routing ceremonies immediately, without waiting for unrelated RPCs to drain. + pub(crate) async fn disconnect_user_verification_connection( + &self, + connection_id: ConnectionId, + ) { + self.verification_connections + .lock() + .await + .remove(&connection_id); + self.request_id_to_callback + .lock() + .await + .retain(|_, entry| entry.verification_owner != Some(connection_id)); + } + + pub(crate) fn watch_user_verification_auth(self: &Arc, auth_manager: Arc) { + let mut changes = auth_manager.auth_change_receiver(); + if self.verification_auth.set(auth_manager).is_err() { + return; + } + let outgoing = Arc::downgrade(self); + tokio::spawn(async move { + while changes.changed().await.is_ok() { + let Some(outgoing) = outgoing.upgrade() else { + break; + }; + // Compare revisions, since watch can coalesce an account switch and switch back. + // Requests created after the change already belong to the new auth revision. + let mut callbacks = outgoing.request_id_to_callback.lock().await; + let revision = outgoing.verification_auth_revision(); + callbacks.retain(|_, entry| { + entry.verification_owner.is_none() + || entry.verification_auth_revision == revision + }); + } + }); + } + + pub(super) fn verification_identity(&self) -> Option { + let auth = self.verification_auth.get()?.auth_cached()?; + Some(Identity { + user_id: auth.get_chatgpt_user_id()?, + account_id: auth.get_account_id()?, + }) + } + + pub(super) fn verification_auth_revision(&self) -> Option { + let changes = self.verification_auth.get()?.auth_change_receiver(); + Some(*changes.borrow()) + } +} diff --git a/codex-rs/app-server/src/user_verification_ownership_tests.rs b/codex-rs/app-server/src/user_verification_ownership_tests.rs new file mode 100644 index 0000000000..a2a0e04d5f --- /dev/null +++ b/codex-rs/app-server/src/user_verification_ownership_tests.rs @@ -0,0 +1,306 @@ +use super::*; +use app_test_support::ChatGptAuthFixture; +use app_test_support::write_chatgpt_auth; +use codex_app_server_protocol::McpServerElicitationRequest; +use codex_app_server_protocol::McpServerElicitationRequestParams; +use codex_config::types::AuthCredentialsStoreMode; +use codex_login::AuthManager; +use codex_login::CodexAuth; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::future::Future; +use std::task::Context; +use std::task::Waker; + +fn request(thread_id: ThreadId) -> ServerRequestPayload { + ServerRequestPayload::McpServerElicitationRequest(McpServerElicitationRequestParams { + thread_id: thread_id.to_string(), + turn_id: Some("turn-1".into()), + server_name: "plugin-service".into(), + request: McpServerElicitationRequest::UserVerification { + title: "Approve".into(), + description: String::new(), + challenge: "AQID".into(), + }, + }) +} + +#[tokio::test] +async fn verification_is_delivered_to_one_owner_and_other_connections_cannot_resolve_it() { + let (sender, mut messages) = mpsc::channel(4); + let outgoing = OutgoingMessageSender::new(sender, AnalyticsEventsClient::disabled()); + outgoing + .enable_user_verification_connection(ConnectionId(1)) + .await; + let thread_id = ThreadId::new(); + let (id, response) = outgoing + .send_request_to_connections( + Some(&[ConnectionId(2), ConnectionId(1)]), + request(thread_id), + Some(thread_id), + ) + .await; + assert!(matches!( + messages.recv().await, + Some(OutgoingEnvelope::ToConnection { + connection_id: ConnectionId(1), + .. + }) + )); + assert!(messages.try_recv().is_err()); + let proof = json!({"action": "accept", "content": {"credentialId": "AQID", "signature": "BAUG"}, "_meta": null}); + outgoing + .notify_client_response(ConnectionId(2), id.clone(), proof.clone()) + .await; + outgoing + .notify_client_error(ConnectionId(2), id.clone(), internal_error("not the owner")) + .await; + assert!( + outgoing + .request_id_to_callback + .lock() + .await + .contains_key(&id) + ); + outgoing + .replay_requests_to_connection_for_thread(ConnectionId(2), thread_id) + .await; + assert!(messages.try_recv().is_err()); + outgoing + .notify_client_response(ConnectionId(1), id.clone(), proof.clone()) + .await; + assert_eq!(response.await.unwrap(), Ok(proof)); + assert!( + !outgoing + .request_id_to_callback + .lock() + .await + .contains_key(&id) + ); +} + +#[tokio::test] +async fn verification_cancels_when_its_owner_disconnects() { + let (sender, mut messages) = mpsc::channel(4); + let outgoing = OutgoingMessageSender::new(sender, AnalyticsEventsClient::disabled()); + outgoing + .enable_user_verification_connection(ConnectionId(1)) + .await; + let thread_id = ThreadId::new(); + let (id, response) = outgoing + .send_request_to_connections( + Some(&[ConnectionId(1)]), + request(thread_id), + Some(thread_id), + ) + .await; + messages.recv().await.unwrap(); + outgoing.connection_closed(ConnectionId(2)).await; + assert!( + outgoing + .request_id_to_callback + .lock() + .await + .contains_key(&id) + ); + outgoing.connection_closed(ConnectionId(1)).await; + assert!(response.await.is_err()); + assert!( + !outgoing + .request_id_to_callback + .lock() + .await + .contains_key(&id) + ); +} + +#[tokio::test] +async fn verification_disconnect_during_registration_leaves_no_callback_or_request() { + let (sender, mut messages) = mpsc::channel(4); + let outgoing = OutgoingMessageSender::new(sender, AnalyticsEventsClient::disabled()); + outgoing + .enable_user_verification_connection(ConnectionId(1)) + .await; + let thread_id = ThreadId::new(); + let connection_ids = [ConnectionId(1)]; + let mut registration = std::pin::pin!(outgoing.send_request_to_connections( + Some(&connection_ids), + request(thread_id), + Some(thread_id), + )); + let mut disconnect = std::pin::pin!(outgoing.connection_closed(ConnectionId(1))); + { + // Suspend registration at the callback lock, then let disconnect remove eligibility. + let _callbacks = outgoing.request_id_to_callback.lock().await; + let mut context = Context::from_waker(Waker::noop()); + assert!(registration.as_mut().poll(&mut context).is_pending()); + assert!(disconnect.as_mut().poll(&mut context).is_pending()); + assert!( + outgoing + .verification_connections + .try_lock() + .expect("registration must release eligibility before waiting for callbacks") + .is_empty() + ); + } + let ((id, response), ()) = tokio::join!(registration, disconnect); + assert!(response.await.is_err()); + assert!(messages.try_recv().is_err()); + assert!( + !outgoing + .request_id_to_callback + .lock() + .await + .contains_key(&id) + ); +} + +#[tokio::test] +async fn verification_without_a_connected_app_has_no_pending_callback() { + let (sender, mut messages) = mpsc::channel(4); + let outgoing = OutgoingMessageSender::new(sender, AnalyticsEventsClient::disabled()); + outgoing + .enable_user_verification_connection(ConnectionId(1)) + .await; + let thread_id = ThreadId::new(); + let (_, response) = outgoing + .send_request_to_connections(Some(&[]), request(thread_id), Some(thread_id)) + .await; + assert!(response.await.is_err()); + assert!(messages.try_recv().is_err()); + assert!(outgoing.request_id_to_callback.lock().await.is_empty()); +} + +async fn auth_manager(home: &std::path::Path) -> Arc { + let auth = AuthManager::from_auth_for_testing_with_home( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + home.to_path_buf(), + ); + write_auth(home, "token-1", "account-1"); + reload_auth(&auth).await; + auth +} + +async fn reload_auth(auth: &AuthManager) { + let changes = auth.auth_change_receiver(); + let revision = *changes.borrow(); + // reload's boolean compares auth modes; the watch revision also detects token/account changes. + auth.reload().await; + assert_eq!(*changes.borrow(), revision + 1); +} + +fn write_auth(home: &std::path::Path, token: &str, account: &str) { + write_chatgpt_auth( + home, + ChatGptAuthFixture::new(token) + .chatgpt_user_id("user-1") + .account_id(account) + .chatgpt_account_id(account), + AuthCredentialsStoreMode::File, + ) + .unwrap(); +} + +#[tokio::test] +async fn auth_changes_cancel_user_verification() { + let home = tempfile::tempdir().unwrap(); + let auth = auth_manager(home.path()).await; + let (sender, mut messages) = mpsc::channel(4); + let outgoing = Arc::new(OutgoingMessageSender::new( + sender, + AnalyticsEventsClient::disabled(), + )); + outgoing.watch_user_verification_auth(Arc::clone(&auth)); + outgoing + .enable_user_verification_connection(ConnectionId(1)) + .await; + let thread_id = ThreadId::new(); + for (token, account) in [("token-2", "account-1"), ("token-3", "account-2")] { + let (_, response) = outgoing + .send_request_to_connections( + Some(&[ConnectionId(1)]), + request(thread_id), + Some(thread_id), + ) + .await; + messages.recv().await.unwrap(); + write_auth(home.path(), token, account); + reload_auth(&auth).await; + assert!( + tokio::time::timeout(std::time::Duration::from_secs(5), response) + .await + .unwrap() + .is_err() + ); + } +} + +#[tokio::test] +async fn account_switch_back_rejects_proof_before_the_auth_watcher_runs() { + let home = tempfile::tempdir().unwrap(); + let auth = auth_manager(home.path()).await; + let (sender, mut messages) = mpsc::channel(4); + let outgoing = OutgoingMessageSender::new(sender, AnalyticsEventsClient::disabled()); + // Leave the watcher stopped to exercise response validation before a coalesced notification. + assert!(outgoing.verification_auth.set(Arc::clone(&auth)).is_ok()); + outgoing + .enable_user_verification_connection(ConnectionId(1)) + .await; + let thread_id = ThreadId::new(); + let (id, response) = outgoing + .send_request_to_connections( + Some(&[ConnectionId(1)]), + request(thread_id), + Some(thread_id), + ) + .await; + messages.recv().await.unwrap(); + for (token, account) in [("token-2", "account-2"), ("token-3", "account-1")] { + write_auth(home.path(), token, account); + reload_auth(&auth).await; + } + let proof = json!({"action": "accept", "content": {"credentialId": "AQID", "signature": "BAUG"}, "_meta": null}); + outgoing + .notify_client_response(ConnectionId(1), id, proof) + .await; + assert!(response.await.is_err()); +} + +#[tokio::test] +async fn account_switch_during_eligibility_lock_wait_does_not_reassign_verification() { + let home = tempfile::tempdir().unwrap(); + let auth = auth_manager(home.path()).await; + let (sender, mut messages) = mpsc::channel(4); + let outgoing = OutgoingMessageSender::new(sender, AnalyticsEventsClient::disabled()); + assert!(outgoing.verification_auth.set(Arc::clone(&auth)).is_ok()); + outgoing + .enable_user_verification_connection(ConnectionId(1)) + .await; + let thread_id = ThreadId::new(); + let connection_ids = [ConnectionId(1)]; + + let mut registration = std::pin::pin!(outgoing.send_request_to_connections( + Some(&connection_ids), + request(thread_id), + Some(thread_id), + )); + { + let eligible = outgoing.verification_connections.lock().await; + let mut context = Context::from_waker(Waker::noop()); + assert!(registration.as_mut().poll(&mut context).is_pending()); + write_auth(home.path(), "token-2", "account-2"); + drop(eligible); + } + reload_auth(&auth).await; + + let (id, response) = registration.await; + assert!(response.await.is_err()); + assert!(messages.try_recv().is_err()); + assert!( + !outgoing + .request_id_to_callback + .lock() + .await + .contains_key(&id) + ); +} diff --git a/codex-rs/app-server/src/user_verification_response.rs b/codex-rs/app-server/src/user_verification_response.rs new file mode 100644 index 0000000000..6d57afcd81 --- /dev/null +++ b/codex-rs/app-server/src/user_verification_response.rs @@ -0,0 +1,45 @@ +//! Device proofs travel only in elicitation content, never in diagnostic metadata. + +use codex_app_server_protocol::McpServerElicitationAction; +use codex_app_server_protocol::McpServerElicitationRequestResponse; +use codex_app_server_protocol::UserVerificationProof; +use serde::Deserialize; +use tokio::sync::oneshot; + +use crate::outgoing_message::ClientRequestResult; + +pub(crate) fn from_client_result( + result: Result, +) -> McpServerElicitationRequestResponse { + let response = result.ok().and_then(Result::ok).and_then(|value| { + serde_json::from_value::(value).ok() + }); + if let Some(mut response) = response { + response.meta = None; + match response.action { + McpServerElicitationAction::Accept => { + if response + .content + .as_ref() + .is_some_and(|content| UserVerificationProof::deserialize(content).is_ok()) + { + // The MCP boundary validates the encoded proof and size limits. + return response; + } + } + McpServerElicitationAction::Decline | McpServerElicitationAction::Cancel => { + response.content = None; + return response; + } + } + } + McpServerElicitationRequestResponse { + action: McpServerElicitationAction::Cancel, + content: None, + meta: None, + } +} + +#[cfg(test)] +#[path = "user_verification_response_tests.rs"] +mod tests; diff --git a/codex-rs/app-server/src/user_verification_response_tests.rs b/codex-rs/app-server/src/user_verification_response_tests.rs new file mode 100644 index 0000000000..57ee37626b --- /dev/null +++ b/codex-rs/app-server/src/user_verification_response_tests.rs @@ -0,0 +1,36 @@ +use super::*; +use pretty_assertions::assert_eq; +use serde_json::json; + +#[test] +fn accepted_device_proof_uses_existing_content_and_discards_metadata() { + assert_eq!( + from_client_result(Ok(Ok(json!({ + "action": "accept", "content": {"credentialId": "AQID", "signature": "BAUG"}, + "_meta": {"ignored": "untrusted"}, + })))), + McpServerElicitationRequestResponse { + action: McpServerElicitationAction::Accept, + content: Some(json!({"credentialId": "AQID", "signature": "BAUG"})), + meta: None, + } + ); +} + +#[test] +fn invalid_response_or_accept_without_proof_cancels() { + for response in [ + json!({"invalid": "response"}), + json!({"action": "accept", "content": null, "_meta": null}), + json!({"action": "accept", "content": {}, "_meta": null}), + ] { + assert_eq!( + from_client_result(Ok(Ok(response))), + McpServerElicitationRequestResponse { + action: McpServerElicitationAction::Cancel, + content: None, + meta: None, + } + ); + } +} diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index bcbfb1dd4a..45deec37ea 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -905,7 +905,11 @@ async fn disabled_permissions_auto_accept_elicitation_with_empty_form_schema() { ElicitationRequestRouter::default(), ); let (tx_event, _rx_event) = async_channel::bounded(1); - let sender = manager.make_sender("server".to_string(), Some(tx_event)); + let sender = manager.make_sender( + "server".to_string(), + Some(tx_event), + &ClientMcpExtensions::default(), + ); let response = sender( NumberOrString::Number(1), @@ -939,7 +943,11 @@ async fn disabled_permissions_do_not_auto_accept_elicitation_with_requested_fiel ElicitationRequestRouter::default(), ); let (tx_event, _rx_event) = async_channel::bounded(1); - let sender = manager.make_sender("server".to_string(), Some(tx_event)); + let sender = manager.make_sender( + "server".to_string(), + Some(tx_event), + &ClientMcpExtensions::default(), + ); let response = sender( NumberOrString::Number(1), @@ -1030,7 +1038,11 @@ async fn assert_elicitation_declined_with_reviewer_calls( full_access_form_input_enabled_router(), ); let (tx_event, rx_event) = async_channel::bounded(1); - let sender = manager.make_sender(server_name.to_string(), Some(tx_event)); + let sender = manager.make_sender( + server_name.to_string(), + Some(tx_event), + &ClientMcpExtensions::default(), + ); let response = tokio::select! { biased; @@ -1070,7 +1082,11 @@ async fn assert_requested_user_input_is_declined( router, ); let (tx_event, rx_event) = async_channel::bounded(1); - let sender = manager.make_sender("server".to_string(), Some(tx_event)); + let sender = manager.make_sender( + "server".to_string(), + Some(tx_event), + &ClientMcpExtensions::default(), + ); let response = tokio::select! { biased; @@ -1157,7 +1173,11 @@ async fn assert_disabled_permissions_surface_requested_user_input( router.clone(), ); let (tx_event, rx_event) = async_channel::bounded(1); - let sender = manager.make_sender("server".to_string(), Some(tx_event)); + let sender = manager.make_sender( + "server".to_string(), + Some(tx_event), + &ClientMcpExtensions::default(), + ); let requested_schema = requested_user_input_schema(); let mut pending = tokio::spawn(sender( NumberOrString::Number(1), @@ -1302,7 +1322,11 @@ async fn disabled_permissions_decline_user_input_without_an_event_channel() { /*lifecycle*/ None, full_access_form_input_enabled_router(), ); - let sender = manager.make_sender("server".to_string(), /*tx_event*/ None); + let sender = manager.make_sender( + "server".to_string(), + /*tx_event*/ None, + &ClientMcpExtensions::default(), + ); let response = sender( NumberOrString::Number(1), @@ -1360,7 +1384,11 @@ async fn concurrent_authority_updates_never_auto_approve_mixed_policy() { )); } }); - let sender = manager.make_sender("server".to_string(), /*tx_event*/ None); + let sender = manager.make_sender( + "server".to_string(), + /*tx_event*/ None, + &ClientMcpExtensions::default(), + ); let elicitation = codex_rmcp_client::Elicitation::Mcp(ElicitRequestParams::FormElicitationParams { meta: None, @@ -1427,8 +1455,16 @@ async fn shared_elicitation_router_targets_the_exact_pending_request() { router.clone(), ); let (tx_event, rx_event) = async_channel::bounded(2); - let sender_a = manager_a.make_sender("server".to_string(), Some(tx_event.clone())); - let sender_b = manager_b.make_sender("server".to_string(), Some(tx_event)); + let sender_a = manager_a.make_sender( + "server".to_string(), + Some(tx_event.clone()), + &ClientMcpExtensions::default(), + ); + let sender_b = manager_b.make_sender( + "server".to_string(), + Some(tx_event), + &ClientMcpExtensions::default(), + ); let elicitation = codex_rmcp_client::Elicitation::Mcp(ElicitRequestParams::FormElicitationParams { meta: None, @@ -1523,7 +1559,11 @@ async fn cancelled_elicitation_is_removed_without_affecting_other_pending_reques router.clone(), ); let (tx_event, rx_event) = async_channel::bounded(2); - let sender = manager.make_sender("server".to_string(), Some(tx_event)); + let sender = manager.make_sender( + "server".to_string(), + Some(tx_event), + &ClientMcpExtensions::default(), + ); let elicitation = codex_rmcp_client::Elicitation::Mcp(ElicitRequestParams::FormElicitationParams { meta: None, diff --git a/codex-rs/codex-mcp/src/elicitation.rs b/codex-rs/codex-mcp/src/elicitation.rs index 7916d6c967..ec3d3f0d0f 100644 --- a/codex-rs/codex-mcp/src/elicitation.rs +++ b/codex-rs/codex-mcp/src/elicitation.rs @@ -21,6 +21,8 @@ use anyhow::anyhow; use async_channel::Sender; use codex_protocol::approvals::ElicitationRequest; use codex_protocol::approvals::ElicitationRequestEvent; +use codex_protocol::mcp::ClientMcpExtensions; +use codex_protocol::mcp::OPENAI_ELICITATION_EXTENSION_ID; use codex_protocol::mcp::RequestId as ProtocolRequestId; use codex_protocol::mcp_approval_meta::APPROVAL_KIND_KEY; use codex_protocol::mcp_approval_meta::APPROVAL_KIND_TOOL_SUGGESTION; @@ -43,6 +45,9 @@ static NEXT_ELICITATION_REQUEST_ID: AtomicU64 = AtomicU64::new(0); const STRICT_AUTO_REVIEW_DECLINE_MESSAGE: &str = "Automated review of this operation failed. Do not proceed without asking the user for explicit approval."; +#[path = "user_verification_elicitation.rs"] +mod user_verification_elicitation; + #[derive(Debug, Clone)] pub struct ElicitationReviewRequest { pub server_name: String, @@ -147,7 +152,7 @@ impl ElicitationRequestRouter { .ok_or_else(|| anyhow!("elicitation request not found"))?; responder .send(response) - .map_err(|e| anyhow!("failed to send elicitation response: {e:?}")) + .map_err(|_| anyhow!("elicitation response receiver closed")) } } @@ -202,7 +207,14 @@ impl ElicitationRequestManager { &self, server_name: String, tx_event: Option>, + client_mcp_extensions: &ClientMcpExtensions, ) -> SendElicitation { + // Event receivers such as codex mcp-server do not necessarily handle verification. + // Only wait for a response when trusted host activation enabled this exact route. + let user_verification_enabled = client_mcp_extensions + .get(OPENAI_ELICITATION_EXTENSION_ID) + .and_then(|settings| settings.get("userVerification")) + .is_some_and(Value::is_object); let router = self.router.clone(); let authority = self.authority.clone(); Box::new(move |id, elicitation| { @@ -211,13 +223,31 @@ impl ElicitationRequestManager { let server_name = server_name.clone(); let authority = authority.clone(); async move { - // Activate only once the typed app-server and UI path is available. - if matches!(&elicitation, Elicitation::UserVerification { .. }) { - return Ok(ElicitationResponse { - action: ElicitationAction::Cancel, - content: None, - meta: None, - }); + if let Elicitation::UserVerification { + title, + description, + challenge, + } = elicitation + { + if !user_verification_enabled { + return Ok(ElicitationResponse { + action: ElicitationAction::Cancel, + content: None, + meta: None, + }); + } + return user_verification_elicitation::route( + router, + tx_event, + authority, + server_name, + ElicitationRequest::UserVerification { + title, + description, + challenge, + }, + ) + .await; } if router.auto_deny() { return Ok(ElicitationResponse { diff --git a/codex-rs/codex-mcp/src/elicitation_tests.rs b/codex-rs/codex-mcp/src/elicitation_tests.rs index 9bd780ccaf..a1a67d30b4 100644 --- a/codex-rs/codex-mcp/src/elicitation_tests.rs +++ b/codex-rs/codex-mcp/src/elicitation_tests.rs @@ -92,7 +92,11 @@ fn elicitation_fixture( ElicitationRequestRouter::default(), ); let (tx_event, events) = async_channel::bounded(1); - let sender = manager.make_sender("independent-mcp".to_string(), Some(tx_event)); + let sender = manager.make_sender( + "independent-mcp".to_string(), + Some(tx_event), + &ClientMcpExtensions::default(), + ); (manager, events, sender) } @@ -210,7 +214,11 @@ async fn strict_auto_review_respects_explicit_elicitation_denials() { ("independent-mcp", Some(json!(false))), ("another-independent-mcp", None), ] { - let sender = manager.make_sender(server_name.into(), /*tx_event*/ None); + let sender = manager.make_sender( + server_name.into(), + /*tx_event*/ None, + &ClientMcpExtensions::default(), + ); assert_eq!( send_elicitation(&sender, marker).await, ElicitationResponse { @@ -334,10 +342,15 @@ async fn reused_elicitation_senders_follow_each_servers_latest_permission_author /*lifecycle*/ None, ElicitationRequestRouter::default(), ); - let attached = manager.make_sender("attached".to_string(), /*tx_event*/ None); + let attached = manager.make_sender( + "attached".to_string(), + /*tx_event*/ None, + &ClientMcpExtensions::default(), + ); let hosted = manager.make_sender( crate::CODEX_APPS_MCP_SERVER_NAME.to_string(), /*tx_event*/ None, + &ClientMcpExtensions::default(), ); assert_eq!( @@ -398,3 +411,179 @@ async fn reused_elicitation_senders_follow_each_servers_latest_permission_author ElicitationAction::Decline ); } + +fn verification_fixture( + approval_policy: AskForApproval, + reviewer: Option>, +) -> (ElicitationRequestManager, Receiver, SendElicitation) { + let mut config = test_elicitation_config( + crate::CODEX_APPS_MCP_SERVER_NAME, + approval_policy, + PermissionProfile::Disabled, + ); + let mut catalog = crate::catalog::ResolvedMcpCatalog::builder(); + catalog.register(crate::catalog::McpServerRegistration::from_hosted_apps( + "verification-test", + /*contribution_order*/ 0, + crate::mcp::codex_apps_mcp_server_config( + "https://example.com", + /*apps_mcp_product_sku*/ None, + /*originator*/ None, + ), + )); + Arc::make_mut(&mut config).mcp_server_catalog = catalog.build(); + let manager = ElicitationRequestManager::new( + config, + reviewer.map(|reviewer| reviewer as Arc), + /*lifecycle*/ None, + ElicitationRequestRouter::default(), + ); + let (tx, events) = async_channel::bounded(1); + let sender = manager.make_sender( + crate::CODEX_APPS_MCP_SERVER_NAME.into(), + Some(tx), + &ClientMcpExtensions::new([( + OPENAI_ELICITATION_EXTENSION_ID.to_string(), + json!({"userVerification": {}}), + )]), + ); + (manager, events, sender) +} + +#[tokio::test] +async fn user_verification_requires_the_app_even_when_policy_would_approve_or_decline() { + for approval_policy in [AskForApproval::Never, AskForApproval::OnRequest] { + let reviewer = RecordingReviewer::new(Ok(Some(approved_response()))); + let (manager, events, sender) = + verification_fixture(approval_policy, Some(reviewer.clone())); + let pending = tokio::spawn(sender( + RequestId::Number(7), + Elicitation::UserVerification { + title: "Approve purchase".into(), + description: "Pay $200".into(), + challenge: "AQID".into(), + }, + )); + let event = events.recv().await.unwrap(); + let EventMsg::ElicitationRequest(request) = event.msg else { + panic!("expected typed elicitation"); + }; + assert_eq!( + request.request, + ElicitationRequest::UserVerification { + title: "Approve purchase".into(), + description: "Pay $200".into(), + challenge: "AQID".into(), + } + ); + assert_eq!(reviewer.calls.load(Relaxed), 0); + let ProtocolRequestId::String(id) = request.id else { + panic!("expected routed request id"); + }; + let proof = ElicitationResponse { + action: ElicitationAction::Accept, + content: Some(json!({"credentialId": "AQID", "signature": "BAUG"})), + meta: None, + }; + manager + .router + .resolve( + request.server_name.clone(), + RequestId::String(id.clone().into()), + proof.clone(), + ) + .await + .unwrap(); + assert_eq!(pending.await.unwrap().unwrap(), proof); + assert!( + manager + .router + .resolve(request.server_name, RequestId::String(id.into()), proof) + .await + .is_err() + ); + } +} + +#[tokio::test] +async fn user_verification_cancels_when_no_app_can_receive_the_request() { + let (manager, _, _) = verification_fixture(AskForApproval::OnRequest, /*reviewer*/ None); + let sender = manager.make_sender( + crate::CODEX_APPS_MCP_SERVER_NAME.into(), + /*tx_event*/ None, + &ClientMcpExtensions::new([( + OPENAI_ELICITATION_EXTENSION_ID.to_string(), + json!({"userVerification": {}}), + )]), + ); + assert_eq!( + sender( + RequestId::Number(7), + Elicitation::UserVerification { + title: "Approve".into(), + description: String::new(), + challenge: "AQID".into(), + } + ) + .await + .unwrap(), + ElicitationResponse { + action: ElicitationAction::Cancel, + content: None, + meta: None + }, + ); +} + +#[tokio::test] +async fn user_verification_cancels_for_an_event_receiver_without_host_activation() { + let (manager, _, _) = verification_fixture(AskForApproval::OnRequest, /*reviewer*/ None); + let (tx, events) = async_channel::bounded(1); + let sender = manager.make_sender( + crate::CODEX_APPS_MCP_SERVER_NAME.into(), + Some(tx), + &ClientMcpExtensions::default(), + ); + let response = tokio::time::timeout( + std::time::Duration::from_secs(1), + sender( + RequestId::Number(7), + Elicitation::UserVerification { + title: "Approve".into(), + description: String::new(), + challenge: "AQID".into(), + }, + ), + ) + .await + .expect("an inactive host must not wait for a response") + .unwrap(); + assert_eq!( + response, + ElicitationResponse { + action: ElicitationAction::Cancel, + content: None, + meta: None, + }, + ); + assert!(events.try_recv().is_err()); + assert!(manager.router.requests.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn user_verification_drops_pending_response_when_the_request_is_cancelled() { + let (manager, events, sender) = + verification_fixture(AskForApproval::OnRequest, /*reviewer*/ None); + let pending = tokio::spawn(sender( + RequestId::Number(7), + Elicitation::UserVerification { + title: "Approve".into(), + description: String::new(), + challenge: "AQID".into(), + }, + )); + events.recv().await.unwrap(); + pending.abort(); + assert!(pending.await.unwrap_err().is_cancelled()); + assert!(manager.router.requests.lock().unwrap().is_empty()); +} diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 8f12ad8de9..0034d36179 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -897,9 +897,10 @@ async fn start_server_task( client_mcp_extensions, catalog_item_limit, } = params; + let send_elicitation = + elicitation_requests.make_sender(server_name.clone(), tx_event, &client_mcp_extensions); let params = mcp_initialize_request_params(client_elicitation_capability, client_mcp_extensions); - let send_elicitation = elicitation_requests.make_sender(server_name.clone(), tx_event); let started_at = Instant::now(); let initialize_result = client diff --git a/codex-rs/codex-mcp/src/user_verification_elicitation.rs b/codex-rs/codex-mcp/src/user_verification_elicitation.rs new file mode 100644 index 0000000000..29226a6f23 --- /dev/null +++ b/codex-rs/codex-mcp/src/user_verification_elicitation.rs @@ -0,0 +1,54 @@ +//! Routes device verification directly to the app, outside automated approval policy. + +use super::*; + +pub(super) async fn route( + router: ElicitationRequestRouter, + events: Option>, + authority: Arc>>, + server_name: String, + request: ElicitationRequest, +) -> Result { + let authority = authority + .lock() + .ok() + .and_then(|authority| authority.clone()); + let Some(events) = events.filter(|_| !router.auto_deny()) else { + return Ok(ElicitationResponse { + action: ElicitationAction::Cancel, + content: None, + meta: None, + }); + }; + let id = format!( + "codex-mcp-elicitation-{}", + NEXT_ELICITATION_REQUEST_ID.fetch_add(1, Ordering::Relaxed) + ); + let key = (server_name.clone(), RequestId::String(id.clone().into())); + let (response, receiver) = oneshot::channel(); + router + .requests + .lock() + .map_err(|_| anyhow!("elicitation request router unavailable"))? + .insert(key.clone(), response); + let _pending = PendingElicitationRequest { router, key }; + let _active = authority + .as_ref() + .and_then(|authority| authority.lifecycle.as_ref()) + .map(ElicitationLifecycle::start); + events + .send(Event { + id: "mcp_elicitation_request".to_string(), + msg: EventMsg::ElicitationRequest(ElicitationRequestEvent { + turn_id: None, + server_name, + id: ProtocolRequestId::String(id), + request, + }), + }) + .await + .context("failed to deliver user-verification request")?; + receiver + .await + .context("user-verification response channel closed") +} diff --git a/codex-rs/core/src/context/mod.rs b/codex-rs/core/src/context/mod.rs index dac2accfe8..d1ac427700 100644 --- a/codex-rs/core/src/context/mod.rs +++ b/codex-rs/core/src/context/mod.rs @@ -45,6 +45,7 @@ mod unsupported_media; mod update_plan_instructions; mod user_instructions; mod user_shell_command; +mod user_verification_notice; pub(crate) mod world_state; pub(crate) use approved_command_prefix_saved::APPROVED_COMMAND_PREFIX_SAVED_MESSAGE_PREFIX; @@ -114,4 +115,5 @@ pub(crate) use unsupported_media::UnsupportedMedia; pub use update_plan_instructions::without_update_plan_instructions; pub(crate) use user_instructions::UserInstructions; pub(crate) use user_shell_command::UserShellCommand; +pub(crate) use user_verification_notice::UserVerificationNotice; pub(crate) use world_state::ManagedDeveloperInstructions; diff --git a/codex-rs/core/src/context/user_verification_notice.rs b/codex-rs/core/src/context/user_verification_notice.rs new file mode 100644 index 0000000000..ea9f381c19 --- /dev/null +++ b/codex-rs/core/src/context/user_verification_notice.rs @@ -0,0 +1,28 @@ +//! A bounded realtime notice that never includes verification challenge or display text. + +use super::ContextualUserFragment; +use codex_protocol::models::ContentItemKind; + +pub(crate) struct UserVerificationNotice; + +impl ContextualUserFragment for UserVerificationNotice { + fn role(&self) -> &'static str { + "developer" + } + + fn content_kind(&self) -> ContentItemKind { + ContentItemKind("user_verification.notice".to_string()) + } + + fn markers(&self) -> (&'static str, &'static str) { + Self::type_markers() + } + + fn type_markers() -> (&'static str, &'static str) { + ("", "") + } + + fn body(&self) -> String { + "User verification is required. Please respond in the app.".to_string() + } +} diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index e4594f78b7..aa7c99700c 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -534,7 +534,11 @@ pub(super) async fn submission_loop( // To break out of this loop, send Op::Shutdown. let mut shutdown_received = false; while let Ok(sub) = rx_sub.recv().await { - debug!(?sub, "Submission"); + if matches!(sub.op, Op::ResolveElicitation { .. }) { + debug!(submission_id = %sub.id, operation = sub.op.kind(), "Submission"); + } else { + debug!(?sub, "Submission"); + } let dispatch_span = submission_dispatch_span(&sub); let should_exit = async { match sub.op { diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 26ff799a44..4719e1d891 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -13,6 +13,7 @@ use crate::compact_remote::run_inline_remote_auto_compact_task; use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_remote_auto_compact_task_v2; use crate::connectors; use crate::context::ContextualUserFragment; +use crate::context::UserVerificationNotice; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::feedback_tags; use crate::hook_runtime::drain_async_hook_results; @@ -1833,6 +1834,14 @@ pub(super) fn agent_message_text(item: &codex_protocol::items::AgentMessageItem) pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<(String, Option)> { match msg { + EventMsg::ElicitationRequest(request) + if matches!( + &request.request, + codex_protocol::approvals::ElicitationRequest::UserVerification { .. } + ) => + { + Some((UserVerificationNotice.render(), None)) + } EventMsg::AgentMessage(event) => Some((event.message.clone(), event.phase.clone())), EventMsg::ItemCompleted(event) => match &event.item { TurnItem::AgentMessage(item) => Some((agent_message_text(item), item.phase.clone())), diff --git a/codex-rs/core/src/session/turn_tests.rs b/codex-rs/core/src/session/turn_tests.rs index d628f601f8..f1829fe561 100644 --- a/codex-rs/core/src/session/turn_tests.rs +++ b/codex-rs/core/src/session/turn_tests.rs @@ -86,3 +86,24 @@ async fn plan_mode_uses_contributed_turn_item_for_last_agent_message() { Some("plan contributed assistant text") ); } + +#[test] +fn realtime_user_verification_notice_excludes_request_payload() { + let event = EventMsg::ElicitationRequest(codex_protocol::approvals::ElicitationRequestEvent { + turn_id: None, + server_name: "private-server-name".to_string(), + id: codex_protocol::mcp::RequestId::String("private-request-id".to_string()), + request: codex_protocol::approvals::ElicitationRequest::UserVerification { + title: "private-title".to_string(), + description: "private-description".to_string(), + challenge: "private-challenge".to_string(), + }, + }); + assert_eq!( + realtime_text_for_event(&event), + Some(( + "User verification is required. Please respond in the app.".to_string(), + None, + )), + ); +} diff --git a/codex-rs/core/tests/suite/mcp_user_verification.rs b/codex-rs/core/tests/suite/mcp_user_verification.rs index 47354a2339..d53b264b77 100644 --- a/codex-rs/core/tests/suite/mcp_user_verification.rs +++ b/codex-rs/core/tests/suite/mcp_user_verification.rs @@ -1,17 +1,19 @@ -//! Before activation, verification never prompts or prevents the enclosing tool/turn completing. +//! Typed verification holds the enclosing tool until the client answers the request. use anyhow::Result; use codex_core::StartThreadOptions; use codex_core::TurnInputRequest; +use codex_protocol::approvals::ElicitationRequest; use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::mcp::OPENAI_ELICITATION_EXTENSION_ID; +use codex_protocol::protocol::ElicitationAction; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; use codex_protocol::user_input::UserInput; use core_test_support::responses; use core_test_support::skip_if_no_network; use core_test_support::skip_if_wine_exec; use core_test_support::test_codex::test_codex; -use core_test_support::wait_for_event; use core_test_support::wait_for_mcp_server; use pretty_assertions::assert_eq; use serde_json::Value; @@ -66,9 +68,9 @@ enum CapabilitySource { } #[test_case(CapabilitySource::HostProjection; "host_projection_does_not_advertise_verification")] -#[test_case(CapabilitySource::ExplicitSession; "explicit_session_cancels_without_ui")] +#[test_case(CapabilitySource::ExplicitSession; "explicit_session_routes_typed_verification")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn mcp_user_verification_before_activation(source: CapabilitySource) -> Result<()> { +async fn mcp_user_verification_transport(source: CapabilitySource) -> Result<()> { skip_if_no_network!(Ok(())); skip_if_wine_exec!(Ok(()), "the MCP fixture requires a host Python interpreter"); let server = responses::start_mock_server().await; @@ -82,8 +84,7 @@ async fn mcp_user_verification_before_activation(source: CapabilitySource) -> Re Some(&declarations), /*legacy_openai_form_elicitation*/ false, ), - // Exercise the manager's fail-closed route even if a host directly supplies - // this capability before the typed transport/UI activation stage lands. + // Exercise the typed route when the host directly supplies this capability. CapabilitySource::ExplicitSession => ClientMcpExtensions::new(declarations), }; let mut config = test.config.clone(); @@ -130,14 +131,47 @@ async fn mcp_user_verification_before_activation(source: CapabilitySource) -> Re text_elements: Vec::new(), }])) .await?; - wait_for_event(&thread, |event| { - assert!( - !matches!(event, EventMsg::ElicitationRequest(_)), - "verification must not prompt before activation" - ); - matches!(event, EventMsg::TurnComplete(_)) + let mut prompted = false; + tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 10), async { + loop { + match thread.next_event().await?.msg { + EventMsg::ElicitationRequest(request) => { + assert!(matches!(source, CapabilitySource::ExplicitSession)); + assert!(!prompted, "verification must only prompt once"); + assert_eq!( + request.request, + ElicitationRequest::UserVerification { + title: "Approve".into(), + description: "".into(), + challenge: "AQID".into(), + } + ); + assert!( + follow_up.requests().is_empty(), + "the tool must wait for the response" + ); + thread + .submit(Op::ResolveElicitation { + server_name: request.server_name, + request_id: request.id, + decision: ElicitationAction::Cancel, + content: None, + meta: None, + }) + .await?; + prompted = true; + } + EventMsg::TurnComplete(_) => break, + _ => {} + } + } + anyhow::Ok(()) }) - .await; + .await??; + assert_eq!( + prompted, + matches!(source, CapabilitySource::ExplicitSession) + ); let output = follow_up .single_request() .function_call_output("verify-call"); diff --git a/codex-rs/protocol/src/approvals.rs b/codex-rs/protocol/src/approvals.rs index e481ccb028..b05c45ce49 100644 --- a/codex-rs/protocol/src/approvals.rs +++ b/codex-rs/protocol/src/approvals.rs @@ -390,6 +390,13 @@ impl ExecApprovalRequestEvent { #[serde(tag = "mode", rename_all = "snake_case")] #[ts(tag = "mode")] pub enum ElicitationRequest { + #[serde(rename = "openai/userVerification")] + #[ts(rename = "openai/userVerification")] + UserVerification { + title: String, + description: String, + challenge: String, + }, Form { #[serde(rename = "_meta", default, skip_serializing_if = "Option::is_none")] #[ts(optional, rename = "_meta")] diff --git a/codex-rs/tui/src/app/agents_overview_details.rs b/codex-rs/tui/src/app/agents_overview_details.rs index 00cb7e76fc..a7529b88b2 100644 --- a/codex-rs/tui/src/app/agents_overview_details.rs +++ b/codex-rs/tui/src/app/agents_overview_details.rs @@ -242,6 +242,7 @@ fn request_preview(request: &ServerRequest) -> Option { | McpServerElicitationRequest::OpenAiForm { message, .. } | McpServerElicitationRequest::OpenAiElicitationForm { message, .. } | McpServerElicitationRequest::Url { message, .. } => Some(message.as_str()), + McpServerElicitationRequest::UserVerification { .. } => None, }, ServerRequest::ToolRequestUserInput { params, .. } => params .questions diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 768322320f..d8174e755b 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -308,6 +308,14 @@ impl App { message: message.clone(), }), )), + codex_app_server_protocol::McpServerElicitationRequest::UserVerification { .. } => { + self.app_event_tx.resolve_elicitation( + thread_id, params.server_name.clone(), request_id.clone(), + codex_app_server_protocol::McpServerElicitationAction::Cancel, + /*content*/ None, /*meta*/ None, + ); + None + } codex_app_server_protocol::McpServerElicitationRequest::OpenAiForm { .. } @@ -1792,6 +1800,9 @@ impl App { }; match ¶ms.request { + codex_app_server_protocol::McpServerElicitationRequest::UserVerification { .. } => { + false + } codex_app_server_protocol::McpServerElicitationRequest::Form { .. } => true, codex_app_server_protocol::McpServerElicitationRequest::OpenAiForm { .. } | codex_app_server_protocol::McpServerElicitationRequest::OpenAiElicitationForm { diff --git a/codex-rs/tui/src/chatwidget/tool_requests.rs b/codex-rs/tui/src/chatwidget/tool_requests.rs index f745dca0f6..8f7a4216b2 100644 --- a/codex-rs/tui/src/chatwidget/tool_requests.rs +++ b/codex-rs/tui/src/chatwidget/tool_requests.rs @@ -373,6 +373,16 @@ impl ChatWidget { self.bottom_pane .push_approval_request(request, &self.config.features); } + McpServerElicitationRequest::UserVerification { .. } => { + self.app_event_tx.resolve_elicitation( + thread_id, + params.server_name, + request_id, + codex_app_server_protocol::McpServerElicitationAction::Cancel, + /*content*/ None, + /*meta*/ None, + ); + } McpServerElicitationRequest::OpenAiForm { .. } | McpServerElicitationRequest::OpenAiElicitationForm { .. } | McpServerElicitationRequest::Url { .. } => {