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

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

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<String>,
+    },
+}
+
+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<Arc<Mutex<State>>>,
+    endpoints: Option<Arc<BTreeSet<Url>>>,
+    account: Option<u64>,
+}
+
+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<DestinationPolicy>,
+    revision: u64,
+    account: u64,
+    permits: Vec<Weak<Permit>>,
+    changes: watch::Sender<()>,
+}
+
+struct Permit {
+    destination: PermitDestination,
+    revoked: watch::Sender<bool>,
+}
+
+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<Mutex<State>>,
+}
+
+/// 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<State>) -> 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<watch::Receiver<()>> {
+        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<NetworkPermit, NetworkPolicyDenied> {
+        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<NetworkPermit, NetworkPolicyDenied> {
+        if self.endpoints.is_some() {
+            return Err(NetworkPolicyDenied::UnsupportedTransport);
+        }
+        self.acquire_destination(PermitDestination::UnrestrictedSdk)
+    }
+
+    fn acquire_destination(
+        &self,
+        destination: PermitDestination,
+    ) -> Result<NetworkPermit, NetworkPolicyDenied> {
+        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<Arc<Permit>>,
+}
+
+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<T>(
+        &self,
+        operation: impl Future<Output = T>,
+    ) -> Result<T, NetworkPolicyDenied> {
+        tokio::select! {
+            biased;
+            _ = self.revoked() => Err(NetworkPolicyDenied::Revoked),
+            result = operation => Ok(result),
+        }
+    }
+}
+
+#[cfg(test)]
+#[path = "network_policy_tests.rs"]
+mod tests;

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)
+    );
+}

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<Arc<ChatGptCookieStore>>,
+    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<Item = HeaderValue>) -> Self {
         let cookies = cookies.into_iter().collect::<Vec<_>>();

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");

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(\"<redacted>\"), .. }"
         )
@@ -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)

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");