Add policy-checked TCP upstream tunneling for SOCKS (#60523)

## Why

Allowing a hostname must not bypass deny or review rules on its resolved IP addresses. Forwarding DNS resolution to an upstream proxy also requires checking whether those address restrictions can still be enforced.

## What changed

- Add `TcpUpstreamProxy` and `set_tcp_upstream_proxy` to require a configured HTTP `CONNECT` next hop for opaque SOCKS TCP traffic. Wait for a successful response before forwarding payloads and preserve bytes read past the response headers.
- Enforce resolved IP deny and review rules at direct connection time, carry destination-scoped authorization through HTTP, SOCKS, and MITM forwarding, and report blocked resolutions as policy denials.
- Recognize equivalent numeric IP spellings for restrictive rules. Permit remote TCP DNS only when local binding is allowed and no deny or review pattern can restrict IP addresses; reject upstream proxy loops.
- Restrict Linux proxy bridges to managed proxy environment keys and supplied loopback ports. Bound concurrent bridge workers and release them after connection failures while allowing queued responses to drain.
- Add test-only Linux DNS name mapping and SOCKS relay helpers with stable synthetic addresses and validation before opening upstream connections.

## Testing

Add regression coverage for CONNECT refusal and buffered tunnel data, resolved-address policy denials, approval handling across policy refreshes, equivalent IP literals, DNS mapping stability, rejected relay requests, and bridge cleanup after resets.

GitOrigin-RevId: bdd6e16de3373d0eb39995f689dad0230680b99e

codex-rs/network-proxy/src/connect_policy.rs

@@ -1,6 +1,13 @@
+use crate::network_policy::BlockDecisionAuditEventArgs;
+use crate::network_policy::NetworkDecisionSource;
+use crate::network_policy::NetworkPolicyDecision;
+use crate::network_policy::emit_block_decision_audit_event;
 use crate::policy::is_non_public_ip;
+use crate::reasons::REASON_POLICY_DENIED;
 use crate::runtime::HostAuthorization;
 use crate::runtime::HostBlockDecision;
+use crate::state::BlockedRequest;
+use crate::state::BlockedRequestArgs;
 use crate::state::NetworkProxyState;
 #[cfg(target_os = "macos")]
 use crate::system_dns::SystemDnsResolver;
@@ -17,12 +24,84 @@ use rama_net::transport::TryRefIntoTransportContext;
 use rama_tcp::TcpStream;
 use rama_tcp::client::TcpStreamConnector;
 use rama_tcp::client::service::TcpConnector;
+use std::error::Error;
 use std::io;
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::sync::atomic::AtomicU8;
+use std::sync::atomic::Ordering;
+
+const BLOCKED: u8 = 1;
+const DIALED: u8 = 2;
+
+#[derive(Debug, thiserror::Error)]
+#[error("all resolved addresses were blocked by network policy")]
+pub(crate) struct ResolvedTargetsDenied;
+
+impl ResolvedTargetsDenied {
+    pub(crate) fn caused(error: &(dyn Error + 'static)) -> bool {
+        let mut cause = Some(error);
+        while let Some(error) = cause {
+            if error.is::<Self>() {
+                return true;
+            }
+            cause = error.source();
+        }
+        false
+    }
+}
+
+/// Metadata from the trusted admission decision, scoped to its logical destination.
+#[derive(Clone, Debug)]
+pub(crate) struct DialPolicy {
+    target: HostWithPort,
+    authorization: HostAuthorization,
+    protocol: crate::network_policy::NetworkProtocol,
+    client: Option<String>,
+    method: Option<String>,
+}
+impl DialPolicy {
+    pub(crate) fn new(
+        target: &HostWithPort,
+        authorization: HostAuthorization,
+        request: &crate::network_policy::NetworkPolicyRequest,
+    ) -> Self {
+        Self {
+            target: target.clone(),
+            authorization,
+            protocol: request.protocol,
+            client: request.client_addr.clone(),
+            method: request.method.clone(),
+        }
+    }
+    // MITM can drop the DNS root dot while reconstructing the dial authority.
+    fn for_target<'a>(
+        extensions: &'a rama_core::extensions::Extensions,
+        target: &HostWithPort,
+    ) -> Option<&'a Self> {
+        extensions.get::<Self>().filter(|policy| {
+            policy.target.port == target.port
+                && match (&policy.target.host, &target.host) {
+                    (Host::Name(a), Host::Name(b)) => a
+                        .as_str()
+                        .trim_end_matches('.')
+                        .eq_ignore_ascii_case(b.as_str().trim_end_matches('.')),
+                    (a, b) => a == b,
+                }
+        })
+    }
+
+    pub(crate) fn authorization(
+        extensions: &rama_core::extensions::Extensions,
+        target: &HostWithPort,
+    ) -> HostAuthorization {
+        Self::for_target(extensions, target)
+            .map_or(HostAuthorization::RequireAllowlist, |p| p.authorization)
+    }
+}
 
 /// Connects traffic whose host policy has already allowed the request.
-/// Resolved non-public targets still undergo explicit-deny and local-address checks.
+/// Apply address-specific restrictions at the actual direct dial, including DNS results.
 #[derive(Clone)]
 pub(crate) struct TargetCheckedTcpConnector {
     state: Arc<NetworkProxyState>,
@@ -57,18 +136,62 @@ where
             .host_with_port()
             .ok_or_else(|| OpaqueError::from_display("network target is missing a port"))?;
 
-        connector
+        let policy = DialPolicy::for_target(input.extensions(), &target).cloned();
+        let attempts = Arc::new(AtomicU8::new(0));
+        let result = connector
             .with_connector(TargetCheckedStreamConnector {
                 state: self.state.clone(),
+                authorization: policy
+                    .as_ref()
+                    .map_or(HostAuthorization::RequireAllowlist, |policy| {
+                        policy.authorization
+                    }),
+                attempts: attempts.clone(),
                 target,
             })
             .serve(input)
-            .await
+            .await;
+        if result.is_err()
+            && attempts.load(Ordering::Relaxed) == BLOCKED
+            && let Some(policy) = policy
+        {
+            let host = crate::policy::normalize_host(&policy.target.host.to_string());
+            emit_block_decision_audit_event(
+                &self.state,
+                BlockDecisionAuditEventArgs {
+                    source: NetworkDecisionSource::BaselinePolicy,
+                    reason: REASON_POLICY_DENIED,
+                    protocol: policy.protocol,
+                    server_address: &host,
+                    server_port: policy.target.port,
+                    method: policy.method.as_deref(),
+                    client_addr: policy.client.as_deref(),
+                },
+            );
+            let _ = self
+                .state
+                .record_blocked(BlockedRequest::new(BlockedRequestArgs {
+                    host,
+                    port: Some(policy.target.port),
+                    client: policy.client,
+                    method: policy.method,
+                    mode: None,
+                    protocol: policy.protocol.as_policy_protocol().into(),
+                    reason: REASON_POLICY_DENIED.into(),
+                    decision: Some(NetworkPolicyDecision::Deny.as_str().into()),
+                    source: Some(NetworkDecisionSource::BaselinePolicy.as_str().into()),
+                }))
+                .await;
+            return Err(ResolvedTargetsDenied.into());
+        }
+        result
     }
 }
 
 #[derive(Clone)]
 struct TargetCheckedStreamConnector {
+    authorization: HostAuthorization,
+    attempts: Arc<AtomicU8>,
     state: Arc<NetworkProxyState>,
     target: HostWithPort,
 }
@@ -77,7 +200,13 @@ impl TcpStreamConnector for TargetCheckedStreamConnector {
     type Error = BoxError;
 
     async fn connect(&self, addr: SocketAddr) -> Result<TcpStream, Self::Error> {
-        if is_non_public_ip(addr.ip()) && !self.allows_non_public_target(addr).await? {
+        if self
+            .state
+            .resolved_target_blocked(&self.target.host.to_string(), addr.ip(), self.authorization)
+            .await?
+            || (is_non_public_ip(addr.ip()) && !self.allows_non_public_target(addr).await?)
+        {
+            self.attempts.fetch_or(BLOCKED, Ordering::Relaxed);
             return Err(io::Error::new(
                 io::ErrorKind::PermissionDenied,
                 "network target rejected by policy",
@@ -85,6 +214,7 @@ impl TcpStreamConnector for TargetCheckedStreamConnector {
             .into());
         }
 
+        self.attempts.fetch_or(DIALED, Ordering::Relaxed);
         tokio::net::TcpStream::connect(addr)
             .await
             .map(TcpStream::from)
@@ -112,7 +242,7 @@ impl TargetCheckedStreamConnector {
                 &self.target.host.to_string(),
                 self.target.port,
                 /*allow_local_binding*/ None,
-                HostAuthorization::Approved,
+                self.authorization,
             )
             .await
             .map(|decision| decision == HostBlockDecision::Allowed)
@@ -152,6 +282,8 @@ mod tests {
     use super::*;
     use crate::config::NetworkProxyConfig;
     use crate::state::network_proxy_state_for_policy;
+    use pretty_assertions::assert_eq;
+    use rama_core::extensions::ExtensionsRef;
     use rama_net::address::HostWithPort;
     use std::net::Ipv4Addr;
     use tokio::net::TcpListener;
@@ -206,7 +338,6 @@ mod tests {
         let target = listener.local_addr().expect("local addr");
         let mut config = NetworkProxyConfig::default();
         config.set_allowed_domains(vec![target.ip().to_string()]);
-        config.review_domains = vec![target.ip().to_string()];
         let state = Arc::new(network_proxy_state_for_policy(config));
         let connector = TargetCheckedTcpConnector::new(Arc::clone(&state));
 
@@ -234,7 +365,6 @@ mod tests {
         let target = listener.local_addr().expect("local addr");
         let mut config = NetworkProxyConfig::default();
         config.set_allowed_domains(vec!["localhost".to_string()]);
-        config.review_domains = vec!["localhost".to_string()];
         let connector =
             TargetCheckedTcpConnector::new(Arc::new(network_proxy_state_for_policy(config)));
 
@@ -248,6 +378,90 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn policy_refresh_cannot_grant_a_missing_dial_approval() {
+        use crate::network_policy::NetworkDecision;
+        use crate::network_policy::NetworkPolicyDecider;
+        use crate::network_policy::NetworkPolicyRequest;
+        use crate::network_policy::NetworkPolicyRequestArgs;
+        use crate::network_policy::NetworkProtocol;
+        use crate::network_policy::evaluate_host_policy_for_dial;
+        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let target = HostWithPort::new(Host::LOCALHOST_NAME, addr.port());
+        let mut config = NetworkProxyConfig {
+            allow_local_binding: Some(true),
+            ..Default::default()
+        };
+        config.set_allowed_domains(vec!["localhost".into()]);
+        config.review_domains = vec![addr.ip().to_string(), "::1".into()];
+        let state = Arc::new(network_proxy_state_for_policy(config.clone()));
+        let request = NetworkPolicyRequest::new(NetworkPolicyRequestArgs {
+            host: "localhost".into(),
+            port: addr.port(),
+            protocol: NetworkProtocol::Socks5Tcp,
+            environment_id: None,
+            client_addr: None,
+            method: None,
+            command: None,
+            exec_policy_hint: None,
+        });
+        let admitted_target =
+            HostWithPort::new(Host::Name("LOCALHOST.".parse().unwrap()), addr.port());
+        let admit = async |decider: Option<Arc<dyn NetworkPolicyDecider>>| {
+            let mut req = rama_tcp::client::Request::new(target.clone());
+            let decision = evaluate_host_policy_for_dial(
+                &state,
+                decider.as_ref(),
+                &request,
+                Some((&admitted_target, req.extensions_mut())),
+            )
+            .await
+            .unwrap();
+            assert_eq!(decision, NetworkDecision::Allow);
+            req
+        };
+        let assert_addresses_blocked = async |authorization| {
+            let dial = TargetCheckedStreamConnector {
+                state: state.clone(),
+                target: target.clone(),
+                attempts: Arc::default(),
+                authorization,
+            };
+            for ip in [addr.ip(), Ipv4Addr::LOCALHOST.to_ipv6_mapped().into()] {
+                let error = dial
+                    .connect(SocketAddr::new(ip, addr.port()))
+                    .await
+                    .unwrap_err();
+                assert_eq!(
+                    error.downcast_ref::<io::Error>().unwrap().kind(),
+                    io::ErrorKind::PermissionDenied
+                );
+            }
+        };
+        let original = admit(None).await;
+        assert_addresses_blocked(HostAuthorization::RequireAllowlist).await;
+        config.review_domains.push("localhost".into());
+        state
+            .replace_config_state(
+                crate::build_config_state(config, Default::default(), crate::Platform::native())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        let approved = admit(Some(Arc::new(|_| async { NetworkDecision::Allow }))).await;
+        let other_port = HostWithPort::new(target.host.clone(), addr.port() ^ 1);
+        assert!(DialPolicy::for_target(approved.extensions(), &other_port).is_none());
+        let connector = TargetCheckedTcpConnector::new(state.clone());
+        assert!(connector.serve(original).await.is_err());
+        assert!(connector.serve(approved).await.is_ok());
+        state
+            .add_denied_domain(&addr.ip().to_string())
+            .await
+            .unwrap();
+        assert_addresses_blocked(HostAuthorization::Approved).await;
+    }
+
     #[test]
     fn resolved_private_address_does_not_match_allowlisted_hostname() {
         let host = Host::Name("example.com".parse().expect("valid domain"));

codex-rs/network-proxy/src/domain_matcher.rs

@@ -20,11 +20,13 @@ pub(crate) enum GlobalWildcard {
 #[derive(Clone)]
 pub struct DomainPatternSet {
     patterns: Arc<[Vec<u8>]>,
+    restricted_ip_literals: Arc<HashSet<IpAddr>>,
 }
 
 impl DomainPatternSet {
     pub(crate) fn new(patterns: &[String], global_wildcard: GlobalWildcard) -> Result<Self> {
         let mut compiled = Vec::new();
+        let mut restricted_ip_literals = HashSet::new();
         let mut seen = HashSet::new();
         for pattern in patterns {
             if global_wildcard == GlobalWildcard::Reject
@@ -73,11 +75,19 @@ impl DomainPatternSet {
                     continue;
                 }
                 ensure!(!candidate.is_empty(), "domain pattern is empty");
+                // Keep exact address equivalence separate from ordinary allowlist
+                // matching. A scoped IP rule must never apply to other interfaces.
+                if !candidate.contains('%')
+                    && let Some(ip) = crate::policy::numeric_host_address(&candidate)
+                {
+                    restricted_ip_literals.insert(ip.to_canonical());
+                }
                 compiled.push(candidate.into_bytes());
             }
         }
         Ok(Self {
             patterns: compiled.into(),
+            restricted_ip_literals: Arc::new(restricted_ip_literals),
         })
     }
 
@@ -85,6 +95,37 @@ impl DomainPatternSet {
         let host = host.as_ref().as_bytes();
         self.patterns.iter().any(|pattern| matches(pattern, host))
     }
+
+    /// Conservatively include wildcard patterns that could match a canonical IP spelling.
+    pub(crate) fn may_restrict_ip(&self) -> bool {
+        !self.restricted_ip_literals.is_empty()
+            || self.patterns.iter().any(|p| {
+                p.iter().any(|b| matches!(b, b'*' | b'?'))
+                    && p.iter()
+                        .all(|b| b.is_ascii_hexdigit() || matches!(b, b':' | b'.' | b'*' | b'?'))
+                    // Dots occur only in the decimal IPv4 tail, including mapped IPv6.
+                    && p.iter().skip_while(|b| **b != b'.')
+                        .all(|b| b.is_ascii_digit() || matches!(b, b'.' | b'*' | b'?'))
+            })
+    }
+
+    /// An IPv4 address and its IPv4-mapped IPv6 form can dial the same peer.
+    /// Check both spellings for restrictive rules, without expanding allowlists.
+    pub(crate) fn restricts_ip(&self, ip: IpAddr) -> bool {
+        if self.patterns.is_empty() {
+            return false;
+        }
+        let ip = ip.to_canonical();
+        if self.restricted_ip_literals.contains(&ip) {
+            return true;
+        }
+        match ip {
+            IpAddr::V4(ip) => {
+                self.is_match(ip.to_string()) || self.is_match(ip.to_ipv6_mapped().to_string())
+            }
+            IpAddr::V6(ip) => self.is_match(ip.to_string()),
+        }
+    }
 }
 
 /// Normalize host fragments for policy matching (trim whitespace, strip ports/brackets, lowercase).

codex-rs/network-proxy/src/http_proxy.rs

@@ -1,5 +1,6 @@
 use crate::attribution::BindConnectionAttribution;
 use crate::config::NetworkMode;
+use crate::connect_policy::ResolvedTargetsDenied;
 use crate::connect_policy::TargetCheckedTcpConnector;
 use crate::connection_lifecycle::CancelOnShutdown;
 use crate::mitm;
@@ -14,7 +15,7 @@ use crate::network_policy::NetworkPolicyRequestArgs;
 use crate::network_policy::NetworkProtocol;
 use crate::network_policy::emit_allow_decision_audit_event;
 use crate::network_policy::emit_block_decision_audit_event;
-use crate::network_policy::evaluate_host_policy;
+use crate::network_policy::evaluate_host_policy_for_dial;
 use crate::policy::normalize_host;
 use crate::reasons::REASON_METHOD_NOT_ALLOWED;
 use crate::reasons::REASON_MITM_HOOK_DENIED;
@@ -252,7 +253,12 @@ async fn http_connect_accept(
     match disconnect
         .track_http_request(
             started_at,
-            evaluate_host_policy(&app_state, policy_decider.as_ref(), &request),
+            evaluate_host_policy_for_dial(
+                &app_state,
+                policy_decider.as_ref(),
+                &request,
+                Some((&authority, req.extensions_mut())),
+            ),
         )
         .await
     {
@@ -788,7 +794,12 @@ async fn http_plain_proxy(
     match disconnect
         .track_http_request(
             started_at,
-            evaluate_host_policy(&app_state, policy_decider.as_ref(), &request),
+            evaluate_host_policy_for_dial(
+                &app_state,
+                policy_decider.as_ref(),
+                &request,
+                Some((&authority, req.extensions_mut())),
+            ),
         )
         .await
     {
@@ -1019,6 +1030,21 @@ async fn http_plain_proxy(
     remove_hop_by_hop_request_headers(req.headers_mut());
     match client.serve(req).await {
         Ok(resp) => Ok(resp),
+        Err(err) if ResolvedTargetsDenied::caused(&err) => {
+            let reason = crate::reasons::REASON_POLICY_DENIED;
+            Ok(json_blocked(
+                &host,
+                reason,
+                Some(&PolicyDecisionDetails {
+                    decision: NetworkPolicyDecision::Deny,
+                    source: NetworkDecisionSource::BaselinePolicy,
+                    reason,
+                    protocol: NetworkProtocol::Http,
+                    host: &host,
+                    port,
+                }),
+            ))
+        }
         Err(err) => {
             warn!("upstream request failed: {err}");
             Ok(text_response(StatusCode::BAD_GATEWAY, "upstream failure"))
@@ -1339,6 +1365,47 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn all_resolved_addresses_blocked_reports_a_policy_denial() {
+        let mut config = NetworkProxyConfig {
+            allow_local_binding: Some(true),
+            allow_upstream_proxy: false,
+            ..Default::default()
+        };
+        config.set_allowed_domains(vec!["localhost".into()]);
+        config.set_denied_domains(vec!["127.0.0.1".into(), "::1".into()]);
+        let mut state = network_proxy_state_for_policy(config);
+        let (tx, audit) = std::sync::mpsc::channel();
+        state.set_policy_audit_observer(Arc::new(move |event| tx.send(event).unwrap()));
+        let state = Arc::new(state);
+        let mut req = Request::builder()
+            .uri("http://localhost:8080/")
+            .header("host", "localhost:8080")
+            .body(Body::empty())
+            .unwrap();
+        req.extensions_mut().insert(state.clone());
+        let response = http_plain_proxy(
+            /*policy_decider*/ None, /*environment_id*/ None, req,
+        )
+        .await
+        .unwrap();
+        assert_eq!(response.status(), StatusCode::FORBIDDEN);
+        let decisions: Vec<_> = audit
+            .try_iter()
+            .map(|event| (event.decision, event.host, event.port))
+            .collect();
+        assert_eq!(
+            decisions,
+            [
+                ("allow".to_string(), "localhost".to_string(), 8080),
+                ("deny".to_string(), "localhost".to_string(), 8080),
+            ]
+        );
+        let blocked = state.drain_blocked().await.unwrap();
+        assert_eq!(blocked.len(), 1);
+        assert_eq!(blocked[0].host, "localhost");
+    }
+
     #[tokio::test]
     async fn http_connect_accept_allows_allowlisted_host_in_full_mode() {
         let policy = {

codex-rs/network-proxy/src/lib.rs

@@ -32,6 +32,7 @@ mod socks5;
 mod state;
 #[cfg(target_os = "macos")]
 mod system_dns;
+mod tcp_upstream;
 mod upstream;
 #[cfg(target_os = "windows")]
 mod windows_proxy_ingress;
@@ -136,3 +137,4 @@ pub use state::NetworkProxyConstraints;
 pub use state::PartialNetworkProxyConfig;
 pub use state::build_config_state;
 pub use state::validate_policy_against_constraints;
+pub use tcp_upstream::TcpUpstreamProxy;

codex-rs/network-proxy/src/mitm.rs

@@ -125,6 +125,7 @@ impl MitmPolicyContext {
 
 #[derive(Clone)]
 struct MitmRequestContext {
+    dial_policy: Option<crate::connect_policy::DialPolicy>,
     policy: MitmPolicyContext,
     mitm: Arc<MitmState>,
     upstream: UpstreamClient,
@@ -233,6 +234,10 @@ where
         )
     };
     let request_ctx = Arc::new(MitmRequestContext {
+        dial_policy: stream
+            .extensions()
+            .get::<crate::connect_policy::DialPolicy>()
+            .cloned(),
         policy: MitmPolicyContext {
             target_host,
             target_port,
@@ -292,6 +297,9 @@ async fn handle_mitm_request(
         .await
     {
         Ok(resp) => resp,
+        Err(err) if crate::connect_policy::ResolvedTargetsDenied::caused(err.as_ref()) => {
+            blocked_text_response(crate::reasons::REASON_POLICY_DENIED)
+        }
         Err(err) => {
             warn!("MITM request handling failed: {err}");
             text_response(StatusCode::BAD_GATEWAY, "mitm upstream error")
@@ -385,6 +393,9 @@ async fn forward_request(req: Request, request_ctx: &MitmRequestContext) -> Resu
         };
         upstream_req = signed;
     }
+    if let Some(policy) = &request_ctx.dial_policy {
+        upstream_req.extensions_mut().insert(policy.clone());
+    }
     let upstream_resp = if request_ctx.policy.scheme == Scheme::HTTP {
         crate::brokered_tunnel::forward_http_request(upstream_req, &request_ctx.upstream).await?
     } else {

codex-rs/network-proxy/src/mitm_forwarding_tests.rs

@@ -118,6 +118,7 @@ async fn tls_origin_with_hook(
         max_body_bytes: MITM_MAX_BODY_BYTES,
     });
     let context = MitmRequestContext {
+        dial_policy: None,
         policy: MitmPolicyContext {
             target_host: "localhost".to_string(),
             target_port: port,

codex-rs/network-proxy/src/network_policy.rs

@@ -365,6 +365,18 @@ pub(crate) async fn evaluate_host_policy(
     state: &NetworkProxyState,
     decider: Option<&Arc<dyn NetworkPolicyDecider>>,
     request: &NetworkPolicyRequest,
+) -> Result<NetworkDecision> {
+    evaluate_host_policy_for_dial(state, decider, request, /*dial*/ None).await
+}
+
+pub(crate) async fn evaluate_host_policy_for_dial(
+    state: &NetworkProxyState,
+    decider: Option<&Arc<dyn NetworkPolicyDecider>>,
+    request: &NetworkPolicyRequest,
+    dial: Option<(
+        &rama_net::address::HostWithPort,
+        &mut rama_core::extensions::Extensions,
+    )>,
 ) -> Result<NetworkDecision> {
     let execution_id = state.execution_id();
     let host_decision = state.host_blocked(&request.host, request.port).await?;
@@ -456,6 +468,20 @@ pub(crate) async fn evaluate_host_policy(
         },
     );
 
+    if let Some((target, extensions)) = dial
+        && matches!(decision, NetworkDecision::Allow)
+    {
+        let authorization = if policy_override {
+            HostAuthorization::Approved
+        } else {
+            HostAuthorization::RequireAllowlist
+        };
+        extensions.insert(crate::connect_policy::DialPolicy::new(
+            target,
+            authorization,
+            request,
+        ));
+    }
     Ok(decision)
 }
 

codex-rs/network-proxy/src/policy.rs

@@ -28,6 +28,26 @@ impl Host {
     }
 }
 
+/// Interpret numeric host spellings without consulting DNS. System resolvers
+/// also accept shortened, integer, octal and hexadecimal IPv4 forms.
+pub(crate) fn numeric_host_address(host: &str) -> Option<IpAddr> {
+    let host = unscoped_ip_literal(host).unwrap_or(host);
+    if let Ok(ip) = host.parse() {
+        return Some(ip);
+    }
+    if !host.as_bytes().first()?.is_ascii_digit()
+        || !host
+            .bytes()
+            .all(|b| b.is_ascii_hexdigit() || matches!(b, b'.' | b'x' | b'X'))
+    {
+        return None;
+    }
+    match UrlHost::parse(host).ok()? {
+        UrlHost::Ipv4(ip) => Some(IpAddr::V4(ip)),
+        _ => None,
+    }
+}
+
 /// Returns true if the host is a loopback hostname or IP literal.
 pub fn is_loopback_host(host: &Host) -> bool {
     let host = host.as_str();

codex-rs/network-proxy/src/runtime.rs

@@ -91,7 +91,7 @@ impl std::fmt::Display for HostBlockReason {
     }
 }
 
-#[derive(Clone, Copy, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(crate) enum HostAuthorization {
     RequireAllowlist,
     Approved,
@@ -287,6 +287,7 @@ pub struct NetworkProxyState {
     /// Belongs to this proxy; config retains omission for launches on other executors.
     pub(crate) local_binding_policy: LocalBindingPolicy,
     pub(crate) proxy_private_ips_via_upstream: bool,
+    pub(crate) tcp_upstream_proxy: Option<crate::TcpUpstreamProxy>,
     reloader: Arc<dyn ConfigReloader>,
     blocked_request_observer: Arc<RwLock<Option<Arc<dyn BlockedRequestObserver>>>>,
     pub(crate) policy_audit_observer: Option<NetworkPolicyAuditObserver>,
@@ -328,6 +329,7 @@ impl Clone for NetworkProxyState {
             state: self.state.clone(),
             local_binding_policy: self.local_binding_policy,
             proxy_private_ips_via_upstream: self.proxy_private_ips_via_upstream,
+            tcp_upstream_proxy: self.tcp_upstream_proxy.clone(),
             reloader: self.reloader.clone(),
             blocked_request_observer: self.blocked_request_observer.clone(),
             policy_audit_observer: self.policy_audit_observer.clone(),
@@ -350,6 +352,20 @@ impl NetworkProxyState {
         self.proxy_private_ips_via_upstream = enabled;
     }
 
+    /// Require the trusted HTTP CONNECT endpoint for opaque SOCKS TCP connections.
+    pub fn set_tcp_upstream_proxy(&mut self, proxy: crate::TcpUpstreamProxy) {
+        self.tcp_upstream_proxy = Some(proxy);
+    }
+
+    /// A remote DNS answer cannot be checked at our direct socket boundary.
+    pub(crate) async fn permits_remote_tcp_dns(&self) -> Result<bool> {
+        self.reload_if_needed().await?;
+        let guard = self.state.read().await;
+        Ok(self.local_binding_policy.resolve(&guard.config)
+            && !guard.deny_set.may_restrict_ip()
+            && !guard.review_set.may_restrict_ip())
+    }
+
     /// Builds runtime state for one executor-local proxy launch.
     /// The launching executor supplies its own OS; the wire policy cannot override it.
     pub fn from_remote_launch_config(
@@ -446,6 +462,7 @@ impl NetworkProxyState {
             state: Arc::new(RwLock::new(state)),
             local_binding_policy,
             proxy_private_ips_via_upstream: false,
+            tcp_upstream_proxy: None,
             reloader,
             blocked_request_observer: Arc::new(RwLock::new(blocked_request_observer)),
             policy_audit_observer: None,
@@ -885,12 +902,16 @@ impl NetworkProxyState {
         //  2) local/private networking is opt-in (defense-in-depth)
         //  3) managed review selectors precede every ordinary allow rule
         //  4) DNS requires an allowlist match or an approval
-        if domains_match_host_or_unscoped(&deny_set, host_str) {
+        let numeric_host = crate::policy::numeric_host_address(host_str);
+        if domains_match_host_or_unscoped(&deny_set, host_str)
+            || numeric_host.is_some_and(|ip| deny_set.restricts_ip(ip))
+        {
             return Ok(HostBlockDecision::Blocked(HostBlockReason::Denied));
         }
 
         let is_authorized = authorization == HostAuthorization::Approved
             || (!domains_match_host_or_unscoped(&review_set, host_str)
+                && !numeric_host.is_some_and(|ip| review_set.restricts_ip(ip))
                 && domains_match_host_or_unscoped(&allow_set, host_str));
         if !allow_local_binding {
             // If the intent is "prevent access to local/internal networks", we must not rely solely
@@ -1052,6 +1073,32 @@ impl NetworkProxyState {
         Ok(guard.config.allow_upstream_proxy)
     }
 
+    /// The logical host has already passed policy. Fail closed if the actual
+    /// socket needs an IP-specific denial or review the hostname did not receive.
+    pub(crate) async fn resolved_target_blocked(
+        &self,
+        host: &str,
+        ip: IpAddr,
+        authorization: HostAuthorization,
+    ) -> Result<bool> {
+        self.reload_if_needed().await?;
+        let guard = self.state.read().await;
+        if guard.deny_set.restricts_ip(ip) {
+            return Ok(true);
+        }
+        if !guard.review_set.restricts_ip(ip) {
+            return Ok(false);
+        }
+        if authorization != HostAuthorization::Approved {
+            return Ok(true);
+        }
+        let host = Host::parse(host)?;
+        let host = host.as_str();
+        Ok(!domains_match_host_or_unscoped(&guard.review_set, host)
+            && crate::policy::numeric_host_address(host)
+                .is_none_or(|target| target.to_canonical() != ip.to_canonical()))
+    }
+
     pub async fn allow_local_binding(&self) -> Result<bool> {
         self.reload_if_needed().await?;
         let guard = self.state.read().await;
@@ -1984,13 +2031,22 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn host_blocked_denies_scoped_ipv6_literal_before_local_binding() {
+    async fn host_blocked_denies_equivalent_ip_literals_before_local_binding() {
         let state = network_proxy_state_for_policy(NetworkProxyConfig {
             allow_local_binding: Some(true),
-            ..network_settings(&["*"], &["fd00::1"])
+            ..network_settings(&["*"], &["fd00::1", "127.0.0.1"])
         });
 
-        for host in ["fd00::1%eth0", "[fd00::1%eth0]", "[fd00::1%25eth0]"] {
+        for host in [
+            "fd00::1%eth0",
+            "[fd00::1%eth0]",
+            "[fd00::1%25eth0]",
+            "::ffff:127.0.0.1",
+            "127.1",
+            "2130706433",
+            "0177.0.0.1",
+            "0x7f000001",
+        ] {
             assert_eq!(
                 state.host_blocked(host, /*port*/ 80).await.unwrap(),
                 HostBlockDecision::Blocked(HostBlockReason::Denied),

codex-rs/network-proxy/src/socks5.rs

@@ -1,5 +1,6 @@
 use crate::attribution::BindConnectionAttribution;
 use crate::config::NetworkMode;
+use crate::connect_policy::DialPolicy;
 use crate::connect_policy::TargetCheckedTcpConnector;
 use crate::connection_lifecycle::CancelOnShutdown;
 use crate::mitm;
@@ -13,6 +14,7 @@ use crate::network_policy::NetworkPolicyRequestArgs;
 use crate::network_policy::NetworkProtocol;
 use crate::network_policy::emit_block_decision_audit_event;
 use crate::network_policy::evaluate_host_policy;
+use crate::network_policy::evaluate_host_policy_for_dial;
 use crate::policy::normalize_host;
 use crate::reasons::REASON_METHOD_NOT_ALLOWED;
 use crate::reasons::REASON_MITM_REQUIRED;
@@ -33,6 +35,7 @@ use rama_core::extensions::ExtensionsRef;
 use rama_core::graceful::ShutdownGuard;
 use rama_core::service::BoxService;
 use rama_core::service::service_fn;
+use rama_core::stream::rewind::Rewind;
 use rama_net::address::HostWithPort;
 use rama_net::client::EstablishedClientConnection;
 use rama_net::proxy::ProxyRequest;
@@ -157,15 +160,13 @@ pub(crate) fn socks5_proxy_service(
     environment_id: Option<String>,
     enable_socks5_udp: bool,
 ) -> BoxService<TcpStream, (), BoxError> {
-    let tcp_connector = TargetCheckedTcpConnector::new(state.clone());
     let policy_tcp_connector = service_fn({
         let policy_decider = policy_decider.clone();
         let environment_id = environment_id.clone();
         move |req: TcpRequest| {
-            let tcp_connector = tcp_connector.clone();
             let policy_decider = policy_decider.clone();
             let environment_id = environment_id.clone();
-            async move { handle_socks5_tcp(req, tcp_connector, policy_decider, environment_id).await }
+            async move { handle_socks5_tcp(req, policy_decider, environment_id).await }
         }
     });
 
@@ -198,8 +199,7 @@ pub(crate) fn socks5_proxy_service(
 }
 
 async fn handle_socks5_tcp(
-    req: TcpRequest,
-    tcp_connector: TargetCheckedTcpConnector,
+    mut req: TcpRequest,
     policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
     environment_id: Option<String>,
 ) -> Result<EstablishedClientConnection<Socks5TcpConnection, TcpRequest>, BoxError> {
@@ -330,7 +330,14 @@ async fn handle_socks5_tcp(
         exec_policy_hint: None,
     });
 
-    match evaluate_host_policy(&app_state, policy_decider.as_ref(), &request).await {
+    match evaluate_host_policy_for_dial(
+        &app_state,
+        policy_decider.as_ref(),
+        &request,
+        Some((&target, req.extensions_mut())),
+    )
+    .await
+    {
         Ok(NetworkDecision::Deny {
             reason,
             source,
@@ -431,21 +438,25 @@ async fn handle_socks5_tcp(
 
     if let Some(mitm_state) = mitm_state {
         let client = client.as_deref().unwrap_or_default();
+        let mut extensions = Extensions::new();
+        if let Some(policy) = req.extensions().get::<DialPolicy>() {
+            extensions.insert(policy.clone());
+        }
         let conn = match socks_mitm_mode {
             SocksMitmMode::Disabled => None,
             SocksMitmMode::Enabled => Some(Socks5TcpConnection::Mitm {
                 target,
                 mode,
                 mitm: mitm_state,
-                extensions: Extensions::new(),
+                extensions,
             }),
             SocksMitmMode::DetectProtocol(protocols) => Some(Socks5TcpConnection::DetectProtocol {
                 protocols,
                 target,
                 mode,
                 mitm: mitm_state,
-                state: app_state,
-                extensions: Extensions::new(),
+                state: app_state.clone(),
+                extensions,
             }),
         };
         if let Some(conn) = conn {
@@ -458,13 +469,7 @@ async fn handle_socks5_tcp(
 
     info!("SOCKS upstream dial started (host={host}, port={port})");
     let connect_started_at = Instant::now();
-    let result = tcp_connector.serve(req).await.map(|connection| {
-        let EstablishedClientConnection { input, conn } = connection;
-        EstablishedClientConnection {
-            input,
-            conn: Socks5TcpConnection::Direct(conn),
-        }
-    });
+    let result = connect_opaque_tcp(app_state, req).await;
     match &result {
         Ok(_) => info!(
             "SOCKS upstream dial established (host={host}, port={port}, elapsed_ms={})",
@@ -478,6 +483,31 @@ async fn handle_socks5_tcp(
     result
 }
 
+async fn connect_opaque_tcp(
+    state: Arc<NetworkProxyState>,
+    req: TcpRequest,
+) -> Result<EstablishedClientConnection<Socks5TcpConnection, TcpRequest>, BoxError> {
+    if let Some(proxy) = &state.tcp_upstream_proxy {
+        let tunnel = proxy
+            .connect(
+                &state,
+                &req.authority,
+                DialPolicy::authorization(req.extensions(), &req.authority),
+            )
+            .await?;
+        return Ok(EstablishedClientConnection {
+            input: req,
+            conn: Socks5TcpConnection::Tunnel(tunnel),
+        });
+    }
+    let EstablishedClientConnection { input, conn } =
+        TargetCheckedTcpConnector::new(state).serve(req).await?;
+    Ok(EstablishedClientConnection {
+        input,
+        conn: Socks5TcpConnection::Direct(conn),
+    })
+}
+
 /// Internal connector output for SOCKS5 TCP. MITM requests do not dial upstream before the
 /// inner HTTPS request is inspected, so they carry the target metadata instead of a socket.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -490,6 +520,7 @@ enum SocksMitmMode {
 #[derive(Debug)]
 enum Socks5TcpConnection {
     Direct(TcpStream),
+    Tunnel(Rewind<TcpStream>),
     Mitm {
         target: HostWithPort,
         mode: NetworkMode,
@@ -514,6 +545,7 @@ impl AsyncRead for Socks5TcpConnection {
     ) -> Poll<io::Result<()>> {
         match self.get_mut() {
             Self::Direct(stream) => Pin::new(stream).poll_read(cx, buf),
+            Self::Tunnel(stream) => Pin::new(stream).poll_read(cx, buf),
             Self::Mitm { .. } | Self::DetectProtocol { .. } => Poll::Ready(Ok(())),
         }
     }
@@ -527,6 +559,7 @@ impl AsyncWrite for Socks5TcpConnection {
     ) -> Poll<io::Result<usize>> {
         match self.get_mut() {
             Self::Direct(stream) => Pin::new(stream).poll_write(cx, buf),
+            Self::Tunnel(stream) => Pin::new(stream).poll_write(cx, buf),
             Self::Mitm { .. } | Self::DetectProtocol { .. } => Poll::Ready(Ok(buf.len())),
         }
     }
@@ -534,6 +567,7 @@ impl AsyncWrite for Socks5TcpConnection {
     fn poll_flush(self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<io::Result<()>> {
         match self.get_mut() {
             Self::Direct(stream) => Pin::new(stream).poll_flush(cx),
+            Self::Tunnel(stream) => Pin::new(stream).poll_flush(cx),
             Self::Mitm { .. } | Self::DetectProtocol { .. } => Poll::Ready(Ok(())),
         }
     }
@@ -541,6 +575,7 @@ impl AsyncWrite for Socks5TcpConnection {
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<io::Result<()>> {
         match self.get_mut() {
             Self::Direct(stream) => Pin::new(stream).poll_shutdown(cx),
+            Self::Tunnel(stream) => Pin::new(stream).poll_shutdown(cx),
             Self::Mitm { .. } | Self::DetectProtocol { .. } => Poll::Ready(Ok(())),
         }
     }
@@ -550,7 +585,7 @@ impl Socket for Socks5TcpConnection {
     fn local_addr(&self) -> io::Result<SocketAddr> {
         match self {
             Self::Direct(stream) => stream.local_addr(),
-            Self::Mitm { .. } | Self::DetectProtocol { .. } => {
+            Self::Tunnel(_) | Self::Mitm { .. } | Self::DetectProtocol { .. } => {
                 Ok(SocketAddr::from(([0, 0, 0, 0], 0)))
             }
         }
@@ -559,7 +594,7 @@ impl Socket for Socks5TcpConnection {
     fn peer_addr(&self) -> io::Result<SocketAddr> {
         match self {
             Self::Direct(stream) => stream.peer_addr(),
-            Self::Mitm { .. } | Self::DetectProtocol { .. } => {
+            Self::Tunnel(_) | Self::Mitm { .. } | Self::DetectProtocol { .. } => {
                 Ok(SocketAddr::from(([0, 0, 0, 0], 0)))
             }
         }
@@ -570,6 +605,7 @@ impl ExtensionsRef for Socks5TcpConnection {
     fn extensions(&self) -> &Extensions {
         match self {
             Self::Direct(stream) => stream.extensions(),
+            Self::Tunnel(stream) => stream.extensions(),
             Self::Mitm { extensions, .. } | Self::DetectProtocol { extensions, .. } => extensions,
         }
     }
@@ -579,6 +615,7 @@ impl ExtensionsMut for Socks5TcpConnection {
     fn extensions_mut(&mut self) -> &mut Extensions {
         match self {
             Self::Direct(stream) => stream.extensions_mut(),
+            Self::Tunnel(stream) => stream.extensions_mut(),
             Self::Mitm { extensions, .. } | Self::DetectProtocol { extensions, .. } => extensions,
         }
     }
@@ -589,13 +626,21 @@ async fn proxy_socks5_tcp(
 ) -> Result<(), BoxError> {
     let ProxyRequest { mut source, target } = request;
     match target {
-        Socks5TcpConnection::Direct(target) => StreamForwardService::default()
-            .serve(ProxyRequest { source, target })
-            .await
-            .map_err(Into::into),
+        target @ (Socks5TcpConnection::Direct(_) | Socks5TcpConnection::Tunnel(_)) => {
+            StreamForwardService::default()
+                .serve(ProxyRequest { source, target })
+                .await
+                .map_err(Into::into)
+        }
         Socks5TcpConnection::Mitm {
-            target, mode, mitm, ..
+            target,
+            mode,
+            mitm,
+            extensions,
         } => {
+            if let Some(policy) = extensions.get::<DialPolicy>() {
+                source.extensions_mut().insert(policy.clone());
+            }
             source.extensions_mut().insert(ProxyTarget(target));
             source.extensions_mut().insert(mode);
             source.extensions_mut().insert(mitm);
@@ -609,8 +654,11 @@ async fn proxy_socks5_tcp(
             mode,
             mitm,
             state,
-            ..
+            extensions,
         } => {
+            if let Some(policy) = extensions.get::<DialPolicy>() {
+                source.extensions_mut().insert(policy.clone());
+            }
             source.extensions_mut().insert(ProxyTarget(target.clone()));
             source.extensions_mut().insert(mode);
             source.extensions_mut().insert(mitm);
@@ -638,10 +686,11 @@ async fn proxy_socks5_tcp(
                     }
                     info!("SOCKS opaque upstream dial started (target={target})");
                     let connect_started_at = Instant::now();
-                    let EstablishedClientConnection { conn: upstream, .. } =
-                        TargetCheckedTcpConnector::new(state)
-                            .serve(TcpRequest::new(target.clone()))
-                            .await?;
+                    let EstablishedClientConnection { conn: upstream, .. } = connect_opaque_tcp(
+                        state,
+                        TcpRequest::new_with_extensions(target.clone(), extensions),
+                    )
+                    .await?;
                     info!(
                         "SOCKS opaque upstream dial established (target={target}, elapsed_ms={})",
                         connect_started_at.elapsed().as_millis()
@@ -927,10 +976,7 @@ mod tests {
 
         let (result, events) = capture_events(|| async {
             handle_socks5_tcp(
-                request,
-                TargetCheckedTcpConnector::new(state.clone()),
-                /*policy_decider*/ None,
-                /*environment_id*/ None,
+                request, /*policy_decider*/ None, /*environment_id*/ None,
             )
             .await
         })
@@ -971,10 +1017,7 @@ mod tests {
         request.extensions_mut().insert(state.clone());
 
         let result = handle_socks5_tcp(
-            request,
-            TargetCheckedTcpConnector::new(state),
-            /*policy_decider*/ None,
-            /*environment_id*/ None,
+            request, /*policy_decider*/ None, /*environment_id*/ None,
         )
         .await
         .expect("limited-mode HTTPS should use MITM");
@@ -997,10 +1040,7 @@ mod tests {
 
         let (result, events) = capture_events(|| async {
             handle_socks5_tcp(
-                request,
-                TargetCheckedTcpConnector::new(state),
-                /*policy_decider*/ None,
-                /*environment_id*/ None,
+                request, /*policy_decider*/ None, /*environment_id*/ None,
             )
             .await
         })
@@ -1048,10 +1088,7 @@ mod tests {
         request.extensions_mut().insert(state.clone());
 
         let result = handle_socks5_tcp(
-            request,
-            TargetCheckedTcpConnector::new(state),
-            /*policy_decider*/ None,
-            /*environment_id*/ None,
+            request, /*policy_decider*/ None, /*environment_id*/ None,
         )
         .await
         .expect("brokered TLS should defer MITM until protocol detection");
@@ -1076,10 +1113,7 @@ mod tests {
         request.extensions_mut().insert(state.clone());
 
         let err = handle_socks5_tcp(
-            request,
-            TargetCheckedTcpConnector::new(state),
-            /*policy_decider*/ None,
-            /*environment_id*/ None,
+            request, /*policy_decider*/ None, /*environment_id*/ None,
         )
         .await
         .expect_err("limited-mode HTTPS requires MITM");
@@ -1114,10 +1148,7 @@ mod tests {
         request.extensions_mut().insert(state.clone());
 
         let result = handle_socks5_tcp(
-            request,
-            TargetCheckedTcpConnector::new(state),
-            /*policy_decider*/ None,
-            /*environment_id*/ None,
+            request, /*policy_decider*/ None, /*environment_id*/ None,
         )
         .await
         .expect("hooked HTTPS should use MITM");
@@ -1149,10 +1180,7 @@ mod tests {
         request.extensions_mut().insert(state.clone());
 
         let err = handle_socks5_tcp(
-            request,
-            TargetCheckedTcpConnector::new(state),
-            /*policy_decider*/ None,
-            /*environment_id*/ None,
+            request, /*policy_decider*/ None, /*environment_id*/ None,
         )
         .await
         .expect_err("hooked non-HTTPS SOCKS should require MITM");

codex-rs/network-proxy/src/tcp_upstream.rs

@@ -0,0 +1,133 @@
+//! Mandatory next hop for opaque SOCKS TCP on executors provisioned with an HTTP CONNECT proxy.
+//! Only the executor selects the endpoint. The sandbox supplies the already-authorized target.
+
+use crate::connect_policy::TargetCheckedTcpConnector;
+use crate::policy::is_non_public_ip;
+use crate::policy::is_private_network_ip;
+use crate::policy::normalize_host;
+use crate::policy::numeric_host_address;
+use crate::state::NetworkProxyState;
+use anyhow::ensure;
+use rama_core::Service;
+use rama_core::error::BoxError;
+use rama_core::extensions::ExtensionsRef;
+use rama_core::stream::rewind::Rewind;
+use rama_http::Body;
+use rama_http::Method;
+use rama_http::Request;
+use rama_http::Version;
+use rama_http::header::HOST;
+use rama_http::io::upgrade;
+use rama_http_backend::client::HttpConnector;
+use rama_net::Protocol;
+use rama_net::address::HostWithPort;
+use rama_net::address::ProxyAddress;
+use rama_net::client::EstablishedClientConnection;
+use rama_net::stream::ClientSocketInfo;
+use rama_tcp::TcpStream;
+use std::pin::Pin;
+use std::str::FromStr;
+use std::sync::Arc;
+use std::time::Duration;
+
+/// Trusted plaintext HTTP CONNECT listener, expressed as a hostname or IP and an explicit port.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TcpUpstreamProxy(HostWithPort);
+
+impl FromStr for TcpUpstreamProxy {
+    type Err = anyhow::Error;
+
+    fn from_str(raw: &str) -> Result<Self, Self::Err> {
+        let endpoint: HostWithPort = raw.parse()?;
+        ensure!(
+            endpoint.port != 0,
+            "TCP upstream proxy port must be nonzero"
+        );
+        Ok(Self(endpoint))
+    }
+}
+
+impl TcpUpstreamProxy {
+    pub(crate) async fn connect(
+        &self,
+        state: &Arc<NetworkProxyState>,
+        target: &HostWithPort,
+        authorization: crate::runtime::HostAuthorization,
+    ) -> Result<Rewind<TcpStream>, BoxError> {
+        if !state.allow_upstream_proxy().await? {
+            return Err("upstream proxies are disabled by network policy".into());
+        }
+
+        let host = normalize_host(&target.host.to_string());
+        let target_ip = numeric_host_address(&host);
+        let same_port = target.port == self.0.port;
+        if same_port && host == normalize_host(&self.0.host.to_string()) {
+            return Err("cannot tunnel back into the TCP upstream proxy".into());
+        }
+        let authority = if let Some(ip) = target_ip {
+            if state
+                .resolved_target_blocked(&host, ip, authorization)
+                .await?
+                || (is_non_public_ip(ip)
+                    && !(state.proxy_private_ips_via_upstream && is_private_network_ip(ip)))
+            {
+                return Err("IP destination cannot use the TCP upstream proxy".into());
+            }
+            HostWithPort::new(ip.into(), target.port).to_string()
+        } else {
+            if host == "localhost"
+                || host.ends_with(".localhost")
+                || !state.permits_remote_tcp_dns().await?
+            {
+                return Err("destination requires local address enforcement".into());
+            }
+            target.to_string()
+        };
+        let request = Request::builder()
+            .method(Method::CONNECT)
+            .version(Version::HTTP_11)
+            .uri(&authority)
+            .header(HOST, &authority)
+            .extension(ProxyAddress {
+                protocol: Some(Protocol::HTTP),
+                address: self.0.clone(),
+                credential: None,
+            })
+            .body(Body::empty())?;
+        // A fresh request carries only the trusted next hop. Rama dials that endpoint and
+        // retains bytes read past the CONNECT headers in the upgraded stream.
+        tokio::time::timeout(Duration::from_secs(10), async {
+            let connector = HttpConnector::new(TargetCheckedTcpConnector::new(state.clone()));
+            let EstablishedClientConnection { input, conn } = connector.serve(request).await?;
+            // Also reject the proxy's actual IP when provisioning supplied a hostname.
+            if same_port && let Some(ip) = target_ip {
+                let peer = conn
+                    .extensions()
+                    .get::<ClientSocketInfo>()
+                    .ok_or("missing TCP upstream peer address")?
+                    .0
+                    .peer_addr();
+                if ip.to_canonical() == peer.ip().to_canonical() {
+                    return Err("cannot tunnel back into the TCP upstream proxy".into());
+                }
+            }
+            let response = conn.serve(input).await?;
+            if !response.status().is_success() {
+                return Err(format!("TCP upstream refused CONNECT ({})", response.status()).into());
+            }
+            let stream = upgrade::handle_upgrade(response).await?;
+            let parts = stream
+                .downcast::<Pin<Box<TcpStream>>>()
+                .map_err(|_| "unexpected transport for HTTP CONNECT")?;
+            Ok(Rewind::new_buffered(
+                *Pin::into_inner(parts.io),
+                parts.read_buf,
+            ))
+        })
+        .await?
+    }
+}
+
+#[cfg(test)]
+#[path = "tcp_upstream_tests.rs"]
+mod tests;

codex-rs/network-proxy/src/tcp_upstream_tests.rs

@@ -0,0 +1,126 @@
+#![cfg(test)]
+
+use crate::NetworkMode;
+use crate::NetworkProxyConfig;
+use crate::socks5::socks5_proxy_service;
+use crate::state::network_proxy_state_for_policy;
+use pretty_assertions::assert_eq;
+use rama_core::Service;
+use std::io::ErrorKind;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::net::TcpListener;
+use tokio::net::TcpStream;
+
+#[tokio::test]
+async fn socks_tcp_waits_for_http_connect_and_preserves_the_tunnel() {
+    let upstream = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let ingress = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let mut config = NetworkProxyConfig {
+        mode: NetworkMode::Full,
+        allow_local_binding: Some(true),
+        allow_upstream_proxy: true,
+        ..Default::default()
+    };
+    config.set_allowed_domains(vec!["db.invalid".into()]);
+    let mut state = network_proxy_state_for_policy(config);
+    state.set_tcp_upstream_proxy(upstream.local_addr().unwrap().to_string().parse().unwrap());
+    let service = socks5_proxy_service(
+        Arc::new(state),
+        /*policy_decider*/ None,
+        /*environment_id*/ None,
+        /*enable_socks5_udp*/ false,
+    );
+
+    for success in [false, true] {
+        let exercise = async {
+            let mut client = TcpStream::connect(ingress.local_addr().unwrap())
+                .await
+                .unwrap();
+            let (socket, _) = ingress.accept().await.unwrap();
+            let peer = async {
+                let (stream, _) = upstream.accept().await.unwrap();
+                let mut proxy = BufReader::new(stream);
+                let mut line = String::new();
+                proxy.read_line(&mut line).await.unwrap();
+                assert_eq!(line, "CONNECT db.invalid:5432 HTTP/1.1\r\n");
+                loop {
+                    line.clear();
+                    assert_ne!(proxy.read_line(&mut line).await.unwrap(), 0);
+                    if line == "\r\n" {
+                        break;
+                    }
+                }
+                assert!(proxy.buffer().is_empty());
+                // Exercise read-ahead and a non-200 success; no payload may escape on refusal.
+                let response = if success {
+                    b"HTTP/1.1 201 Created\r\n\r\nhello".as_slice()
+                } else {
+                    b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\n\r\n"
+                };
+                proxy.get_mut().write_all(response).await.unwrap();
+                let mut received = Vec::new();
+                proxy.read_to_end(&mut received).await.unwrap();
+                assert_eq!(received, if success { b"private".as_slice() } else { &[] });
+            };
+            let caller = async {
+                client.write_all(&[5, 1, 0]).await.unwrap();
+                let mut greeting = [0; 2];
+                client.read_exact(&mut greeting).await.unwrap();
+                assert_eq!(greeting, [5, 0]);
+                // Pipeline the payload with CONNECT, before the upstream has replied.
+                client
+                    .write_all(b"\x05\x01\x00\x03\x0adb.invalid\x15\x38private")
+                    .await
+                    .unwrap();
+                let mut reply = [0; 10];
+                match client.read_exact(&mut reply).await {
+                    Ok(_) => assert_eq!(reply[1] == 0, success),
+                    // Closing a socket with unread pipelined data can reset it on Windows.
+                    Err(err) if !success => assert_eq!(err.kind(), ErrorKind::ConnectionReset),
+                    Err(err) => panic!("missing SOCKS reply: {err}"),
+                }
+                if success {
+                    client.shutdown().await.unwrap();
+                    let mut bytes = Vec::new();
+                    client.read_to_end(&mut bytes).await.unwrap();
+                    assert_eq!(bytes, b"hello");
+                }
+            };
+            let (result, (), ()) = tokio::join!(service.serve(socket.into()), peer, caller);
+            assert_eq!(result.is_ok(), success);
+        };
+        tokio::time::timeout(Duration::from_secs(5), exercise)
+            .await
+            .unwrap();
+    }
+}
+
+#[tokio::test]
+async fn remote_dns_requires_policy_independent_of_the_resolved_ip() {
+    for (allow_local, deny, allow_remote_dns) in [
+        (false, "denied.example", false),
+        (true, "denied.example", true),
+        (true, "0177.0.0.1", false),
+        (true, "10.*", false),
+        (true, "*.de", true),
+        (true, "*.10", false),
+        (true, "[::ffff:192.0.2.1]", false),
+    ] {
+        let mut config = NetworkProxyConfig {
+            allow_local_binding: Some(allow_local),
+            ..Default::default()
+        };
+        config.set_denied_domains(vec![deny.into()]);
+        let state = network_proxy_state_for_policy(config);
+        assert_eq!(
+            state.permits_remote_tcp_dns().await.unwrap(),
+            allow_remote_dns,
+            "{deny}"
+        );
+    }
+}