From b313cca1cfe06d3b2c2af37daa6935175673318e Mon Sep 17 00:00:00 2001 From: jamy-OAI <332197178+jamy-OAI@users.noreply.github.com> Date: Thu, 5 Nov 2026 22:32:25 +0000 Subject: [PATCH] Enforce host-required cyber access programs per thread (#59754) ## What changed - Expose `RequiredCyberAccessPrograms` so hosts can require an explicit allowed program on a thread. Snapshot the requirement before extension callbacks; hosts must reapply it on cold resume. - Validate resolved `access_programs` before HTTP and WebSocket model requests, rejecting omitted or disallowed programs. Explicit `Standard` remains distinct from omission. Reject memory summarization and realtime requests when a requirement is present. - Keep required programs scoped to the parent thread by preventing Guardian reviewers from inheriting its program, including Guardian v2 reviews. ## Testing Add HTTP and WebSocket coverage for allowed programs, omission, disallowed programs, unsupported providers, and connection reuse. Extend Guardian tests to verify parent-only program scope and that removing the public attachment does not erase the startup snapshot. GitOrigin-RevId: 9306a9af0e03e7a1728dfeb610cd7ec7517367fd --- codex-rs/core-api/src/lib.rs | 1 + codex-rs/core/src/client.rs | 29 ++ codex-rs/core/src/cyber_access_program.rs | 21 ++ codex-rs/core/src/guardian/review_session.rs | 7 +- codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/realtime_conversation.rs | 3 + codex-rs/core/src/session/session.rs | 10 + .../guardian_cyber_access_program_tests.rs | 51 ++- codex-rs/core/tests/suite/mod.rs | 2 + .../suite/required_access_program_tests.rs | 334 ++++++++++++++++++ .../src/async_scorer/extension_tests.rs | 17 +- .../guardian-v2/src/async_scorer/sampler.rs | 6 +- .../src/async_scorer/sampler_tests.rs | 3 + .../guardian-v2/src/async_scorer/startup.rs | 3 + 14 files changed, 472 insertions(+), 16 deletions(-) create mode 100644 codex-rs/core/tests/suite/required_access_program_tests.rs diff --git a/codex-rs/core-api/src/lib.rs b/codex-rs/core-api/src/lib.rs index 45613aef42..222b8f2f49 100644 --- a/codex-rs/core-api/src/lib.rs +++ b/codex-rs/core-api/src/lib.rs @@ -52,6 +52,7 @@ pub use codex_core::NotSubmittedReason; pub use codex_core::PromptCacheKey; pub use codex_core::PromptCacheTtl; pub use codex_core::RecoverTurnRequest; +pub use codex_core::RequiredCyberAccessPrograms; pub use codex_core::StartIfIdleSubmission; pub use codex_core::StartThreadOptions; pub use codex_core::StartupModels; diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 9d13e2eb8d..86c41c9617 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -310,6 +310,7 @@ pub struct ModelClient { executed_tool_calls: Option, // Resolved once when the session is created, like other session feature flags. api_key_cyber_access_programs: cyber_access_program::ApiKeyCyberAccessPrograms, + required_cyber_access_programs: Option>, } /// A turn-scoped streaming session created from a [`ModelClient`]. @@ -656,6 +657,7 @@ impl ModelClient { executed_tool_calls: None, api_key_cyber_access_programs: cyber_access_program::ApiKeyCyberAccessPrograms::UnsupportedProvider, + required_cyber_access_programs: None, } } @@ -682,15 +684,36 @@ impl ModelClient { event_sender: Sender, codex_responses_headers: Option>, api_key_cyber_access_programs: cyber_access_program::ApiKeyCyberAccessPrograms, + required_cyber_access_programs: Option< + Arc, + >, ) -> Self { self.prompt_cache_key_policy = prompt_cache_key; self.session_id_override = session_id_override; self.event_sender = Some(event_sender); self.codex_responses_headers = codex_responses_headers; self.api_key_cyber_access_programs = api_key_cyber_access_programs; + self.required_cyber_access_programs = required_cyber_access_programs; self } + pub(crate) fn validate_required_cyber_access_program( + &self, + programs: Option, + ) -> Result<()> { + if let Some(required) = &self.required_cyber_access_programs + && !required + .allowed + .iter() + .any(|program| programs == Some((*program).into())) + { + return Err(CodexErr::InvalidRequest( + "this thread requires an allowed explicit access program".into(), + )); + } + Ok(()) + } + pub(crate) fn prompt_cache_key_policy(&self) -> PromptCacheKey { self.prompt_cache_key_policy.clone() } @@ -863,6 +886,7 @@ impl ModelClient { if raw_memories.is_empty() { return Ok(Vec::new()); } + self.validate_required_cyber_access_program(/*programs*/ None)?; let client_setup = self .current_client_setup(ClientRouting::ConfiguredProvider) @@ -1200,6 +1224,7 @@ impl ModelClient { // Realtime keeps provider auth until its runtime integration lands later in the stack. async fn realtime_client_setup(&self) -> Result { + self.validate_required_cyber_access_program(/*programs*/ None)?; let provider = if self.state.provider.runtime_auth().is_some() { create_model_provider( self.state.provider.info().clone(), @@ -1916,6 +1941,8 @@ impl ResponsesHttpExecutor<'_> { prompt.cyber_access_program, self.client.api_key_cyber_access_programs, )?; + self.client + .validate_required_cyber_access_program(request.access_programs)?; self.client .prepare_response_items_for_request(&mut request.input); let request_session_telemetry = @@ -2076,6 +2103,8 @@ impl ModelClientSession { prompt.cyber_access_program, self.client.api_key_cyber_access_programs, )?; + self.client + .validate_required_cyber_access_program(request.access_programs)?; let mut websocket_metadata = responses_metadata.clone(); websocket_metadata.routing_hint = self.client.build_routing_hint_header( client_setup.auth.as_ref(), diff --git a/codex-rs/core/src/cyber_access_program.rs b/codex-rs/core/src/cyber_access_program.rs index 6e6eb421f7..409ccc0e40 100644 --- a/codex-rs/core/src/cyber_access_program.rs +++ b/codex-rs/core/src/cyber_access_program.rs @@ -6,6 +6,27 @@ use codex_login::CodexAuth; use codex_protocol::error::CodexErr; use codex_protocol::error::Result; use codex_protocol::turn_input::CyberAccessProgram; +use std::sync::Arc; + +/// Host requirement for every model request issued by this thread. +/// +/// Install in thread extension data before startup and reapply on cold resume. +/// Core snapshots it before extension callbacks; it is not inherited or persisted. +/// Every request must resolve to one listed program; backend authorization remains authoritative. +#[derive(Clone, Debug)] +pub struct RequiredCyberAccessPrograms { + pub allowed: Vec, +} + +pub(crate) struct RequiredCyberAccessProgramsSnapshot(pub Arc); + +impl RequiredCyberAccessPrograms { + /// Read Core's startup snapshot, independent of later edits to the public attachment. + pub fn for_thread(data: &codex_extension_api::ExtensionData) -> Option> { + data.get::() + .map(|snapshot| Arc::clone(&snapshot.0)) + } +} /// Configured support for explicit Cyber programs on API-key requests. #[derive(Clone, Copy, Debug)] diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index cf99c7a539..c72fd9447d 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -588,7 +588,12 @@ async fn run_review_on_session( personality: params.personality, model: review_model.model.clone(), reasoning_effort: review_model.reasoning_effort.clone(), - cyber_access_program: parent_turn.cyber_access_program, + cyber_access_program: parent_turn.cyber_access_program.filter(|_| { + crate::RequiredCyberAccessPrograms::for_thread( + ¶ms.parent_session.services.thread_extension_data, + ) + .is_none() + }), parent_response_id: params .parent_context .parent_response diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index b18d77ac22..0df7afaefc 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -93,6 +93,7 @@ pub mod context; mod context_manager; mod current_time; pub mod cyber_access_program; +pub use cyber_access_program::RequiredCyberAccessPrograms; mod elicitation; mod environment_selection; pub mod exec; diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index a139b0e96f..e92fa13ebb 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -1396,6 +1396,9 @@ async fn prepare_realtime_start( sess: &Arc, params: ConversationStartParams, ) -> CodexResult { + sess.services + .model_client + .validate_required_cyber_access_program(/*programs*/ None)?; let provider = sess.provider().await; let auth_manager = sess .services diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 275f9b4b48..be85f1fb3c 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1012,6 +1012,8 @@ impl Session { )); } }; + let required_cyber_access_programs = + thread_extension_init.get::(); let inherited_session_id = thread_extension_init .remove::() .map(|inherited| inherited.0); @@ -1150,6 +1152,13 @@ impl Session { thread_id.to_string(), thread_extension_init, ); + if let Some(required) = &required_cyber_access_programs { + thread_extension_data.insert( + crate::cyber_access_program::RequiredCyberAccessProgramsSnapshot(Arc::clone( + required, + )), + ); + } thread_extension_data.insert(crate::context::GuardianReviewEvidence::default()); crate::model_search::SearchOwner::restore(&thread_extension_data, &initial_history); // Kick off independent async setup tasks in parallel to reduce startup latency. @@ -1924,6 +1933,7 @@ impl Session { tx_event.clone(), codex_responses_headers, crate::cyber_access_program::ApiKeyCyberAccessPrograms::from_config(&config), + required_cyber_access_programs, ); // Once tracing starts, startup cleanup must own the session before the next await. let rollout_thread_trace = if matches!( diff --git a/codex-rs/core/tests/suite/guardian_cyber_access_program_tests.rs b/codex-rs/core/tests/suite/guardian_cyber_access_program_tests.rs index f8d5d88556..cc54d9e183 100644 --- a/codex-rs/core/tests/suite/guardian_cyber_access_program_tests.rs +++ b/codex-rs/core/tests/suite/guardian_cyber_access_program_tests.rs @@ -1,4 +1,4 @@ -//! Verifies that reused Guardian reviewers use Blue for Daybreak without changing parent programs. +//! Verifies reviewer program selection and parent-only enrollment across reviewer reuse. use anyhow::Result; use codex_core::TurnInputRequest; @@ -21,11 +21,14 @@ use pretty_assertions::assert_eq; use serde_json::json; use test_case::test_case; -#[test_case(CodexAuth::from_api_key("test-api-key"); "api_key")] -#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(); "chatgpt")] +#[test_case(CodexAuth::from_api_key("test-api-key"), false; "api_key")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), false; "chatgpt")] +#[test_case(CodexAuth::from_api_key("test-api-key"), true; "api_key_parent_only")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), true; "chatgpt_parent_only")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn guardian_reused_reviewer_uses_blue_without_changing_parent_cyber_access_program( +async fn guardian_reused_reviewer_respects_parent_program_scope( auth: CodexAuth, + parent_only: bool, ) -> Result<()> { skip_if_no_network!(Ok(())); skip_if_wine_exec!( @@ -34,7 +37,7 @@ async fn guardian_reused_reviewer_uses_blue_without_changing_parent_cyber_access ); let server = responses::start_mock_server().await; - let test = test_codex() + let mut test = test_codex() .with_auth(auth) .with_model_info_override("gpt-6.1-sol", |_| {}) .with_model_info_override("gpt-5.5", |model| { @@ -54,8 +57,25 @@ async fn guardian_reused_reviewer_uses_blue_without_changing_parent_cyber_access }) .build_with_auto_env(&server) .await?; + if parent_only { + let mut options = test.start_thread_options().await; + options + .thread_extension_init + .insert(codex_core::RequiredCyberAccessPrograms { + allowed: vec![ + CyberAccessProgram::DaybreakBlue, + CyberAccessProgram::DaybreakRed, + CyberAccessProgram::Standard, + ], + }); + test.codex.shutdown_and_wait().await?; + test.codex = test.thread_manager.start_thread(options).await?.thread; + test.codex + .thread_extension_data() + .remove::(); + } - let cases = [ + let mut cases = vec![ ( Some(CyberAccessProgram::DaybreakBlue), Some("daybreak_blue"), @@ -64,6 +84,9 @@ async fn guardian_reused_reviewer_uses_blue_without_changing_parent_cyber_access (Some(CyberAccessProgram::Standard), Some("standard")), (None, None), ]; + if parent_only { + cases.pop(); + } let mut events = Vec::new(); for (index, _) in cases.iter().enumerate() { events.push(responses::sse(vec![ @@ -131,12 +154,16 @@ async fn guardian_reused_reviewer_uses_blue_without_changing_parent_cyber_access .iter() .map(|(_, name)| name.map(|name| json!({ "cyber": name }))) .collect::>(); - let expected_guardian_programs = [ - Some(json!({ "cyber": "daybreak_blue" })), - Some(json!({ "cyber": "daybreak_blue" })), - Some(json!({ "cyber": "standard" })), - None, - ]; + let expected_guardian_programs = if parent_only { + vec![None; cases.len()] + } else { + vec![ + Some(json!({ "cyber": "daybreak_blue" })), + Some(json!({ "cyber": "daybreak_blue" })), + Some(json!({ "cyber": "standard" })), + None, + ] + }; assert_eq!( guardian_requests .iter() diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 687a49d97d..59eb54f175 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -199,6 +199,8 @@ mod request_permissions_tool; mod request_plugin_install; mod request_user_input; mod request_user_input_async; +#[path = "required_access_program_tests.rs"] +mod required_access_program; mod responses_api_proxy_headers; mod responses_lite; #[cfg(target_os = "linux")] diff --git a/codex-rs/core/tests/suite/required_access_program_tests.rs b/codex-rs/core/tests/suite/required_access_program_tests.rs new file mode 100644 index 0000000000..ad18a0728f --- /dev/null +++ b/codex-rs/core/tests/suite/required_access_program_tests.rs @@ -0,0 +1,334 @@ +//! Selected threads require an allowed program on every HTTP and WebSocket model request. + +use anyhow::Result; +use codex_core::RequiredCyberAccessPrograms; +use codex_core::TurnInputRequest; +use codex_core::TurnStartOptions; +use codex_login::AuthHeaders; +use codex_login::CodexAuth; +use codex_protocol::protocol::EventMsg; +use codex_protocol::turn_input::CyberAccessProgram; +use codex_protocol::user_input::UserInput; +use core_test_support::responses; +use core_test_support::test_codex::TestCodex; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use http::HeaderMap; +use http::HeaderValue; +use pretty_assertions::assert_eq; +use serde_json::json; +use test_case::test_case; + +const ACTOR: &str = "x-openai-actor-authorization"; +const ACCOUNT: &str = "chatgpt-account-id"; + +fn actor_headers() -> HeaderMap { + HeaderMap::from_iter([ + ( + http::HeaderName::from_static(ACTOR), + HeaderValue::from_static("verified-actor"), + ), + ( + http::HeaderName::from_static(ACCOUNT), + HeaderValue::from_static("workspace-1"), + ), + ]) +} + +fn actor_auth() -> CodexAuth { + CodexAuth::Headers(AuthHeaders::new(actor_headers())) +} + +async fn submit(test: &TestCodex, program: Option) -> Result<()> { + test.codex + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "hello".to_owned(), + text_elements: Vec::new(), + }]) + .on_start(TurnStartOptions { + cyber_access_program: program, + ..Default::default() + }), + ) + .await?; + let event = wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_) | EventMsg::Error(_)) + }) + .await; + if let EventMsg::Error(error) = event { + anyhow::bail!("actor program request failed: {error:?}"); + } + Ok(()) +} + +/// A Blue/Red startup requirement rejects Standard, omission, and unsupported providers. +#[test_case(None, true, None; "omitted")] +#[test_case(Some(CyberAccessProgram::Standard), true, None; "standard_rejected")] +#[test_case(Some(CyberAccessProgram::DaybreakBlue), true, Some("daybreak_blue"); "blue")] +#[test_case(Some(CyberAccessProgram::DaybreakRed), true, Some("daybreak_red"); "red")] +#[test_case(Some(CyberAccessProgram::DaybreakBlue), false, None; "unsupported_provider")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn required_actor_program_http( + program: Option, + provider_enabled: bool, + expected_wire_program: Option<&str>, +) -> Result<()> { + core_test_support::skip_if_no_network!(Ok(())); + let server = responses::start_mock_server().await; + let requests = responses::mount_sse_once(&server, responses::sse_completed("resp-1")).await; + let test = test_codex() + .with_auth(actor_auth()) + .with_config(move |config| { + config.model_provider_id = "local".to_owned(); + config + .model_provider + .capabilities + .get_or_insert_default() + .cyber_access_programs = Some(provider_enabled); + }) + .build_with_auto_env(&server) + .await?; + let mut options = test.start_thread_options().await; + options + .thread_extension_init + .insert(RequiredCyberAccessPrograms { + allowed: vec![ + CyberAccessProgram::DaybreakBlue, + CyberAccessProgram::DaybreakRed, + ], + }); + let thread = test.thread_manager.start_thread(options).await?.thread; + thread + .thread_extension_data() + .remove::(); + thread + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "hello".into(), + text_elements: Vec::new(), + }]) + .on_start(TurnStartOptions { + cyber_access_program: program, + ..Default::default() + }), + ) + .await?; + let event = wait_for_event(&thread, |event| { + matches!(event, EventMsg::TurnComplete(_) | EventMsg::Error(_)) + }) + .await; + if let Some(expected_wire_program) = expected_wire_program { + assert!(matches!(event, EventMsg::TurnComplete(_))); + assert_eq!( + requests + .single_request() + .body_json() + .get("access_programs") + .cloned(), + Some(json!({"cyber": expected_wire_program})) + ); + } else { + let EventMsg::Error(error) = event else { + panic!("host requirement must refuse request") + }; + assert!( + error + .message + .contains("this thread requires an allowed explicit access program") + ); + assert!(requests.requests().is_empty()); + } + thread.shutdown_and_wait().await?; + test.codex.shutdown_and_wait().await?; + Ok(()) +} + +/// A Blue/Red requirement survives WebSocket reuse and rejects Standard or omission. +#[test_case(CyberAccessProgram::DaybreakBlue, "daybreak_blue", Some(CyberAccessProgram::Standard); "blue_then_standard")] +#[test_case(CyberAccessProgram::DaybreakRed, "daybreak_red", Some(CyberAccessProgram::Standard); "red_then_standard")] +#[test_case(CyberAccessProgram::DaybreakBlue, "daybreak_blue", None; "blue_then_omitted")] +#[test_case(CyberAccessProgram::DaybreakRed, "daybreak_red", None; "red_then_omitted")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn required_actor_program_websocket( + program: CyberAccessProgram, + expected_wire_program: &str, + rejected_program: Option, +) -> Result<()> { + core_test_support::skip_if_no_network!(Ok(())); + let server = responses::start_websocket_server(vec![ + vec![vec![responses::ev_completed("resp-warmup")]], + vec![vec![responses::ev_completed("resp-allowed")]], + ]) + .await; + let base_url = format!("{}/v1", server.uri().replacen("ws://", "http://", 1)); + let mut test = test_codex() + .with_auth(actor_auth()) + .with_config(move |config| { + config.model_provider_id = "local".to_owned(); + config + .model_provider + .capabilities + .get_or_insert_default() + .cyber_access_programs = Some(true); + config.model_provider.base_url = Some(base_url); + }) + .build_with_websocket_server(&server) + .await?; + server + .wait_for_request(/*connection_index*/ 0, /*request_index*/ 0) + .await; + let mut options = test.start_thread_options().await; + options + .thread_extension_init + .insert(RequiredCyberAccessPrograms { + allowed: vec![ + CyberAccessProgram::DaybreakBlue, + CyberAccessProgram::DaybreakRed, + ], + }); + test.codex.shutdown_and_wait().await?; + test.codex = test.thread_manager.start_thread(options).await?.thread; + submit(&test, Some(program)).await?; + let error = submit(&test, rejected_program) + .await + .expect_err("Standard or omission must not satisfy a Blue/Red requirement"); + assert!( + error + .to_string() + .contains("this thread requires an allowed explicit access program") + ); + assert_eq!( + server + .connections() + .into_iter() + .flatten() + .map(|request| request.body_json().get("access_programs").cloned()) + .collect::>(), + vec![None, Some(json!({"cyber": expected_wire_program}))] + ); + test.codex.shutdown_and_wait().await?; + server.shutdown().await; + Ok(()) +} + +/// A Standard-only host requirement accepts explicit Standard, never omission. +#[test_case(Some(CyberAccessProgram::Standard), Some("standard"); "standard")] +#[test_case(None, None; "omitted")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn required_actor_program_http_standard_only( + program: Option, + expected_wire_program: Option<&str>, +) -> Result<()> { + core_test_support::skip_if_no_network!(Ok(())); + let server = responses::start_mock_server().await; + let requests = responses::mount_sse_once(&server, responses::sse_completed("resp-1")).await; + let mut test = test_codex() + .with_auth(actor_auth()) + .with_config(|config| { + config.model_provider_id = "local".to_owned(); + config + .model_provider + .capabilities + .get_or_insert_default() + .cyber_access_programs = Some(true); + }) + .build_with_auto_env(&server) + .await?; + let mut options = test.start_thread_options().await; + options + .thread_extension_init + .insert(RequiredCyberAccessPrograms { + allowed: vec![CyberAccessProgram::Standard], + }); + test.codex.shutdown_and_wait().await?; + test.codex = test.thread_manager.start_thread(options).await?.thread; + let result = submit(&test, program).await; + if let Some(expected_wire_program) = expected_wire_program { + result?; + assert_eq!( + requests + .single_request() + .body_json() + .get("access_programs") + .cloned(), + Some(json!({"cyber": expected_wire_program})) + ); + } else { + let error = result.expect_err("omission must not silently select Standard"); + assert!( + error + .to_string() + .contains("this thread requires an allowed explicit access program") + ); + assert!(requests.requests().is_empty()); + } + test.codex.shutdown_and_wait().await?; + Ok(()) +} + +/// A Standard-only WebSocket requirement also distinguishes Standard from omission. +#[test_case(Some(CyberAccessProgram::Standard), Some("standard"); "standard")] +#[test_case(None, None; "omitted")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn required_actor_program_websocket_standard_only( + program: Option, + expected_wire_program: Option<&str>, +) -> Result<()> { + core_test_support::skip_if_no_network!(Ok(())); + let server = responses::start_websocket_server(vec![ + vec![vec![responses::ev_completed("resp-warmup")]], + vec![vec![responses::ev_completed("resp-allowed")]], + ]) + .await; + let base_url = format!("{}/v1", server.uri().replacen("ws://", "http://", 1)); + let mut test = test_codex() + .with_auth(actor_auth()) + .with_config(move |config| { + config.model_provider_id = "local".to_owned(); + config + .model_provider + .capabilities + .get_or_insert_default() + .cyber_access_programs = Some(true); + config.model_provider.base_url = Some(base_url); + }) + .build_with_websocket_server(&server) + .await?; + server + .wait_for_request(/*connection_index*/ 0, /*request_index*/ 0) + .await; + let mut options = test.start_thread_options().await; + options + .thread_extension_init + .insert(RequiredCyberAccessPrograms { + allowed: vec![CyberAccessProgram::Standard], + }); + test.codex.shutdown_and_wait().await?; + test.codex = test.thread_manager.start_thread(options).await?.thread; + let result = submit(&test, program).await; + let mut expected_programs = vec![None]; + if let Some(expected_wire_program) = expected_wire_program { + result?; + expected_programs.push(Some(json!({"cyber": expected_wire_program}))); + } else { + let error = result.expect_err("omission must not silently select Standard"); + assert!( + error + .to_string() + .contains("this thread requires an allowed explicit access program") + ); + } + assert_eq!( + server + .connections() + .into_iter() + .flatten() + .map(|request| request.body_json().get("access_programs").cloned()) + .collect::>(), + expected_programs + ); + test.codex.shutdown_and_wait().await?; + server.shutdown().await; + Ok(()) +} diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs index ff16ff40ca..d08a3453f1 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs @@ -264,7 +264,8 @@ async fn installed_extension_warms_connections_without_blocking_thread_start() - } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Result<()> { +async fn installed_extension_keeps_parent_program_out_of_websocket_and_http_reviews() -> Result<()> +{ skip_if_no_network!(Ok(())); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("original")); @@ -272,11 +273,19 @@ async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Resul .set_external_auth(Arc::new(RefreshableAuth(std::sync::Mutex::new("original")))) .await?; let thread_server = responses::start_mock_server().await; - let test = test_codex() + let mut test = test_codex() .with_auth_manager(Arc::clone(&auth_manager)) .with_config(|config| config.approvals_reviewer = ApprovalsReviewer::AutoReview) .build_with_auto_env(&thread_server) .await?; + let mut options = test.start_thread_options().await; + options + .thread_extension_init + .insert(codex_core::RequiredCyberAccessPrograms { + allowed: vec![codex_protocol::turn_input::CyberAccessProgram::DaybreakRed], + }); + test.codex.shutdown_and_wait().await?; + test.codex = test.thread_manager.start_thread(options).await?.thread; let events = vec![ ev_assistant_message("sample", "low"), ev_completed("response-1"), @@ -307,6 +316,8 @@ async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Resul let registry = builder.build(); let session_store = ExtensionData::new("session-1"); let thread_store = test.codex.thread_extension_data(); + // An earlier extension cannot erase Core's captured enrollment policy. + thread_store.remove::(); let mut model = test .thread_manager .get_models_manager() @@ -337,6 +348,7 @@ async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Resul .expect("Guardian v2 should initialize"); let turn_store = ExtensionData::new("turn-1"); turn_store.insert_arc(thread_store.get::().expect("captured test config")); + turn_store.insert(codex_protocol::turn_input::CyberAccessProgram::DaybreakRed); let tool_name = ToolName::namespaced("mcp__node_repl__", "js"); let payload = ToolPayload::Function { arguments: r#"{"path":"README.md"}"#.to_owned(), @@ -406,6 +418,7 @@ async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Resul assert_eq!(requests.len(), 1); requests.push(http_request.body_json()); for request in &requests { + assert!(request.get("access_programs").is_none()); responses::assert_parent_turn(request, Some("turn-1"))?; responses::assert_root_turn(request, /*expected*/ None)?; } diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs index 505315f22f..fc10f402ba 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs @@ -48,6 +48,8 @@ pub struct LunaSamplerConfig { pub provider: SharedModelProvider, /// Provider and feature policy shared with ordinary model requests. pub api_key_cyber_access_programs: ApiKeyCyberAccessPrograms, + /// False when the parent host confines its explicit program to that thread. + pub inherit_parent_cyber_access_program: bool, /// Routing scope and retained configuration layers for the owning thread. pub workspace_routing: WorkspaceRoutingContext, /// Effective proxy, custom-CA, and cookie configuration. @@ -253,7 +255,9 @@ impl LunaSampler { execution::SamplingExecution { auth_owner_generation, cyber_access_program: codex_guardian_reviewer::guardian_cyber_access_program( - request.cyber_access_program, + request + .cyber_access_program + .filter(|_| self.config.inherit_parent_cyber_access_program), ), config: Arc::clone(&self.config), connections: Arc::clone(&self.connections), diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs index ff1069f077..16d761c275 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs @@ -242,6 +242,7 @@ pub(in crate::async_scorer) async fn proxy_websocket_servers_with_http( pub(in crate::async_scorer) fn sampler_config(base_url: String) -> LunaSamplerConfig { LunaSamplerConfig { api_key_cyber_access_programs: ApiKeyCyberAccessPrograms::Disabled, + inherit_parent_cyber_access_program: true, workspace_routing: codex_model_provider::WorkspaceRoutingContext::new( "https://chatgpt.com/backend-api".into(), ), @@ -658,6 +659,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications let sampler = connect_sampler(LunaSamplerConfig { api_key_cyber_access_programs: ApiKeyCyberAccessPrograms::Disabled, + inherit_parent_cyber_access_program: true, workspace_routing: codex_model_provider::WorkspaceRoutingContext::new( "https://chatgpt.com/backend-api".into(), ), @@ -840,6 +842,7 @@ async fn sampler_returns_classification_token_before_terminal_response_events() ); let sampler = connect_sampler(LunaSamplerConfig { api_key_cyber_access_programs: ApiKeyCyberAccessPrograms::Disabled, + inherit_parent_cyber_access_program: true, workspace_routing: codex_model_provider::WorkspaceRoutingContext::new( "https://chatgpt.com/backend-api".into(), ), diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/startup.rs b/codex-rs/ext/guardian-v2/src/async_scorer/startup.rs index 5027026e45..b774caa402 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/startup.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/startup.rs @@ -79,6 +79,8 @@ pub(super) async fn sampler_config( input: &ThreadStartInput<'_, Config>, manager: Option>, ) -> LunaSamplerConfig { + let inherit_parent_cyber_access_program = + codex_core::RequiredCyberAccessPrograms::for_thread(input.thread_store).is_none(); let luna_model = if let Some(thread_manager) = manager { let mut model_config = input.config.to_models_manager_config(); model_config.model_context_window = None; @@ -107,6 +109,7 @@ pub(super) async fn sampler_config( .expect("Core installs provider auth before starting Guardian"); LunaSamplerConfig { api_key_cyber_access_programs: ApiKeyCyberAccessPrograms::from_config(input.config), + inherit_parent_cyber_access_program, workspace_routing: input .thread_store .get_or_init(|| input.config.workspace_routing_context())