NetworkProxy架构

Codex 的 NetworkProxy 不是一个只负责转发字节的 listener。它同时承担配置状态、域名与方法策略、HTTP/SOCKS 服务、环境级代理地址、执行 attribution、managed network sandbox context 和 upstream connector 的装配。一个请求从 child process 的代理环境变量进入 listener 后，必须经过 state 绑定、目标策略检查、非公开地址检查和 upstream 连接；只有连接成功，才进入 MITM 或 opaque forwarding。

本文承接NetworkPolicy决策模型和Linux Seccomp与Namespace。前文解释单次策略如何得出 Allow/Deny/Ask，本篇继续追踪这个决定如何被 HTTP、SOCKS5、环境代理和 executor 消费；不展开 TLS 证书签发和 SOCKS5 wire handshake 的每个字段。

1. Builder装配状态

NetworkProxyBuilder 要求调用方显式提供 Arc<NetworkProxyState>，可选注入 policy decider 和 blocked-request observer。build 先把 observer 写入 state，再解析当前配置，决定 managed listener 地址；Windows 会使用共享 ingress，其他平台预留 loopback listener。这个对象因此是“配置状态 + 服务装配”的组合，而不是简单的 TcpListener 包装。

源码位置：codex-rs/network-proxy/src/proxy.rs :: NetworkProxyBuilder、NetworkProxyBuilder::build

#[derive(Clone)]
pub struct NetworkProxyBuilder {
    state: Option<Arc<NetworkProxyState>>,
    http_addr: Option<SocketAddr>,
    socks_addr: Option<SocketAddr>,
    managed_by_codex: bool,
    policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
    blocked_request_observer: Option<Arc<dyn BlockedRequestObserver>>,
}

impl NetworkProxyBuilder {
    pub fn state(mut self, state: Arc<NetworkProxyState>) -> Self {
        self.state = Some(state);
        self
    }

    pub fn policy_decider<D>(mut self, decider: D) -> Self
    where
        D: NetworkPolicyDecider,
    {
        self.policy_decider = Some(Arc::new(decider));
        self
    }

    pub fn blocked_request_observer<O>(mut self, observer: O) -> Self
    where
        O: BlockedRequestObserver,
    {
        self.blocked_request_observer = Some(Arc::new(observer));
        self
    }

    pub async fn build(self) -> Result<NetworkProxy> {
        let state = self.state.ok_or_else(|| {
            anyhow::anyhow!(
                "NetworkProxyBuilder requires a state; supply one via builder.state(...)"
            )
        })?;
        state
            .set_blocked_request_observer(self.blocked_request_observer.clone())
            .await;
        let current_cfg = state.current_cfg().await?;

构建结果保存 environment_proxies 和可选 execution_scope。这两个字段决定同一个 proxy 实例能否为不同 environment/exec token 生成不同的 child 环境和策略身份。

源码位置：codex-rs/network-proxy/src/proxy.rs :: NetworkProxy

#[derive(Clone)]
pub struct NetworkProxy {
    state: Arc<NetworkProxyState>,
    http_addr: SocketAddr,
    socks_addr: SocketAddr,
    socks_enabled: bool,
    socks5_udp_enabled: bool,
    runtime_settings: Arc<RwLock<NetworkProxyRuntimeSettings>>,
    reserved_listeners: Option<Arc<ReservedListeners>>,
    policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
    environment_proxies: Arc<Mutex<HashMap<String, EnvironmentProxy>>>,
    execution_scope: Option<Arc<ExecutionScope>>,
}

2. Listener服务

HTTP 和 SOCKS5 都把 state、decider、environment id 传进 service。SOCKS5 额外根据 enable_socks5_udp 安装 UDP inspector；TCP 与 UDP 最终都共享同一类 policy request，但错误响应和 relay 返回类型不同。

源码位置：codex-rs/network-proxy/src/http_proxy.rs :: http_proxy_service

pub(crate) fn http_proxy_service(
    state: Arc<NetworkProxyState>,
    policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
    environment_id: Option<String>,
) -> BoxService<TcpStream, (), BoxError> {
    let http_service = HttpServer::http1().service(
        (
            UpgradeLayer::new(
                MethodMatcher::CONNECT,
                service_fn({
                    let policy_decider = policy_decider.clone();
                    let environment_id = environment_id.clone();
                    move |req| {
                        http_connect_accept(policy_decider.clone(), environment_id.clone(), req)
                    }
                }),
                service_fn(http_connect_proxy),
            ),
            RemoveResponseHeaderLayer::hop_by_hop(),
        )
            .into_layer(service_fn({
                let policy_decider = policy_decider.clone();
                let environment_id = environment_id.clone();
                move |req| http_plain_proxy(policy_decider.clone(), environment_id.clone(), req)
            })),
    );

    BindConnectionAttribution::new(http_service, state, environment_id).boxed()
}

源码位置：codex-rs/network-proxy/src/socks5.rs :: socks5_proxy_service

pub(crate) fn socks5_proxy_service(
    state: Arc<NetworkProxyState>,
    policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
    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 }
        }
    });

    let socks_proxy = service_fn(|request| async move { proxy_socks5_tcp(request).await });
    let socks_connector = DefaultConnector::default()
        .with_connector(policy_tcp_connector)
        .with_service(socks_proxy);
    let base = Socks5Acceptor::new().with_connector(socks_connector);

3. Child环境

prepare_for_addrs 是 proxy 和执行 sandbox 的连接点。它重写 HTTP、HTTPS、WebSocket、npm、Docker 等代理变量，设置 NO_PROXY、Node proxy 开关和 managed CA；随后虚拟化 broker credentials、注入 execution attribution token，并返回 loopback ports 与 allow_local_binding 组成的 ManagedNetworkSandboxContext。

源码位置：codex-rs/network-proxy/src/proxy.rs :: ManagedNetworkSandboxContext、PreparedManagedNetwork、prepare_for_addrs

#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ManagedNetworkSandboxContext {
    #[serde(default)]
    pub loopback_ports: Vec<u16>,
    #[serde(default)]
    pub allow_local_binding: bool,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PreparedManagedNetwork {
    pub env: HashMap<String, String>,
    pub sandbox_context: ManagedNetworkSandboxContext,
}

源码位置：codex-rs/network-proxy/src/proxy.rs :: prepare_for_addrs

fn prepare_for_addrs(
    &self,
    mut env: HashMap<String, String>,
    addrs: EnvironmentProxyAddrs,
    #[cfg_attr(not(target_os = "windows"), allow(unused_variables))]
    client: EnvironmentProxyClient,
) -> PreparedManagedNetwork {
    let runtime_settings = self.runtime_settings();
    apply_proxy_env_overrides(
        &mut env,
        addrs.http_addr,
        addrs.socks_addr,
        self.socks_enabled,
        runtime_settings.allow_local_binding,
        runtime_settings.mitm_ca_trust_bundle.as_ref(),
    );
    self.state.virtualize_child_credentials(&mut env);
    if let Some(execution_scope) = self.execution_scope.as_ref() {
        env.insert(
            PROXY_ATTRIBUTION_TOKEN_ENV_KEY.to_string(),
            execution_scope.attribution_token.clone(),
        );
    } else {
        env.remove(PROXY_ATTRIBUTION_TOKEN_ENV_KEY);
    }

生成的 ports 会排序去重；Windows sandbox child 还会收到 CODEX_WINDOWS_SANDBOX_PROXY_PORTS，用于 setup 层计算允许连接的 loopback 端口。Trusted bridge 与 sandboxed process 使用不同的 client 类型，前者不会错误地暴露共享 ingress 的 sandbox 端口。

源码位置：codex-rs/network-proxy/src/proxy.rs :: prepare_for_optional_environment

pub fn prepare_for_optional_environment(
    &self,
    env: HashMap<String, String>,
    environment_id: Option<&str>,
) -> Result<PreparedManagedNetwork> {
    let addrs = self.environment_proxy_addrs(
        environment_id,
        EnvironmentProxyClient::SandboxedProcess,
    )?;
    Ok(self.prepare_for_addrs(env, addrs, EnvironmentProxyClient::SandboxedProcess))
}

4. Attribution绑定

受管环境的 child 通过 CODEX_NETWORK_PROXY_ATTRIBUTION 发送二进制 preface。BindConnectionAttribution 在 HTTP/SOCKS service 之前读取 token，用 state 的 token registry 换回执行状态，并检查 environment id 是否匹配；未知 token、环境不匹配、长度错误或超时都会拒绝连接。

源码位置：codex-rs/network-proxy/src/attribution.rs :: BindConnectionAttribution::serve

async fn serve(&self, mut stream: TcpStream) -> Result<Self::Output, Self::Error> {
    let state = match read_attribution_token(&mut stream).await? {
        Some(token) => self.state.for_execution_token(&token).ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::PermissionDenied,
                "unknown network proxy attribution token",
            )
        })?,
        None => self.state.as_ref().clone(),
    };
    if let Some(expected_environment_id) = self.environment_id.as_deref()
        && state
            .environment_id()
            .is_some_and(|actual| actual != expected_environment_id)
    {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "network proxy attribution environment mismatch",
        )
        .into());
    }
    stream.extensions_mut().insert(Arc::new(state));
    self.inner.serve(stream).await.map_err(Into::into)
}

源码位置：codex-rs/network-proxy/src/attribution.rs :: read_attribution_token

let token = tokio::time::timeout(ATTRIBUTION_FRAME_TIMEOUT, async {
    let mut magic = [0_u8; ATTRIBUTION_FRAME_MAGIC.len()];
    stream.read_exact(&mut magic).await?;
    if &magic != ATTRIBUTION_FRAME_MAGIC {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "invalid network proxy attribution frame",
        ));
    }

    let token_len = stream.read_u16().await? as usize;
    if token_len == 0 || token_len > MAX_ATTRIBUTION_TOKEN_LEN {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "invalid network proxy attribution token length",
        ));
    }

这层保护解释了为什么只知道 listener 端口还不够：同一个共享 ingress 需要先恢复执行身份，后面的 policy state 才能使用正确的 environment/execution scope。

5. CONNECT转发

CONNECT handler 先检查 proxy enabled，构造带 disconnect hook 的 NetworkPolicyRequest，再等待 evaluate_host_policy。Allow 之后还要读取 network mode、MITM requirement 和 upstream proxy setting；只有 connector 成功建立目标连接，才进入 StreamForwardService。

源码位置：codex-rs/network-proxy/src/http_proxy.rs :: http_connect_accept

let disconnect = NetworkRequestDisconnect::default();
let mut request = NetworkPolicyRequest::new(NetworkPolicyRequestArgs {
    protocol: NetworkProtocol::HttpsConnect,
    host: host.clone(),
    port: authority.port,
    environment_id,
    client_addr: client.clone(),
    method: Some("CONNECT".to_string()),
    command: None,
    exec_policy_hint: None,
});

request.disconnect = Some(disconnect.clone());
match disconnect
    .track_http_request(
        started_at,
        evaluate_host_policy(&app_state, policy_decider.as_ref(), &request),
    )
    .await
{
    Ok(NetworkDecision::Deny {
        reason,
        source,
        decision,
    }) => {
        let details = PolicyDecisionDetails {
            decision,
            reason: &reason,
            source,
            protocol: NetworkProtocol::HttpsConnect,
            host: &host,
            port: authority.port,
        };
        let _ = app_state
            .record_blocked(BlockedRequest::new(BlockedRequestArgs {
                host: host.clone(),
                reason: reason.clone(),
                client: client.clone(),
                method: Some("CONNECT".to_string()),
                mode: None,
                protocol: "http-connect".to_string(),
                decision: Some(details.decision.as_str().to_string()),
                source: Some(details.source.as_str().to_string()),
                port: Some(authority.port),
            }))
            .await;
        return Err(blocked_text_with_details(&reason, &details));
    }
    Ok(NetworkDecision::Allow) => {}
    Err(err) => {
        return Err(text_response(StatusCode::INTERNAL_SERVER_ERROR, "error"));
    }
}

源码位置：codex-rs/network-proxy/src/http_proxy.rs :: http_connect_proxy

let proxy_connector = HttpProxyConnector::optional(TargetCheckedTcpConnector::new(app_state));
let tls_config = TlsConnectorDataBuilder::new()
    .with_alpn_protocols_http_auto()
    .build();
let connector = TlsConnectorLayer::tunnel(None)
    .with_connector_data(tls_config)
    .into_layer(proxy_connector);
let EstablishedClientConnection { conn: target, .. } = connector.connect(req).await?;
let proxy_req = ProxyRequest {
    source: upgraded,
    target,
};
StreamForwardService::default().serve(proxy_req).await?;

策略 deny/ask 在 upstream connect 之前形成响应；DNS、TCP、TLS 或 forwarding error 则发生在 Allow 之后，不能被改写成 policy deny。

6. SOCKS5复用策略

SOCKS5 service 为 TCP connector 注入 TargetCheckedTcpConnector，并按配置决定是否安装 UDP relay inspector。TCP handler 从 request authority 规范化 host、获取 client address，再执行 enabled、network mode 和 domain policy 检查；UDP inspector 走同样的 state，但返回 RelayResponse。

源码位置：codex-rs/network-proxy/src/socks5.rs :: socks5_proxy_service

if enable_socks5_udp {
    let udp_state = state.clone();
    let udp_decider = policy_decider.clone();
    let udp_relay = DefaultUdpRelay::default().with_async_inspector(service_fn({
        let environment_id = environment_id.clone();
        move |request: RelayRequest| {
            let udp_state = udp_state.clone();
            let udp_decider = udp_decider.clone();
            let environment_id = environment_id.clone();
            async move {
                inspect_socks5_udp(request, udp_state, udp_decider, environment_id).await
            }
        }
    }));
    let socks_acceptor = base.with_udp_associator(udp_relay);
    BindConnectionAttribution::new(socks_acceptor, state, environment_id).boxed()
} else {
    BindConnectionAttribution::new(base, state, environment_id).boxed()
}

limited mode 下，SOCKS5 UDP 和非 HTTPS TCP 会被 ModeGuard 阻断；HTTPS TCP 则要求 MITM inspection。这个方法级别的 guard 发生在 domain host policy 之前或之外，因此 source 可能是 ModeGuard 或 ProxyState。

7. Upstream检查

TargetCheckedTcpConnector 只在没有预先注入 ProxyAddress 时检查目标；已经选择 upstream proxy 的连接则由 proxy connector 继续处理。目标解析出的 non-public IP 如果不被 allow_local_binding 或 host policy 明确允许，会在真正 TcpStream::connect 之前返回 PermissionDenied。

源码位置：codex-rs/network-proxy/src/connect_policy.rs :: TargetCheckedTcpConnector::serve、TargetCheckedStreamConnector::connect

async fn serve(&self, input: Input) -> Result<Self::Output, Self::Error> {
    if input.extensions().get::<ProxyAddress>().is_some() {
        return TcpConnector::new().serve(input).await;
    }

    let target = input
        .try_ref_into_transport_ctx()
        .map_err(|err| OpaqueError::from_boxed(err.into()).context("read network target"))?
        .host_with_port()
        .ok_or_else(|| OpaqueError::from_display("network target is missing a port"))?;

    TcpConnector::new()
        .with_connector(TargetCheckedStreamConnector {
            state: self.state.clone(),
            target,
        })
        .serve(input)
        .await
}

源码位置：codex-rs/network-proxy/src/connect_policy.rs :: TargetCheckedStreamConnector::connect

async fn connect(&self, addr: SocketAddr) -> Result<TcpStream, Self::Error> {
    if is_non_public_ip(addr.ip()) && !self.allows_non_public_target(addr).await? {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "network target rejected by policy",
        )
        .into());
    }

    tokio::net::TcpStream::connect(addr)
        .await
        .map(TcpStream::from)
        .map_err(Into::into)
}

8. Core与exec消费

Core session 启动 managed proxy 时会把 exec-policy network rules、PermissionProfile、decider、blocked observer 和 audit metadata 一起交给 start_proxy；权限 profile 变化时，session 通过 semaphore 串行刷新配置，禁用的 spec 会清掉 active proxy。exec-server 则从 remote launch config 重建本地 state、builder 和 proxy handle，并为 Windows 返回 restricting SID。

源码位置：codex-rs/core/src/session/mod.rs :: start_managed_network_proxy、refresh_managed_network_proxy_for_current_permission_profile

let spec = spec
    .with_exec_policy_network_rules(exec_policy)
    .map_err(|err| {
        tracing::warn!(
            "failed to apply execpolicy network rules to managed proxy; continuing with configured network policy: {err}"
        );
        err
    })
    .unwrap_or_else(|_| spec.clone());
let network_proxy = spec
    .start_proxy(
        permission_profile,
        network_policy_decider,
        blocked_request_observer,
        managed_network_requirements_enabled,
        audit_metadata,
    )
    .await
    .map_err(|err| anyhow::anyhow!("failed to start managed network proxy: {err}"))?;

源码位置：codex-rs/exec-server/src/process_sandbox.rs :: prepare_network_proxy

let mut state = NetworkProxyState::from_remote_launch_config(network_proxy)
    .map_err(|err| invalid_params(format!("invalid network proxy config: {err}")))?;
if let Some(observer) = network_policy_audit_observer {
    state.set_policy_audit_observer(observer);
}
let mut builder = NetworkProxy::builder().state(Arc::new(state));
if let Some(network_policy_decider) = network_policy_decider {
    builder = builder.policy_decider_arc(network_policy_decider);
}
let proxy = builder
    .build()
    .await
    .map_err(|err| internal_error(format!("failed to build executor network proxy: {err}")))?;
let handle = proxy
    .run()
    .await
    .map_err(|err| internal_error(format!("failed to start executor network proxy: {err}")))?;
let prepared = proxy
    .prepare_for_optional_environment(env, /*environment_id*/ None)
    .map_err(|err| internal_error(format!("failed to prepare executor network proxy: {err}")))?;

9. 测试与边界

相关测试覆盖四个层次：policy decision 的 baseline/decider 分支，HTTP/SOCKS 的 blocked/allow 路径，attribution token 的格式与环境绑定，以及环境变量和 remote launch config 的 round trip。测试能证明状态和调用顺序，但不能把本地 listener 成功绑定等同于目标服务器可达。

源码位置：codex-rs/network-proxy/src/attribution_tests.rs :: attribution token tests

let state = network_proxy_state_for_policy(NetworkProxyConfig::default());
let actual = state
    .for_execution_token("token-1")
    .expect("registered execution should resolve");
assert_eq!(actual.environment_id(), Some("local"));
assert_eq!(actual.execution_id().as_deref(), Some("execution-1"));

源码位置：codex-rs/network-proxy/src/network_policy.rs :: policy_audit_observer_receives_domain_and_non_domain_decisions

let events: Vec<_> = captured_rx.try_iter().collect();
assert_eq!(
    events
        .iter()
        .map(|event| (event.scope.as_str(), event.decision.as_str()))
        .collect::<Vec<_>>(),
    vec![("domain", "allow"), ("non_domain", "deny")]
);

输入、断言和限制必须分开理解：listener fixture 验证服务能否装配；policy fixture 验证决定与来源；attribution fixture 验证执行身份绑定；connector/HTTP integration 才能验证连接前拒绝和 forwarding。它们不能证明 DNS、TLS、MITM 证书信任、外部代理策略或目标服务行为。

在源码仓库中可运行：

cd codex-rs
cargo test -p codex-network-proxy --lib network_policy -- --test-threads=1
cargo test -p codex-network-proxy --lib attribution -- --test-threads=1
cargo test -p codex-network-proxy --lib upstream -- --test-threads=1

10. 阅读闭环

建议按 NetworkProxyBuilder → NetworkProxyState → HTTP/SOCKS service → attribution → child environment/context → evaluate_host_policy → target-checked connector → upstream/MITM/forwarder → Core/exec-server consumer 阅读。读完后应能解释：为什么 state 必须和 listener 一起传递；为什么 attribution token 是共享 ingress 的身份边界；为什么 child sandbox context 要由 proxy 生成；以及 policy deny、target reject、upstream error 和 forwarding error 分别发生在哪一层。

下一篇进入 SOCKS 与 HTTP 转发的握手、目标解析和错误传播。