From c79dfb1d9cc531e0b091b802edd44665edee9b81 Mon Sep 17 00:00:00 2001 From: viyatb-oai Date: Fri, 13 Nov 2026 08:37:58 +0000 Subject: [PATCH] Centralize managed network proxy startup (#61337) ## What changed Add `NetworkProxy::start_managed` and use it in core and exec-server to share proxy construction, startup, and error context. Keep approval-policy selection in core and pass the active decider and blocked-request observer to the shared helper. ## Testing Expand session coverage for a full-access to workspace-write profile change: verify decider calls with approvals enabled, approvals disabled, and a final managed allowlist, and verify blocked-request notifications in each case. GitOrigin-RevId: 97abdaa9af698166f46d74e0f231ec557d346200 --- .../core/src/config/network_proxy_spec.rs | 46 ++++----- .../core/src/session/network_policy_tests.rs | 94 ++++++++++++++++++- codex-rs/core/src/session/tests.rs | 76 --------------- codex-rs/exec-server/src/process_sandbox.rs | 22 ++--- codex-rs/network-proxy/src/proxy.rs | 27 ++++++ 5 files changed, 147 insertions(+), 118 deletions(-) diff --git a/codex-rs/core/src/config/network_proxy_spec.rs b/codex-rs/core/src/config/network_proxy_spec.rs index f7d2c2b621..f47a25ea1a 100644 --- a/codex-rs/core/src/config/network_proxy_spec.rs +++ b/codex-rs/core/src/config/network_proxy_spec.rs @@ -229,33 +229,25 @@ impl NetworkProxySpec { Platform::native(), local_binding_policy, )?; - let mut builder = NetworkProxy::builder() - .state(Arc::new(state)) - .managed_proxy_routing(managed_proxy_routing); - let retained_policy_decider = policy_decider.as_ref().map(Arc::clone); - if enable_network_approval_flow && !self.hard_deny_allowlist_misses { - if let Some(policy_decider) = policy_decider { - builder = builder.policy_decider_arc(policy_decider); - } else if Self::managed_sandbox_active(permission_profile) { - builder = builder - .policy_decider(|_request| async { NetworkDecision::ask("not_allowed") }); - } - } - if let Some(blocked_request_observer) = blocked_request_observer { - builder = builder.blocked_request_observer_arc(blocked_request_observer); - } - let proxy = builder.build().await.map_err(|err| { - std::io::Error::other(format!("failed to build network proxy: {err}")) - })?; - let handle = proxy - .run() - .await - .map_err(|err| std::io::Error::other(format!("failed to run network proxy: {err}")))?; - Ok(StartedNetworkProxy::new( - proxy, - handle, - retained_policy_decider, - )) + let active_decider: Option> = + if !enable_network_approval_flow || self.hard_deny_allowlist_misses { + None + } else if policy_decider.is_none() && Self::managed_sandbox_active(permission_profile) { + Some(Arc::new(|_request| async { + NetworkDecision::ask("not_allowed") + })) + } else { + policy_decider.clone() + }; + let (proxy, handle) = NetworkProxy::start_managed( + state, + managed_proxy_routing, + active_decider, + blocked_request_observer, + ) + .await + .map_err(std::io::Error::other)?; + Ok(StartedNetworkProxy::new(proxy, handle, policy_decider)) } pub(crate) fn recompute_for_permission_profile( diff --git a/codex-rs/core/src/session/network_policy_tests.rs b/codex-rs/core/src/session/network_policy_tests.rs index 1cd31e405e..82cc0a3c40 100644 --- a/codex-rs/core/src/session/network_policy_tests.rs +++ b/codex-rs/core/src/session/network_policy_tests.rs @@ -1,6 +1,6 @@ #![cfg(test)] -//! Compares session network enforcement with the policy sent to MCP servers. +//! Tests session proxy policy across permission changes and MCP launches. use super::*; use pretty_assertions::assert_eq; @@ -186,3 +186,95 @@ async fn failed_network_rule_write_keeps_native_denial_in_mcp_policy( ); Ok(()) } + +#[test_case(true, false, 1; "approvals enabled")] +#[test_case(false, false, 0; "approvals disabled")] +#[test_case(true, true, 0; "managed allowlist is final")] +#[tokio::test] +async fn managed_network_proxy_respects_approval_policy_after_profile_change( + approvals_enabled: bool, + managed_allowed_domains_only: bool, + expected_decider_calls: usize, +) -> anyhow::Result<()> { + let full_access_permission_profile = PermissionProfile::Disabled; + let spec = crate::config::NetworkProxySpec::from_config_and_constraints( + NetworkProxyConfig::default(), + Some(NetworkConstraints { + enabled: Some(true), + managed_allowed_domains_only: Some(managed_allowed_domains_only), + ..Default::default() + }), + &full_access_permission_profile, + )?; + let exec_policy = Policy::empty(); + let decider_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let network_policy_decider: Arc = Arc::new({ + let decider_calls = Arc::clone(&decider_calls); + move |_request| { + decider_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async { codex_network_proxy::NetworkDecision::ask("not_allowed") } + } + }); + + let (blocked_tx, mut blocked_rx) = tokio::sync::mpsc::unbounded_channel(); + let observer = move |request: codex_network_proxy::BlockedRequest| { + blocked_tx + .send(request.host) + .expect("blocked request receiver"); + std::future::ready(()) + }; + let (started_proxy, _) = Session::start_managed_network_proxy( + &spec, + &exec_policy, + &full_access_permission_profile, + SandboxType::None, + Some(network_policy_decider), + Some(Arc::new(observer)), + approvals_enabled, + crate::config::NetworkProxyAuditMetadata::default(), + ) + .await?; + + let spec = spec.recompute_for_permission_profile(&PermissionProfile::workspace_write())?; + spec.apply_to_started_proxy(&started_proxy).await?; + let current_cfg = started_proxy.proxy().current_cfg().await?; + assert_eq!(current_cfg.allowed_domains(), None); + + use tokio::io::AsyncReadExt as _; + use tokio::io::AsyncWriteExt as _; + + let prepared = started_proxy + .proxy() + .prepare_for_remote_environment(std::collections::HashMap::new(), "test-bridge")?; + let proxy_addr = prepared.env["HTTP_PROXY"] + .strip_prefix("http://") + .expect("HTTP proxy URL") + .parse::()?; + let mut stream = tokio::net::TcpStream::connect(proxy_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 mut buffer = [0_u8; 4096]; + let bytes_read = tokio::time::timeout(StdDuration::from_secs(2), stream.read(&mut buffer)) + .await + .expect("timed out waiting for proxy response")?; + let response = String::from_utf8_lossy(&buffer[..bytes_read]); + + assert!( + response.starts_with("HTTP/1.1 403 Forbidden"), + "unexpected proxy response: {response}" + ); + assert!( + response.contains("x-proxy-error: blocked-by-allowlist"), + "unexpected proxy response: {response}" + ); + assert_eq!(blocked_rx.try_recv()?, "example.com"); + assert_eq!( + decider_calls.load(std::sync::atomic::Ordering::SeqCst), + expected_decider_calls, + "unexpected proxy response: {response}" + ); + Ok(()) +} diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index df78439180..8076bb618e 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -1221,82 +1221,6 @@ async fn start_managed_network_proxy_applies_execpolicy_network_rules() -> anyho Ok(()) } -#[tokio::test] -async fn managed_network_proxy_decider_survives_full_access_start() -> anyhow::Result<()> { - let full_access_permission_profile = PermissionProfile::Disabled; - let spec = crate::config::NetworkProxySpec::from_config_and_constraints( - NetworkProxyConfig::default(), - Some(NetworkConstraints { - enabled: Some(true), - ..Default::default() - }), - &full_access_permission_profile, - )?; - let exec_policy = Policy::empty(); - let decider_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); - let network_policy_decider: Arc = Arc::new({ - let decider_calls = Arc::clone(&decider_calls); - move |_request| { - decider_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); - async { codex_network_proxy::NetworkDecision::ask("not_allowed") } - } - }); - - let (started_proxy, _) = Session::start_managed_network_proxy( - &spec, - &exec_policy, - &full_access_permission_profile, - SandboxType::None, - Some(network_policy_decider), - /*blocked_request_observer*/ None, - /*managed_network_requirements_enabled*/ true, - crate::config::NetworkProxyAuditMetadata::default(), - ) - .await?; - - let spec = spec.recompute_for_permission_profile(&PermissionProfile::workspace_write())?; - spec.apply_to_started_proxy(&started_proxy).await?; - let current_cfg = started_proxy.proxy().current_cfg().await?; - assert_eq!(current_cfg.allowed_domains(), None); - - use tokio::io::AsyncReadExt as _; - use tokio::io::AsyncWriteExt as _; - - let prepared = started_proxy - .proxy() - .prepare_for_remote_environment(std::collections::HashMap::new(), "test-bridge")?; - let proxy_addr = prepared.env["HTTP_PROXY"] - .strip_prefix("http://") - .expect("HTTP proxy URL") - .parse::()?; - let mut stream = tokio::net::TcpStream::connect(proxy_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 mut buffer = [0_u8; 4096]; - let bytes_read = tokio::time::timeout(StdDuration::from_secs(2), stream.read(&mut buffer)) - .await - .expect("timed out waiting for proxy response")?; - let response = String::from_utf8_lossy(&buffer[..bytes_read]); - - assert!( - response.starts_with("HTTP/1.1 403 Forbidden"), - "unexpected proxy response: {response}" - ); - assert!( - response.contains("x-proxy-error: blocked-by-allowlist"), - "unexpected proxy response: {response}" - ); - assert_eq!( - decider_calls.load(std::sync::atomic::Ordering::SeqCst), - 1, - "unexpected proxy response: {response}" - ); - Ok(()) -} - #[tokio::test] async fn new_turn_refreshes_managed_network_proxy_for_sandbox_change() -> anyhow::Result<()> { let (session, _turn_context) = make_session_and_context().await; diff --git a/codex-rs/exec-server/src/process_sandbox.rs b/codex-rs/exec-server/src/process_sandbox.rs index 057f4fda79..05bb9b232b 100644 --- a/codex-rs/exec-server/src/process_sandbox.rs +++ b/codex-rs/exec-server/src/process_sandbox.rs @@ -366,20 +366,14 @@ async fn prepare_managed_network( registration_id: registration.executor_registration_id.clone(), }), }); - let mut builder = NetworkProxy::builder() - .state(Arc::new(state)) - .managed_proxy_routing(routing); - if let Some(network_policy_decider) = network_policy_decider { - builder = builder.policy_decider_arc(network_policy_decider); - } - let proxy = builder - .build() - .await - .map_err(|err| internal_error(format!("failed to build executor network proxy: {err}")))?; - let handle = proxy - .run() - .await - .map_err(|err| internal_error(format!("failed to start executor network proxy: {err}")))?; + let (proxy, handle) = NetworkProxy::start_managed( + state, + routing, + network_policy_decider, + /*blocked_request_observer*/ None, + ) + .await + .map_err(|err| internal_error(err.to_string()))?; #[cfg(target_os = "windows")] let network_proxy_restricting_sid = if routing == ManagedProxyRouting::SharedIngress { Some( diff --git a/codex-rs/network-proxy/src/proxy.rs b/codex-rs/network-proxy/src/proxy.rs index 6c7f1c5d36..a886857265 100644 --- a/codex-rs/network-proxy/src/proxy.rs +++ b/codex-rs/network-proxy/src/proxy.rs @@ -886,6 +886,33 @@ impl NetworkProxy { NetworkProxyBuilder::default() } + /// Starts a Codex-managed proxy from resolved policy. Keep the handle for as long as + /// clients need the proxy. A disabled policy returns the proxy and an inactive handle. + pub async fn start_managed( + state: NetworkProxyState, + routing: ManagedProxyRouting, + policy_decider: Option>, + blocked_request_observer: Option>, + ) -> Result<(Self, NetworkProxyHandle)> { + // Startup errors can cross the executor RPC boundary. Keep the stage and immediate cause. + let proxy = NetworkProxyBuilder { + state: Some(Arc::new(state)), + managed_by_codex: true, + managed_proxy_routing: routing, + policy_decider, + blocked_request_observer, + ..Default::default() + } + .build() + .await + .map_err(|err| anyhow::anyhow!("failed to build network proxy: {err}"))?; + let handle = proxy + .run() + .await + .map_err(|err| anyhow::anyhow!("failed to run network proxy: {err}"))?; + Ok((proxy, handle)) + } + pub fn http_addr(&self) -> SocketAddr { self.http_addr }