From 80f469ce8aeadcdb0b79b1a1daf6b88d92ff8799 Mon Sep 17 00:00:00 2001 From: Charlie Marsh Date: Fri, 30 Oct 2026 14:29:59 +0000 Subject: [PATCH] Serialize manual network approval prompts per endpoint (#58351) ## Why Concurrent executions accessing the same endpoint can produce duplicate network approval prompts before a session decision is recorded. ## What changed - Queue manual prompts by environment, host, protocol, and port while keeping execution decisions separate. - After a one-time approval or denial, let the next execution prompt. Reuse committed session grants and deny rules without another prompt. - Keep tool results paused while requests are queued and attribute the wait to the triggering tool call. ## Testing Extend coverage for concurrent requests, one-time and session decisions, independent ports, queued cancellation and timing, and isolation between local and remote environments. GitOrigin-RevId: ebf580dddc39a9a9dc3b0d1acfa21438fbc6e100 --- codex-rs/core/src/session/mod.rs | 16 + codex-rs/core/src/tools/approvals.rs | 54 ++- codex-rs/core/src/tools/approvals_tests.rs | 6 +- codex-rs/core/src/tools/network_approval.rs | 139 ++++-- .../core/src/tools/network_approval_tests.rs | 134 ++++- codex-rs/core/tests/common/tracing.rs | 43 ++ codex-rs/core/tests/suite/network_approval.rs | 457 ++++++++++-------- 7 files changed, 587 insertions(+), 262 deletions(-) diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 2ae6f480c2..3c1889b25a 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3070,6 +3070,22 @@ impl Session { timing } + /// Keep tool results paused and attribute a queued network prompt to its triggering call. + pub(crate) fn hold_queued_network_approval( + &self, + turn: &TurnContext, + call_id: Option<&str>, + ) -> ( + crate::elicitation::ElicitationRegistration, + Option, + ) { + let owner = self.user_wait_timing_owner(turn, call_id); + ( + self.services.elicitations.register(), + self.start_user_wait_timing(turn, owner.as_deref()), + ) + } + /// Emit an exec approval request event and await the user's decision. /// /// The request is keyed by `call_id` + `approval_id` so matching responses diff --git a/codex-rs/core/src/tools/approvals.rs b/codex-rs/core/src/tools/approvals.rs index 9e1cd00013..31280867b3 100644 --- a/codex-rs/core/src/tools/approvals.rs +++ b/codex-rs/core/src/tools/approvals.rs @@ -474,6 +474,14 @@ pub(crate) struct ApprovalResolution { } impl ApprovalResolution { + pub(super) fn user(decision: ReviewDecision) -> Self { + Self { + decision, + source: ApprovalResolutionSource::User, + cancellation_reason: None, + } + } + fn into_tool_result(self, model_info: &ModelInfo) -> Result { let source = self.source; match self.decision { @@ -538,7 +546,21 @@ impl Session { action: ApprovalAction, ctx: ApprovalContext, ) -> ApprovalResult { - if let Err(err) = self.check_approval_requirement(&ctx).await { + self.request_approval_with_user_prompt( + &action, + &ctx, + self.request_user_approval(&action, &ctx), + ) + .await + } + + pub(super) async fn request_approval_with_user_prompt( + self: &Arc, + action: &ApprovalAction, + ctx: &ApprovalContext, + user_prompt: impl Future, + ) -> ApprovalResult { + if let Err(err) = self.check_approval_requirement(ctx).await { return ApprovalResult { result: Err(err), source: None, @@ -548,7 +570,7 @@ impl Session { // Retained authority outside current grants needs a new sandbox approval. // Current grants permit review of the input; they never approve the input itself. let policy = ctx.review_context.turn().approval_policy(); - if matches!(&action, ApprovalAction::WriteStdin { sandbox_permissions, additional_permissions_preapproved, .. } + if matches!(action, ApprovalAction::WriteStdin { sandbox_permissions, additional_permissions_preapproved, .. } if sandbox_permissions.requests_sandbox_override() && (!additional_permissions_preapproved || matches!(policy, AskForApproval::Never))) && !(ctx.strict_auto_review && matches!(policy, AskForApproval::Never)) @@ -561,9 +583,9 @@ impl Session { cancellation_reason: None, }; } - let is_mcp_tool_call = matches!(&action, ApprovalAction::McpToolCall { .. }); - let is_network_approval = matches!(&action, ApprovalAction::NetworkAccess { .. }); - let permission_request_run_id = match &action { + let is_mcp_tool_call = matches!(action, ApprovalAction::McpToolCall { .. }); + let is_network_approval = matches!(action, ApprovalAction::NetworkAccess { .. }); + let permission_request_run_id = match action { ApprovalAction::NetworkAccess { hook_run_id, .. } => hook_run_id.clone(), _ if ctx.retry_reason.is_some() => format!("{}:retry", ctx.call_id), _ => ctx.call_id.clone(), @@ -593,11 +615,12 @@ impl Session { cancellation_reason: None, }, Some(PermissionRequestDecision::Allow) | None => { - self.request_reviewer_approval(action, &ctx).await + self.request_reviewer_approval(action, ctx, user_prompt) + .await } }; if resolution.decision == ReviewDecision::Approved { - if let Err(ToolError::Rejected(reason)) = self.check_approval_requirement(&ctx).await { + if let Err(ToolError::Rejected(reason)) = self.check_approval_requirement(ctx).await { resolution.decision = ReviewDecision::denied(reason); } if ctx.requires_human_approval() @@ -611,7 +634,7 @@ impl Session { } // Network approvals record their final telemetry after validation and persistence. if !is_network_approval { - record_resolution(&ctx, &resolution); + record_resolution(ctx, &resolution); } let source = Some(resolution.source); let cancellation_reason = resolution.cancellation_reason; @@ -657,11 +680,12 @@ impl Session { async fn request_reviewer_approval( self: &Arc, - action: ApprovalAction, + action: &ApprovalAction, ctx: &ApprovalContext, + user_prompt: impl Future, ) -> ApprovalResolution { if ctx.requires_human_approval() { - self.request_user_approval(&action, ctx).await + user_prompt.await } else if let Some(decision) = self.request_guardian_approval(action.clone(), ctx).await { ApprovalResolution { cancellation_reason: (decision == ReviewDecision::Abort) @@ -670,7 +694,7 @@ impl Session { source: ApprovalResolutionSource::Guardian, } } else { - self.request_user_approval(&action, ctx).await + user_prompt.await } } @@ -754,7 +778,7 @@ impl Session { } } - async fn request_user_approval( + pub(super) async fn request_user_approval( &self, action: &ApprovalAction, ctx: &ApprovalContext, @@ -948,11 +972,7 @@ impl Session { unreachable!("permission requests are routed directly to Guardian") } }; - ApprovalResolution { - decision, - source: ApprovalResolutionSource::User, - cancellation_reason: None, - } + ApprovalResolution::user(decision) } } diff --git a/codex-rs/core/src/tools/approvals_tests.rs b/codex-rs/core/src/tools/approvals_tests.rs index 12c49cff64..7a87b9f90d 100644 --- a/codex-rs/core/src/tools/approvals_tests.rs +++ b/codex-rs/core/src/tools/approvals_tests.rs @@ -154,7 +154,8 @@ async fn non_utf8_cwd_preserves_approval_routing( tty: false, proposed_execpolicy_amendment: None, }; - let approval = session.request_reviewer_approval(action, &context); + let prompt = session.request_user_approval(&action, &context); + let approval = session.request_reviewer_approval(&action, &context, prompt); tokio::pin!(approval); let expected = if reviewer == ApprovalsReviewer::User { tokio::select! { @@ -225,8 +226,9 @@ async fn explicit_mcp_reviewer_override_takes_precedence_over_action_context() { network_approval_context: None, }; + let prompt = session.request_user_approval(&action, &context); tokio::select! { - resolution = session.request_reviewer_approval(action, &context) => { + resolution = session.request_reviewer_approval(&action, &context, prompt) => { panic!("expected a user approval request, got {resolution:?}"); } event = events.recv() => { diff --git a/codex-rs/core/src/tools/network_approval.rs b/codex-rs/core/src/tools/network_approval.rs index 4b3a5b5aba..e24db1a780 100644 --- a/codex-rs/core/src/tools/network_approval.rs +++ b/codex-rs/core/src/tools/network_approval.rs @@ -8,6 +8,7 @@ use crate::session::turn_context::TurnContext; use crate::session::turn_context::TurnEnvironment; use crate::tools::approvals::ApprovalAction; use crate::tools::approvals::ApprovalContext; +use crate::tools::approvals::ApprovalResolution; use crate::tools::events::truncate_rejection_message; use crate::tools::sandboxing::ToolError; use codex_features::Feature; @@ -47,12 +48,15 @@ use std::sync::OnceLock as SyncOnceLock; use tokio::sync::Mutex; use tokio::sync::Notify; use tokio::sync::OnceCell; +use tokio::sync::OwnedSemaphorePermit; use tokio::sync::RwLock; +use tokio::sync::Semaphore; use tokio_util::sync::CancellationToken; use tracing::error; use tracing::warn; use uuid::Uuid; +const REASON_NOT_ALLOWED: &str = "not_allowed"; const ABANDONED_NETWORK_APPROVAL_MESSAGE: &str = "network approval was cancelled before a decision was returned"; @@ -204,7 +208,7 @@ impl PendingApprovalDecision { fn to_network_decision(self) -> NetworkDecision { match self { Self::AllowOnce | Self::AllowForSession => NetworkDecision::Allow, - Self::Deny => NetworkDecision::deny("not_allowed"), + Self::Deny => NetworkDecision::deny(REASON_NOT_ALLOWED), } } } @@ -268,11 +272,17 @@ struct NetworkApprovalCallState { call_outcomes: HashMap, } +#[derive(Default)] +struct PendingNetworkApprovals { + executions: HashMap>, + manual_prompts: HashMap>, +} + pub(crate) struct NetworkApprovalService { // Pending-request cleanup must publish a tool outcome synchronously before cancelling it. calls: SyncMutex, // Owner cleanup runs from Drop, so this lock cannot require an async task. - pending_host_approvals: SyncMutex>>, + pending_host_approvals: SyncMutex, // Keep persisted session policy and the in-memory approval caches in the same order. session_policy_commit_lock: Mutex<()>, session_approved_hosts: Mutex>, @@ -291,6 +301,7 @@ struct PendingHostApprovalOwner<'a> { disconnect: Option, cancellation: Option, decision_on_drop: PendingApprovalDecision, + prompt_permit: Option, completed: bool, } @@ -309,10 +320,62 @@ impl<'a> PendingHostApprovalOwner<'a> { disconnect: None, cancellation: None, decision_on_drop: PendingApprovalDecision::Deny, + prompt_permit: None, completed: false, } } + /// Keep the endpoint's prompt until this execution commits or abandons its decision. + async fn review_user_prompt( + &mut self, + session: &Session, + turn: &TurnContext, + timing_call_id: Option<&str>, + policy_denial: &str, + prompt: impl Future, + ) -> ApprovalResolution { + let queued = { + let mut approvals = self + .service + .pending_host_approvals + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let queue = approvals + .manual_prompts + .entry(self.key.host.clone()) + .or_insert_with(|| Arc::new(Semaphore::new(/*permits*/ 1))); + if let Ok(permit) = Arc::clone(queue).try_acquire_owned() { + self.prompt_permit = Some(permit); + None + } else { + Some(( + Arc::clone(queue), + session.hold_queued_network_approval(turn, timing_call_id), + )) + } + }; + // Keep the queue's elicitation and timing guards through any handoff to the visible prompt. + if let Some((queue, _hold)) = &queued { + tracing::debug!( + name: "codex.network_approval.queued", + execution_id = self.key.execution_id.as_deref(), + "waiting for the current endpoint approval" + ); + let Ok(permit) = Arc::clone(queue).acquire_owned().await else { + // The prompt owner closes the queue when it commits a session grant. + return ApprovalResolution::user(ReviewDecision::Approved); + }; + self.prompt_permit = Some(permit); + } + match self.service.session_host_decision(&self.key.host).await { + Some(NetworkDecision::Allow) => ApprovalResolution::user(ReviewDecision::Approved), + Some(NetworkDecision::Deny { .. }) => { + ApprovalResolution::user(ReviewDecision::denied(policy_denial)) + } + None => prompt.await, + } + } + fn set_decision_on_drop(&mut self, decision: PendingApprovalDecision) { self.decision_on_drop = decision; } @@ -331,7 +394,7 @@ impl<'a> PendingHostApprovalOwner<'a> { } } - fn publish_and_remove(&self, decision: PendingApprovalDecision) { + fn publish_and_remove(&mut self, decision: PendingApprovalDecision) { // Remove this generation before waking its waiters so a concurrent retry // cannot attach to an already-completed approval. let mut approvals = self @@ -339,11 +402,24 @@ impl<'a> PendingHostApprovalOwner<'a> { .pending_host_approvals .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(queue) = approvals.manual_prompts.get(&self.key.host) { + let permit = self.prompt_permit.take(); + if permit.is_some() && decision == PendingApprovalDecision::AllowForSession { + queue.close(); + } + // Returning the permit lets one waiter take over the prompt. + drop(permit); + // Each queued or active execution holds a reference in addition to this map. + if queue.is_closed() || Arc::strong_count(queue) == 1 { + approvals.manual_prompts.remove(&self.key.host); + } + } if approvals + .executions .get(&self.key) .is_some_and(|current| Arc::ptr_eq(current, &self.pending)) { - approvals.remove(&self.key); + approvals.executions.remove(&self.key); } drop(approvals); self.pending.set_decision(decision); @@ -386,7 +462,7 @@ impl Default for NetworkApprovalService { fn default() -> Self { Self { calls: SyncMutex::new(NetworkApprovalCallState::default()), - pending_host_approvals: SyncMutex::new(HashMap::new()), + pending_host_approvals: SyncMutex::default(), session_policy_commit_lock: Mutex::new(()), session_approved_hosts: Mutex::new(HashSet::new()), session_denied_hosts: Mutex::new(HashSet::new()), @@ -520,15 +596,31 @@ impl NetworkApprovalService { .pending_host_approvals .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if let Some(existing) = pending.get(&key).cloned() { + if let Some(existing) = pending.executions.get(&key).cloned() { return (existing, false); } let created = Arc::new(PendingHostApproval::new()); - pending.insert(key, Arc::clone(&created)); + pending.executions.insert(key, Arc::clone(&created)); (created, true) } + #[expect( + clippy::await_holding_invalid_type, + reason = "read both host caches under the session policy commit lock" + )] + async fn session_host_decision(&self, key: &HostApprovalKey) -> Option { + let _commit_guard = self.session_policy_commit_lock.lock().await; + if self.session_denied_hosts.lock().await.contains(key) { + return Some(NetworkDecision::deny(REASON_NOT_ALLOWED)); + } + self.session_approved_hosts + .lock() + .await + .contains(key) + .then_some(NetworkDecision::Allow) + } + #[cfg(test)] async fn take_call_outcome(&self, registration_id: &str) -> Option { let mut calls = self @@ -622,17 +714,11 @@ impl NetworkApprovalService { ) } - #[expect( - clippy::await_holding_invalid_type, - reason = "persisted session policy and its in-memory caches must commit atomically" - )] pub(crate) async fn handle_inline_policy_request( &self, session: Arc, request: NetworkPolicyRequest, ) -> NetworkDecision { - const REASON_NOT_ALLOWED: &str = "not_allowed"; - let protocol = match request.protocol { NetworkProtocol::Http => NetworkApprovalProtocol::Http, NetworkProtocol::HttpsConnect => NetworkApprovalProtocol::Https, @@ -663,19 +749,8 @@ impl NetworkApprovalService { }; let key = HostApprovalKey::from_request(&request, protocol, environment_id.clone()); - { - let _commit_guard = self.session_policy_commit_lock.lock().await; - { - let denied_hosts = self.session_denied_hosts.lock().await; - if denied_hosts.contains(&key) { - return NetworkDecision::deny(REASON_NOT_ALLOWED); - } - } - - let approved_hosts = self.session_approved_hosts.lock().await; - if approved_hosts.contains(&key) { - return NetworkDecision::Allow; - } + if let Some(decision) = self.session_host_decision(&key).await { + return decision; } let target = Self::format_network_target(key.protocol, request.host.as_str(), key.port); @@ -809,7 +884,17 @@ impl NetworkApprovalService { retry_reason: Some(policy_denial_message.clone()), network_approval_context: Some(network_approval_context.clone()), }; - let approval = session.request_approval(action, approval_context); + let user_prompt = pending_owner.review_user_prompt( + &session, + &turn_context, + owner_call + .as_ref() + .map(|call| call.trigger.call_id.as_str()), + &policy_denial_message, + session.request_user_approval(&action, &approval_context), + ); + let approval = + session.request_approval_with_user_prompt(&action, &approval_context, user_prompt); let approval_result = if let Some(owner_call) = &owner_call { // A ready review result takes precedence over concurrent cancellation. let Some(result) = owner_call diff --git a/codex-rs/core/src/tools/network_approval_tests.rs b/codex-rs/core/src/tools/network_approval_tests.rs index 350ac5ed38..facc4421d2 100644 --- a/codex-rs/core/src/tools/network_approval_tests.rs +++ b/codex-rs/core/src/tools/network_approval_tests.rs @@ -1,15 +1,18 @@ use super::*; use crate::sandboxing::SandboxPermissions; +use codex_direct_tool_timing::ToolBatchTiming; use codex_network_proxy::BlockedRequestArgs; use codex_protocol::models::PermissionProfile; use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::AskForApproval; +use codex_utils_critical_path::CriticalPath; use codex_utils_path_uri::PathUri; use core_test_support::PathBufExt; use core_test_support::test_path_buf; use futures::poll; use pretty_assertions::assert_eq; use std::time::Duration; +use std::time::Instant; use test_case::test_case; use tokio::time::timeout; use tokio_util::sync::CancellationToken; @@ -60,7 +63,7 @@ async fn execution_cancellation_respects_network_approval_boundary( } let (session, _, events) = make_session_and_context_with_rx().await; - let (turn, _) = session + let (mut turn, _) = session .new_turn_with_sub_id( "active-turn".to_string(), crate::session::SessionSettingsUpdate { @@ -75,9 +78,17 @@ async fn execution_cancellation_respects_network_approval_boundary( ) .await .unwrap(); + Arc::make_mut(&mut Arc::get_mut(&mut turn).expect("unique test turn").config) + .features + .enable(Feature::CriticalPathTiming) + .unwrap(); let finish_turn = CancellationToken::new(); session - .start_task(turn, Vec::new(), PendingTask(finish_turn.clone())) + .start_task( + Arc::clone(&turn), + Vec::new(), + PendingTask(finish_turn.clone()), + ) .await; let service = &session.services.network_approval; register_call_with_default_shell_trigger(service, "execution-1").await; @@ -110,6 +121,7 @@ async fn execution_cancellation_respects_network_approval_boundary( }) .await .unwrap(); + let mut queued = None; let expected_decision = match timing { ExecutionCancellationTiming::TurnFinished => { finish_turn.cancel(); @@ -127,6 +139,33 @@ async fn execution_cancellation_respects_network_approval_boundary( NetworkDecision::deny("not_allowed") } ExecutionCancellationTiming::AcceptedReview => { + let mut queue_events = core_test_support::tracing::NamedTraceEvents::subscribe( + "codex.network_approval.queued", + ); + let started = Instant::now(); + let timing = turn + .turn_timing_state + .direct_tool_timings + .start_turn_direct_call( + "execution-2", + &turn.sub_id, + ToolBatchTiming::default(), + started, + Some(CriticalPath::default()), + ); + register_call_with_default_shell_trigger(service, "execution-2").await; + let mut waiter = Box::pin(service.handle_inline_policy_request( + Arc::clone(&session), + NetworkPolicyRequest { + execution_id: Some("execution-2".to_string()), + ..request.clone() + }, + )); + tokio::select! { + result = waiter.as_mut() => panic!("expected queued request, got {result:?}"), + event = queue_events.next() => event.expect("execution joins the queue"), + } + queued = Some((waiter, timing, started)); let commit_guard = service.session_policy_commit_lock.lock().await; session .notify_approval( @@ -136,6 +175,8 @@ async fn execution_cancellation_respects_network_approval_boundary( .await; // Consume the approval and stop at the held policy commit lock. assert!(poll!(decision.as_mut()).is_pending()); + // The visible prompt closed, but the queued execution still holds tool results. + assert!(*session.subscribe_elicitation_pause_state().borrow()); service.record_call_outcome("execution-1", "original execution denial".to_string()); assert!(poll!(decision.as_mut()).is_pending()); drop(commit_guard); @@ -148,6 +189,21 @@ async fn execution_cancellation_respects_network_approval_boundary( .expect("finish cancelled or accepted review"), expected_decision, ); + if let Some((queued, timing, started)) = queued { + assert_eq!( + timeout(Duration::from_secs(5), queued).await.unwrap(), + NetworkDecision::Allow + ); + assert!(!*session.subscribe_elicitation_pause_state().borrow()); + let work = timing.finish().unwrap(); + assert!( + work.path + .unwrap() + .durations(started, work.ended_at) + .user_wait + > Duration::ZERO + ); + } let key = HostApprovalKey::from_request(&request, NetworkApprovalProtocol::Http, "local".to_string()); assert_eq!( @@ -166,7 +222,11 @@ async fn execution_cancellation_respects_network_approval_boundary( service.take_call_outcome("execution-1").await, Some("original execution denial".to_string()) ); - assert!(service.pending_host_approvals.lock().unwrap().is_empty()); + { + let pending = service.pending_host_approvals.lock().unwrap(); + assert!(pending.executions.is_empty()); + assert!(pending.manual_prompts.is_empty()); + } let mut resolved = Vec::new(); while let Ok(event) = events.try_recv() { match event.msg { @@ -438,19 +498,67 @@ async fn sync_session_approved_hosts_to_replaces_existing_target_hosts() { ); } +#[test_case(PendingApprovalDecision::AllowOnce, 1; "approve once wakes one waiter")] +#[test_case(PendingApprovalDecision::Deny, 1; "denial wakes one waiter")] +#[test_case(PendingApprovalDecision::AllowForSession, 3; "session grant wakes all waiters")] #[tokio::test] -async fn pending_waiters_receive_owner_decision() { - let pending = Arc::new(PendingHostApproval::new()); +async fn manual_prompt_wakeups(decision: PendingApprovalDecision, expected: usize) { + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + use std::task::Context; + use std::task::Wake; + use std::task::Waker; - let waiter = { - let pending = Arc::clone(&pending); - tokio::spawn(async move { pending.wait_for_decision().await }) + #[derive(Default)] + struct WakeCount(AtomicUsize); + impl Wake for WakeCount { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } + } + + let (session, turn, _events) = crate::session::tests::make_session_and_context_with_rx().await; + let service = &session.services.network_approval; + let owner = |id: &str| { + let host = HostApprovalKey { + environment_id: "local".to_string(), + host: "example.com".to_string(), + protocol: "https", + port: 443, + }; + let key = pending_key(host, &turn.sub_id, id); + let (pending, _) = service.get_or_create_pending_approval(key.clone()); + PendingHostApprovalOwner::new(service, key, pending, /*execution_cancellation*/ None) }; + let mut first = owner("first"); + first + .review_user_prompt( + &session, + &turn, + /*timing_call_id*/ None, + "denied", + std::future::ready(ApprovalResolution::user(ReviewDecision::Approved)), + ) + .await; + let mut owners = ["second", "third", "fourth"].map(owner); + let mut waiters = owners.each_mut().map(|owner| { + Box::pin(owner.review_user_prompt( + &session, + &turn, + /*timing_call_id*/ None, + "denied", + std::future::pending(), + )) + }); + let wakes = Arc::new(WakeCount::default()); + let waker = Waker::from(Arc::clone(&wakes)); + let mut cx = Context::from_waker(&waker); + for waiter in &mut waiters { + assert!(waiter.as_mut().poll(&mut cx).is_pending()); + } - pending.set_decision(PendingApprovalDecision::AllowOnce); - - let decision = waiter.await.expect("waiter should complete"); - assert_eq!(decision, PendingApprovalDecision::AllowOnce); + first.complete(decision); + assert_eq!(wakes.0.load(Ordering::Relaxed), expected); } #[tokio::test] @@ -659,7 +767,7 @@ async fn register_call_with_default_shell_trigger( turn_id: "turn-1".to_string(), tool_name: ToolName::plain("exec_command"), trigger: GuardianNetworkAccessTrigger { - call_id: "call-1".to_string(), + call_id: registration_id.to_string(), tool_name: "exec_command".to_string(), command: vec!["curl".to_string(), "https://example.com".to_string()], cwd: PathUri::from_abs_path(&test_path_buf("/tmp").abs()), diff --git a/codex-rs/core/tests/common/tracing.rs b/codex-rs/core/tests/common/tracing.rs index 5470e0d313..79e576f171 100644 --- a/codex-rs/core/tests/common/tracing.rs +++ b/codex-rs/core/tests/common/tracing.rs @@ -1,3 +1,4 @@ +use anyhow::Context; use opentelemetry::global; use opentelemetry::trace::TracerProvider as _; use opentelemetry_sdk::propagation::TraceContextPropagator; @@ -6,6 +7,48 @@ use tracing::dispatcher::DefaultGuard; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; +/// Receives named trace events on a test's current-thread runtime. +pub struct NamedTraceEvents { + events: tokio::sync::mpsc::UnboundedReceiver<()>, + _subscriber: DefaultGuard, +} + +impl NamedTraceEvents { + pub fn subscribe(name: &'static str) -> Self { + struct EventLayer { + name: &'static str, + sender: tokio::sync::mpsc::UnboundedSender<()>, + } + impl tracing_subscriber::Layer for EventLayer { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + if event.metadata().name() == self.name { + let _ = self.sender.send(()); + } + } + } + + let (sender, events) = tokio::sync::mpsc::unbounded_channel(); + let subscriber = tracing_subscriber::registry() + .with(EventLayer { name, sender }) + .set_default(); + Self { + events, + _subscriber: subscriber, + } + } + + pub async fn next(&mut self) -> anyhow::Result<()> { + tokio::time::timeout(std::time::Duration::from_secs(10), self.events.recv()) + .await + .context("timed out waiting for trace event")? + .context("trace collector closed") + } +} + pub struct TestTracingContext { _provider: SdkTracerProvider, _guard: DefaultGuard, diff --git a/codex-rs/core/tests/suite/network_approval.rs b/codex-rs/core/tests/suite/network_approval.rs index f3b40af714..9a8d63ed46 100644 --- a/codex-rs/core/tests/suite/network_approval.rs +++ b/codex-rs/core/tests/suite/network_approval.rs @@ -11,10 +11,8 @@ use codex_core::config::NetworkProxySpec; use codex_core::shell::ShellType; use codex_core::shell::get_shell; use codex_core::windows_sandbox::WindowsSandboxLevelExt; -use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::LOCAL_ENVIRONMENT_ID; use codex_exec_server::REMOTE_ENVIRONMENT_ID; -use codex_exec_server::RemoveOptions; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::ToolFinishInput; use codex_extension_api::ToolLifecycleContributor; @@ -70,7 +68,6 @@ use codex_protocol::request_permissions::RequestPermissionProfile; use codex_protocol::request_permissions::RequestPermissionsResponse; use codex_protocol::user_input::UserInput; use codex_thread_store::LoadThreadHistoryParams; -use codex_utils_path_uri::PathUri; use core_test_support::PathBufExt; use core_test_support::PathExt; use core_test_support::managed_network_requirements_loader; @@ -89,6 +86,7 @@ use core_test_support::responses::start_mock_server; use core_test_support::skip_if_host_windows; use core_test_support::skip_if_no_network; use core_test_support::skip_if_no_remote_env; +use core_test_support::skip_if_remote; use core_test_support::skip_if_sandbox; use core_test_support::skip_if_target_windows; use core_test_support::test_codex::TestCodex; @@ -97,17 +95,18 @@ use core_test_support::test_codex::local; use core_test_support::test_codex::local_request; use core_test_support::test_codex::test_codex; use core_test_support::test_codex::turn_permission_fields; +use core_test_support::tracing::NamedTraceEvents; use core_test_support::wait_for_event; use core_test_support::wait_for_event_with_timeout; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; +use std::collections::HashMap; +use std::collections::HashSet; use std::fs; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; -use std::time::SystemTime; -use std::time::UNIX_EPOCH; use tempfile::TempDir; use test_case::test_case; use tokio::io::AsyncReadExt; @@ -1143,7 +1142,7 @@ async fn background_network_approval_uses_current_review_settings_and_original_e #[test_case("experimental_network"; "legacy_requirements")] #[test_case("network_proxy"; "proxy_requirements")] #[test_case("permissions.network"; "canonical_requirements")] -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] #[cfg_attr( not(target_os = "linux"), ignore = "requires the trusted Linux proxy bridge" @@ -1151,50 +1150,52 @@ async fn background_network_approval_uses_current_review_settings_and_original_e async fn user_network_approval_once_session_and_denial_semantics( requirements_table: &str, ) -> Result<()> { + skip_if_remote!( + Ok(()), + "remote executors forward network decisions only to Guardian" + ); skip_if_target_windows!(Ok(()), "uses the POSIX/Python network fixture"); skip_if_host_windows!(Ok(())); skip_if_no_network!(Ok(())); skip_if_sandbox!(Ok(())); let server = start_mock_server().await; - let test = managed_network_unified_exec_builder(&[])? - .with_cloud_config_bundle( - CloudConfigBundleFixture::loader_with_enterprise_requirement(format!( - "[{requirements_table}]\nenabled = true\nallow_local_binding = true\n" - )), - ) - .build_with_remote_and_local_env(&server) - .await?; + // Keep the test's trace subscriber installed until the queued tool tasks finish. + let mut queued = NamedTraceEvents::subscribe("codex.network_approval.queued"); + let requirements = CloudConfigBundleFixture::loader_with_enterprise_requirement(format!( + "[{requirements_table}]\nenabled = true\nallow_local_binding = true\n" + )); + let matching_calls = ["network-first", "network-second", "network-third"]; + let (test, responses, mut prompts) = start_concurrent_network_requests( + managed_network_unified_exec_builder(&[])?.with_cloud_config_bundle(requirements), + &server, + &[ + (matching_calls[0], 80), + (matching_calls[1], 80), + (matching_calls[2], 80), + ("user-network-port", 81), + ], + &mut queued, + ) + .await?; assert!(test.config.managed_network_requirements_enabled()); assert!(test.config.permissions.network.is_some()); assert!(test.session_configured.network_proxy.is_some()); - let environments = vec![local(test.config.cwd.clone())]; - - mount_exec_network_turn( - &server, - "resp-user-network-once-1", - "user-network-once-1", - network_fetch_args(LOCAL_ENVIRONMENT_ID), - ) - .await?; - submit_managed_network_turn( - &test, - "approve this network request once", - environments.clone(), - ApprovalsReviewer::User, - AskForApproval::OnRequest, - ) - .await?; - let approval = expect_network_approval(&test, LOCAL_ENVIRONMENT_ID).await?; - assert!( - approval - .call_id - .starts_with("network#local#http#codex-network-test.invalid#80#") - ); + let environment = test.executor_environment().selection(); + let approval = prompts + .remove(NETWORK_TEST_TARGET) + .context("missing matching-host prompt")?; + let other_port = prompts + .remove(&format!("http://{NETWORK_TEST_HOST}:81")) + .context("missing independent port prompt")?; + assert!(approval.call_id.starts_with(&format!( + "network#{}#http#{NETWORK_TEST_HOST}#80#", + environment.environment_id + ))); assert_eq!(approval.approval_id.as_deref(), None); let first_approval_call_id = approval.call_id.clone(); assert!(!approval.turn_id.is_empty()); - assert_eq!(approval.cwd, test.config.cwd.clone().into()); + assert_eq!(approval.cwd, environment.cwd.clone().into()); assert_eq!( approval.reason.as_deref(), Some("Network access to codex-network-test.invalid requires approval") @@ -1206,93 +1207,85 @@ async fn user_network_approval_once_session_and_denial_semantics( decision: ReviewDecision::Approved, }) .await?; - wait_for_turn_complete(&test).await; - - mount_exec_network_turn( - &server, - "resp-user-network-once-2", - "user-network-once-2", - network_fetch_args(LOCAL_ENVIRONMENT_ID), - ) - .await?; - submit_managed_network_turn( - &test, - "the once decision must prompt again", - environments.clone(), - ApprovalsReviewer::User, - AskForApproval::OnRequest, - ) - .await?; - let approval = expect_network_approval(&test, LOCAL_ENVIRONMENT_ID).await?; - assert_eq!(approval.approval_id.as_deref(), None); - assert_ne!(approval.call_id, first_approval_call_id); + // The first execution may finish before or after the next manual prompt arrives. + let mut pending = HashSet::from(matching_calls); + let mut granted_for_session = false; + tokio::time::timeout(Duration::from_secs(30), async { + while !pending.is_empty() { + match test.codex.next_event().await?.msg { + EventMsg::ExecApprovalRequest(approval) => { + assert!(!granted_for_session, "duplicate prompt after session grant"); + assert_eq!(approval.command, ["network-access", NETWORK_TEST_TARGET]); + assert_eq!( + approval.environment_id.as_ref(), + Some(&environment.environment_id) + ); + assert_eq!(approval.approval_id, None); + assert_ne!(approval.call_id, first_approval_call_id); + test.codex + .submit(Op::ExecApproval { + id: approval.effective_approval_id(), + turn_id: Some(approval.turn_id), + decision: ReviewDecision::ApprovedForSession, + }) + .await?; + granted_for_session = true; + } + EventMsg::ExecCommandEnd(end) => { + assert!( + pending.remove(end.call_id.as_str()), + "unexpected completion: {end:?}" + ); + assert_eq!(end.exit_code, 0); + } + EventMsg::TurnComplete(_) => { + anyhow::bail!("turn completed with port 81 still pending") + } + _ => {} + } + } + anyhow::Ok(()) + }) + .await + .context("matching executions did not finish while port 81 was pending")??; + assert!(granted_for_session); test.codex .submit(Op::ExecApproval { - id: approval.effective_approval_id(), - turn_id: Some(approval.turn_id), - decision: ReviewDecision::ApprovedForSession, + id: other_port.effective_approval_id(), + turn_id: Some(other_port.turn_id), + decision: ReviewDecision::denied("rejected by user"), }) .await?; - wait_for_turn_complete(&test).await; + wait_for_completion_without_network_prompt(&test).await; + for id in matching_calls { + assert_network_probe_allowed(&responses, id); + } + let denied_output = responses + .function_call_output_text("user-network-port") + .context("expected user-denied network output")?; + assert!(denied_output.contains("rejected by user")); + assert!(!denied_output.contains("blocked by policy")); - mount_exec_network_turn( + let session_probe = mount_exec_network_turn( &server, "resp-user-network-session", "user-network-session", - network_fetch_args(LOCAL_ENVIRONMENT_ID), + network_probe_args( + &remote_network_proxy_request_command("PROXY"), + &environment.environment_id, + ), ) .await?; submit_managed_network_turn( &test, "the session decision must bypass another prompt", - environments.clone(), + vec![environment.clone()], ApprovalsReviewer::User, AskForApproval::OnRequest, ) .await?; wait_for_completion_without_network_prompt(&test).await; - - let different_port_target = format!("http://{NETWORK_TEST_HOST}:81"); - let different_port_command = format!( - "python3 -c \"import urllib.request; urllib.request.build_opener(urllib.request.ProxyHandler()).open('{different_port_target}', timeout=2).read()\"" - ); - let denied_responses = mount_exec_network_turn( - &server, - "resp-user-network-port", - "user-network-port", - network_exec_args(&different_port_command), - ) - .await?; - submit_managed_network_turn( - &test, - "a different port must prompt", - environments, - ApprovalsReviewer::User, - AskForApproval::OnRequest, - ) - .await?; - let approval = expect_network_approval_target( - &test, - LOCAL_ENVIRONMENT_ID, - &different_port_target, - NetworkApprovalProtocol::Http, - ) - .await?; - test.codex - .submit(Op::ExecApproval { - id: approval.effective_approval_id(), - turn_id: Some(approval.turn_id), - decision: ReviewDecision::denied("rejected by user"), - }) - .await?; - wait_for_turn_complete(&test).await; - let denied_output = denied_responses - .requests() - .iter() - .find_map(|request| request.function_call_output_text("user-network-port")) - .context("expected user-denied network output")?; - assert!(denied_output.contains("rejected by user")); - assert!(!denied_output.contains("blocked by policy")); + assert_network_probe_allowed(&session_probe, "user-network-session"); let socks_target = format!("socks5-tcp://{NETWORK_TEST_HOST}:443"); let socks_command = format!( @@ -1305,7 +1298,10 @@ async fn user_network_approval_once_session_and_denial_semantics( ev_function_call( "user-network-abort", "exec_command", - &serde_json::to_string(&network_exec_args(&socks_command))?, + &serde_json::to_string(&network_probe_args( + &socks_command, + &environment.environment_id, + ))?, ), ev_completed("resp-user-network-abort"), ]), @@ -1314,14 +1310,14 @@ async fn user_network_approval_once_session_and_denial_semantics( submit_managed_network_turn( &test, "a different protocol must prompt and the user abort must stay a user outcome", - vec![local(test.config.cwd.clone())], + vec![environment.clone()], ApprovalsReviewer::User, AskForApproval::OnRequest, ) .await?; let approval = expect_network_approval_target( &test, - LOCAL_ENVIRONMENT_ID, + &environment.environment_id, &socks_target, NetworkApprovalProtocol::Socks5Tcp, ) @@ -1525,7 +1521,10 @@ PY"# &server, "network-session-grant-probe", probe_call_id, - network_exec_args(&remote_network_proxy_request_command("REUSED-GRANT")), + network_probe_args( + &remote_network_proxy_request_command("PROXY"), + LOCAL_ENVIRONMENT_ID, + ), ) .await?; submit_managed_network_turn( @@ -1537,14 +1536,7 @@ PY"# ) .await?; wait_for_completion_without_network_prompt(&test).await; - let output = probe - .function_call_output_text(probe_call_id) - .context("expected network output from the new execution")?; - assert!( - output.contains("REUSED-GRANT:HTTP/1.1 200") - || output.contains("REUSED-GRANT:HTTP/1.1 502"), - "{output}", - ); + assert_network_probe_allowed(&probe, probe_call_id); } Ok(()) @@ -1637,35 +1629,33 @@ async fn allowing_network_policy_amendment_persists_context_and_bypasses_prompt( Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] #[cfg_attr( not(target_os = "linux"), ignore = "requires the trusted Linux proxy bridge" )] async fn denying_network_policy_amendment_persists_and_blocks_request() -> Result<()> { + skip_if_remote!( + Ok(()), + "remote executors forward network decisions only to Guardian" + ); skip_if_target_windows!(Ok(()), "uses the POSIX/Python network fixture"); skip_if_host_windows!(Ok(())); skip_if_no_network!(Ok(())); skip_if_sandbox!(Ok(())); let server = start_mock_server().await; - let test = managed_network_unified_exec_test(&server).await?; - let responses = mount_exec_network_turn( + let mut queued = NamedTraceEvents::subscribe("codex.network_approval.queued"); + let (test, responses, mut prompts) = start_concurrent_network_requests( + managed_network_unified_exec_builder(&[])?, &server, - "resp-network-deny-amendment", - "network-deny-amendment", - network_fetch_args(LOCAL_ENVIRONMENT_ID), + &[("network-deny-first", 80), ("network-deny-second", 80)], + &mut queued, ) .await?; - submit_managed_network_turn( - &test, - "persist a deny rule for this host", - vec![local(test.config.cwd.clone())], - ApprovalsReviewer::User, - AskForApproval::OnRequest, - ) - .await?; - let approval = expect_network_approval(&test, LOCAL_ENVIRONMENT_ID).await?; + let approval = prompts + .remove(NETWORK_TEST_TARGET) + .context("missing prompt")?; test.codex .submit(Op::ExecApproval { id: approval.effective_approval_id(), @@ -1678,18 +1668,30 @@ async fn denying_network_policy_amendment_persists_and_blocks_request() -> Resul }, }) .await?; - wait_for_turn_complete(&test).await; + wait_for_completion_without_network_prompt(&test).await; let policy = fs::read_to_string(test.home.path().join("rules/default.rules"))?; assert!(policy.contains( r#"network_rule(host="codex-network-test.invalid", protocol="http", decision="deny""# )); - let output = responses - .requests() - .iter() - .find_map(|request| request.function_call_output_text("network-deny-amendment")) - .context("expected denied network tool output")?; - assert!(output.contains("rejected by user")); + let outputs = ["network-deny-first", "network-deny-second"].map(|id| { + responses + .function_call_output_text(id) + .expect("denied network tool output") + }); + assert!( + outputs + .iter() + .any(|output| output.contains("rejected by user")), + "{outputs:?}" + ); + assert!( + outputs + .iter() + .any(|output| output.contains(NETWORK_TEST_TARGET) + && output.contains("was blocked by policy.")), + "{outputs:?}" + ); Ok(()) } @@ -3130,7 +3132,7 @@ async fn escalated_owner_network_terminal_requires_stdin_approval( } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn approved_network_host_for_one_environment_still_prompts_in_another() -> Result<()> { +async fn local_session_grant_does_not_allow_the_same_host_on_a_remote_executor() -> Result<()> { skip_if_target_windows!(Ok(()), "uses the POSIX/Python network fixture"); skip_if_host_windows!(Ok(())); skip_if_no_network!(Ok(())); @@ -3140,38 +3142,16 @@ async fn approved_network_host_for_one_environment_still_prompts_in_another() -> let server = start_mock_server().await; let test = managed_network_unified_exec_test(&server).await?; let local_cwd = TempDir::new()?; - let remote_cwd = PathBuf::from(format!( - "/tmp/codex-network-approval-{}", - SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() - )) - .abs(); - let remote_cwd_uri = PathUri::from_host_native_path(&remote_cwd)?; - test.fs() - .create_directory( - &remote_cwd_uri, - CreateDirectoryOptions { - recursive: true, - follow_symlinks: true, - }, - /*sandbox*/ None, - ) - .await?; let environments = vec![ local(local_cwd.path().abs()), - TurnEnvironmentSelection { - selected_capability_roots: Default::default(), - environment_id: REMOTE_ENVIRONMENT_ID.to_string(), - cwd: PathUri::from_abs_path(&remote_cwd), - workspace_roots: vec![PathUri::from_abs_path(&remote_cwd)], - config: EnvironmentConfigState::FromThread, - }, + test.executor_environment().selection().clone(), ]; - - mount_exec_network_turn( + let command = remote_network_proxy_request_command("PROXY"); + let local_responses = mount_exec_network_turn( &server, "resp-network-local", "exec-network-local", - network_fetch_args(LOCAL_ENVIRONMENT_ID), + network_probe_args(&command, LOCAL_ENVIRONMENT_ID), ) .await?; submit_managed_network_turn( @@ -3179,7 +3159,7 @@ async fn approved_network_host_for_one_environment_still_prompts_in_another() -> "fetch from the local environment", environments.clone(), ApprovalsReviewer::User, - AskForApproval::UnlessTrusted, + AskForApproval::OnRequest, ) .await?; let approval = expect_network_approval(&test, LOCAL_ENVIRONMENT_ID).await?; @@ -3190,49 +3170,30 @@ async fn approved_network_host_for_one_environment_still_prompts_in_another() -> decision: ReviewDecision::ApprovedForSession, }) .await?; - wait_for_turn_complete(&test).await; + wait_for_completion_without_network_prompt(&test).await; + assert_network_probe_allowed(&local_responses, "exec-network-local"); let remote_responses = mount_exec_network_turn( &server, "resp-network-remote", "exec-network-remote", - network_fetch_args(REMOTE_ENVIRONMENT_ID), + network_probe_args(&command, REMOTE_ENVIRONMENT_ID), ) .await?; submit_managed_network_turn( &test, "fetch from the remote environment", - environments.clone(), + environments, ApprovalsReviewer::User, - AskForApproval::UnlessTrusted, + AskForApproval::OnRequest, ) .await?; - let approval = expect_network_approval(&test, REMOTE_ENVIRONMENT_ID).await?; - let rejection = "approval request failed because the client disconnected"; - test.codex - .submit(Op::ExecApproval { - id: approval.effective_approval_id(), - turn_id: None, - decision: ReviewDecision::denied(rejection), - }) - .await?; - wait_for_turn_complete(&test).await; - assert_eq!( - remote_responses.function_call_output_text("exec-network-remote"), - Some(rejection.to_string()) - ); - - test.fs() - .remove( - &remote_cwd_uri, - RemoveOptions { - recursive: true, - force: true, - follow_symlinks: true, - }, - /*sandbox*/ None, - ) - .await?; + // Remote executors do not forward manual prompts. The proxy must deny this request. + wait_for_completion_without_network_prompt(&test).await; + let output = remote_responses + .function_call_output_text("exec-network-remote") + .context("missing remote proxy response")?; + assert!(output.contains("PROXY:HTTP/1.1 403 Forbidden"), "{output}"); Ok(()) } @@ -3403,6 +3364,69 @@ pub(super) fn guardian_parent_catalog() -> ModelsResponse { } } +async fn start_concurrent_network_requests( + mut builder: TestCodexBuilder, + server: &wiremock::MockServer, + requests: &[(&str, u16)], + queued: &mut NamedTraceEvents, +) -> Result<( + TestCodex, + ResponseMock, + HashMap, +)> { + let test = builder.build_with_auto_env(server).await?; + let environment = test.executor_environment().selection(); + let mut calls = Vec::new(); + for &(id, port) in requests { + let command = + network_proxy_request_command(&format!("{NETWORK_TEST_HOST}:{port}"), "PROXY"); + let args = network_probe_args(&command, &environment.environment_id); + calls.push(ev_function_call(id, "exec_command", &args.to_string())); + } + calls.push(ev_completed("network-requests")); + let responses = mount_sse_sequence( + server, + vec![sse(calls), sse(vec![ev_completed("network-done")])], + ) + .await; + submit_managed_network_turn( + &test, + "review these concurrent network requests", + vec![environment.clone()], + ApprovalsReviewer::User, + AskForApproval::OnRequest, + ) + .await?; + + let ports: HashSet<_> = requests.iter().map(|(_, port)| port).collect(); + let mut prompts = HashMap::new(); + wait_for_event(&test.codex, |event| { + match event { + EventMsg::ExecApprovalRequest(approval) => { + assert!( + prompts + .insert(approval.command[1].clone(), approval.clone()) + .is_none(), + "duplicate prompt for {:?}", + approval.command + ); + } + EventMsg::TurnComplete(_) => panic!( + "turn completed before network approval: {:?}", + responses.function_call_output_text(requests[0].0) + ), + _ => {} + } + prompts.len() == ports.len() + }) + .await; + // Every matching execution has joined its endpoint queue before the first decision. + for _ in ports.len()..requests.len() { + queued.next().await?; + } + Ok((test, responses, prompts)) +} + async fn managed_network_unified_exec_test(server: &wiremock::MockServer) -> Result { managed_network_unified_exec_test_with_features(server, &[]).await } @@ -3422,18 +3446,26 @@ async fn managed_network_unified_exec_test_with_features( fn managed_network_unified_exec_builder(features: &[Feature]) -> Result { let home = Arc::new(TempDir::new()?); + // Limited mode uses TLS interception, which remote executors cannot enforce. + let mode = if core_test_support::is_remote_test_environment() { + "full" + } else { + "limited" + }; fs::write( home.path().join("config.toml"), - r#"default_permissions = "workspace" + format!( + r#"default_permissions = "workspace" [permissions.workspace.filesystem] ":minimal" = "read" [permissions.workspace.network] enabled = true -mode = "limited" +mode = "{mode}" allow_local_binding = true -"#, +"# + ), )?; let approval_policy = AskForApproval::OnRequest; let permission_profile = PermissionProfile::workspace_write_with( @@ -3501,8 +3533,27 @@ fn network_exec_args(command: &str) -> Value { }) } +fn network_probe_args(command: &str, environment_id: &str) -> Value { + // Remote executors accept only their configured shell. + json!({ "cmd": command, "environment_id": environment_id, "login": false, "yield_time_ms": 10_000 }) +} + +fn assert_network_probe_allowed(responses: &ResponseMock, id: &str) { + let output = responses + .function_call_output_text(id) + .unwrap_or_else(|| panic!("missing network output for {id}")); + // The fixture host need not resolve. A gateway error still means the proxy allowed it. + assert!( + output.contains("PROXY:HTTP/1.1 502") || output.contains("PROXY:HTTP/1.1 200"), + "{id}: {output}" + ); +} + fn remote_network_proxy_request_command(marker: &str) -> String { - let host = NETWORK_TEST_HOST; + network_proxy_request_command(NETWORK_TEST_HOST, marker) +} + +fn network_proxy_request_command(host: &str, marker: &str) -> String { format!( "python3 -c \"import os,socket,urllib.parse; proxy = urllib.parse.urlparse(os.environ['HTTP_PROXY']); sock = socket.create_connection((proxy.hostname, proxy.port), timeout=10); sock.sendall(b'GET http://{host} HTTP/1.1\\r\\nHost: {host}\\r\\nConnection: close\\r\\n\\r\\n'); print('{marker}:' + sock.makefile('rb').readline().decode(errors='replace'))\"" )