From 973ec2942c33159a188ea056b1cb1b5ec3d31ed6 Mon Sep 17 00:00:00 2001 From: acrognale-oai Date: Mon, 14 Sep 2026 19:16:41 +0000 Subject: [PATCH] Add revocable network policy primitives to the HTTP client (#45503) ## What changed - Add `NetworkPolicyController` and `NetworkPolicy` APIs for publishing destination policies, checking access, and observing policy changes. Restricted policies permit only `https` and `wss` URLs with exact allowed hosts. - Add revocable `NetworkPermit` values with cancellation support. Reject stale policy publications, revoke permits when access is removed, and prevent account-bound policy handles from regaining access after invalidation. - Provide a permit API for SDK transports without destination enforcement that denies access under restricted policies. - Let `HttpClientFactory` carry a network policy, defaulting to unmanaged access, and include policy identity in factory equality. ## Testing Add tests for secure host matching, invalidation, stale publication, recovery after policy load failure, account isolation, and SDK permit revocation. Make accepted sockets blocking in redirect test servers to handle macOS socket inheritance, and add a missing read timeout. GitOrigin-RevId: c79231527bc5e6954cda6581c0b9f43c90335755 --- codex-rs/http-client/src/lib.rs | 7 + codex-rs/http-client/src/network_policy.rs | 353 ++++++++++++++++++ .../http-client/src/network_policy_tests.rs | 110 ++++++ codex-rs/http-client/src/outbound_proxy.rs | 15 + .../outbound_proxy_redirect_coverage_tests.rs | 4 + .../src/route_aware_client_pool_tests.rs | 10 +- .../route_aware_redirect_integration_tests.rs | 4 + 7 files changed, 502 insertions(+), 1 deletion(-) create mode 100644 codex-rs/http-client/src/network_policy.rs create mode 100644 codex-rs/http-client/src/network_policy_tests.rs diff --git a/codex-rs/http-client/src/lib.rs b/codex-rs/http-client/src/lib.rs index d3e8ac85a6..f9cdb1fb20 100644 --- a/codex-rs/http-client/src/lib.rs +++ b/codex-rs/http-client/src/lib.rs @@ -4,6 +4,7 @@ mod client; mod client_builder; mod custom_ca; mod error; +mod network_policy; mod outbound_proxy; mod request; mod route_aware_client_pool; @@ -31,6 +32,12 @@ pub use crate::custom_ca::build_rustls_client_config_with_custom_ca; pub use crate::custom_ca::maybe_build_rustls_client_config_with_custom_ca; pub use crate::error::StreamError; pub use crate::error::TransportError; +pub use crate::network_policy::DestinationPolicy; +pub use crate::network_policy::NetworkPermit; +pub use crate::network_policy::NetworkPolicy; +pub use crate::network_policy::NetworkPolicyController; +pub use crate::network_policy::NetworkPolicyDenied; +pub use crate::network_policy::NetworkPolicyRevision; pub use crate::outbound_proxy::BuildRouteAwareHttpClientError; pub use crate::outbound_proxy::ClientRouteClass; pub use crate::outbound_proxy::HttpClientFactory; diff --git a/codex-rs/http-client/src/network_policy.rs b/codex-rs/http-client/src/network_policy.rs new file mode 100644 index 0000000000..fa92421d60 --- /dev/null +++ b/codex-rs/http-client/src/network_policy.rs @@ -0,0 +1,353 @@ +//! Application destination checks and revocation of outstanding network operations. +//! +//! The configuration owner publishes an already-composed policy. Transports retain only its +//! read side and a permit for each request; invalidation cannot leave an old request authorized. + +use std::collections::BTreeSet; +use std::fmt; +use std::future::Future; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::MutexGuard; +use std::sync::Weak; + +use reqwest::Url; +use tokio::sync::watch; + +/// Effective application destinations, after the requirements owner has applied precedence. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum DestinationPolicy { + Unrestricted, + /// Only HTTPS and WSS requests to these exact, normalized hosts are permitted. + Restricted { + allowed_hosts: BTreeSet, + }, +} + +impl DestinationPolicy { + fn allows(&self, url: &Url) -> bool { + match self { + Self::Unrestricted => true, + Self::Restricted { allowed_hosts } => { + matches!(url.scheme(), "https" | "wss") + && url.host_str().is_some_and(|host| { + allowed_hosts.contains(host.strip_suffix('.').unwrap_or(host)) + }) + } + } + } +} + +/// A deterministic denial, distinct from retryable network failures. +#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)] +pub enum NetworkPolicyDenied { + #[error("application network policy is unavailable")] + Unavailable, + #[error("destination denied by application network policy")] + Destination, + #[error("application network permission was revoked")] + Revoked, + #[error("this SDK transport is disabled by application network restrictions")] + UnsupportedTransport, +} + +/// Read access to application policy. Clones observe the same updates. +#[derive(Clone, Default)] +pub struct NetworkPolicy { + state: Option>>, + endpoints: Option>>, + account: Option, +} + +impl fmt::Debug for NetworkPolicy { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("NetworkPolicy") + .field("managed", &self.state.is_some()) + .finish_non_exhaustive() + } +} + +impl PartialEq for NetworkPolicy { + fn eq(&self, other: &Self) -> bool { + self.endpoints == other.endpoints + && self.account == other.account + && match (&self.state, &other.state) { + (Some(left), Some(right)) => Arc::ptr_eq(left, right), + (None, None) => true, + (Some(_), None) | (None, Some(_)) => false, + } + } +} + +impl Eq for NetworkPolicy {} + +struct State { + policy: Option, + revision: u64, + account: u64, + permits: Vec>, + changes: watch::Sender<()>, +} + +struct Permit { + destination: PermitDestination, + revoked: watch::Sender, +} + +enum PermitDestination { + Url(Url), + UnrestrictedSdk, +} + +impl PermitDestination { + fn allowed_by(&self, policy: &DestinationPolicy) -> bool { + match self { + Self::Url(url) => policy.allows(url), + Self::UnrestrictedSdk => matches!(policy, DestinationPolicy::Unrestricted), + } + } +} + +/// Publication access retained by the account/configuration owner. +#[derive(Clone)] +pub struct NetworkPolicyController { + state: Arc>, +} + +/// Identifies the account/configuration generation that a load is allowed to publish into. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct NetworkPolicyRevision(u64); + +impl Default for NetworkPolicyController { + fn default() -> Self { + Self { + state: Arc::new(Mutex::new(State { + policy: None, + revision: 0, + account: 0, + permits: Vec::new(), + changes: watch::channel(()).0, + })), + } + } +} + +fn lock_state(state: &Mutex) -> MutexGuard<'_, State> { + state + .lock() + .unwrap_or_else(|_| panic!("application network policy lock poisoned")) +} + +impl NetworkPolicyController { + pub fn policy(&self) -> NetworkPolicy { + NetworkPolicy { + state: Some(self.state.clone()), + endpoints: None, + account: None, + } + } + + /// Fails a requirements load without changing the current account identity. + pub fn unavailable(&self, revision: NetworkPolicyRevision) { + let mut state = lock_state(&self.state); + if state.revision != revision.0 { + return; + } + if state.policy.take().is_some() { + state.changes.send_replace(()); + } + for permit in state + .permits + .drain(..) + .filter_map(|permit| permit.upgrade()) + { + permit.revoked.send_replace(/*value*/ true); + } + } + + /// Publishes a successful load only if its account/configuration generation is still current. + pub fn publish(&self, revision: NetworkPolicyRevision, policy: DestinationPolicy) -> bool { + let mut state = lock_state(&self.state); + if state.revision != revision.0 { + return false; + } + state.permits.retain(|permit| { + let Some(permit) = permit.upgrade() else { + return false; + }; + if !permit.destination.allowed_by(&policy) { + permit.revoked.send_replace(/*value*/ true); + } + true + }); + if state.policy.as_ref() != Some(&policy) { + state.policy = Some(policy); + state.changes.send_replace(()); + } + true + } +} + +impl NetworkPolicy { + pub const fn unmanaged() -> Self { + Self { + state: None, + endpoints: None, + account: None, + } + } + + /// Binds a content client to the current account so retained credentials cannot + /// be used after another workspace's policy has been installed. + pub fn for_current_account(mut self) -> Self { + self.account = self.state.as_ref().map(|state| lock_state(state).account); + self + } + + /// Whether this transport participates in the application policy lifecycle. + pub fn is_managed(&self) -> bool { + self.state.is_some() + } + + /// Watches effective changes; unmanaged callers have no application policy owner. + pub fn changes(&self) -> Option> { + self.state + .as_ref() + .map(|state| lock_state(state).changes.subscribe()) + } + + pub fn revision(&self) -> NetworkPolicyRevision { + NetworkPolicyRevision( + self.state + .as_ref() + .map_or(0, |state| lock_state(state).revision), + ) + } + + /// Revokes outstanding operations before an account change or a failed requirements load. + /// This read-side operation can only remove access; publication stays with the controller. + pub fn invalidate(&self) { + if let Some(state) = &self.state { + let mut state = lock_state(state); + state.revision += 1; + state.account += 1; + state.policy = None; + state.changes.send_replace(()); + for permit in state + .permits + .drain(..) + .filter_map(|permit| permit.upgrade()) + { + permit.revoked.send_replace(/*value*/ true); + } + } + } + + /// Checks a destination before proxy resolution, DNS, or transport work starts. + pub fn acquire(&self, url: &Url) -> Result { + if self + .endpoints + .as_ref() + .is_some_and(|endpoints| !endpoints.contains(url)) + { + return Err(NetworkPolicyDenied::Destination); + } + self.acquire_destination(PermitDestination::Url(url.clone())) + } + + /// Disables SDKs without destination enforcement whenever restrictions apply. + /// The returned permit also cancels work if an unrestricted policy changes. + pub fn acquire_for_unsupported_sdk(&self) -> Result { + if self.endpoints.is_some() { + return Err(NetworkPolicyDenied::UnsupportedTransport); + } + self.acquire_destination(PermitDestination::UnrestrictedSdk) + } + + fn acquire_destination( + &self, + destination: PermitDestination, + ) -> Result { + let Some(state) = &self.state else { + return Ok(NetworkPermit { inner: None }); + }; + let mut state = lock_state(state); + if self.account.is_some_and(|account| account != state.account) { + return Err(NetworkPolicyDenied::Revoked); + } + let policy = state + .policy + .as_ref() + .ok_or(NetworkPolicyDenied::Unavailable)?; + if !destination.allowed_by(policy) { + return Err(match destination { + PermitDestination::Url(_) => NetworkPolicyDenied::Destination, + PermitDestination::UnrestrictedSdk => NetworkPolicyDenied::UnsupportedTransport, + }); + } + let (revoked, _) = watch::channel(/*init*/ false); + let permit = Arc::new(Permit { + destination, + revoked, + }); + state.permits.retain(|permit| permit.strong_count() != 0); + state.permits.push(Arc::downgrade(&permit)); + Ok(NetworkPermit { + inner: Some(permit), + }) + } +} + +/// Authorization retained for the entire operation, including response or WebSocket streaming. +#[derive(Clone)] +pub struct NetworkPermit { + inner: Option>, +} + +impl fmt::Debug for NetworkPermit { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("NetworkPermit").finish_non_exhaustive() + } +} + +impl NetworkPermit { + pub fn check(&self) -> Result<(), NetworkPolicyDenied> { + if self + .inner + .as_ref() + .is_some_and(|permit| *permit.revoked.borrow()) + { + Err(NetworkPolicyDenied::Revoked) + } else { + Ok(()) + } + } + + pub async fn revoked(&self) { + match &self.inner { + Some(permit) => { + let _ = permit + .revoked + .subscribe() + .wait_for(|revoked| *revoked) + .await; + } + None => std::future::pending().await, + } + } + + pub async fn run( + &self, + operation: impl Future, + ) -> Result { + tokio::select! { + biased; + _ = self.revoked() => Err(NetworkPolicyDenied::Revoked), + result = operation => Ok(result), + } + } +} + +#[cfg(test)] +#[path = "network_policy_tests.rs"] +mod tests; diff --git a/codex-rs/http-client/src/network_policy_tests.rs b/codex-rs/http-client/src/network_policy_tests.rs new file mode 100644 index 0000000000..dd97871365 --- /dev/null +++ b/codex-rs/http-client/src/network_policy_tests.rs @@ -0,0 +1,110 @@ +//! Destination matching, bootstrap scope, and account revocation behavior. + +use std::collections::BTreeSet; + +use pretty_assertions::assert_eq; +use reqwest::Url; + +use super::*; + +#[test] +fn restricted_policy_matches_exact_secure_hosts() { + let controller = NetworkPolicyController::default(); + let policy = controller.policy(); + assert!(controller.publish( + policy.revision(), + DestinationPolicy::Restricted { + allowed_hosts: BTreeSet::from(["example.com".to_string()]), + } + )); + for (url, expected) in [ + ("https://EXAMPLE.com/path", true), + ("https://example.com.:8443/path", true), + ("wss://example.com/path", true), + ("http://example.com/path", false), + ("ws://example.com/path", false), + ("https://sub.example.com/path", false), + ("https://example.com.evil.test/path", false), + ("https://example.com@evil.test/path", false), + ("https://127.0.0.1/path", false), + ] { + assert_eq!( + policy.acquire(&Url::parse(url).unwrap()).is_ok(), + expected, + "{url}" + ); + } +} + +#[tokio::test] +async fn invalidation_revokes_existing_permits_and_rejects_stale_publication() { + let controller = NetworkPolicyController::default(); + let policy = controller.policy(); + let url = Url::parse("https://example.com").unwrap(); + assert_eq!( + policy.acquire(&url).unwrap_err(), + NetworkPolicyDenied::Unavailable + ); + let previous = policy.revision(); + assert!(controller.publish(previous, DestinationPolicy::Unrestricted)); + let permit = policy.acquire(&url).unwrap(); + policy.invalidate(); + assert!(!controller.publish(previous, DestinationPolicy::Unrestricted)); + assert!(controller.publish(policy.revision(), DestinationPolicy::Unrestricted)); + assert_eq!( + permit.run(async { "should not run" }).await, + Err(NetworkPolicyDenied::Revoked) + ); + assert!(policy.acquire(&url).is_ok()); +} + +#[tokio::test] +async fn load_failure_recovers_current_account_but_cannot_revive_previous_account() { + let controller = NetworkPolicyController::default(); + let policy = controller.policy(); + let account = policy.clone().for_current_account(); + let url = Url::parse("https://example.com").unwrap(); + let mut changes = policy.changes().unwrap(); + controller.publish(policy.revision(), DestinationPolicy::Unrestricted); + changes.changed().await.unwrap(); + controller.publish(policy.revision(), DestinationPolicy::Unrestricted); + assert!(!changes.has_changed().unwrap()); + let old_request = account.acquire(&url).unwrap(); + controller.unavailable(policy.revision()); + assert_eq!( + account.acquire(&url).unwrap_err(), + NetworkPolicyDenied::Unavailable + ); + controller.publish(policy.revision(), DestinationPolicy::Unrestricted); + assert!(account.acquire(&url).is_ok()); + assert_eq!(old_request.check(), Err(NetworkPolicyDenied::Revoked)); + policy.invalidate(); + controller.publish(policy.revision(), DestinationPolicy::Unrestricted); + assert_eq!( + account.acquire(&url).unwrap_err(), + NetworkPolicyDenied::Revoked + ); + assert!(policy.for_current_account().acquire(&url).is_ok()); +} + +#[tokio::test] +async fn unsupported_sdk_work_is_denied_and_running_work_is_cancelled() { + let controller = NetworkPolicyController::default(); + let policy = controller.policy(); + controller.publish(policy.revision(), DestinationPolicy::Unrestricted); + let permit = policy.acquire_for_unsupported_sdk().unwrap(); + controller.publish( + policy.revision(), + DestinationPolicy::Restricted { + allowed_hosts: BTreeSet::from(["example.com".to_string()]), + }, + ); + assert_eq!( + policy.acquire_for_unsupported_sdk().unwrap_err(), + NetworkPolicyDenied::UnsupportedTransport + ); + assert_eq!( + permit.run(std::future::pending::<()>()).await, + Err(NetworkPolicyDenied::Revoked) + ); +} diff --git a/codex-rs/http-client/src/outbound_proxy.rs b/codex-rs/http-client/src/outbound_proxy.rs index 27d1a80587..64713478ec 100644 --- a/codex-rs/http-client/src/outbound_proxy.rs +++ b/codex-rs/http-client/src/outbound_proxy.rs @@ -18,6 +18,7 @@ use tokio::sync::Semaphore; use http::HeaderValue; +use crate::NetworkPolicy; use crate::chatgpt_cloudflare_cookies::ChatGptCookieStore; use crate::custom_ca::BuildCustomCaTransportError; use crate::custom_ca::build_reqwest_client_with_custom_ca; @@ -166,11 +167,13 @@ impl fmt::Debug for OutboundProxyRoute { pub struct HttpClientFactory { outbound_proxy_policy: OutboundProxyPolicy, chatgpt_cookie_store: Option>, + network_policy: NetworkPolicy, } impl PartialEq for HttpClientFactory { fn eq(&self, other: &Self) -> bool { self.outbound_proxy_policy == other.outbound_proxy_policy + && self.network_policy == other.network_policy && self .chatgpt_cookie_store .as_ref() @@ -188,6 +191,7 @@ impl fmt::Debug for HttpClientFactory { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("HttpClientFactory") .field("outbound_proxy_policy", &self.outbound_proxy_policy) + .field("network_policy", &self.network_policy) .finish() } } @@ -198,9 +202,20 @@ impl HttpClientFactory { Self { outbound_proxy_policy, chatgpt_cookie_store: None, + network_policy: NetworkPolicy::unmanaged(), } } + /// Carries the account/configuration owner's application policy into every transport. + pub fn with_network_policy(mut self, policy: NetworkPolicy) -> Self { + self.network_policy = policy; + self + } + + pub fn network_policy(&self) -> &NetworkPolicy { + &self.network_policy + } + /// Adds process-scoped cookies to requests made by ChatGPT cookie-store clients. pub fn with_chatgpt_cookies(mut self, cookies: impl IntoIterator) -> Self { let cookies = cookies.into_iter().collect::>(); diff --git a/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs b/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs index ed10af742e..8aefa93f1e 100644 --- a/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs +++ b/codex-rs/http-client/src/outbound_proxy_redirect_coverage_tests.rs @@ -208,6 +208,10 @@ fn spawn_failing_listener() -> (std::net::SocketAddr, std::thread::JoinHandle<() Err(error) => panic!("failing listener should accept: {error}"), } }; + // Accepted sockets inherit the listener's nonblocking mode on macOS. + stream + .set_nonblocking(false) + .expect("failing stream should become blocking"); stream .set_read_timeout(Some(Duration::from_secs(2))) .expect("failing stream should get a read timeout"); diff --git a/codex-rs/http-client/src/route_aware_client_pool_tests.rs b/codex-rs/http-client/src/route_aware_client_pool_tests.rs index e3848397bd..1be6042edb 100644 --- a/codex-rs/http-client/src/route_aware_client_pool_tests.rs +++ b/codex-rs/http-client/src/route_aware_client_pool_tests.rs @@ -123,7 +123,8 @@ fn request_builder_debug_redacts_url_secrets() { format!("{request:?}"), concat!( "RouteAwareRequestBuilder { pool: RouteAwareClientPool { ", - "http_client_factory: HttpClientFactory { outbound_proxy_policy: ReqwestDefault }, ", + "http_client_factory: HttpClientFactory { outbound_proxy_policy: ReqwestDefault, ", + "network_policy: NetworkPolicy { managed: false, .. } }, ", "route_class: Api, .. }, method: Some(GET), ", "url: Some(\"\"), .. }" ) @@ -402,6 +403,13 @@ async fn reqwest_default_route_preserves_transport_redirects() { Err(error) => panic!("redirect server should accept: {error}"), } }; + // Accepted sockets inherit the listener's nonblocking mode on macOS. + stream + .set_nonblocking(false) + .expect("redirect stream should become blocking"); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("redirect stream should get a read timeout"); let mut buffer = [0_u8; 1024]; let size = stream .read(&mut buffer) diff --git a/codex-rs/http-client/src/route_aware_redirect_integration_tests.rs b/codex-rs/http-client/src/route_aware_redirect_integration_tests.rs index 91119ca08c..344e8a7f6b 100644 --- a/codex-rs/http-client/src/route_aware_redirect_integration_tests.rs +++ b/codex-rs/http-client/src/route_aware_redirect_integration_tests.rs @@ -98,6 +98,10 @@ fn spawn_response( Err(error) => panic!("HTTP listener should accept: {error}"), } }; + // Accepted sockets inherit the listener's nonblocking mode on macOS. + stream + .set_nonblocking(false) + .expect("set accepted stream blocking"); stream .set_read_timeout(Some(Duration::from_secs(2))) .expect("read timeout");