diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index b33c7c25b8..5166ffad28 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -1698,7 +1698,7 @@ impl CodexMessageProcessor { None => self.config.sandbox_policy.get().clone(), }; let started_network_proxy = match self.config.network.as_ref() { - Some(spec) => match spec.start_proxy(&effective_policy).await { + Some(spec) => match spec.start_proxy(&effective_policy, None).await { Ok(started) => Some(started), Err(err) => { let error = JSONRPCErrorError { @@ -1721,6 +1721,7 @@ impl CodexMessageProcessor { network: started_network_proxy .as_ref() .map(codex_core::config::StartedNetworkProxy::proxy), + network_attempt_id: None, sandbox_permissions: SandboxPermissions::UseDefault, windows_sandbox_level, justification: None, diff --git a/codex-rs/cli/src/debug_sandbox.rs b/codex-rs/cli/src/debug_sandbox.rs index 82fd655d41..c9b9001b09 100644 --- a/codex-rs/cli/src/debug_sandbox.rs +++ b/codex-rs/cli/src/debug_sandbox.rs @@ -216,7 +216,7 @@ async fn run_command_under_sandbox( // This proxy should only live for the lifetime of the child process. let network_proxy = match config.permissions.network.as_ref() { Some(spec) => Some( - spec.start_proxy(config.sandbox_policy.get()) + spec.start_proxy(config.sandbox_policy.get(), None) .await .map_err(|err| anyhow::anyhow!("failed to start managed network proxy: {err}"))?, ), diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index ef42524cb2..412aefa9f1 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -47,6 +47,10 @@ use codex_hooks::HookEventAfterAgent; use codex_hooks::HookPayload; use codex_hooks::Hooks; use codex_hooks::HooksConfig; +use codex_network_proxy::NetworkDecision; +use codex_network_proxy::NetworkPolicyDecider; +use codex_network_proxy::NetworkPolicyRequest; +use codex_network_proxy::NetworkProtocol; use codex_network_proxy::NetworkProxy; use codex_protocol::ThreadId; use codex_protocol::approvals::ExecPolicyAmendment; @@ -171,6 +175,7 @@ use crate::protocol::EventMsg; use crate::protocol::ExecApprovalRequestEvent; use crate::protocol::McpServerRefreshConfig; use crate::protocol::NetworkApprovalContext; +use crate::protocol::NetworkApprovalProtocol; use crate::protocol::Op; use crate::protocol::PlanDeltaEvent; use crate::protocol::RateLimitSnapshot; @@ -514,12 +519,22 @@ pub(crate) struct Session { /// session. features: Features, pending_mcp_server_refresh_config: Mutex>, + network_approval_attempts: Mutex>>, + network_session_approved_hosts: Mutex>, pub(crate) active_turn: Mutex>, pub(crate) services: SessionServices, js_repl: Arc, next_internal_sub_id: AtomicU64, } +struct NetworkApprovalAttempt { + turn_id: String, + call_id: String, + command: Vec, + cwd: PathBuf, + approved_hosts: Mutex>, +} + const SEARCH_TOOL_DEVELOPER_INSTRUCTIONS: &str = include_str!("../templates/search_tool/developer_instructions.md"); @@ -1175,13 +1190,29 @@ impl Session { }; session_configuration.thread_name = thread_name.clone(); let mut state = SessionState::new(session_configuration.clone()); + let inline_network_decider_session = + Arc::new(RwLock::new(std::sync::Weak::::new())); + let inline_network_decider: Arc = Arc::new({ + let inline_network_decider_session = Arc::clone(&inline_network_decider_session); + move |request: NetworkPolicyRequest| { + let inline_network_decider_session = Arc::clone(&inline_network_decider_session); + async move { + let Some(session) = inline_network_decider_session.read().await.upgrade() + else { + return NetworkDecision::ask("not_allowed"); + }; + session.handle_inline_network_policy_request(request).await + } + } + }); let network_proxy = match config.network.as_ref() { Some(spec) => Some( - spec.start_proxy(config.sandbox_policy.get()) - .await - .map_err(|err| { - anyhow::anyhow!("failed to start managed network proxy: {err}") - })?, + spec.start_proxy( + config.sandbox_policy.get(), + Some(Arc::clone(&inline_network_decider)), + ) + .await + .map_err(|err| anyhow::anyhow!("failed to start managed network proxy: {err}"))?, ), None => None, }; @@ -1262,11 +1293,17 @@ impl Session { state: Mutex::new(state), features: config.features.clone(), pending_mcp_server_refresh_config: Mutex::new(None), + network_approval_attempts: Mutex::new(HashMap::new()), + network_session_approved_hosts: Mutex::new(HashSet::new()), active_turn: Mutex::new(None), services, js_repl, next_internal_sub_id: AtomicU64::new(0), }); + { + let mut guard = inline_network_decider_session.write().await; + *guard = Arc::downgrade(&sess); + } // Dispatch the SessionConfiguredEvent first and then report any errors. // If resuming, include converted initial messages in the payload so UIs can render them immediately. @@ -2083,6 +2120,146 @@ impl Session { } } + pub(crate) async fn register_network_approval_attempt( + &self, + attempt_id: String, + turn_id: String, + call_id: String, + command: Vec, + cwd: PathBuf, + ) { + let mut attempts = self.network_approval_attempts.lock().await; + attempts.insert( + attempt_id, + Arc::new(NetworkApprovalAttempt { + turn_id, + call_id, + command, + cwd, + approved_hosts: Mutex::new(HashSet::new()), + }), + ); + } + + pub(crate) async fn unregister_network_approval_attempt(&self, attempt_id: &str) { + let mut attempts = self.network_approval_attempts.lock().await; + attempts.remove(attempt_id); + } + + async fn resolve_network_approval_attempt( + &self, + request: &NetworkPolicyRequest, + ) -> Option> { + let attempts = self.network_approval_attempts.lock().await; + + if let Some(attempt_id) = request.attempt_id.as_deref() { + if let Some(attempt) = attempts.get(attempt_id).cloned() { + return Some(attempt); + } + tracing::debug!( + "inline network approval decider did not find attempt context for {attempt_id}" + ); + } else { + tracing::debug!( + "inline network approval decider received request without attempt_id for host {}", + request.host + ); + } + + if attempts.len() == 1 + && let Some(attempt) = attempts.values().next().cloned() + { + tracing::debug!( + "inline network approval decider falling back to only active attempt for host {}", + request.host + ); + return Some(attempt); + } else { + tracing::debug!( + "inline network approval decider cannot disambiguate attempt for host {} (active_attempts={})", + request.host, + attempts.len() + ); + } + + None + } + + async fn handle_inline_network_policy_request( + &self, + request: NetworkPolicyRequest, + ) -> NetworkDecision { + const REASON_NOT_ALLOWED: &str = "not_allowed"; + + { + let approved_hosts = self.network_session_approved_hosts.lock().await; + if approved_hosts.contains(request.host.as_str()) { + return NetworkDecision::Allow; + } + } + + let Some(attempt) = self.resolve_network_approval_attempt(&request).await else { + return NetworkDecision::ask(REASON_NOT_ALLOWED); + }; + + { + let approved_hosts = attempt.approved_hosts.lock().await; + if approved_hosts.contains(request.host.as_str()) { + return NetworkDecision::Allow; + } + } + + let protocol = match request.protocol { + NetworkProtocol::Http => NetworkApprovalProtocol::Http, + NetworkProtocol::HttpsConnect => NetworkApprovalProtocol::Https, + NetworkProtocol::Socks5Tcp | NetworkProtocol::Socks5Udp => { + return NetworkDecision::deny(REASON_NOT_ALLOWED); + } + }; + + let Some(turn_context) = self.turn_context_for_sub_id(&attempt.turn_id).await else { + tracing::debug!( + "inline network approval decider could not resolve turn context for {}", + attempt.turn_id + ); + return NetworkDecision::ask(REASON_NOT_ALLOWED); + }; + + let approval_decision = self + .request_command_approval( + turn_context.as_ref(), + attempt.call_id.clone(), + attempt.command.clone(), + attempt.cwd.clone(), + Some(format!( + "Network access to \"{}\" is blocked by policy.", + request.host + )), + Some(NetworkApprovalContext { + host: request.host.clone(), + protocol, + }), + None, + ) + .await; + + match approval_decision { + ReviewDecision::Approved | ReviewDecision::ApprovedExecpolicyAmendment { .. } => { + let mut approved_hosts = attempt.approved_hosts.lock().await; + approved_hosts.insert(request.host); + NetworkDecision::Allow + } + ReviewDecision::ApprovedForSession => { + let mut approved_hosts = self.network_session_approved_hosts.lock().await; + approved_hosts.insert(request.host); + NetworkDecision::Allow + } + ReviewDecision::Denied | ReviewDecision::Abort => { + NetworkDecision::deny(REASON_NOT_ALLOWED) + } + } + } + /// Emit an exec approval request event and await the user's decision. /// /// The request is keyed by `call_id` so matching responses are delivered @@ -6865,6 +7042,8 @@ mod tests { state: Mutex::new(state), features: config.features.clone(), pending_mcp_server_refresh_config: Mutex::new(None), + network_approval_attempts: Mutex::new(HashMap::new()), + network_session_approved_hosts: Mutex::new(HashSet::new()), active_turn: Mutex::new(None), services, js_repl, @@ -7011,6 +7190,8 @@ mod tests { state: Mutex::new(state), features: config.features.clone(), pending_mcp_server_refresh_config: Mutex::new(None), + network_approval_attempts: Mutex::new(HashMap::new()), + network_session_approved_hosts: Mutex::new(HashSet::new()), active_turn: Mutex::new(None), services, js_repl, @@ -7066,6 +7247,112 @@ mod tests { assert!(!new_token.is_cancelled()); } + #[tokio::test] + async fn resolve_network_approval_attempt_falls_back_to_only_active_attempt() { + let (session, _turn_context) = make_session_and_context().await; + session + .register_network_approval_attempt( + "attempt-1".to_string(), + "turn-1".to_string(), + "call-1".to_string(), + vec!["curl".to_string(), "google.com".to_string()], + std::env::temp_dir(), + ) + .await; + + let request = codex_network_proxy::NetworkPolicyRequest::new( + codex_network_proxy::NetworkPolicyRequestArgs { + protocol: codex_network_proxy::NetworkProtocol::Http, + host: "google.com".to_string(), + port: 80, + client_addr: None, + method: Some("GET".to_string()), + command: None, + exec_policy_hint: None, + attempt_id: None, + }, + ); + + let resolved = session.resolve_network_approval_attempt(&request).await; + assert!(resolved.is_some()); + + session + .unregister_network_approval_attempt("attempt-1") + .await; + } + + #[tokio::test] + async fn resolve_network_approval_attempt_returns_none_when_ambiguous() { + let (session, _turn_context) = make_session_and_context().await; + session + .register_network_approval_attempt( + "attempt-1".to_string(), + "turn-1".to_string(), + "call-1".to_string(), + vec!["curl".to_string(), "google.com".to_string()], + std::env::temp_dir(), + ) + .await; + session + .register_network_approval_attempt( + "attempt-2".to_string(), + "turn-2".to_string(), + "call-2".to_string(), + vec!["curl".to_string(), "robinhood.com".to_string()], + std::env::temp_dir(), + ) + .await; + + let request = codex_network_proxy::NetworkPolicyRequest::new( + codex_network_proxy::NetworkPolicyRequestArgs { + protocol: codex_network_proxy::NetworkProtocol::Http, + host: "google.com".to_string(), + port: 80, + client_addr: None, + method: Some("GET".to_string()), + command: None, + exec_policy_hint: None, + attempt_id: None, + }, + ); + + let resolved = session.resolve_network_approval_attempt(&request).await; + assert!(resolved.is_none()); + + session + .unregister_network_approval_attempt("attempt-1") + .await; + session + .unregister_network_approval_attempt("attempt-2") + .await; + } + + #[tokio::test] + async fn inline_network_decider_allows_session_approved_host_without_attempt() { + let (session, _turn_context) = make_session_and_context().await; + { + let mut approved_hosts = session.network_session_approved_hosts.lock().await; + approved_hosts.insert("openai.com".to_string()); + } + + let decision = session + .handle_inline_network_policy_request(codex_network_proxy::NetworkPolicyRequest::new( + codex_network_proxy::NetworkPolicyRequestArgs { + protocol: codex_network_proxy::NetworkProtocol::Http, + host: "openai.com".to_string(), + port: 80, + client_addr: None, + method: Some("GET".to_string()), + command: None, + exec_policy_hint: None, + attempt_id: None, + }, + )) + .await; + + assert_eq!(decision, codex_network_proxy::NetworkDecision::Allow); + } + #[tokio::test] async fn record_model_warning_appends_user_message() { let (mut session, turn_context) = make_session_and_context().await; @@ -7676,6 +7963,7 @@ mod tests { expiration: timeout_ms.into(), env: HashMap::new(), network: None, + network_attempt_id: None, sandbox_permissions, windows_sandbox_level: turn_context.windows_sandbox_level, justification: Some("test".to_string()), @@ -7689,6 +7977,7 @@ mod tests { expiration: timeout_ms.into(), env: HashMap::new(), network: None, + network_attempt_id: None, windows_sandbox_level: turn_context.windows_sandbox_level, justification: params.justification.clone(), arg0: None, diff --git a/codex-rs/core/src/config/network_proxy_spec.rs b/codex-rs/core/src/config/network_proxy_spec.rs index 5bc9dfcb33..3686e96357 100644 --- a/codex-rs/core/src/config/network_proxy_spec.rs +++ b/codex-rs/core/src/config/network_proxy_spec.rs @@ -4,6 +4,7 @@ use async_trait::async_trait; use codex_network_proxy::ConfigReloader; use codex_network_proxy::ConfigState; use codex_network_proxy::NetworkDecision; +use codex_network_proxy::NetworkPolicyDecider; use codex_network_proxy::NetworkProxy; use codex_network_proxy::NetworkProxyConfig; use codex_network_proxy::NetworkProxyConstraints; @@ -97,6 +98,7 @@ impl NetworkProxySpec { pub async fn start_proxy( &self, sandbox_policy: &SandboxPolicy, + policy_decider: Option>, ) -> std::io::Result { let state = build_config_state(self.config.clone(), self.constraints.clone()).map_err(|err| { @@ -106,11 +108,15 @@ impl NetworkProxySpec { let state = NetworkProxyState::with_reloader(state, reloader); let mut builder = NetworkProxy::builder().state(Arc::new(state)); if should_ask_on_allowlist_miss(sandbox_policy) { - builder = builder.policy_decider(|_request| async { - // In restricted sandbox modes, allowlist misses should ask for - // explicit network approval instead of hard-denying. - NetworkDecision::ask("not_allowed") - }); + if let Some(policy_decider) = policy_decider { + builder = builder.policy_decider_arc(policy_decider); + } else { + builder = builder.policy_decider(|_request| async { + // In restricted sandbox modes, allowlist misses should ask for + // explicit network approval instead of hard-denying. + NetworkDecision::ask("not_allowed") + }); + } } let proxy = builder.build().await.map_err(|err| { std::io::Error::other(format!("failed to build network proxy: {err}")) diff --git a/codex-rs/core/src/exec.rs b/codex-rs/core/src/exec.rs index 5a42e8b7ff..35c523f204 100644 --- a/codex-rs/core/src/exec.rs +++ b/codex-rs/core/src/exec.rs @@ -15,6 +15,7 @@ use tokio::io::AsyncReadExt; use tokio::io::BufReader; use tokio::process::Child; use tokio_util::sync::CancellationToken; +use uuid::Uuid; use crate::error::CodexErr; use crate::error::Result; @@ -67,6 +68,7 @@ pub struct ExecParams { pub expiration: ExecExpiration, pub env: HashMap, pub network: Option, + pub network_attempt_id: Option, pub sandbox_permissions: SandboxPermissions, pub windows_sandbox_level: codex_protocol::config_types::WindowsSandboxLevel, pub justification: Option, @@ -185,13 +187,14 @@ pub async fn process_exec_tool_call( mut env, expiration, network, + network_attempt_id, sandbox_permissions, windows_sandbox_level, justification, arg0: _, } = params; if let Some(network) = network.as_ref() { - network.apply_to_env(&mut env); + network.apply_to_env_for_attempt(&mut env, network_attempt_id.as_deref()); } let (program, args) = command.split_first().ok_or_else(|| { CodexErr::Io(io::Error::new( @@ -239,6 +242,7 @@ pub(crate) async fn execute_exec_env( cwd, env, network, + network_attempt_id, expiration, sandbox, windows_sandbox_level, @@ -247,12 +251,26 @@ pub(crate) async fn execute_exec_env( arg0, } = env; + let network_attempt_id = + network_attempt_id.or_else(|| network.as_ref().map(|_| Uuid::new_v4().to_string())); + let blocked_cursor = match network.as_ref() { + Some(network) => match network.blocked_requests_cursor().await { + Ok(cursor) => Some(cursor), + Err(err) => { + tracing::debug!("failed to read blocked telemetry cursor before exec: {err:#}"); + None + } + }, + None => None, + }; + let params = ExecParams { command, cwd, expiration, env, network: network.clone(), + network_attempt_id: network_attempt_id.clone(), sandbox_permissions, windows_sandbox_level, justification, @@ -260,35 +278,24 @@ pub(crate) async fn execute_exec_env( }; let start = Instant::now(); - let blocked_cursor = match network.as_ref() { - Some(network) => network.blocked_requests_cursor().await.ok(), - None => None, - }; - match (network.as_ref(), blocked_cursor) { - (Some(_), Some(cursor)) => { - tracing::debug!("captured blocked telemetry cursor before exec: {cursor}"); - } - (Some(_), None) => { - tracing::debug!( - "managed network is present but failed to capture blocked telemetry cursor" - ); - } - (None, _) => {} - } let raw_output_result = exec(params, sandbox, sandbox_policy, stdout_stream).await; let duration = start.elapsed(); let finalized = finalize_exec_result(raw_output_result, sandbox, duration); - let telemetry_decision = match (network.as_ref(), blocked_cursor) { - (Some(network), Some(cursor)) => { - blocking_network_policy_decision_from_blocked_queue(network, cursor, sandbox_policy) - .await + let network_policy_decision_for_exec = match network.as_ref() { + Some(network) => { + blocking_network_policy_decision_from_attempt_or_cursor( + network, + network_attempt_id.as_deref(), + blocked_cursor, + sandbox_policy, + ) + .await } - (None, _) => None, - (Some(_), None) => None, + None => None, }; match finalized { Ok(exec_output) => { - if let Some(policy_decision) = telemetry_decision { + if let Some(policy_decision) = network_policy_decision_for_exec { tracing::debug!( "promoting successful exec result to sandbox denied based on structured network telemetry (decision={}, source={}, host={:?}, protocol={:?}, port={:?})", policy_decision.decision, @@ -308,7 +315,7 @@ pub(crate) async fn execute_exec_env( output, network_policy_decision, })) => { - let merged_decision = network_policy_decision.or(telemetry_decision); + let merged_decision = network_policy_decision.or(network_policy_decision_for_exec); if let Some(payload) = merged_decision.as_ref() { tracing::debug!( "sandbox-denied exec result includes structured network decision (decision={}, source={}, host={:?}, protocol={:?}, port={:?})", @@ -413,12 +420,13 @@ async fn exec_windows_sandbox( cwd, mut env, network, + network_attempt_id, expiration, windows_sandbox_level, .. } = params; if let Some(network) = network.as_ref() { - network.apply_to_env(&mut env); + network.apply_to_env_for_attempt(&mut env, network_attempt_id.as_deref()); } // TODO(iceweasel-oai): run_windows_sandbox_capture should support all @@ -656,28 +664,26 @@ pub(crate) fn is_likely_sandbox_denied( false } -pub(crate) async fn blocking_network_policy_decision_from_blocked_queue( +pub(crate) async fn blocking_network_policy_decision_from_attempt( network: &NetworkProxy, - blocked_cursor: u64, + attempt_id: &str, sandbox_policy: &SandboxPolicy, ) -> Option { - let blocked = match network.blocked_requests_since(blocked_cursor).await { - Ok(blocked) => blocked, + let entry = match network.latest_blocked_request_for_attempt(attempt_id).await { + Ok(entry) => entry, Err(err) => { tracing::debug!( - "failed to load blocked telemetry since cursor {blocked_cursor}: {err:#}" + "failed to load blocked telemetry for network attempt {attempt_id}: {err:#}" ); return None; } }; - tracing::debug!( - "loaded {} blocked telemetry entries since cursor {blocked_cursor}", - blocked.len() - ); - let selected = select_network_policy_decision_from_blocked_entries(blocked, sandbox_policy); - if let Some(payload) = selected.as_ref() { + let payload = entry + .as_ref() + .and_then(|entry| network_policy_decision_from_blocked_entry(entry, sandbox_policy)); + if let Some(payload) = payload.as_ref() { tracing::debug!( - "selected telemetry network decision (decision={}, source={}, host={:?}, protocol={:?}, port={:?})", + "selected telemetry network decision for attempt {attempt_id} (decision={}, source={}, host={:?}, protocol={:?}, port={:?})", payload.decision, payload.source, payload.host, @@ -685,44 +691,60 @@ pub(crate) async fn blocking_network_policy_decision_from_blocked_queue( payload.port ); } else { - tracing::debug!("no telemetry network decision selected since cursor {blocked_cursor}"); + tracing::debug!("no blocked telemetry entry found for network attempt {attempt_id}"); } - selected + payload } -pub(crate) fn select_network_policy_decision_from_blocked_entries( - blocked: Vec, +pub(crate) async fn blocking_network_policy_decision_from_attempt_or_cursor( + network: &NetworkProxy, + attempt_id: Option<&str>, + blocked_cursor: Option, sandbox_policy: &SandboxPolicy, ) -> Option { - let mut latest_blocking_decision = None; - - for entry in blocked.into_iter().rev() { - let Some(payload) = network_policy_decision_from_blocked_entry(&entry, sandbox_policy) - else { - continue; - }; - - // If the command produced an ask-from-decider block, prefer that for retry - // prompting even if a later deny was also recorded. - if payload.is_ask_from_decider() { - tracing::debug!( - "selected ask-from-decider telemetry decision for host {:?}", - payload.host - ); - return Some(payload); - } - if latest_blocking_decision.is_none() { - tracing::debug!( - "recorded fallback telemetry decision candidate (decision={}, source={}, host={:?})", - payload.decision, - payload.source, - payload.host - ); - latest_blocking_decision = Some(payload); - } + if let Some(attempt_id) = attempt_id + && let Some(payload) = + blocking_network_policy_decision_from_attempt(network, attempt_id, sandbox_policy).await + { + return Some(payload); } - latest_blocking_decision + let blocked_cursor = blocked_cursor?; + blocking_network_policy_decision_from_blocked_queue(network, blocked_cursor, sandbox_policy) + .await +} + +pub(crate) async fn blocking_network_policy_decision_from_blocked_queue( + network: &NetworkProxy, + cursor: u64, + sandbox_policy: &SandboxPolicy, +) -> Option { + let entries = match network.blocked_requests_since(cursor).await { + Ok(entries) => entries, + Err(err) => { + tracing::debug!( + "failed to load blocked telemetry entries since cursor {cursor}: {err:#}" + ); + return None; + } + }; + select_network_policy_decision_from_blocked_entries(&entries, sandbox_policy) +} + +fn select_network_policy_decision_from_blocked_entries( + entries: &[codex_network_proxy::BlockedRequest], + sandbox_policy: &SandboxPolicy, +) -> Option { + let mapped: Vec = entries + .iter() + .filter_map(|entry| network_policy_decision_from_blocked_entry(entry, sandbox_policy)) + .collect(); + mapped + .iter() + .rev() + .find(|payload| payload.is_ask_from_decider()) + .cloned() + .or_else(|| mapped.last().cloned()) } fn network_policy_decision_from_blocked_entry( @@ -902,14 +924,19 @@ async fn exec( let ExecParams { command, cwd, - env, + mut env, network, + network_attempt_id, arg0, expiration, windows_sandbox_level: _, .. } = params; + if let Some(network) = network.as_ref() { + network.apply_to_env_for_attempt(&mut env, network_attempt_id.as_deref()); + } + let (program, args) = command.split_first().ok_or_else(|| { CodexErr::Io(io::Error::new( io::ErrorKind::InvalidInput, @@ -923,7 +950,7 @@ async fn exec( arg0: arg0_ref, cwd, sandbox_policy, - network: network.as_ref(), + network: None, stdio_policy: StdioPolicy::RedirectForShellTool, env, }) @@ -1213,6 +1240,7 @@ mod tests { method: Some("GET".to_string()), mode: None, protocol: "http-connect".to_string(), + attempt_id: None, decision: Some("ask".to_string()), source: Some("decider".to_string()), port: Some(443), @@ -1244,6 +1272,7 @@ mod tests { method: Some("GET".to_string()), mode: None, protocol: "http".to_string(), + attempt_id: None, decision: Some("allow".to_string()), source: Some("decider".to_string()), port: Some(80), @@ -1266,6 +1295,7 @@ mod tests { method: Some("GET".to_string()), mode: None, protocol: "http".to_string(), + attempt_id: None, decision: Some("ask".to_string()), source: None, port: Some(80), @@ -1279,44 +1309,6 @@ mod tests { assert_eq!(payload.source, "decider"); } - #[test] - fn select_network_policy_decision_prefers_ask_from_decider_over_newer_deny() { - let blocked = vec![ - BlockedRequest { - host: "google.com".to_string(), - reason: "not_allowed".to_string(), - client: None, - method: Some("GET".to_string()), - mode: None, - protocol: "http".to_string(), - decision: Some("ask".to_string()), - source: Some("decider".to_string()), - port: Some(80), - timestamp: 200, - }, - BlockedRequest { - host: "google.com".to_string(), - reason: "not_allowed".to_string(), - client: None, - method: Some("GET".to_string()), - mode: None, - protocol: "http".to_string(), - decision: Some("deny".to_string()), - source: Some("baseline_policy".to_string()), - port: Some(80), - timestamp: 201, - }, - ]; - - let decision = select_network_policy_decision_from_blocked_entries( - blocked, - &restricted_sandbox_policy(), - ) - .expect("an ask decision should be selected"); - assert_eq!(decision.decision, "ask"); - assert_eq!(decision.source, "decider"); - } - #[test] fn yolo_sandbox_denies_allowlist_miss() { let entry = BlockedRequest { @@ -1326,6 +1318,7 @@ mod tests { method: Some("GET".to_string()), mode: None, protocol: "http".to_string(), + attempt_id: None, decision: Some("ask".to_string()), source: Some("decider".to_string()), port: Some(80), @@ -1348,6 +1341,88 @@ mod tests { ); } + #[test] + fn select_network_policy_decision_prefers_ask_from_decider() { + let entries = vec![ + BlockedRequest { + host: "google.com".to_string(), + reason: "not_allowed".to_string(), + client: None, + method: Some("GET".to_string()), + mode: None, + protocol: "http".to_string(), + attempt_id: Some("attempt-1".to_string()), + decision: Some("ask".to_string()), + source: Some("decider".to_string()), + port: Some(80), + timestamp: 100, + }, + BlockedRequest { + host: "example.com".to_string(), + reason: "denied".to_string(), + client: None, + method: Some("GET".to_string()), + mode: None, + protocol: "http".to_string(), + attempt_id: Some("attempt-2".to_string()), + decision: Some("deny".to_string()), + source: Some("baseline_policy".to_string()), + port: Some(80), + timestamp: 200, + }, + ]; + + let selected = select_network_policy_decision_from_blocked_entries( + &entries, + &restricted_sandbox_policy(), + ) + .expect("expected a structured decision from blocked entries"); + assert_eq!(selected.decision, "ask"); + assert_eq!(selected.source, "decider"); + assert_eq!(selected.host.as_deref(), Some("google.com")); + } + + #[test] + fn select_network_policy_decision_uses_newest_when_no_ask() { + let entries = vec![ + BlockedRequest { + host: "old.example.com".to_string(), + reason: "denied".to_string(), + client: None, + method: Some("GET".to_string()), + mode: None, + protocol: "http".to_string(), + attempt_id: None, + decision: Some("deny".to_string()), + source: Some("baseline_policy".to_string()), + port: Some(80), + timestamp: 100, + }, + BlockedRequest { + host: "new.example.com".to_string(), + reason: "method_not_allowed".to_string(), + client: None, + method: Some("CONNECT".to_string()), + mode: None, + protocol: "http-connect".to_string(), + attempt_id: None, + decision: Some("deny".to_string()), + source: Some("mode_guard".to_string()), + port: Some(443), + timestamp: 200, + }, + ]; + + let selected = select_network_policy_decision_from_blocked_entries( + &entries, + &SandboxPolicy::DangerFullAccess, + ) + .expect("expected a structured decision from blocked entries"); + assert_eq!(selected.decision, "deny"); + assert_eq!(selected.host.as_deref(), Some("new.example.com")); + assert_eq!(selected.protocol.as_deref(), Some("https_connect")); + } + #[tokio::test] async fn read_capped_limits_retained_bytes() { let (mut writer, reader) = tokio::io::duplex(1024); @@ -1471,6 +1546,7 @@ mod tests { expiration: 500.into(), env, network: None, + network_attempt_id: None, sandbox_permissions: SandboxPermissions::UseDefault, windows_sandbox_level: codex_protocol::config_types::WindowsSandboxLevel::Disabled, justification: None, @@ -1524,6 +1600,7 @@ mod tests { expiration: ExecExpiration::Cancellation(cancel_token), env, network: None, + network_attempt_id: None, sandbox_permissions: SandboxPermissions::UseDefault, windows_sandbox_level: codex_protocol::config_types::WindowsSandboxLevel::Disabled, justification: None, diff --git a/codex-rs/core/src/sandboxing/mod.rs b/codex-rs/core/src/sandboxing/mod.rs index 9a72b6e249..487278b619 100644 --- a/codex-rs/core/src/sandboxing/mod.rs +++ b/codex-rs/core/src/sandboxing/mod.rs @@ -46,6 +46,7 @@ pub struct ExecRequest { pub cwd: PathBuf, pub env: HashMap, pub network: Option, + pub network_attempt_id: Option, pub expiration: ExecExpiration, pub sandbox: SandboxType, pub windows_sandbox_level: WindowsSandboxLevel, @@ -214,6 +215,7 @@ impl SandboxManager { cwd: spec.cwd, env, network: network.cloned(), + network_attempt_id: None, expiration: spec.expiration, sandbox, windows_sandbox_level, diff --git a/codex-rs/core/src/tools/handlers/shell.rs b/codex-rs/core/src/tools/handlers/shell.rs index b9e2a97d6d..2540426ec1 100644 --- a/codex-rs/core/src/tools/handlers/shell.rs +++ b/codex-rs/core/src/tools/handlers/shell.rs @@ -54,6 +54,7 @@ impl ShellHandler { expiration: params.timeout_ms.into(), env: create_env(&turn_context.shell_environment_policy, Some(thread_id)), network: turn_context.network.clone(), + network_attempt_id: None, sandbox_permissions: params.sandbox_permissions.unwrap_or_default(), windows_sandbox_level: turn_context.windows_sandbox_level, justification: params.justification.clone(), @@ -83,6 +84,7 @@ impl ShellCommandHandler { expiration: params.timeout_ms.into(), env: create_env(&turn_context.shell_environment_policy, Some(thread_id)), network: turn_context.network.clone(), + network_attempt_id: None, sandbox_permissions: params.sandbox_permissions.unwrap_or_default(), windows_sandbox_level: turn_context.windows_sandbox_level, justification: params.justification.clone(), diff --git a/codex-rs/core/src/tools/runtimes/shell.rs b/codex-rs/core/src/tools/runtimes/shell.rs index c4047dd750..ecd0d273ba 100644 --- a/codex-rs/core/src/tools/runtimes/shell.rs +++ b/codex-rs/core/src/tools/runtimes/shell.rs @@ -28,6 +28,7 @@ use codex_network_proxy::NetworkProxy; use codex_protocol::protocol::ReviewDecision; use futures::future::BoxFuture; use std::path::PathBuf; +use uuid::Uuid; #[derive(Clone, Debug)] pub struct ShellRequest { @@ -168,12 +169,35 @@ impl ToolRuntime for ShellRuntime { req.sandbox_permissions, req.justification.clone(), )?; - let env = attempt + let mut env = attempt .env_for(spec, req.network.as_ref()) .map_err(|err| ToolError::Codex(err.into()))?; + + let network_attempt_id = req.network.as_ref().map(|_| Uuid::new_v4().to_string()); + if let Some(attempt_id) = network_attempt_id.as_ref() { + ctx.session + .register_network_approval_attempt( + attempt_id.clone(), + ctx.turn.sub_id.clone(), + ctx.call_id.clone(), + req.command.clone(), + req.cwd.clone(), + ) + .await; + } + env.network_attempt_id = network_attempt_id.clone(); + let out = execute_env(env, attempt.policy, Self::stdout_stream(ctx)) .await - .map_err(ToolError::Codex)?; + .map_err(ToolError::Codex); + + if let Some(attempt_id) = network_attempt_id.as_deref() { + ctx.session + .unregister_network_approval_attempt(attempt_id) + .await; + } + + let out = out?; Ok(out) } } diff --git a/codex-rs/core/src/tools/runtimes/unified_exec.rs b/codex-rs/core/src/tools/runtimes/unified_exec.rs index fbd72db7d3..eb45ccc3f5 100644 --- a/codex-rs/core/src/tools/runtimes/unified_exec.rs +++ b/codex-rs/core/src/tools/runtimes/unified_exec.rs @@ -40,6 +40,7 @@ pub struct UnifiedExecRequest { pub cwd: PathBuf, pub env: HashMap, pub network: Option, + pub network_attempt_id: Option, pub tty: bool, pub sandbox_permissions: SandboxPermissions, pub justification: Option, @@ -166,7 +167,7 @@ impl<'a> ToolRuntime for UnifiedExecRunt let mut env = req.env.clone(); if let Some(network) = req.network.as_ref() { - network.apply_to_env(&mut env); + network.apply_to_env_for_attempt(&mut env, req.network_attempt_id.as_deref()); } let spec = build_command_spec( &command, diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index 7caf6a0836..8eb1367e62 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -11,8 +11,9 @@ use tokio::sync::mpsc; use tokio::time::Duration; use tokio::time::Instant; use tokio_util::sync::CancellationToken; +use uuid::Uuid; -use crate::exec::blocking_network_policy_decision_from_blocked_queue; +use crate::exec::blocking_network_policy_decision_from_attempt_or_cursor; use crate::exec_env::create_env; use crate::exec_policy::ExecApprovalRequest; use crate::network_policy_decision::NetworkPolicyDecisionPayload; @@ -202,10 +203,31 @@ impl UnifiedExecProcessManager { None }; + let network_attempt_id = request.network.as_ref().map(|_| Uuid::new_v4().to_string()); let blocked_cursor = match request.network.as_ref() { - Some(network) => network.blocked_requests_cursor().await.ok(), + Some(network) => match network.blocked_requests_cursor().await { + Ok(cursor) => Some(cursor), + Err(err) => { + tracing::debug!( + "failed to read blocked telemetry cursor before unified exec: {err:#}" + ); + None + } + }, None => None, }; + if let Some(attempt_id) = network_attempt_id.as_ref() { + context + .session + .register_network_approval_attempt( + attempt_id.clone(), + context.turn.sub_id.clone(), + context.call_id.clone(), + request.command.clone(), + cwd.clone(), + ) + .await; + } let process = self .open_session_with_sandbox( @@ -213,12 +235,19 @@ impl UnifiedExecProcessManager { cwd.clone(), context, retried_after_network_approval, + network_attempt_id.clone(), ) .await; let process = match process { Ok(process) => Arc::new(process), Err(err) => { + if let Some(attempt_id) = network_attempt_id.as_deref() { + context + .session + .unregister_network_approval_attempt(attempt_id) + .await; + } if let Some((network, host)) = temporary_allowed_host { network.revoke_temporary_allowed_host(&host).await; } @@ -279,17 +308,24 @@ impl UnifiedExecProcessManager { ) .await; - let network_policy_decision = match (request.network.as_ref(), blocked_cursor) { - (Some(network), Some(cursor)) => { - blocking_network_policy_decision_from_blocked_queue( + let network_policy_decision = match request.network.as_ref() { + Some(network) => { + blocking_network_policy_decision_from_attempt_or_cursor( network, - cursor, + network_attempt_id.as_deref(), + blocked_cursor, &context.turn.sandbox_policy, ) .await } - _ => None, + None => None, }; + if let Some(attempt_id) = network_attempt_id.as_deref() { + context + .session + .unregister_network_approval_attempt(attempt_id) + .await; + } if let Some((network, host)) = temporary_allowed_host { network.revoke_temporary_allowed_host(&host).await; @@ -370,6 +406,12 @@ impl UnifiedExecProcessManager { if let Some((network, host)) = temporary_allowed_host { network.revoke_temporary_allowed_host(&host).await; } + if let Some(attempt_id) = network_attempt_id.as_deref() { + context + .session + .unregister_network_approval_attempt(attempt_id) + .await; + } let original_token_count = approx_token_count(&text); let response = UnifiedExecResponse { @@ -641,6 +683,7 @@ impl UnifiedExecProcessManager { cwd: PathBuf, context: &UnifiedExecContext, skip_command_approval: bool, + network_attempt_id: Option, ) -> Result { let env = apply_unified_exec_env(create_env( &context.turn.shell_environment_policy, @@ -674,6 +717,7 @@ impl UnifiedExecProcessManager { cwd, env, network: request.network.clone(), + network_attempt_id, tty: request.tty, sandbox_permissions: request.sandbox_permissions, justification: request.justification.clone(), diff --git a/codex-rs/core/tests/suite/exec.rs b/codex-rs/core/tests/suite/exec.rs index c3cb15875b..c857affd2b 100644 --- a/codex-rs/core/tests/suite/exec.rs +++ b/codex-rs/core/tests/suite/exec.rs @@ -37,6 +37,7 @@ async fn run_test_cmd(tmp: TempDir, cmd: Vec<&str>) -> Result