openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 17:28:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Harden network approval cancellation and concurrency (#35267)

Browse Source 
## What changed

- Scope pending network approvals to a turn and execution while still coalescing duplicate requests within one execution.
- Fail abandoned approval owners closed, cancel denied executions, resolve any waiting requests, and cancel in-flight Guardian reviews when their owner is dropped.
- Serialize session policy updates with approval caches so concurrent allow and deny decisions cannot leave enforcement and cached decisions inconsistent.
- Give user approval requests and permission hooks unique per-execution identifiers.

## Testing

- Cover approval deduplication boundaries, abandoned-owner cleanup, cancellation outcomes, replacement requests, and Guardian cancellation events.

GitOrigin-RevId: 11e184dd1c70294a99581b6ff0d4b6607d9ea1d9
This commit is contained in:
viyatb-oai
2026-07-25 00:10:48 +00:00
committed by copyberry
parent cba0e2701c
commit 63fe5a6b71
5 changed files with 556 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/guardian/mod.rs
-1
View File
@@ -35,7 +35,6 @@ pub(crate) use review::new_guardian_review_id;
#[cfg(test)]
pub(crate) use review::record_guardian_denial_for_test;
pub(crate) use review::review_approval_request;
#[cfg(test)]
pub(crate) use review::review_approval_request_with_cancel;
pub(crate) use review::routes_approval_policy_to_guardian;
pub(crate) use review::routes_approval_to_guardian;
codex-rs/core/src/session/mod.rs
+12
View File
@@ -2184,6 +2184,7 @@ impl Session {
&self,
amendment: &NetworkPolicyAmendment,
network_approval_context: &NetworkApprovalContext,
on_policy_applied: impl FnOnce() + Send,
) -> anyhow::Result<()> {
let _refresh_guard = self
.managed_network_proxy_refresh_lock
@@ -2201,6 +2202,7 @@ impl Session {
.clone();
let execpolicy_amendment =
execpolicy_network_rule_amendment(amendment, network_approval_context, &host);
let mut on_policy_applied = Some(on_policy_applied);
if let Some(started_network_proxy) = self.services.network_proxy.load_full() {
let proxy = started_network_proxy.proxy();
@@ -2214,6 +2216,11 @@ impl Session {
.await
.map_err(|err| anyhow::anyhow!("failed to update runtime denylist: {err}"))?,
}
// Active enforcement changed successfully. Notify the owner before
// the next fallible await so cancellation cannot contradict it.
if let Some(on_policy_applied) = on_policy_applied.take() {
on_policy_applied();
}
}
self.services
@@ -2230,6 +2237,11 @@ impl Session {
anyhow::anyhow!("failed to persist network policy amendment to execpolicy: {err}")
})?;
// Without a running proxy, persistence is the first effective policy change.
if let Some(on_policy_applied) = on_policy_applied {
on_policy_applied();
}
Ok(())
}
codex-rs/core/src/tools/network_approval.rs
+287 -103
View File
@@ -1,7 +1,8 @@
use crate::guardian::GuardianApprovalRequest;
use crate::guardian::GuardianNetworkAccessTrigger;
use crate::guardian::GuardianReviewOptions;
use crate::guardian::new_guardian_review_id;
use crate::guardian::review_approval_request;
use crate::guardian::review_approval_request_with_cancel;
use crate::guardian::routes_approval_to_guardian;
use crate::hook_runtime::run_permission_request_hooks;
use crate::network_policy_decision::denied_network_policy_message;
@@ -9,6 +10,7 @@ use crate::session::session::Session;
use crate::tools::events::truncate_rejection_message;
use crate::tools::sandboxing::PermissionRequestPayload;
use crate::tools::sandboxing::ToolError;
use codex_analytics::GuardianApprovalRequestSource;
use codex_hooks::PermissionRequestDecision;
use codex_network_proxy::BlockedRequest;
use codex_network_proxy::BlockedRequestObserver;
@@ -31,6 +33,8 @@ use std::collections::HashMap;
use std::collections::HashSet;
use std::io;
use std::sync::Arc;
use std::sync::Mutex as SyncMutex;
use std::sync::OnceLock as SyncOnceLock;
use tokio::sync::Mutex;
use tokio::sync::Notify;
use tokio::sync::OnceCell;
@@ -39,6 +43,9 @@ use tokio_util::sync::CancellationToken;
use tracing::warn;
use uuid::Uuid;
const ABANDONED_NETWORK_APPROVAL_MESSAGE: &str =
"network approval was cancelled before a decision was returned";
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum NetworkApprovalMode {
Immediate,
@@ -81,6 +88,8 @@ impl DeferredNetworkApproval {
.get_or_init(|| async { service.finish_call_outcome(&self.registration_id).await })
.await
.clone();
let outcome =
outcome.or_else(|| abandoned_network_approval_outcome(&self.cancellation_token));
network_approval_outcome_to_result(outcome)
}
}
@@ -135,6 +144,13 @@ struct HostApprovalKey {
port: u16,
}
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct PendingHostApprovalKey {
host: HostApprovalKey,
turn_id: String,
execution_id: Option<String>,
}
impl HostApprovalKey {
fn from_request(
request: &NetworkPolicyRequest,
@@ -186,6 +202,14 @@ fn network_approval_outcome_to_result(
}
}
fn abandoned_network_approval_outcome(
cancellation_token: &CancellationToken,
) -> Option<NetworkApprovalOutcome> {
cancellation_token.is_cancelled().then(|| {
NetworkApprovalOutcome::DeniedByPolicy(ABANDONED_NETWORK_APPROVAL_MESSAGE.to_string())
})
}
/// Whether an allowlist miss may be reviewed instead of hard-denied.
fn allows_network_approval_flow(policy: AskForApproval) -> bool {
!matches!(policy, AskForApproval::Never)
@@ -205,14 +229,14 @@ impl PendingApprovalDecision {
}
struct PendingHostApproval {
decision: Mutex<Option<PendingApprovalDecision>>,
decision: SyncOnceLock<PendingApprovalDecision>,
notify: Notify,
}
impl PendingHostApproval {
fn new() -> Self {
Self {
decision: Mutex::new(None),
decision: SyncOnceLock::new(),
notify: Notify::new(),
}
}
@@ -220,19 +244,17 @@ impl PendingHostApproval {
async fn wait_for_decision(&self) -> PendingApprovalDecision {
loop {
let notified = self.notify.notified();
if let Some(decision) = *self.decision.lock().await {
return decision;
if let Some(decision) = self.decision.get() {
return *decision;
}
notified.await;
}
}
async fn set_decision(&self, decision: PendingApprovalDecision) {
{
let mut current = self.decision.lock().await;
*current = Some(decision);
fn set_decision(&self, decision: PendingApprovalDecision) {
if self.decision.set(decision).is_ok() {
self.notify.notify_waiters();
}
self.notify.notify_waiters();
}
}
@@ -264,16 +286,96 @@ struct NetworkApprovalCallState {
pub(crate) struct NetworkApprovalService {
calls: Mutex<NetworkApprovalCallState>,
pending_host_approvals: Mutex<HashMap<HostApprovalKey, Arc<PendingHostApproval>>>,
// Owner cleanup runs from Drop, so this lock cannot require an async task.
pending_host_approvals: SyncMutex<HashMap<PendingHostApprovalKey, Arc<PendingHostApproval>>>,
// Keep persisted session policy and the in-memory approval caches in the same order.
session_policy_commit_lock: Mutex<()>,
session_approved_hosts: Mutex<HashSet<HostApprovalKey>>,
session_denied_hosts: Mutex<HashSet<HostApprovalKey>>,
}
/// Removes and resolves the exact pending generation created by one request.
///
/// Dropping an unfinished owner fails its existing waiters closed without
/// deleting a newer same-host request.
struct PendingHostApprovalOwner<'a> {
service: &'a NetworkApprovalService,
key: PendingHostApprovalKey,
pending: Arc<PendingHostApproval>,
execution_cancellation: Option<CancellationToken>,
decision_on_drop: PendingApprovalDecision,
completed: bool,
}
impl<'a> PendingHostApprovalOwner<'a> {
fn new(
service: &'a NetworkApprovalService,
key: PendingHostApprovalKey,
pending: Arc<PendingHostApproval>,
execution_cancellation: Option<CancellationToken>,
) -> Self {
Self {
service,
key,
pending,
execution_cancellation,
decision_on_drop: PendingApprovalDecision::Deny,
completed: false,
}
}
fn set_decision_on_drop(&mut self, decision: PendingApprovalDecision) {
self.decision_on_drop = decision;
}
fn complete(mut self, decision: PendingApprovalDecision) {
self.cancel_execution_if_denied(decision);
self.publish_and_remove(decision);
self.completed = true;
}
fn cancel_execution_if_denied(&self, decision: PendingApprovalDecision) {
if matches!(decision, PendingApprovalDecision::Deny)
&& let Some(execution_cancellation) = &self.execution_cancellation
{
execution_cancellation.cancel();
}
}
fn publish_and_remove(&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
.service
.pending_host_approvals
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if approvals
.get(&self.key)
.is_some_and(|current| Arc::ptr_eq(current, &self.pending))
{
approvals.remove(&self.key);
}
drop(approvals);
self.pending.set_decision(decision);
}
}
impl Drop for PendingHostApprovalOwner<'_> {
fn drop(&mut self) {
if !self.completed {
self.cancel_execution_if_denied(self.decision_on_drop);
self.publish_and_remove(self.decision_on_drop);
}
}
}
impl Default for NetworkApprovalService {
fn default() -> Self {
Self {
calls: Mutex::new(NetworkApprovalCallState::default()),
pending_host_approvals: Mutex::new(HashMap::new()),
pending_host_approvals: SyncMutex::new(HashMap::new()),
session_policy_commit_lock: Mutex::new(()),
session_approved_hosts: Mutex::new(HashSet::new()),
session_denied_hosts: Mutex::new(HashSet::new()),
}
@@ -283,7 +385,12 @@ impl Default for NetworkApprovalService {
impl NetworkApprovalService {
/// Replace the target session's approval cache with the source session's
/// currently approved hosts.
#[expect(
clippy::await_holding_invalid_type,
reason = "the approval snapshot must not interleave with a session policy commit"
)]
pub(crate) async fn sync_session_approved_hosts_to(&self, other: &Self) {
let _commit_guard = self.session_policy_commit_lock.lock().await;
let approved_hosts = self.session_approved_hosts.lock().await.clone();
let mut other_approved_hosts = other.session_approved_hosts.lock().await;
other_approved_hosts.clear();
@@ -401,11 +508,14 @@ impl NetworkApprovalService {
}
}
async fn get_or_create_pending_approval(
fn get_or_create_pending_approval(
&self,
key: HostApprovalKey,
key: PendingHostApprovalKey,
) -> (Arc<PendingHostApproval>, bool) {
let mut pending = self.pending_host_approvals.lock().await;
let mut pending = self
.pending_host_approvals
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(existing) = pending.get(&key).cloned() {
return (existing, false);
}
@@ -450,8 +560,16 @@ impl NetworkApprovalService {
self.remove_call(registration_id).await
}
async fn finish_call(&self, registration_id: &str) -> Result<(), ToolError> {
network_approval_outcome_to_result(self.finish_call_outcome(registration_id).await)
async fn finish_call(
&self,
registration_id: &str,
cancellation_token: &CancellationToken,
) -> Result<(), ToolError> {
let outcome = self
.finish_call_outcome(registration_id)
.await
.or_else(|| abandoned_network_approval_outcome(cancellation_token));
network_approval_outcome_to_result(outcome)
}
pub(crate) async fn record_blocked_request(&self, blocked: BlockedRequest) {
@@ -505,6 +623,10 @@ 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<Session>,
@@ -537,32 +659,26 @@ impl NetworkApprovalService {
let key = HostApprovalKey::from_request(&request, protocol, environment_id.clone());
{
let denied_hosts = self.session_denied_hosts.lock().await;
if denied_hosts.contains(&key) {
return NetworkDecision::deny(REASON_NOT_ALLOWED);
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;
}
}
let (pending, is_owner) = self.get_or_create_pending_approval(key.clone()).await;
if !is_owner {
return pending.wait_for_decision().await.to_network_decision();
}
let target = Self::format_network_target(key.protocol, request.host.as_str(), key.port);
let policy_denial_message =
format!("Network access to \"{target}\" was blocked by policy.");
let prompt_reason = format!("{} is not in the allowed_domains", request.host);
let Some(turn_context) = turn_context else {
pending.set_decision(PendingApprovalDecision::Deny).await;
self.pending_host_approvals.lock().await.remove(&key);
if let Some(owner_call) = owner_call.as_ref() {
self.record_call_outcome(
&owner_call.registration_id,
@@ -572,9 +688,29 @@ impl NetworkApprovalService {
}
return NetworkDecision::deny(REASON_NOT_ALLOWED);
};
let pending_key = PendingHostApprovalKey {
host: key.clone(),
turn_id: owner_call
.as_ref()
.map_or_else(|| turn_context.sub_id.clone(), |call| call.turn_id.clone()),
execution_id: owner_call
.as_ref()
.map(|owner_call| owner_call.registration_id.clone()),
};
let (pending, is_owner) = self.get_or_create_pending_approval(pending_key.clone());
if !is_owner {
return pending.wait_for_decision().await.to_network_decision();
}
let mut pending_owner = PendingHostApprovalOwner::new(
self,
pending_key,
Arc::clone(&pending),
owner_call
.as_ref()
.map(|call| call.cancellation_token.clone()),
);
if !permission_profile_allows_network_approval_flow(&turn_context.permission_profile()) {
pending.set_decision(PendingApprovalDecision::Deny).await;
self.pending_host_approvals.lock().await.remove(&key);
if let Some(owner_call) = owner_call.as_ref() {
self.record_call_outcome(
&owner_call.registration_id,
@@ -582,11 +718,10 @@ impl NetworkApprovalService {
)
.await;
}
pending_owner.complete(PendingApprovalDecision::Deny);
return NetworkDecision::deny(REASON_NOT_ALLOWED);
}
if !allows_network_approval_flow(turn_context.approval_policy.value()) {
pending.set_decision(PendingApprovalDecision::Deny).await;
self.pending_host_approvals.lock().await.remove(&key);
if let Some(owner_call) = owner_call.as_ref() {
self.record_call_outcome(
&owner_call.registration_id,
@@ -594,6 +729,7 @@ impl NetworkApprovalService {
)
.await;
}
pending_owner.complete(PendingApprovalDecision::Deny);
return NetworkDecision::deny(REASON_NOT_ALLOWED);
}
@@ -602,63 +738,78 @@ impl NetworkApprovalService {
protocol,
};
let guardian_approval_id = Self::approval_id_for_key(&key);
let hook_run_id_suffix = owner_call.as_ref().map_or_else(
|| guardian_approval_id.clone(),
|call| format!("{guardian_approval_id}#{}", call.registration_id),
);
let prompt_command = vec!["network-access".to_string(), target.clone()];
let command = owner_call
.as_ref()
.map_or_else(|| prompt_command.join(" "), |call| call.command.clone());
if let Some(permission_request_decision) = run_permission_request_hooks(
let hook_approval_decision = match run_permission_request_hooks(
&session,
&turn_context,
&guardian_approval_id,
&hook_run_id_suffix,
PermissionRequestPayload::bash(command, Some(format!("network-access {target}"))),
)
.await
{
match permission_request_decision {
PermissionRequestDecision::Allow => {
pending
.set_decision(PendingApprovalDecision::AllowOnce)
.await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
return NetworkDecision::Allow;
}
PermissionRequestDecision::Deny { message } => {
if let Some(owner_call) = owner_call.as_ref() {
self.record_call_outcome(
&owner_call.registration_id,
NetworkApprovalOutcome::DeniedByPolicy(message),
)
.await;
}
pending.set_decision(PendingApprovalDecision::Deny).await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
return NetworkDecision::deny(REASON_NOT_ALLOWED);
Some(PermissionRequestDecision::Allow) => Some(ReviewDecision::Approved),
Some(PermissionRequestDecision::Deny { message }) => {
if let Some(owner_call) = owner_call.as_ref() {
self.record_call_outcome(
&owner_call.registration_id,
NetworkApprovalOutcome::DeniedByPolicy(message),
)
.await;
}
pending_owner.complete(PendingApprovalDecision::Deny);
return NetworkDecision::deny(REASON_NOT_ALLOWED);
}
}
None => None,
};
let use_guardian = routes_approval_to_guardian(&turn_context);
let guardian_review_id = use_guardian.then(new_guardian_review_id);
let approval_decision = if let Some(review_id) = guardian_review_id.clone() {
review_approval_request(
&session,
&turn_context,
review_id,
GuardianApprovalRequest::NetworkAccess {
id: guardian_approval_id.clone(),
turn_id: owner_call
.as_ref()
.map_or_else(|| turn_context.sub_id.clone(), |call| call.turn_id.clone()),
target,
host: request.host,
protocol,
port: key.port,
trigger: owner_call.as_ref().map(|call| call.trigger.clone()),
},
Some(policy_denial_message.clone()),
)
.await
let approval_decision = if let Some(hook_approval_decision) = hook_approval_decision {
hook_approval_decision
} else if let Some(review_id) = guardian_review_id.clone() {
let review_cancel = CancellationToken::new();
let review_cancel_guard = review_cancel.clone().drop_guard();
let review_session = Arc::clone(&session);
let review_turn = Arc::clone(&turn_context);
let review_request = GuardianApprovalRequest::NetworkAccess {
id: guardian_approval_id.clone(),
turn_id: owner_call
.as_ref()
.map_or_else(|| turn_context.sub_id.clone(), |call| call.turn_id.clone()),
target: target.clone(),
host: request.host.clone(),
protocol,
port: key.port,
trigger: owner_call.as_ref().map(|call| call.trigger.clone()),
};
let retry_reason = Some(policy_denial_message.clone());
let review = tokio::spawn(async move {
review_approval_request_with_cancel(
&review_session,
&review_turn,
review_id,
review_request,
retry_reason,
GuardianReviewOptions {
plugin_attribution_override: None,
approval_request_source: GuardianApprovalRequestSource::MainTurn,
external_cancel: Some(review_cancel),
},
)
.await
});
let decision = review.await.unwrap_or_else(|err| {
warn!("network Guardian review task failed: {err}");
ReviewDecision::denied("automatic approval review could not complete")
});
drop(review_cancel_guard.disarm());
decision
} else {
let available_decisions = None;
let cwd = if let Some(owner_call) = owner_call.as_ref() {
@@ -674,10 +825,11 @@ impl NetworkApprovalService {
turn_context.cwd.clone()
})
};
let approval_call_id = format!("{guardian_approval_id}#{}", Uuid::new_v4());
session
.request_command_approval(
turn_context.as_ref(),
guardian_approval_id,
approval_call_id,
/*approval_id*/ None,
Some(environment_id),
prompt_command,
@@ -692,12 +844,47 @@ impl NetworkApprovalService {
.await
};
let mut cache_session_deny = false;
let _session_policy_commit_guard = if matches!(
&approval_decision,
ReviewDecision::Approved
| ReviewDecision::ApprovedExecpolicyAmendment { .. }
| ReviewDecision::ApprovedForSession
| ReviewDecision::NetworkPolicyAmendment { .. }
) {
Some(self.session_policy_commit_lock.lock().await)
} else {
None
};
let resolved = match approval_decision {
ReviewDecision::Approved | ReviewDecision::ApprovedExecpolicyAmendment { .. } => {
PendingApprovalDecision::AllowOnce
if self.session_denied_hosts.lock().await.contains(&key) {
if let Some(owner_call) = owner_call.as_ref() {
self.record_call_outcome(
&owner_call.registration_id,
NetworkApprovalOutcome::DeniedByPolicy(policy_denial_message.clone()),
)
.await;
}
PendingApprovalDecision::Deny
} else {
PendingApprovalDecision::AllowOnce
}
}
ReviewDecision::ApprovedForSession => {
if self.session_denied_hosts.lock().await.contains(&key) {
if let Some(owner_call) = owner_call.as_ref() {
self.record_call_outcome(
&owner_call.registration_id,
NetworkApprovalOutcome::DeniedByPolicy(policy_denial_message.clone()),
)
.await;
}
PendingApprovalDecision::Deny
} else {
self.session_approved_hosts.lock().await.insert(key.clone());
PendingApprovalDecision::AllowForSession
}
}
ReviewDecision::ApprovedForSession => PendingApprovalDecision::AllowForSession,
ReviewDecision::NetworkPolicyAmendment {
network_policy_amendment,
} => match network_policy_amendment.action {
@@ -706,6 +893,10 @@ impl NetworkApprovalService {
.persist_network_policy_amendment(
&network_policy_amendment,
&network_approval_context,
|| {
pending_owner
.set_decision_on_drop(PendingApprovalDecision::AllowForSession);
},
)
.await
{
@@ -729,6 +920,11 @@ impl NetworkApprovalService {
.await;
}
}
{
let mut denied_hosts = self.session_denied_hosts.lock().await;
denied_hosts.remove(&key);
}
self.session_approved_hosts.lock().await.insert(key.clone());
PendingApprovalDecision::AllowForSession
}
NetworkPolicyRuleAction::Deny => {
@@ -736,6 +932,9 @@ impl NetworkApprovalService {
.persist_network_policy_amendment(
&network_policy_amendment,
&network_approval_context,
|| {
pending_owner.set_decision_on_drop(PendingApprovalDecision::Deny);
},
)
.await
{
@@ -768,7 +967,11 @@ impl NetworkApprovalService {
)
.await;
}
cache_session_deny = true;
{
let mut approved_hosts = self.session_approved_hosts.lock().await;
approved_hosts.remove(&key);
}
self.session_denied_hosts.lock().await.insert(key.clone());
PendingApprovalDecision::Deny
}
},
@@ -817,28 +1020,9 @@ impl NetworkApprovalService {
PendingApprovalDecision::Deny
}
};
pending_owner.set_decision_on_drop(resolved);
if matches!(resolved, PendingApprovalDecision::AllowForSession) {
{
let mut denied_hosts = self.session_denied_hosts.lock().await;
denied_hosts.remove(&key);
}
let mut approved_hosts = self.session_approved_hosts.lock().await;
approved_hosts.insert(key.clone());
}
if cache_session_deny {
{
let mut approved_hosts = self.session_approved_hosts.lock().await;
approved_hosts.remove(&key);
}
let mut denied_hosts = self.session_denied_hosts.lock().await;
denied_hosts.insert(key.clone());
}
pending.set_decision(resolved).await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
pending_owner.complete(resolved);
resolved.to_network_decision()
}
@@ -938,7 +1122,7 @@ pub(crate) async fn finish_immediate_network_approval(
session
.services
.network_approval
.finish_call(registration_id)
.finish_call(registration_id, &active.cancellation_token)
.await
}
codex-rs/core/src/tools/network_approval_tests.rs
+228 -32
View File
@@ -6,73 +6,114 @@ use codex_protocol::permissions::NetworkSandboxPolicy;
use codex_protocol::protocol::AskForApproval;
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 tokio::time::timeout;
use tokio_util::sync::CancellationToken;
#[tokio::test]
async fn pending_approvals_are_deduped_per_host_protocol_and_port() {
let service = NetworkApprovalService::default();
let key = HostApprovalKey {
environment_id: "local".to_string(),
host: "example.com".to_string(),
protocol: "http",
port: 443,
};
fn pending_key(host: HostApprovalKey, turn_id: &str, execution_id: &str) -> PendingHostApprovalKey {
PendingHostApprovalKey {
host,
turn_id: turn_id.to_string(),
execution_id: Some(execution_id.to_string()),
}
}
let (first, first_is_owner) = service.get_or_create_pending_approval(key.clone()).await;
let (second, second_is_owner) = service.get_or_create_pending_approval(key).await;
#[test]
fn pending_approvals_are_deduped_within_one_execution() {
let service = NetworkApprovalService::default();
let key = pending_key(
HostApprovalKey {
environment_id: "local".to_string(),
host: "example.com".to_string(),
protocol: "http",
port: 443,
},
"turn-1",
"execution-1",
);
let (first, first_is_owner) = service.get_or_create_pending_approval(key.clone());
let (second, second_is_owner) = service.get_or_create_pending_approval(key);
assert!(first_is_owner);
assert!(!second_is_owner);
assert!(Arc::ptr_eq(&first, &second));
}
#[tokio::test]
async fn pending_approvals_do_not_dedupe_across_ports() {
#[test]
fn pending_approvals_do_not_dedupe_across_ports() {
let service = NetworkApprovalService::default();
let first_key = HostApprovalKey {
let first_host = HostApprovalKey {
environment_id: "local".to_string(),
host: "example.com".to_string(),
protocol: "https",
port: 443,
};
let second_key = HostApprovalKey {
environment_id: "local".to_string(),
host: "example.com".to_string(),
protocol: "https",
let second_host = HostApprovalKey {
port: 8443,
..first_host.clone()
};
let (first, first_is_owner) = service.get_or_create_pending_approval(first_key).await;
let (second, second_is_owner) = service.get_or_create_pending_approval(second_key).await;
let (first, first_is_owner) =
service.get_or_create_pending_approval(pending_key(first_host, "turn-1", "execution-1"));
let (second, second_is_owner) =
service.get_or_create_pending_approval(pending_key(second_host, "turn-1", "execution-1"));
assert!(first_is_owner);
assert!(second_is_owner);
assert!(!Arc::ptr_eq(&first, &second));
}
#[tokio::test]
async fn pending_approvals_do_not_dedupe_across_environments() {
#[test]
fn pending_approvals_do_not_dedupe_across_environments() {
let service = NetworkApprovalService::default();
let first_key = HostApprovalKey {
let first_host = HostApprovalKey {
environment_id: "local".to_string(),
host: "example.com".to_string(),
protocol: "https",
port: 443,
};
let second_key = HostApprovalKey {
let second_host = HostApprovalKey {
environment_id: "remote".to_string(),
..first_key.clone()
..first_host.clone()
};
let (first, first_is_owner) = service.get_or_create_pending_approval(first_key).await;
let (second, second_is_owner) = service.get_or_create_pending_approval(second_key).await;
let (first, first_is_owner) =
service.get_or_create_pending_approval(pending_key(first_host, "turn-1", "execution-1"));
let (second, second_is_owner) =
service.get_or_create_pending_approval(pending_key(second_host, "turn-1", "execution-1"));
assert!(first_is_owner);
assert!(second_is_owner);
assert!(!Arc::ptr_eq(&first, &second));
}
#[test]
fn pending_approvals_do_not_dedupe_across_execution_or_turn() {
let service = NetworkApprovalService::default();
let host = HostApprovalKey {
environment_id: "remote".to_string(),
host: "example.com".to_string(),
protocol: "https",
port: 443,
};
let (first, first_is_owner) =
service.get_or_create_pending_approval(pending_key(host.clone(), "turn-1", "execution-1"));
let (second, second_is_owner) =
service.get_or_create_pending_approval(pending_key(host.clone(), "turn-1", "execution-2"));
let (third, third_is_owner) =
service.get_or_create_pending_approval(pending_key(host, "turn-2", "execution-1"));
assert!(first_is_owner);
assert!(second_is_owner);
assert!(third_is_owner);
assert!(!Arc::ptr_eq(&first, &second));
assert!(!Arc::ptr_eq(&first, &third));
}
#[tokio::test]
async fn session_approved_hosts_are_scoped_by_environment() {
let service = NetworkApprovalService::default();
@@ -226,14 +267,150 @@ async fn pending_waiters_receive_owner_decision() {
tokio::spawn(async move { pending.wait_for_decision().await })
};
pending
.set_decision(PendingApprovalDecision::AllowOnce)
.await;
pending.set_decision(PendingApprovalDecision::AllowOnce);
let decision = waiter.await.expect("waiter should complete");
assert_eq!(decision, PendingApprovalDecision::AllowOnce);
}
#[tokio::test]
async fn dropping_pending_owner_denies_waiters_and_preserves_replacement() {
let service = NetworkApprovalService::default();
let execution_cancellation =
register_call_with_default_shell_trigger(&service, "execution-1").await;
let deferred = DeferredNetworkApproval {
registration_id: "execution-1".to_string(),
cancellation_token: execution_cancellation.clone(),
finish_outcome: Arc::new(OnceCell::new()),
_execution_proxy: None,
};
let key = pending_key(
HostApprovalKey {
environment_id: "remote".to_string(),
host: "example.com".to_string(),
protocol: "https",
port: 443,
},
"turn-1",
"execution-1",
);
let (pending, is_owner) = service.get_or_create_pending_approval(key.clone());
assert!(is_owner);
let owner = PendingHostApprovalOwner::new(
&service,
key.clone(),
Arc::clone(&pending),
Some(execution_cancellation.clone()),
);
let first_waiter = pending.wait_for_decision();
let second_waiter = pending.wait_for_decision();
tokio::pin!(first_waiter, second_waiter);
assert!(poll!(first_waiter.as_mut()).is_pending());
assert!(poll!(second_waiter.as_mut()).is_pending());
drop(owner);
let decisions = timeout(Duration::from_secs(1), async {
tokio::join!(first_waiter, second_waiter)
})
.await
.expect("coalesced waiters should fail closed when their owner is dropped");
assert_eq!(
decisions,
(PendingApprovalDecision::Deny, PendingApprovalDecision::Deny)
);
assert!(execution_cancellation.is_cancelled());
let error = deferred
.finish(&service)
.await
.expect_err("abandoned approval should fail its execution closed");
assert!(matches!(
error,
ToolError::Rejected(message) if message == ABANDONED_NETWORK_APPROVAL_MESSAGE
));
let (replacement, is_owner) = service.get_or_create_pending_approval(key.clone());
assert!(is_owner);
assert!(!Arc::ptr_eq(&pending, &replacement));
let replacement_owner = PendingHostApprovalOwner::new(
&service,
key.clone(),
Arc::clone(&replacement),
/*execution_cancellation*/ None,
);
let stale = Arc::new(PendingHostApproval::new());
drop(PendingHostApprovalOwner::new(
&service,
key.clone(),
Arc::clone(&stale),
/*execution_cancellation*/ None,
));
assert_eq!(
stale.wait_for_decision().await,
PendingApprovalDecision::Deny
);
let (current, is_owner) = service.get_or_create_pending_approval(key);
assert!(!is_owner);
assert!(Arc::ptr_eq(&current, &replacement));
replacement_owner.complete(PendingApprovalDecision::AllowOnce);
assert_eq!(
replacement.wait_for_decision().await,
PendingApprovalDecision::AllowOnce
);
}
#[tokio::test]
async fn pending_owner_cancels_execution_only_for_denial() {
let service = NetworkApprovalService::default();
let key = pending_key(
HostApprovalKey {
environment_id: "remote".to_string(),
host: "example.com".to_string(),
protocol: "https",
port: 443,
},
"turn-1",
"execution-1",
);
let (pending, is_owner) = service.get_or_create_pending_approval(key.clone());
assert!(is_owner);
let execution_cancellation = CancellationToken::new();
let mut owner = PendingHostApprovalOwner::new(
&service,
key.clone(),
Arc::clone(&pending),
Some(execution_cancellation.clone()),
);
owner.set_decision_on_drop(PendingApprovalDecision::AllowForSession);
drop(owner);
assert!(!execution_cancellation.is_cancelled());
assert_eq!(
pending.wait_for_decision().await,
PendingApprovalDecision::AllowForSession
);
let (pending, is_owner) = service.get_or_create_pending_approval(key.clone());
assert!(is_owner);
let execution_cancellation = CancellationToken::new();
PendingHostApprovalOwner::new(
&service,
key,
Arc::clone(&pending),
Some(execution_cancellation.clone()),
)
.complete(PendingApprovalDecision::Deny);
assert!(execution_cancellation.is_cancelled());
assert_eq!(
pending.wait_for_decision().await,
PendingApprovalDecision::Deny
);
}
#[test]
fn allow_once_and_allow_for_session_both_allow_network() {
assert_eq!(
@@ -449,7 +626,8 @@ fn approval_denial_messages_are_bounded_for_model_context() {
#[tokio::test]
async fn finish_call_returns_denial_and_unregisters_active_call() {
let service = NetworkApprovalService::default();
register_call_with_default_shell_trigger(&service, "registration-1").await;
let cancellation_token =
register_call_with_default_shell_trigger(&service, "registration-1").await;
service
.record_call_outcome(
@@ -459,7 +637,7 @@ async fn finish_call_returns_denial_and_unregisters_active_call() {
.await;
let err = service
.finish_call("registration-1")
.finish_call("registration-1", &cancellation_token)
.await
.expect_err("denial should be returned");
@@ -468,6 +646,24 @@ async fn finish_call_returns_denial_and_unregisters_active_call() {
assert_eq!(service.take_call_outcome("registration-1").await, None);
}
#[tokio::test]
async fn finish_call_reports_abandoned_network_approval() {
let service = NetworkApprovalService::default();
let cancellation_token =
register_call_with_default_shell_trigger(&service, "registration-1").await;
cancellation_token.cancel();
let err = service
.finish_call("registration-1", &cancellation_token)
.await
.expect_err("abandoned approval should be returned");
assert!(matches!(
err,
ToolError::Rejected(message) if message == ABANDONED_NETWORK_APPROVAL_MESSAGE
));
}
#[tokio::test]
async fn deferred_finish_reuses_denial_result_after_first_consumer() {
let service = NetworkApprovalService::default();
codex-rs/core/tests/suite/network_approval.rs
+29 -10
View File
@@ -16,6 +16,7 @@ use codex_protocol::permissions::NetworkSandboxPolicy;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExecApprovalRequestEvent;
use codex_protocol::protocol::GuardianAssessmentStatus;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::TurnEnvironmentSelection;
@@ -255,10 +256,20 @@ async fn cancelled_guardian_network_review_fails_closed_without_rewriting_turn_s
.await?;
wait_for_response_request(&pending_guardian).await;
test.codex.submit(Op::Interrupt).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnAborted(_))
})
.await;
let mut saw_turn_aborted = false;
let mut saw_guardian_aborted = false;
while !saw_turn_aborted || !saw_guardian_aborted {
let event = tokio::time::timeout(Duration::from_secs(5), test.codex.next_event())
.await
.context("timed out waiting for parent and Guardian cancellation")?
.context("event stream ended while waiting for cancellation")?;
saw_turn_aborted |= matches!(&event.msg, EventMsg::TurnAborted(_));
saw_guardian_aborted |= matches!(
&event.msg,
EventMsg::GuardianAssessment(assessment)
if assessment.status == GuardianAssessmentStatus::Aborted
);
}
let state_check = mount_sse_once_match(
&server,
@@ -427,11 +438,13 @@ async fn user_network_approval_once_session_and_denial_semantics() -> Result<()>
)
.await?;
let approval = expect_network_approval(&test, LOCAL_ENVIRONMENT_ID).await?;
assert_eq!(
approval.call_id,
"network#local#http#codex-network-test.invalid#80"
assert!(
approval
.call_id
.starts_with("network#local#http#codex-network-test.invalid#80#")
);
assert_eq!(approval.approval_id, None);
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);
assert_eq!(
@@ -463,6 +476,8 @@ async fn user_network_approval_once_session_and_denial_semantics() -> Result<()>
)
.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);
test.codex
.submit(Op::ExecApproval {
id: approval.effective_approval_id(),
@@ -855,6 +870,10 @@ async fn guardian_receives_exact_triggers_for_concurrent_network_requests() -> R
};
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 mut first_args = network_exec_args(&first_command);
first_args["yield_time_ms"] = json!(10_000);
let mut second_args = network_exec_args(&second_command);
second_args["yield_time_ms"] = json!(10_000);
let first_denial = "first concurrent network request denied";
let second_denial = "second concurrent network request denied";
mount_sse_once_match(
@@ -869,12 +888,12 @@ 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(&first_command))?,
&serde_json::to_string(&first_args)?,
),
ev_function_call(
"exec-network-second",
"exec_command",
&serde_json::to_string(&network_exec_args(&second_command))?,
&serde_json::to_string(&second_args)?,
),
ev_completed("resp-network-concurrent"),
]),