openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-04 15:08:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Harden exec-scoped proxy attribution

Browse Source
This commit is contained in:
Winston Howes
2026-06-23 04:23:39 -04:00
parent 2a058af7a7
commit 313ff96e8e
6 changed files with 197 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
codex-rs/core/src/tools/network_approval.rs
View File

@@ -60,6 +60,7 @@ pub(crate) struct DeferredNetworkApproval {
registration_id: String,
cancellation_token: CancellationToken,
finish_outcome: Arc<OnceCell<Option<NetworkApprovalOutcome>>>,
_network: Option<NetworkProxy>,
}
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<String, Arc<ActiveNetworkApprovalCall>>,
call_outcomes: HashMap<String, NetworkApprovalOutcome>,
// Keep each listener alive for the same lifetime as its approval registration.
request_scopes: HashMap<String, NetworkProxy>,
}
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<NetworkApprovalOutcome> {
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),

1
codex-rs/core/src/tools/network_approval_tests.rs
View File

@@ -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(

85
codex-rs/core/tests/suite/network_approval.rs
View File

@@ -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::<Result<Vec<_>>>()?;
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<TestCodex> {
async fn managed_network_unified_exec_test(
server: &wiremock::MockServer,
allow_local_binding: bool,
) -> Result<TestCodex> {
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<TurnEnvironmentSelection>,
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,

163
codex-rs/network-proxy/src/proxy.rs
View File

@@ -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<JoinHandle<Result<()>>>,
}
struct RequestScopedProxy {
environment_id: String,
http_task: JoinHandle<Result<()>>,
socks_task: Option<JoinHandle<Result<()>>>,
}
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<NetworkProxyState>,
@@ -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<Self> {
#[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<config::NetworkProxyConfig> {
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 {

107
codex-rs/network-proxy/src/proxy/request_scope.rs Normal file
View File

@@ -0,0 +1,107 @@
use super::*;
pub(super) struct RequestScopedProxy {
pub(super) environment_id: String,
http_task: JoinHandle<Result<()>>,
socks_task: Option<JoinHandle<Result<()>>>,
}
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<Self> {
#[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;

42
codex-rs/network-proxy/src/proxy/request_scope_tests.rs Normal file
View File

@@ -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(())
}