diff --git a/codex-rs/core/src/tools/network_approval.rs b/codex-rs/core/src/tools/network_approval.rs index f5acdd3395..3f9cb0a276 100644 --- a/codex-rs/core/src/tools/network_approval.rs +++ b/codex-rs/core/src/tools/network_approval.rs @@ -60,6 +60,7 @@ pub(crate) struct DeferredNetworkApproval { registration_id: String, cancellation_token: CancellationToken, finish_outcome: Arc>>, + _network: Option, } impl DeferredNetworkApproval { @@ -111,7 +112,7 @@ impl ActiveNetworkApproval { registration_id, mode, cancellation_token, - network: _, + network, } = self; match (mode, registration_id) { (NetworkApprovalMode::Deferred, Some(registration_id)) => { @@ -119,6 +120,7 @@ impl ActiveNetworkApproval { registration_id, cancellation_token, finish_outcome: Arc::new(OnceCell::new()), + _network: Some(network), }) } _ => None, @@ -252,8 +254,6 @@ enum ActiveNetworkApprovalAttribution { struct NetworkApprovalCallState { active_calls: IndexMap>, call_outcomes: HashMap, - // Keep each listener alive for the same lifetime as its approval registration. - request_scopes: HashMap, } pub(crate) struct NetworkApprovalService { @@ -294,9 +294,8 @@ impl NetworkApprovalService { cancellation_token: CancellationToken, ) { let mut calls = self.calls.lock().await; - let key = registration_id.clone(); calls.active_calls.insert( - key, + registration_id.clone(), Arc::new(ActiveNetworkApprovalCall { registration_id, turn_id, @@ -385,7 +384,6 @@ impl NetworkApprovalService { async fn remove_call(&self, registration_id: &str) -> Option { let mut calls = self.calls.lock().await; calls.active_calls.shift_remove(registration_id); - calls.request_scopes.remove(registration_id); calls.call_outcomes.remove(registration_id) } @@ -869,14 +867,6 @@ pub(crate) async fn begin_network_approval( cancellation_token.clone(), ) .await; - session - .services - .network_approval - .calls - .lock() - .await - .request_scopes - .insert(registration_id.clone(), network.clone()); Ok(Some(ActiveNetworkApproval { registration_id: Some(registration_id), diff --git a/codex-rs/core/src/tools/network_approval_tests.rs b/codex-rs/core/src/tools/network_approval_tests.rs index 60c7215b1b..48857dcea6 100644 --- a/codex-rs/core/src/tools/network_approval_tests.rs +++ b/codex-rs/core/src/tools/network_approval_tests.rs @@ -430,6 +430,7 @@ async fn deferred_finish_reuses_denial_result_after_first_consumer() { registration_id: "registration-1".to_string(), cancellation_token, finish_outcome: Arc::new(OnceCell::new()), + _network: None, }; service .record_call_outcome( diff --git a/codex-rs/core/tests/suite/network_approval.rs b/codex-rs/core/tests/suite/network_approval.rs index 2c63cd071c..a0ef2ad6c2 100644 --- a/codex-rs/core/tests/suite/network_approval.rs +++ b/codex-rs/core/tests/suite/network_approval.rs @@ -1,5 +1,6 @@ use anyhow::Context; use anyhow::Result; +use codex_config::test_support::CloudConfigBundleFixture; use codex_config::types::ApprovalsReviewer; use codex_core::config::Constrained; use codex_exec_server::CreateDirectoryOptions; @@ -23,7 +24,6 @@ use codex_utils_path_uri::PathUri; use core_test_support::PathBufExt; use core_test_support::PathExt; use core_test_support::get_remote_test_env; -use core_test_support::managed_network_requirements_loader; use core_test_support::responses::ResponseMock; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; @@ -55,8 +55,6 @@ use tempfile::TempDir; const NETWORK_TEST_HOST: &str = "codex-network-test.invalid"; const NETWORK_TEST_TARGET: &str = "http://codex-network-test.invalid:80"; -const FIRST_CONCURRENT_NETWORK_TEST_HOST: &str = "first.codex-network-test.invalid"; -const SECOND_CONCURRENT_NETWORK_TEST_HOST: &str = "second.codex-network-test.invalid"; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardian_receives_exact_triggers_for_concurrent_network_requests() -> Result<()> { @@ -66,7 +64,7 @@ async fn guardian_receives_exact_triggers_for_concurrent_network_requests() -> R skip_if_windows!(Ok(())); let server = start_mock_server().await; - let test = managed_network_unified_exec_test(&server).await?; + let test = managed_network_unified_exec_test(&server, /*allow_local_binding*/ false).await?; let barrier_dir = TempDir::new_in(test.cwd.path())?; let first_marker = barrier_dir.path().join("first"); let second_marker = barrier_dir.path().join("second"); @@ -77,16 +75,8 @@ async fn guardian_receives_exact_triggers_for_concurrent_network_requests() -> R peer_marker.display(), ) }; - let first_command = network_command( - &first_marker, - &second_marker, - FIRST_CONCURRENT_NETWORK_TEST_HOST, - ); - let second_command = network_command( - &second_marker, - &first_marker, - SECOND_CONCURRENT_NETWORK_TEST_HOST, - ); + let first_command = network_command(&first_marker, &second_marker, "1.1.1.1"); + let second_command = network_command(&second_marker, &first_marker, "8.8.8.8"); let responses = mount_sse_sequence( &server, vec![ @@ -95,29 +85,23 @@ async fn guardian_receives_exact_triggers_for_concurrent_network_requests() -> R ev_function_call( "exec-network-first", "exec_command", - &serde_json::to_string(&network_exec_args( - LOCAL_ENVIRONMENT_ID, - &first_command, - ))?, + &serde_json::to_string(&network_exec_args(&first_command))?, ), ev_function_call( "exec-network-second", "exec_command", - &serde_json::to_string(&network_exec_args( - LOCAL_ENVIRONMENT_ID, - &second_command, - ))?, + &serde_json::to_string(&network_exec_args(&second_command))?, ), ev_completed("resp-network-concurrent"), ]), sse(vec![ ev_response_created("resp-network-guardian-1"), - ev_assistant_message("msg-network-guardian-1", r#"{"outcome":"allow"}"#), + ev_assistant_message("msg-network-guardian-1", r#"{"outcome":"deny"}"#), ev_completed("resp-network-guardian-1"), ]), sse(vec![ ev_response_created("resp-network-guardian-2"), - ev_assistant_message("msg-network-guardian-2", r#"{"outcome":"allow"}"#), + ev_assistant_message("msg-network-guardian-2", r#"{"outcome":"deny"}"#), ev_completed("resp-network-guardian-2"), ]), sse(vec![ @@ -134,6 +118,7 @@ async fn guardian_receives_exact_triggers_for_concurrent_network_requests() -> R "run both network requests", vec![local(test.config.cwd.clone())], ApprovalsReviewer::AutoReview, + AskForApproval::OnRequest, ) .await?; wait_for_turn_complete(&test).await; @@ -153,15 +138,14 @@ async fn guardian_receives_exact_triggers_for_concurrent_network_requests() -> R .context("expected network access JSON in Guardian request")? .trim(), )?; - let trigger = &action["trigger"]; Ok(( - trigger["callId"] - .as_str() + action + .pointer("/trigger/callId") + .and_then(Value::as_str) .context("expected exact trigger call id")? .to_string(), - trigger["command"] - .as_array() - .and_then(|command| command.last()) + action + .pointer("/trigger/command/2") .and_then(Value::as_str) .context("expected exact trigger command")? .to_string(), @@ -169,12 +153,13 @@ async fn guardian_receives_exact_triggers_for_concurrent_network_requests() -> R }) .collect::>>()?; actual_triggers.sort_unstable(); - let mut expected_triggers = vec![ - ("exec-network-first".to_string(), first_command), - ("exec-network-second".to_string(), second_command), - ]; - expected_triggers.sort_unstable(); - assert_eq!(actual_triggers, expected_triggers); + assert_eq!( + actual_triggers, + vec![ + ("exec-network-first".to_string(), first_command), + ("exec-network-second".to_string(), second_command), + ] + ); Ok(()) } @@ -190,7 +175,7 @@ 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 test = managed_network_unified_exec_test(&server, /*allow_local_binding*/ true).await?; let local_cwd = TempDir::new()?; let remote_cwd = PathBuf::from(format!( "/tmp/codex-network-approval-{}", @@ -225,6 +210,7 @@ async fn approved_network_host_for_one_environment_still_prompts_in_another() -> "fetch from the local environment", environments.clone(), ApprovalsReviewer::User, + AskForApproval::OnFailure, ) .await?; let approval = expect_network_approval(&test, LOCAL_ENVIRONMENT_ID).await?; @@ -249,6 +235,7 @@ async fn approved_network_host_for_one_environment_still_prompts_in_another() -> "fetch from the remote environment", environments.clone(), ApprovalsReviewer::User, + AskForApproval::OnFailure, ) .await?; let approval = expect_network_approval(&test, REMOTE_ENVIRONMENT_ID).await?; @@ -275,7 +262,10 @@ async fn approved_network_host_for_one_environment_still_prompts_in_another() -> Ok(()) } -async fn managed_network_unified_exec_test(server: &wiremock::MockServer) -> Result { +async fn managed_network_unified_exec_test( + server: &wiremock::MockServer, + allow_local_binding: bool, +) -> Result { let home = Arc::new(TempDir::new()?); fs::write( home.path().join("config.toml"), @@ -287,7 +277,6 @@ async fn managed_network_unified_exec_test(server: &wiremock::MockServer) -> Res [permissions.workspace.network] enabled = true mode = "limited" -allow_local_binding = true "#, )?; let approval_policy = AskForApproval::OnFailure; @@ -300,7 +289,11 @@ allow_local_binding = true let permission_profile_for_config = permission_profile.clone(); let mut builder = test_codex() .with_home(home) - .with_cloud_config_bundle(managed_network_requirements_loader()) + .with_cloud_config_bundle( + CloudConfigBundleFixture::loader_with_enterprise_requirement(format!( + "[experimental_network]\nenabled = true\nallow_local_binding = {allow_local_binding}" + )), + ) .with_config(move |config| { config.use_experimental_unified_exec_tool = true; config @@ -355,16 +348,15 @@ fn network_fetch_args(environment_id: &str) -> Value { let command = format!( "python3 -c \"import urllib.request; opener = urllib.request.build_opener(urllib.request.ProxyHandler()); print('OK:' + opener.open('http://{NETWORK_TEST_HOST}', timeout=2).read().decode(errors='replace'))\"" ); - network_exec_args(environment_id, &command) + let mut args = network_exec_args(&command); + args["environment_id"] = json!(environment_id); + args } -fn network_exec_args(environment_id: &str, command: &str) -> Value { +fn network_exec_args(command: &str) -> Value { json!({ - "shell": "/bin/sh", "cmd": command, - "login": false, "yield_time_ms": 1_000, - "environment_id": environment_id, }) } @@ -373,6 +365,7 @@ async fn submit_managed_network_turn( prompt: &str, environments: Vec, approvals_reviewer: ApprovalsReviewer, + approval_policy: AskForApproval, ) -> Result<()> { let permission_profile = PermissionProfile::workspace_write_with( &[], @@ -396,7 +389,7 @@ async fn submit_managed_network_turn( additional_context: Default::default(), thread_settings: codex_protocol::protocol::ThreadSettingsOverrides { environments: Some(turn_environment_selections), - approval_policy: Some(AskForApproval::OnFailure), + approval_policy: Some(approval_policy), approvals_reviewer: Some(approvals_reviewer), sandbox_policy: Some(sandbox_policy), permission_profile, diff --git a/codex-rs/network-proxy/src/proxy.rs b/codex-rs/network-proxy/src/proxy.rs index 56edb621ee..c442087b38 100644 --- a/codex-rs/network-proxy/src/proxy.rs +++ b/codex-rs/network-proxy/src/proxy.rs @@ -1,3 +1,5 @@ +mod request_scope; + use crate::config; use crate::http_proxy; use crate::network_policy::NetworkPolicyDecider; @@ -19,6 +21,8 @@ use std::sync::RwLock; use tokio::task::JoinHandle; use tracing::warn; +use self::request_scope::RequestScopedProxy; + #[derive(Debug, Clone, Parser)] #[command(name = "codex-network-proxy", about = "Codex network sandbox proxy")] pub struct Args {} @@ -335,21 +339,6 @@ struct EnvironmentProxy { socks_task: Option>>, } -struct RequestScopedProxy { - environment_id: String, - http_task: JoinHandle>, - socks_task: Option>>, -} - -impl Drop for RequestScopedProxy { - fn drop(&mut self) { - self.http_task.abort(); - if let Some(socks_task) = self.socks_task.as_ref() { - socks_task.abort(); - } - } -} - #[derive(Clone)] pub struct NetworkProxy { state: Arc, @@ -635,94 +624,6 @@ impl NetworkProxy { self.socks_addr } - /// Creates a proxy listener pair whose requests carry one execution-specific origin. - /// The listeners remain active until every clone of the returned proxy is dropped. - /// On Windows this currently returns the unscoped proxy because its sandbox firewall is shared. - pub fn scope_for_request(&self, environment_id: &str, request_origin: String) -> Result { - #[cfg(target_os = "windows")] - { - // Windows' offline firewall policy is shared by every sandboxed exec. Distinct - // endpoints therefore cannot be isolated safely between concurrent processes yet. - let _ = (environment_id, request_origin); - Ok(self.clone()) - } - - #[cfg(not(target_os = "windows"))] - { - anyhow::ensure!( - self.request_scope.is_none(), - "cannot create a request scope from an already scoped network proxy" - ); - - let runtime = tokio::runtime::Handle::try_current() - .context("failed to create request-scoped network proxy")?; - let listeners = reserve_loopback_ephemeral_listeners(self.socks_enabled) - .context("failed to reserve request-scoped network proxy")?; - let http_addr = listeners - .http_addr() - .context("failed to read request-scoped HTTP proxy address")?; - let socks_addr = listeners - .socks_addr(self.socks_addr) - .context("failed to read request-scoped SOCKS proxy address")?; - let ReservedListenerSet { - http_listener, - socks_listener, - } = listeners; - - let state = Arc::new(self.state.with_request_origin(request_origin)); - let http_state = Arc::clone(&state); - let http_decider = self.policy_decider.clone(); - let http_environment_id = Some(environment_id.to_string()); - let http_task = runtime.spawn(async move { - http_proxy::run_http_proxy_with_std_listener( - http_state, - http_listener, - http_decider, - http_environment_id, - ) - .await - }); - - let socks_task = if self.socks_enabled { - let socks_state = Arc::clone(&state); - let socks_decider = self.policy_decider.clone(); - let socks_environment_id = Some(environment_id.to_string()); - let socks5_udp_enabled = self.socks5_udp_enabled; - socks_listener.map(|listener| { - runtime.spawn(async move { - socks5::run_socks5_with_std_listener( - socks_state, - listener, - socks_decider, - socks_environment_id, - socks5_udp_enabled, - ) - .await - }) - }) - } else { - None - }; - - Ok(Self { - state, - http_addr, - socks_addr, - socks_enabled: self.socks_enabled, - socks5_udp_enabled: self.socks5_udp_enabled, - runtime_settings: Arc::clone(&self.runtime_settings), - reserved_listeners: None, - policy_decider: self.policy_decider.clone(), - environment_proxies: Arc::new(Mutex::new(HashMap::new())), - request_scope: Some(Arc::new(RequestScopedProxy { - environment_id: environment_id.to_string(), - http_task, - socks_task, - })), - }) - } - } - pub async fn current_cfg(&self) -> Result { self.state.current_cfg().await } @@ -1119,19 +1020,11 @@ impl Drop for NetworkProxyHandle { mod tests { use super::*; use crate::config::NetworkProxySettings; - #[cfg(not(target_os = "windows"))] - use crate::network_policy::NetworkDecision; use crate::state::network_proxy_state_for_policy; use pretty_assertions::assert_eq; use std::net::IpAddr; use std::net::Ipv4Addr; use std::path::Path; - #[cfg(not(target_os = "windows"))] - use std::time::Duration; - #[cfg(not(target_os = "windows"))] - use tokio::io::AsyncWriteExt; - #[cfg(not(target_os = "windows"))] - use tokio::time::timeout; #[tokio::test] async fn managed_proxy_builder_uses_loopback_ports() { @@ -1226,54 +1119,6 @@ mod tests { Ok(()) } - #[cfg(not(target_os = "windows"))] - #[tokio::test] - async fn request_scoped_proxy_stamps_policy_and_blocked_requests() -> Result<()> { - let (policy_tx, mut policy_rx) = tokio::sync::mpsc::unbounded_channel(); - let (blocked_tx, mut blocked_rx) = tokio::sync::mpsc::unbounded_channel(); - let state = Arc::new(network_proxy_state_for_policy( - NetworkProxySettings::default(), - )); - let proxy = NetworkProxy::builder() - .state(state) - .policy_decider( - move |request: crate::network_policy::NetworkPolicyRequest| { - let policy_tx = policy_tx.clone(); - async move { - let _ = policy_tx.send(request.request_origin); - NetworkDecision::deny("not_allowed") - } - }, - ) - .blocked_request_observer(move |request: crate::runtime::BlockedRequest| { - let blocked_tx = blocked_tx.clone(); - async move { - let _ = blocked_tx.send(request.request_origin); - } - }) - .build() - .await?; - let scoped = proxy.scope_for_request("local", "exec-1".to_string())?; - - assert_ne!(scoped.http_addr(), proxy.http_addr()); - let mut stream = tokio::net::TcpStream::connect(scoped.http_addr()).await?; - stream - .write_all( - b"GET http://example.com/ HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n", - ) - .await?; - - assert_eq!( - timeout(Duration::from_secs(2), policy_rx.recv()).await?, - Some(Some("exec-1".to_string())) - ); - assert_eq!( - timeout(Duration::from_secs(2), blocked_rx.recv()).await?, - Some(Some("exec-1".to_string())) - ); - Ok(()) - } - #[tokio::test] async fn managed_proxy_builder_does_not_reserve_socks_listener_when_disabled() { let settings = NetworkProxySettings { diff --git a/codex-rs/network-proxy/src/proxy/request_scope.rs b/codex-rs/network-proxy/src/proxy/request_scope.rs new file mode 100644 index 0000000000..19dc288b4d --- /dev/null +++ b/codex-rs/network-proxy/src/proxy/request_scope.rs @@ -0,0 +1,107 @@ +use super::*; + +pub(super) struct RequestScopedProxy { + pub(super) environment_id: String, + http_task: JoinHandle>, + socks_task: Option>>, +} + +impl Drop for RequestScopedProxy { + fn drop(&mut self) { + self.http_task.abort(); + if let Some(socks_task) = self.socks_task.as_ref() { + socks_task.abort(); + } + } +} + +impl NetworkProxy { + /// Creates an execution-scoped proxy that lives until all returned clones are dropped. + /// Windows, and macOS with local binding, fall back because loopback listeners are shared. + pub fn scope_for_request(&self, environment_id: &str, request_origin: String) -> Result { + #[cfg(target_os = "windows")] + { + // Windows' firewall is shared, so distinct endpoints cannot be isolated between execs. + let _ = (environment_id, request_origin); + Ok(self.clone()) + } + + #[cfg(target_os = "macos")] + if self.allow_local_binding() { + return Ok(self.clone()); + } + + #[cfg(not(target_os = "windows"))] + { + anyhow::ensure!( + self.request_scope.is_none(), + "cannot create a request scope from an already scoped network proxy" + ); + + let runtime = tokio::runtime::Handle::try_current()?; + let listeners = reserve_loopback_ephemeral_listeners(self.socks_enabled)?; + let http_addr = listeners.http_addr()?; + let socks_addr = listeners.socks_addr(self.socks_addr)?; + let ReservedListenerSet { + http_listener, + socks_listener, + } = listeners; + + let state = Arc::new(self.state.with_request_origin(request_origin)); + let http_state = Arc::clone(&state); + let http_decider = self.policy_decider.clone(); + let http_environment_id = Some(environment_id.to_string()); + let http_task = runtime.spawn(async move { + http_proxy::run_http_proxy_with_std_listener( + http_state, + http_listener, + http_decider, + http_environment_id, + ) + .await + }); + + let socks_task = if self.socks_enabled { + let socks_state = Arc::clone(&state); + let socks_decider = self.policy_decider.clone(); + let socks_environment_id = Some(environment_id.to_string()); + let socks5_udp_enabled = self.socks5_udp_enabled; + socks_listener.map(|listener| { + runtime.spawn(async move { + socks5::run_socks5_with_std_listener( + socks_state, + listener, + socks_decider, + socks_environment_id, + socks5_udp_enabled, + ) + .await + }) + }) + } else { + None + }; + + Ok(Self { + state, + http_addr, + socks_addr, + socks_enabled: self.socks_enabled, + socks5_udp_enabled: self.socks5_udp_enabled, + runtime_settings: Arc::clone(&self.runtime_settings), + reserved_listeners: None, + policy_decider: self.policy_decider.clone(), + environment_proxies: Arc::clone(&self.environment_proxies), + request_scope: Some(Arc::new(RequestScopedProxy { + environment_id: environment_id.to_string(), + http_task, + socks_task, + })), + }) + } + } +} + +#[cfg(all(test, not(target_os = "windows")))] +#[path = "request_scope_tests.rs"] +mod tests; diff --git a/codex-rs/network-proxy/src/proxy/request_scope_tests.rs b/codex-rs/network-proxy/src/proxy/request_scope_tests.rs new file mode 100644 index 0000000000..3294bb83b3 --- /dev/null +++ b/codex-rs/network-proxy/src/proxy/request_scope_tests.rs @@ -0,0 +1,42 @@ +use super::*; +use crate::state::network_proxy_state_for_policy; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::AsyncWriteExt; +use tokio::time::timeout; + +#[tokio::test] +async fn request_scoped_proxy_stamps_blocked_requests() -> Result<()> { + let (blocked_tx, mut blocked_rx) = tokio::sync::mpsc::unbounded_channel(); + let proxy = NetworkProxy::builder() + .state(Arc::new(network_proxy_state_for_policy(Default::default()))) + .blocked_request_observer(move |request: crate::runtime::BlockedRequest| { + let blocked_tx = blocked_tx.clone(); + async move { + let _ = blocked_tx.send(request.request_origin); + } + }) + .build() + .await?; + let scoped = proxy.scope_for_request("local", "exec-1".to_string())?; + let mut stream = tokio::net::TcpStream::connect(scoped.http_addr()).await?; + stream + .write_all( + b"GET http://example.com/ HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n", + ) + .await?; + + let blocked = timeout(Duration::from_secs(2), blocked_rx.recv()).await?; + assert_eq!(blocked, Some(Some("exec-1".to_string()))); + + #[cfg(target_os = "macos")] + { + proxy.runtime_settings.write().unwrap().allow_local_binding = true; + assert_eq!( + proxy.scope_for_request("local", "exec-2".to_string())?, + proxy + ); + } + Ok(()) +}