diff --git a/codex-rs/app-server/tests/suite/v2/rate_limits.rs b/codex-rs/app-server/tests/suite/v2/rate_limits.rs index a675c21ece..122d44c396 100644 --- a/codex-rs/app-server/tests/suite/v2/rate_limits.rs +++ b/codex-rs/app-server/tests/suite/v2/rate_limits.rs @@ -1,3 +1,4 @@ +use anyhow::Context; use anyhow::Result; use app_test_support::ChatGptAuthFixture; use app_test_support::ChatGptIdTokenClaims; @@ -5,7 +6,9 @@ use app_test_support::TestAppServer; use app_test_support::write_chatgpt_auth; use codex_app_server_protocol::AddCreditsNudgeCreditType; use codex_app_server_protocol::AddCreditsNudgeEmailStatus; +use codex_app_server_protocol::GetAccountParams; use codex_app_server_protocol::GetAccountRateLimitsResponse; +use codex_app_server_protocol::GetAccountResponse; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::LoginAccountResponse; use codex_app_server_protocol::RateLimitReachedType; @@ -745,3 +748,242 @@ fn write_chatgpt_base_url(codex_home: &Path, base_url: &str) -> std::io::Result< let config_toml = codex_home.join("config.toml"); std::fs::write(config_toml, format!("chatgpt_base_url = \"{base_url}\"\n")) } + +#[test_case("disabled"; "disabled")] +#[test_case("dual_header"; "dual_header")] +#[test_case("assertion_only"; "assertion_only")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn account_requests_share_runtime_and_rotate_on_restart(mode: &str) -> Result<()> { + use base64::Engine; + use core_test_support::runtime_auth::PROBE_SERVER; + use core_test_support::runtime_auth::run_harness_through_proxy; + use std::collections::BTreeSet; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + + let enabled = mode != "disabled"; + let token = format!("e30.{}.c2ln", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode( + serde_json::to_vec(&json!({ + "exp": chrono::Utc::now().timestamp() + 3600, + "session_id": "session-a", "iss": "https://auth.openai.com", + "https://api.openai.com/auth": {"chatgpt_user_id": "user-123", "chatgpt_account_id": "account-123"}, + }))? + )); + if std::env::var(PROBE_SERVER).is_ok() { + let home = TempDir::new()?; + write_chatgpt_auth( + home.path(), + ChatGptAuthFixture::new(&token) + .account_id("account-123") + .chatgpt_user_id("user-123") + .plan_type("pro"), + AuthCredentialsStoreMode::File, + )?; + let before = std::fs::read(home.path().join("auth.json"))?; + std::fs::write( + home.path().join("config.toml"), + format!( + "chatgpt_base_url = \"https://chatgpt.com/backend-api\"\nfeatures.connectors = true\n[features.runtime_agent_instances]\nenabled = {enabled}\nmode = {mode:?}\n", + mode = if enabled { mode } else { "dual_header" }, + ), + )?; + for _ in 0..2 { + let mut app = TestAppServer::builder() + .with_codex_home(home.path()) + .without_auto_env() + .with_env_overrides(&[("OPENAI_API_KEY", None)]) + .build_initialized_with_timeout(DEFAULT_READ_TIMEOUT) + .await?; + let id = app + .send_get_account_request(GetAccountParams { + refresh_token: false, + }) + .await?; + let account: GetAccountResponse = + timeout(DEFAULT_READ_TIMEOUT, app.read_response(id)).await??; + assert_eq!( + account + .workspace_routing + .context("workspace routing")? + .chatgpt_account_id, + "account-123" + ); + for _ in 0..2 { + let id = app.send_get_account_rate_limits_request().await?; + let result: GetAccountRateLimitsResponse = + timeout(DEFAULT_READ_TIMEOUT, app.read_response(id)).await??; + assert_eq!(result.rate_limits.plan_type, Some(AccountPlanType::Pro)); + assert_eq!( + result + .rate_limits + .primary + .context("primary rate limit")? + .used_percent, + 42 + ); + } + let id = app + .send_apps_list_request(codex_app_server_protocol::AppsListParams { + limit: None, + cursor: None, + thread_id: None, + force_refetch: true, + }) + .await?; + let apps: codex_app_server_protocol::AppsListResponse = + timeout(DEFAULT_READ_TIMEOUT, app.read_response(id)).await??; + assert!(!apps.data.is_empty(), "Apps list must return the mock app"); + let id = app + .send_apps_read_request(codex_app_server_protocol::AppsReadParams { + app_ids: vec!["details-only".into()], + thread_id: None, + include_tools: false, + }) + .await?; + let apps: codex_app_server_protocol::AppsReadResponse = + timeout(DEFAULT_READ_TIMEOUT, app.read_response(id)).await??; + assert_eq!(apps.apps[0].id, "details-only"); + } + assert_eq!(std::fs::read(home.path().join("auth.json"))?, before); + return Ok(()); + } + + let server = MockServer::start().await; + app_test_support::mount_workspace_routing(&server).await; + core_test_support::apps_test_server::AppsTestServer::mount(&server).await?; + Mock::given(method("GET")).and(wiremock::matchers::path_regex(".*/connectors/directory/(list|list_workspace)$")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "apps": [{"id": "directory-app", "name": "Directory app", "description": "An app from the directory"}], + "nextToken": null, + }))).mount(&server).await; + Mock::given(method("POST")).and(wiremock::matchers::path_regex(".*/ps/apps/batch$")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"apps": [{ + "id": "details-only", "name": "Details", "description": "An app read through the public RPC", + }]}))).expect(2).mount(&server).await; + let registrations = AtomicUsize::new(0); + Mock::given(method("POST")) + .and(path("/agent/register")) + .respond_with(move |_: &wiremock::Request| { + let id = registrations.fetch_add(1, Ordering::Relaxed) + 1; + ResponseTemplate::new(200).set_body_json(json!({ + "runtime_agent_instance_id": format!("runtime-{id}"), "agent_host_id": "host-a", + "actor": {"kind": "chatgpt", "user_id": "user-123", "account_id": "account-123"}, + })) + }) + .expect(if enabled { 2 } else { 0 }) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/backend-api/wham/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "account_id": "account-123", "user_id": "user-123", "plan_type": "pro", + "rate_limit": {"allowed": true, "limit_reached": false, "primary_window": { + "used_percent": 42, "limit_window_seconds": 3600, + "reset_after_seconds": 120, "reset_at": 1900000000, + }}, + }))) + .expect(4) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/backend-api/wham/rate-limit-reset-credits")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "credits": [], "available_count": 0, "total_earned_count": 0, + }))) + .mount(&server) + .await; + run_harness_through_proxy( + &server, + &format!( + "suite::v2::rate_limits::account_requests_share_runtime_and_rotate_on_restart::{mode}" + ), + /*realtime_target*/ None, + ) + .await?; + + let requests = server + .received_requests() + .await + .context("captured requests")?; + let registrations: Vec<_> = requests + .iter() + .filter(|request| request.url.path() == "/agent/register") + .collect(); + if enabled { + let first: serde_json::Value = registrations[0].body_json()?; + let second: serde_json::Value = registrations[1].body_json()?; + assert_eq!(first["agent_host_auth"], second["agent_host_auth"]); + assert_ne!(first["public_key"], second["public_key"]); + assert_ne!( + registrations[0].headers["idempotency-key"], + registrations[1].headers["idempotency-key"] + ); + for request in registrations { + assert!(!request.headers.contains_key("authorization")); + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + } + } + let mut proofs = BTreeSet::new(); + let mut runtimes = BTreeSet::new(); + for request in requests.iter().filter(|request| { + let path = request.url.path(); + path.starts_with("/backend-api/wham/") + || path.contains("/connectors/") + || path.contains("/ps/") + }) { + assert_eq!(request.headers["chatgpt-account-id"], "account-123"); + let authorization = request.headers["authorization"].to_str()?; + let proof = if mode == "assertion_only" { + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + authorization + .strip_prefix("AgentAssertion ") + .with_context(|| { + format!( + "assertion-only authentication for {} {}", + request.method, + request.url.path() + ) + })? + } else { + assert!(authorization.starts_with("Bearer e30.")); + if !enabled { + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + continue; + } + request + .headers + .get("x-openai-agent-assertion") + .with_context(|| { + format!( + "runtime assertion for {} {}", + request.method, + request.url.path() + ) + })? + .to_str()? + }; + assert!( + proofs.insert(proof.to_string()), + "each request must get a fresh assertion" + ); + let payload: serde_json::Value = serde_json::from_slice( + &base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(proof.split('.').nth(2).context("assertion payload")?)?, + )?; + runtimes.insert( + payload["runtime_agent_instance"] + .as_str() + .context("runtime ID")? + .to_string(), + ); + } + if enabled { + assert!(proofs.len() >= 4); + assert_eq!( + runtimes, + BTreeSet::from(["runtime-1".to_string(), "runtime-2".to_string()]) + ); + } + server.verify().await; + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index 4fbdd37934..48f782673f 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -4577,3 +4577,289 @@ fn assert_invalid_request(error: JSONRPCError, message: String) { assert_eq!(error.error.message, message); assert_eq!(error.error.data, None); } + +#[test_case("websocket", "dual_header"; "websocket_dual_header")] +#[test_case("websocket", "assertion_only"; "websocket_assertion_only")] +#[test_case("webrtc", "dual_header"; "webrtc_reconnect_dual_header")] +#[test_case("webrtc", "assertion_only"; "webrtc_reconnect_assertion_only")] +#[test_case("existing_call", "dual_header"; "existing_call_reconnect_dual_header")] +#[test_case("existing_call", "assertion_only"; "existing_call_reconnect_assertion_only")] +#[test_case("voice_key", "dual_header"; "voice_key_dual_header")] +#[test_case("voice_key", "assertion_only"; "voice_key_assertion_only")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn realtime_requests_use_runtime_auth(transport: &str, mode: &str) -> Result<()> { + use base64::Engine; + use core_test_support::runtime_auth::PROBE_SERVER; + use core_test_support::runtime_auth::run_harness_through_proxy; + use std::collections::BTreeSet; + + skip_if_no_network!(Ok(())); + let sideband = matches!(transport, "webrtc" | "existing_call"); + if let Ok(server_uri) = std::env::var(PROBE_SERVER) { + let home = TempDir::new()?; + std::fs::write( + home.path().join("config.toml"), + format!( + "experimental_realtime_ws_base_url = \"wss://chatgpt.com/v1\"\n\ + experimental_realtime_webrtc_call_base_url = \"https://api.openai.com/v1\"\n\ + experimental_realtime_ws_startup_context = \"startup context\"\n\ + [features.runtime_agent_instances]\nenabled = true\nmode = {mode:?}\n\ + [realtime]\nversion = \"v3\"\ntype = \"conversational\"\n" + ), + )?; + let mut app = TestAppServer::builder() + .with_codex_home(home.path()) + .with_env_overrides(&[( + "OPENAI_API_KEY", + (transport == "voice_key").then_some("voice-key"), + )]) + .build_initialized_with_timeout(DEFAULT_TIMEOUT) + .await?; + if transport == "voice_key" { + // External ChatGPT credentials own the harness; Realtime can still use the + // explicit voice API key supplied by the embedding application. + let claims = json!({ + "iss": "https://auth.openai.com/", "session_id": "voice-test-login", + "https://api.openai.com/auth": {"user_id": "actor", "chatgpt_account_id": "account"}, + }); + let token = format!( + "header.{}.signature", + base64::engine::general_purpose::URL_SAFE_NO_PAD + .encode(serde_json::to_vec(&claims)?) + ); + let id = app + .send_chatgpt_auth_tokens_login_request( + token, + "account".into(), + /*chatgpt_plan_type*/ None, + ) + .await?; + let _: serde_json::Value = timeout(DEFAULT_TIMEOUT, app.read_response(id)).await??; + } else { + login_with_api_key(&mut app, "actor-key").await?; + } + let id = app + .send_thread_start_request_with_auto_env(ThreadStartParams::default()) + .await?; + let thread: ThreadStartResponse = timeout(DEFAULT_TIMEOUT, app.read_response(id)).await??; + { + let id = app + .send_thread_realtime_start_request(ThreadRealtimeStartParams { + thread_id: thread.thread.id.clone(), + client_managed_handoffs: None, + delegation_ack_filler: None, + flush_transcript_tail_on_session_end: None, + codex_response_item_prefix: None, + codex_response_handoff_mode: None, + backend_reasoning_status: false, + codex_response_handoff_channel_prefixes: None, + codex_responses_as_items: None, + model: None, + output_modality: RealtimeOutputModality::Audio, + include_startup_context: None, + initial_items: None, + realtime_start_instructions: None, + realtime_end_instructions: None, + prompt: None, + realtime_session_id: None, + voice_session_id: None, + transport: match transport { + "webrtc" => Some(ThreadRealtimeStartTransport::Webrtc { + sdp: "v=offer\r\n".into(), + }), + "existing_call" => Some(ThreadRealtimeStartTransport::ExistingCall { + call_id: "rtc_existing".into(), + }), + _ => None, + }, + version: Some(if sideband { + RealtimeConversationVersion::V3 + } else { + RealtimeConversationVersion::V2 + }), + voice: None, + }) + .await?; + let _: ThreadRealtimeStartResponse = + timeout(DEFAULT_TIMEOUT, app.read_response(id)).await??; + let _: ThreadRealtimeStartedNotification = + read_notification(&mut app, "thread/realtime/started").await?; + if sideband { + // The parent acknowledges the second handshake, caused by a server disconnect. + let client = codex_login::default_client::create_client_without_request_logging(); + timeout(DEFAULT_TIMEOUT, async { + loop { + if client + .get(format!("{server_uri}/realtime-ready")) + .send() + .await? + .status() + .is_success() + { + return Ok::<_, anyhow::Error>(()); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await??; + } + let id = app + .send_thread_realtime_stop_request(ThreadRealtimeStopParams { + thread_id: thread.thread.id.clone(), + }) + .await?; + let _: ThreadRealtimeStopResponse = + timeout(DEFAULT_TIMEOUT, app.read_response(id)).await??; + let _: ThreadRealtimeClosedNotification = + read_notification(&mut app, "thread/realtime/closed").await?; + } + return Ok(()); + } + + let server = MockServer::start().await; + Mock::given(method("POST")).and(path("/agent/register")) + .respond_with(|request: &WiremockRequest| { + let body: Value = request.body_json().expect("valid registration body"); + let voice = body["actor_auth"]["credential"] == "voice-key"; + let actor = if body["actor_auth"]["type"] == "oauth" { + json!({"kind": "chatgpt", "user_id": "actor", "account_id": "account"}) + } else { + json!({"kind": "platform", "user_id": if voice {"voice"} else {"actor"}, "organization_id": "org-a", "project_id": null}) + }; + ResponseTemplate::new(200).set_body_json(json!({ + "runtime_agent_instance_id": if voice {"runtime-voice"} else {"runtime-actor"}, + "agent_host_id": "host-a", "actor": actor, + })) + }).mount(&server).await; + Mock::given(method("POST")) + .and(path("/v1/live")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("Location", "/v1/live/rtc_created") + .set_body_string("v=answer\r\n"), + ) + .expect(if transport == "webrtc" { 1 } else { 0 }) + .mount(&server) + .await; + let expected_handshakes = if sideband { 2 } else { 1 }; + let realtime = start_websocket_server_with_headers( + (0..expected_handshakes) + .map(|index| WebSocketConnectionConfig { + requests: if sideband { + vec![] + } else { + vec![vec![json!({ + "type": "session.updated", "session": {"id": "session-test"}, + })]] + }, + response_headers: Vec::new(), + accept_delay: None, + close_after_requests: sideband && index == 0, + }) + .collect(), + ) + .await; + let target = + url::Url::parse(realtime.uri())?.socket_addrs(/*default_port_number*/ || None)?[0]; + let case = match transport { + "webrtc" => "webrtc_reconnect", + "existing_call" => "existing_call_reconnect", + other => other, + }; + let test_name = format!( + "suite::v2::realtime_conversation::realtime_requests_use_runtime_auth::{case}_{mode}" + ); + let (result, ()) = tokio::join!( + run_harness_through_proxy(&server, &test_name, Some(target)), + async { + assert!( + realtime + .wait_for_handshakes(expected_handshakes, Duration::from_secs(30)) + .await, + "missing authenticated Realtime handshake" + ); + Mock::given(path("/realtime-ready")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + } + ); + result?; + let mut proofs = BTreeSet::new(); + let expected_runtime = if transport == "voice_key" { + "runtime-voice" + } else { + "runtime-actor" + }; + for handshake in realtime.handshakes() { + let authorization = handshake + .header("authorization") + .context("handshake auth")?; + let proof = if mode == "dual_header" { + let credential = if transport == "voice_key" { + "voice-key" + } else { + "actor-key" + }; + assert_eq!(authorization, format!("Bearer {credential}")); + handshake + .header("x-openai-agent-assertion") + .context("runtime proof")? + } else { + assert!(handshake.header("x-openai-agent-assertion").is_none()); + authorization + .strip_prefix("AgentAssertion ") + .context("assertion-only auth")? + .to_string() + }; + assert!( + proofs.insert(proof.clone()), + "a reconnect must sign a fresh assertion" + ); + let payload: Value = serde_json::from_slice( + &base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(proof.split('.').nth(2).context("assertion payload")?)?, + )?; + assert_eq!(payload["runtime_agent_instance"], expected_runtime); + } + let requests = server + .received_requests() + .await + .context("captured requests")?; + let registrations = requests + .iter() + .filter(|request| request.url.path() == "/agent/register") + .count(); + assert_eq!( + registrations, + if transport == "voice_key" { 2 } else { 1 }, + "reconnects reuse the runtime" + ); + for request in requests + .iter() + .filter(|request| request.url.path() == "/v1/live") + { + let authorization = request.headers["authorization"].to_str()?; + let proof = if mode == "dual_header" { + assert_eq!(authorization, "Bearer actor-key"); + request.headers["x-openai-agent-assertion"].to_str()? + } else { + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + authorization + .strip_prefix("AgentAssertion ") + .context("call-create auth")? + }; + assert!( + proofs.insert(proof.to_string()), + "call creation and sideband cannot share an assertion" + ); + let payload: Value = serde_json::from_slice( + &base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(proof.split('.').nth(2).context("assertion payload")?)?, + )?; + assert_eq!(payload["runtime_agent_instance"], "runtime-actor"); + } + server.verify().await; + realtime.shutdown().await; + Ok(()) +} diff --git a/codex-rs/config/src/config_requirements.rs b/codex-rs/config/src/config_requirements.rs index 523f5299fd..ff8f5bd638 100644 --- a/codex-rs/config/src/config_requirements.rs +++ b/codex-rs/config/src/config_requirements.rs @@ -1,5 +1,7 @@ use crate::ApplicationRequirementsToml; use codex_features::FeatureToml; +use codex_features::RuntimeAgentInstancesConfigToml; +use codex_features::RuntimeAuthMode; use codex_model_provider_info::ModelProviderInfo; pub use codex_model_provider_info::ResidencyRequirement; use codex_protocol::config_types::ApprovalsReviewer; @@ -954,14 +956,46 @@ impl WindowsRequirementsToml { } #[derive(Deserialize, Debug, Clone, Default, PartialEq, Eq)] +#[serde(from = "FeatureRequirementsInput")] pub struct FeatureRequirementsToml { - #[serde(flatten)] pub entries: BTreeMap, + pub runtime_auth_mode: Option, +} + +// Keep boolean requirements unchanged while accepting the same typed runtime setting as config. +#[derive(Deserialize)] +struct FeatureRequirementsInput { + runtime_agent_instances: Option>, + #[serde(flatten)] + entries: BTreeMap, +} + +impl From for FeatureRequirementsToml { + fn from(input: FeatureRequirementsInput) -> Self { + let mut entries = input.entries; + if let Some(enabled) = input + .runtime_agent_instances + .as_ref() + .and_then(FeatureToml::enabled) + { + entries.insert("runtime_agent_instances".to_string(), enabled); + } + let runtime_auth_mode = input + .runtime_agent_instances + .and_then(|feature| match feature { + FeatureToml::Enabled(_) => None, + FeatureToml::Config(config) => config.mode, + }); + Self { + entries, + runtime_auth_mode, + } + } } impl FeatureRequirementsToml { pub fn is_empty(&self) -> bool { - self.entries.is_empty() + self.entries.is_empty() && self.runtime_auth_mode.is_none() } } @@ -3319,6 +3353,7 @@ tracestate = { managed = { tenant = "enterprise", region = "us" }, other = { kee ]; let feature_requirements = FeatureRequirementsToml { entries: BTreeMap::from([("personality".to_string(), true)]), + ..Default::default() }; let browser_use = BrowserUseRequirementsToml { extension: None, @@ -4824,6 +4859,7 @@ allowed_approvals_reviewers = ["user"] ("apps".to_string(), false), ("personality".to_string(), true), ]), + ..Default::default() }, RequirementSource::Unknown, )) diff --git a/codex-rs/config/src/guardian_tests.rs b/codex-rs/config/src/guardian_tests.rs index d21c42f1ac..8f32aababb 100644 --- a/codex-rs/config/src/guardian_tests.rs +++ b/codex-rs/config/src/guardian_tests.rs @@ -20,6 +20,7 @@ fn explicit_catalog_policy_preserves_precedence_over_the_legacy_switch() { Sourced::new( FeatureRequirementsToml { entries: [("guardianv2".to_owned(), enabled)].into(), + ..Default::default() }, RequirementSource::Unknown, ) diff --git a/codex-rs/config/src/requirements_layers/stack_tests.rs b/codex-rs/config/src/requirements_layers/stack_tests.rs index 4a638b26c7..85b28885e4 100644 --- a/codex-rs/config/src/requirements_layers/stack_tests.rs +++ b/codex-rs/config/src/requirements_layers/stack_tests.rs @@ -725,6 +725,7 @@ alpha = true Sourced::new( crate::FeatureRequirementsToml { entries: BTreeMap::from([("alpha".to_string(), true), ("beta".to_string(), true),]), + ..Default::default() }, RequirementSource::composite([high_source, low_source]), ) diff --git a/codex-rs/core/README.md b/codex-rs/core/README.md index 278c614dfd..1dc53ef655 100644 --- a/codex-rs/core/README.md +++ b/codex-rs/core/README.md @@ -96,3 +96,33 @@ instead of running with weaker enforcement. Expects the binary containing `codex-core` to simulate the virtual `apply_patch` CLI when `arg1` is `--codex-run-as-apply-patch`. See the `codex-arg0` crate for details. + +## Runtime authentication (experimental) + +Runtime authentication is off by default. To enable it in `config.toml`: + +```toml +[features.runtime_agent_instances] +enabled = true +mode = "dual_header" +``` + +`dual_header` sends the existing actor credential and a fresh agent assertion. +`assertion_only` sends only the assertion. The latter requires backend support; +registration failures do not fall back to credential-only requests. Non-OpenAI +providers and third-party MCP authentication keep their existing behavior. + +The same table in `requirements.toml` pins enablement and mode. A conflicting +explicit mode is rejected. The older boolean `runtime_agent_instances = true` +still works and uses the default mode (`dual_header`) unless requirements pin it. +Deliver typed requirements only to client versions that understand them. + +The harness owns the runtime and its memory-only signing key. Ordinary token +refreshes reuse it; a different credential source or actor context requires a new +runtime. When runtime authentication is selected, legacy `AgentIdentity` +credentials migrate to saved OAuth credentials for the same actor and remove the +persisted signing key. If no matching actor credential exists, a supported +sign-in is required. Ordinary credential loading does not perform this migration. + +Cloud-config bundle and requirements fetches retain their endpoint-restricted +bootstrap authentication, including refreshes after runtime initialization. diff --git a/codex-rs/core/src/config/auth_keyring_tests.rs b/codex-rs/core/src/config/auth_keyring_tests.rs index bdd62f2eef..1c93c2f213 100644 --- a/codex-rs/core/src/config/auth_keyring_tests.rs +++ b/codex-rs/core/src/config/auth_keyring_tests.rs @@ -49,6 +49,7 @@ fn resolve_bootstrap_auth_keyring_backend_kind_uses_secret_auth_storage_feature( let requirements = Sourced::new( FeatureRequirementsToml { entries: BTreeMap::from([("secret_auth_storage".to_string(), true)]), + ..Default::default() }, RequirementSource::Unknown, ); diff --git a/codex-rs/core/src/config/config_loader_tests.rs b/codex-rs/core/src/config/config_loader_tests.rs index 16811b4c11..23612f7dc9 100644 --- a/codex-rs/core/src/config/config_loader_tests.rs +++ b/codex-rs/core/src/config/config_loader_tests.rs @@ -1431,6 +1431,7 @@ personality = true .map(|requirements| requirements.value.clone()), Some(codex_config::FeatureRequirementsToml { entries: BTreeMap::from([("personality".to_string(), true)]), + ..Default::default() }) ); let config_requirements: ConfigRequirements = config_requirements_toml.try_into()?; @@ -1471,6 +1472,7 @@ personality = true .map(|requirements| requirements.value.clone()), Some(codex_config::FeatureRequirementsToml { entries: BTreeMap::from([("personality".to_string(), true)]), + ..Default::default() }) ); Ok(()) diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 1837b8ed6b..362f09a85c 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -2233,6 +2233,7 @@ respect_system_proxy = true let disabled = Sourced::new( FeatureRequirementsToml { entries: BTreeMap::from([("respect_system_proxy".to_string(), false)]), + ..Default::default() }, RequirementSource::Unknown, ); @@ -2255,6 +2256,7 @@ respect_system_proxy = true let enabled = Sourced::new( FeatureRequirementsToml { entries: BTreeMap::from([("respect_system_proxy".to_string(), true)]), + ..Default::default() }, RequirementSource::Unknown, ); @@ -2356,6 +2358,7 @@ async fn system_proxy_fallback_honors_feature_requirements() -> std::io::Result< let requirements = Sourced::new( FeatureRequirementsToml { entries: BTreeMap::from([("system_proxy_fallback".to_string(), required)]), + ..Default::default() }, RequirementSource::Unknown, ); @@ -12454,6 +12457,7 @@ fn retired_feature_requirements_do_not_pin_configured_values( (key.to_string(), required), ("shell_tool".to_string(), false), ]), + ..Default::default() }, RequirementSource::EnterpriseManaged { id: "enterprise-id".to_string(), diff --git a/codex-rs/core/src/config/managed_features.rs b/codex-rs/core/src/config/managed_features.rs index fba265828e..a50c7fed19 100644 --- a/codex-rs/core/src/config/managed_features.rs +++ b/codex-rs/core/src/config/managed_features.rs @@ -14,7 +14,9 @@ use codex_config::config_toml::ConfigToml; use codex_features::Feature; use codex_features::FeatureConfigSource; use codex_features::FeatureOverrides; +use codex_features::FeatureToml; use codex_features::Features; +use codex_features::RuntimeAuthMode; use codex_features::canonical_feature_for_key; use codex_features::feature_for_key; @@ -359,10 +361,42 @@ fn explicit_feature_settings_in_config(cfg: &ConfigToml) -> Vec<(String, Feature explicit_settings } +pub(crate) fn resolve_runtime_auth_mode( + cfg: &ConfigToml, + requirements: Option<&Sourced>, +) -> std::io::Result { + let configured = cfg + .features + .as_ref() + .and_then(|features| features.runtime_agent_instances.as_ref()) + .and_then(|feature| match feature { + FeatureToml::Enabled(_) => None, + FeatureToml::Config(config) => config.mode, + }); + if let Some(requirements) = requirements + && let Some(required) = requirements.value.runtime_auth_mode + { + if configured.is_some_and(|mode| mode != required) { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + ConstraintError::InvalidValue { + field_name: "features.runtime_agent_instances.mode", + candidate: format!("{:?}", configured.unwrap_or_default()), + allowed: format!("{required:?}"), + requirement_source: requirements.source.clone(), + }, + )); + } + return Ok(required); + } + Ok(configured.unwrap_or_default()) +} + pub(crate) fn validate_explicit_feature_settings_in_config_toml( cfg: &ConfigToml, feature_requirements: Option<&Sourced>, ) -> std::io::Result<()> { + resolve_runtime_auth_mode(cfg, feature_requirements)?; let Some(Sourced { value: feature_requirements, source, diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 704219bd6f..74ba3d96bb 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -3619,6 +3619,9 @@ impl Config { { configured_features.enable(Feature::Collab); } + let runtime_auth_mode = managed_features::resolve_runtime_auth_mode( + &cfg, feature_requirements.as_ref(), + )?; let features = ManagedFeatures::from_configured_with_warnings( configured_features, feature_requirements, @@ -4087,15 +4090,6 @@ impl Config { FeatureToml::Config(config) => config.mode, }) .unwrap_or_default(); - let runtime_auth_mode = cfg - .features - .as_ref() - .and_then(|features| features.runtime_agent_instances.as_ref()) - .and_then(|feature| match feature { - FeatureToml::Enabled(_) => None, - FeatureToml::Config(config) => config.mode, - }) - .unwrap_or_default(); let terminal_resize_reflow = resolve_terminal_resize_reflow_config(&cfg); let agent_roles = @@ -5349,3 +5343,7 @@ mod tests; #[cfg(test)] #[path = "config_loader_tests.rs"] mod config_loader_tests; + +#[cfg(test)] +#[path = "runtime_auth_tests.rs"] +mod runtime_auth_tests; diff --git a/codex-rs/core/src/config/runtime_auth_tests.rs b/codex-rs/core/src/config/runtime_auth_tests.rs new file mode 100644 index 0000000000..05ae256f8c --- /dev/null +++ b/codex-rs/core/src/config/runtime_auth_tests.rs @@ -0,0 +1,101 @@ +//! Verifies managed runtime-auth defaults and rejection of conflicting user overrides. + +#![cfg(test)] + +use super::*; +use codex_config::RequirementSource; +use pretty_assertions::assert_eq; +use tempfile::tempdir; + +#[tokio::test] +async fn managed_runtime_mode_pins_defaults_and_rejects_overrides() -> anyhow::Result<()> { + for (name, mode) in [ + ("dual_header", RuntimeAuthMode::DualHeader), + ("assertion_only", RuntimeAuthMode::AssertionOnly), + ] { + let requirement: codex_config::ConfigRequirementsToml = toml::from_str(&format!( + "[features.runtime_agent_instances]\nenabled = true\nmode = '{name}'" + ))?; + for local in ["", "[features]\nruntime_agent_instances = true"] { + let mut sourced = codex_config::ConfigRequirementsWithSources::default(); + sourced.merge_unset_fields(RequirementSource::Unknown, requirement.clone()); + let stack = codex_config::ConfigLayerStack::new( + Vec::new(), + sourced.try_into()?, + requirement.clone(), + )?; + let home = tempdir()?; + let config = Config::load_config_with_layer_stack( + LOCAL_FS.as_ref(), + toml::from_str(local)?, + ConfigOverrides::default(), + AbsolutePathBuf::try_from(home.path())?, + stack, + ) + .await?; + assert_eq!(config.runtime_auth_mode, mode); + assert!(config.features.enabled(Feature::RuntimeAgentInstances)); + } + let different = if name == "dual_header" { + "assertion_only" + } else { + "dual_header" + }; + let configured = toml::from_str(&format!( + "[features.runtime_agent_instances]\nmode = '{different}'" + ))?; + let requirements = Sourced::new( + requirement.feature_requirements.clone().unwrap(), + RequirementSource::Unknown, + ); + assert!( + managed_features::resolve_runtime_auth_mode(&configured, Some(&requirements)).is_err() + ); + assert!( + managed_features::validate_explicit_feature_settings_in_config_toml( + &configured, + Some(&requirements) + ) + .is_err() + ); + } + Ok(()) +} + +#[test] +fn unmanaged_runtime_modes_and_legacy_boolean_requirements_work() -> anyhow::Result<()> { + for (text, expected) in [ + ("", RuntimeAuthMode::DualHeader), + ( + "[features]\nruntime_agent_instances = true", + RuntimeAuthMode::DualHeader, + ), + ( + "[features.runtime_agent_instances]\nmode = 'assertion_only'", + RuntimeAuthMode::AssertionOnly, + ), + ] { + assert_eq!( + managed_features::resolve_runtime_auth_mode( + &toml::from_str(text)?, + /*requirements*/ None + )?, + expected + ); + } + let requirement: codex_config::ConfigRequirementsToml = + toml::from_str("[features]\nruntime_agent_instances = false")?; + let requirement = requirement.feature_requirements.unwrap(); + assert_eq!( + requirement.entries.get("runtime_agent_instances"), + Some(&false) + ); + assert_eq!(requirement.runtime_auth_mode, None); + assert!( + toml::from_str::( + "[features.runtime_agent_instances]\nmode = 'unknown'" + ) + .is_err() + ); + Ok(()) +} diff --git a/codex-rs/core/src/runtime_auth.rs b/codex-rs/core/src/runtime_auth.rs index cba4b0f43d..ad26b763a3 100644 --- a/codex-rs/core/src/runtime_auth.rs +++ b/codex-rs/core/src/runtime_auth.rs @@ -67,21 +67,5 @@ pub(crate) fn is_openai_realtime_url(base_url: &str) -> CodexResult { } #[cfg(test)] -mod tests { - use super::is_openai_realtime_url; - - #[test] - fn realtime_runtime_auth_rejects_unsupported_openai_origins() { - for url in [ - "ws://api.openai.com/v1", - "http://api.openai.com/v1", - "wss://api.openai.com:8443/v1", - "wss://unknown.openai.com/v1", - ] { - assert!(is_openai_realtime_url(url).is_err(), "{url}"); - } - assert!(is_openai_realtime_url("wss://api.openai.com/v1").unwrap()); - assert!(is_openai_realtime_url("https://api.openai.org/v1").unwrap()); - assert!(!is_openai_realtime_url("wss://provider.example.com/v1").unwrap()); - } -} +#[path = "runtime_auth_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/runtime_auth_tests.rs b/codex-rs/core/src/runtime_auth_tests.rs new file mode 100644 index 0000000000..3269cc7e23 --- /dev/null +++ b/codex-rs/core/src/runtime_auth_tests.rs @@ -0,0 +1,62 @@ +//! Verifies the supported destinations and environment boundaries for Realtime auth. + +#![cfg(test)] + +use super::*; +use codex_model_provider::validate_runtime_environment; +use pretty_assertions::assert_eq; + +#[test] +fn realtime_auth_accepts_only_secure_first_party_destinations() { + for url in [ + "https://api.openai.com/v1", + "https://api.openai.org/v1", + "wss://api.openai.com/v1", + "wss://chatgpt.com/backend-api", + ] { + assert!(is_openai_realtime_url(url).unwrap(), "{url}"); + } + for url in [ + "ws://api.openai.com/v1", + "http://api.openai.com/v1", + "wss://unknown.openai.com/v1", + "wss://api.openai.com:8443/v1", + "wss://user:secret@api.openai.com/v1", + ] { + assert!(is_openai_realtime_url(url).is_err(), "{url}"); + } + assert!(!is_openai_realtime_url("wss://voice.example.com").unwrap()); +} + +#[test] +fn realtime_rejects_a_different_registration_environment() { + for (chatgpt_base_url, expected_staging) in [ + ("https://chatgpt.com/backend-api", false), + ("https://chatgpt-staging.com/backend-api", true), + ] { + let runtime = RuntimeAuthManager::for_chatgpt( + AuthManager::from_auth_for_testing(codex_login::CodexAuth::from_api_key("test-key")), + uuid::Uuid::nil().to_string(), + chatgpt_base_url.to_owned(), + RuntimeRegistrationMetadata { + harness: None, + harness_version: None, + role: RuntimeRole::Orchestrator, + }, + ); + for (destination, staging) in [ + ("wss://api.openai.com/v1", false), + ("https://chatgpt.com/backend-api", false), + ("wss://api.openai-staging.com/v1", true), + ("https://api.openai.org/v1", true), + ("https://workspace.chatgpt-staging.com/backend-api", true), + ("https://chatgpt-staging.com/backend-api", true), + ] { + assert_eq!( + validate_runtime_environment(&runtime, Some(destination)).is_ok(), + staging == expected_staging, + "{chatgpt_base_url} -> {destination}", + ); + } + } +} diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index b44e0277dd..bcff2ed2ae 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -635,6 +635,7 @@ async fn reviewer_tool_policy_respect_managed_shell_restrictions() { Some(codex_config::Sourced::new( codex_config::FeatureRequirementsToml { entries: BTreeMap::from([(feature.key().to_string(), false)]), + ..Default::default() }, codex_config::RequirementSource::Unknown, )), diff --git a/codex-rs/core/tests/suite/runtime_auth.rs b/codex-rs/core/tests/suite/runtime_auth.rs index 48e10c3773..1c49cc3676 100644 --- a/codex-rs/core/tests/suite/runtime_auth.rs +++ b/codex-rs/core/tests/suite/runtime_auth.rs @@ -24,6 +24,8 @@ use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::user_input::UserInput; use core_test_support::responses; +use core_test_support::runtime_auth::PROBE_SERVER; +use core_test_support::runtime_auth::run_harness_through_proxy; use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; @@ -32,15 +34,6 @@ use serde_json::Value; use serde_json::json; use std::collections::BTreeSet; use std::sync::Arc; -use std::time::Duration; -use tokio::io::AsyncReadExt; -use tokio::io::AsyncWriteExt; -use tokio::net::TcpListener; -use tokio::net::TcpStream; -use tokio::process::Command; -use tokio::task::JoinSet; -use tokio_rustls::TlsAcceptor; -use tokio_rustls::rustls; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; @@ -49,7 +42,6 @@ use wiremock::matchers::header_regex; use wiremock::matchers::method; use wiremock::matchers::path; -const PROBE_SERVER: &str = "CODEX_RUNTIME_AUTH_TEST_SERVER"; const TEST_NAME: &str = "suite::runtime_auth::harness_shares_runtime_and_honors_thread_auth_config"; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -161,7 +153,7 @@ async fn harness_shares_runtime_and_honors_thread_auth_config() -> Result<()> { .mount(&server) .await; - run_harness_through_proxy(&server, TEST_NAME).await?; + run_harness_through_proxy(&server, TEST_NAME, /*realtime_target*/ None).await?; let requests = responses.requests(); let expected = [ @@ -677,87 +669,6 @@ async fn upload_with_harness( Ok(()) } -async fn run_harness_through_proxy(server: &MockServer, test_name: &str) -> Result<()> { - // The production destination check stays intact. A test-only CA and CONNECT proxy route - // the real host names to wiremock, only inside this isolated test subprocess. - codex_utils_rustls_provider::ensure_rustls_crypto_provider(); - let rcgen::CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![ - "api.openai.com".into(), - "auth.openai.com".into(), - "chatgpt.com".into(), - "us.chatgpt.com".into(), - ])?; - let ca_dir = tempfile::tempdir()?; - let ca_path = ca_dir.path().join("ca.pem"); - std::fs::write(&ca_path, cert.pem())?; - let acceptor = TlsAcceptor::from(Arc::new( - rustls::ServerConfig::builder() - .with_no_client_auth() - .with_single_cert(vec![cert.der().clone()], signing_key.into())?, - )); - let proxy = TcpListener::bind("127.0.0.1:0").await?; - let proxy_url = format!("http://{}", proxy.local_addr()?); - let target = *server.address(); - let proxy_task = tokio::spawn(async move { - let mut connections = JoinSet::new(); - loop { - let (mut stream, _) = proxy.accept().await?; - let acceptor = acceptor.clone(); - connections.spawn(async move { - let mut headers = Vec::new(); - while !headers.ends_with(b"\r\n\r\n") && headers.len() < 8192 { - headers.push(stream.read_u8().await?); - } - let headers = std::str::from_utf8(&headers)?; - anyhow::ensure!( - matches!( - headers.lines().next(), - Some( - "CONNECT api.openai.com:443 HTTP/1.1" - | "CONNECT auth.openai.com:443 HTTP/1.1" - | "CONNECT chatgpt.com:443 HTTP/1.1" - | "CONNECT us.chatgpt.com:443 HTTP/1.1" - ) - ), - "unexpected proxy destination" - ); - stream - .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n") - .await?; - let mut stream = acceptor.accept(stream).await?; - let mut target = TcpStream::connect(target).await?; - tokio::io::copy_bidirectional(&mut stream, &mut target).await?; - Ok::<_, anyhow::Error>(()) - }); - } - #[allow(unreachable_code)] - Ok::<_, anyhow::Error>(()) - }); - let mut command = Command::new(std::env::current_exe()?); - command - .args(["--exact", test_name, "--nocapture"]) - .kill_on_drop(true); - for &key in codex_network_proxy::PROXY_ENV_KEYS { - command.env_remove(key); - } - command - .env(PROBE_SERVER, server.uri()) - .env("HTTPS_PROXY", proxy_url) - .env("NO_PROXY", codex_network_proxy::DEFAULT_NO_PROXY_VALUE) - .env("CODEX_CA_CERTIFICATE", ca_path); - let output = tokio::time::timeout(Duration::from_secs(90), command.output()).await; - proxy_task.abort(); - let output = output??; - assert!( - output.status.success(), - "harness subprocess failed\n{}\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ); - - Ok(()) -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn mcp_uploads_follow_a_resumed_actors_runtime() -> Result<()> { use core_test_support::apps_test_server::*; @@ -863,6 +774,7 @@ async fn mcp_uploads_follow_a_resumed_actors_runtime() -> Result<()> { run_harness_through_proxy( &server, "suite::runtime_auth::mcp_uploads_follow_a_resumed_actors_runtime", + /*realtime_target*/ None, ) .await?; assert_eq!(responses.requests().len(), 9); @@ -1111,3 +1023,135 @@ async fn exercise_mcp_uploads() -> Result<()> { } Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn daybreak_requests_use_runtime_auth() -> Result<()> { + use core_test_support::stdio_server_bin; + skip_if_no_network!(Ok(())); + let original = mcp_actor_token("user-a"); + if std::env::var(PROBE_SERVER).is_ok() { + let actor = |token: &str| { + CodexAuth::from_external_chatgpt_tokens( + token, + "account_id", + /*chatgpt_plan_type*/ None, + ) + }; + let home = Arc::new(tempfile::TempDir::new()?); + let plugin = super::plugins::write_sample_plugin_manifest_and_config(&home); + std::fs::write( + plugin.join(".mcp.json"), + serde_json::to_vec(&json!({ + "mcpServers":{"sample":{"command":stdio_server_bin()?,"cwd":plugin, + "environment_id":"local","default_tools_approval_mode":"approve", + "env":{"MCP_TEST_DAYBREAK_READ_ONLY":"true"}}} + }))?, + )?; + let server = MockServer::start().await; + let test = test_codex() + .with_home(home) + .with_auth(actor(&original)?) + .with_config(|config| { + config.chatgpt_base_url = "https://chatgpt.com".into(); + config + .features + .enable(Feature::RuntimeAgentInstances) + .unwrap(); + config.features.disable(Feature::Apps).unwrap(); + }) + .build_with_remote_and_local_env(&server) + .await?; + core_test_support::wait_for_mcp_server(&test.codex, "sample").await?; + responses::mount_sse_sequence( + &server, + vec![ + responses::sse(vec![ + responses::ev_tool_search_call( + "search", + &json!({"query":"get_codex_security_daybreak_access"}), + ), + responses::ev_completed("response-1"), + ]), + responses::sse(vec![ + responses::ev_function_call_with_namespace( + "access", + "mcp__sample", + "get_codex_security_daybreak_access", + "{}", + ), + responses::ev_completed("response-2"), + ]), + responses::sse(vec![ + responses::ev_assistant_message("done", "done"), + responses::ev_completed("response-3"), + ]), + ], + ) + .await; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Call the sample Daybreak access tool".into(), + text_elements: Vec::new(), + }])) + .await?; + let event = wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::McpToolCallEnd(_)) + }) + .await; + let EventMsg::McpToolCallEnd(event) = event else { + unreachable!() + }; + assert!( + event.result.is_ok(), + "the authenticated tool call must succeed" + ); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.codex.shutdown_and_wait().await?; + return Ok(()); + } + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/agent/register")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "runtime_agent_instance_id":"runtime-access", "agent_host_id":"host-a", + "actor":{"kind":"chatgpt","user_id":"user-a","account_id":"account_id"}, + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")).and(path("/accounts/verified_access")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"programs":[{ + "program":"cyber", "state":"active", "grants":[{"level":"tac1","source":"individual"}] + }]}))).expect(1).mount(&server).await; + run_harness_through_proxy( + &server, + "suite::runtime_auth::daybreak_requests_use_runtime_auth", + /*realtime_target*/ None, + ) + .await?; + let requests = server.received_requests().await.unwrap(); + let access: Vec<_> = requests + .iter() + .filter(|request| request.url.path() == "/accounts/verified_access") + .collect(); + let mut proofs = BTreeSet::new(); + for request in access { + assert_eq!( + request.headers["authorization"], + format!("Bearer {original}") + ); + assert_eq!(request.headers["chatgpt-account-id"], "account_id"); + let proof = request.headers["x-openai-agent-assertion"].to_str()?; + assert!(proofs.insert(proof.to_string())); + let payload: Value = serde_json::from_slice( + &base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(proof.split('.').nth(2).unwrap())?, + )?; + assert_eq!(payload["runtime_agent_instance"], "runtime-access"); + } + server.verify().await; + Ok(()) +} diff --git a/codex-rs/ext/guardian-v2/src/sync_reviewer/reviewer_config_tests.rs b/codex-rs/ext/guardian-v2/src/sync_reviewer/reviewer_config_tests.rs index bdea3a1f98..7c918977e3 100644 --- a/codex-rs/ext/guardian-v2/src/sync_reviewer/reviewer_config_tests.rs +++ b/codex-rs/ext/guardian-v2/src/sync_reviewer/reviewer_config_tests.rs @@ -19,6 +19,7 @@ async fn guardian_review_session_config_allows_pinned_disabled_feature() { Some(Sourced { value: FeatureRequirementsToml { entries: BTreeMap::from([("multi_agent".to_string(), true)]), + ..Default::default() }, source: RequirementSource::Unknown, }), diff --git a/codex-rs/rmcp-client/src/http_client_adapter_tests.rs b/codex-rs/rmcp-client/src/http_client_adapter_tests.rs index 7d1d06a018..01b24851fb 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter_tests.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter_tests.rs @@ -468,3 +468,144 @@ async fn requests_await_auth_and_stop_assertion_redirects() -> anyhow::Result<() server.verify().await; Ok(()) } + +#[tokio::test] +async fn automatic_event_stream_reconnect_refreshes_request_auth() -> anyhow::Result<()> { + use codex_exec_server::RouteAwareHttpClient; + use codex_http_client::HttpClientFactory; + use codex_http_client::OutboundProxyPolicy; + use rmcp::model::ClientCapabilities; + use rmcp::model::Implementation; + use rmcp::model::InitializeRequestParams; + use serde_json::json; + use std::sync::Arc; + use std::time::Duration; + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::ResponseTemplate; + use wiremock::matchers::method; + use wiremock::matchers::path; + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(|request: &wiremock::Request| { + let body: serde_json::Value = request.body_json().unwrap(); + let result = match body["method"].as_str() { + Some("initialize") => json!({ + "protocolVersion": body["params"]["protocolVersion"], + "capabilities": {"tools": {}}, + "serverInfo": {"name": "reconnect-test", "version": "1"}, + }), + Some("tools/call") => json!({ + "content": [], "structuredContent": {"echo": "after reconnect"}, + }), + _ => return ResponseTemplate::new(202), + }; + ResponseTemplate::new(200) + .insert_header("mcp-session-id", "session-a") + .set_body_json(json!({"jsonrpc": "2.0", "id": body["id"], "result": result})) + }) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/mcp")) + .respond_with( + ResponseTemplate::new(200).set_body_raw( + concat!( + "retry: 1\nid: event-1\nevent: message\ndata: ", + r#"{"jsonrpc":"2.0","method":"notifications/tools/list_changed"}"#, + "\n\n", + ) + .as_bytes(), + "text/event-stream", + ), + ) + .up_to_n_times(1) + .with_priority(1) + .mount(&server) + .await; + // Leave the reconnect pending until shutdown instead of starting a tight EOF loop. + Mock::given(method("GET")) + .and(path("/mcp")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_delay(Duration::from_secs(60)), + ) + .with_priority(2) + .mount(&server) + .await; + let client = crate::RmcpClient::new_streamable_http_client( + "reconnect-test", + &format!("{}/mcp", server.uri()), + /*bearer_token*/ None, + /*http_headers*/ None, + /*env_http_headers*/ None, + Default::default(), + Default::default(), + Arc::new(RouteAwareHttpClient::new(HttpClientFactory::new( + OutboundProxyPolicy::ReqwestDefault, + ))), + Some(Arc::new(AsyncRequestAuth { + requests: Default::default(), + })), + ) + .await?; + client + .initialize( + InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("codex-test", "1"), + ), + Some(Duration::from_secs(5)), + Box::new(|_, _| Box::pin(async { anyhow::bail!("unexpected elicitation") })), + ) + .await?; + + // The server ends the stream. Only the protocol engine, not the test, reopens it. + let streams = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let streams: Vec<_> = server + .received_requests() + .await + .unwrap() + .into_iter() + .filter(|request| request.method == "GET") + .collect(); + if streams.len() >= 2 { + break streams; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + assert_eq!(streams[0].headers["mcp-session-id"], "session-a"); + assert_eq!(streams[1].headers["last-event-id"], "event-1"); + assert_ne!( + streams[0].headers["authorization"], + streams[1].headers["authorization"] + ); + for request in &streams { + assert!( + request.headers["authorization"] + .to_str()? + .starts_with("AgentAssertion proof-") + ); + assert!(!request.headers.contains_key("x-openai-agent-assertion")); + } + let result = client + .call_tool( + "echo".to_string(), + Some(json!({"message": "after reconnect"})), + |_| Ok(None), + Some(Duration::from_secs(5)), + ) + .await?; + assert_eq!( + result.structured_content, + Some(json!({"echo": "after reconnect"})) + ); + client.shutdown().await; + Ok(()) +} diff --git a/codex-rs/tui/src/debug_config.rs b/codex-rs/tui/src/debug_config.rs index 73927c5545..7334869b00 100644 --- a/codex-rs/tui/src/debug_config.rs +++ b/codex-rs/tui/src/debug_config.rs @@ -920,6 +920,7 @@ interrupt_message = false feature_requirements: Some(Sourced::new( FeatureRequirementsToml { entries: BTreeMap::from([("guardian_approval".to_string(), true)]), + ..Default::default() }, RequirementSource::LegacyManagedConfigTomlFromMdm, )), @@ -1000,6 +1001,7 @@ interrupt_message = false guardian_extra_policy: None, feature_requirements: Some(FeatureRequirementsToml { entries: BTreeMap::from([("guardian_approval".to_string(), true)]), + ..Default::default() }), hooks: None, mcp_servers: Some(BTreeMap::from([(