SOCKS与HTTP转发

HTTP 和 SOCKS5 在 NetworkProxy 中共享 state、attribution 和目标策略，但它们的“转发”并不是同一条函数链：HTTP CONNECT 先返回升级响应，再在 opaque tunnel、TLS 检测或 MITM 之间分流；HTTP plain proxy 还要验证 absolute-form URI 与 Host header；SOCKS5 TCP 通过 connector 返回连接，UDP 则逐个 datagram 交给 inspector。所有路径都把 policy deny、mode guard、目标连接失败和 forwarding failure 分开处理。

本文承接NetworkProxy架构和NetworkPolicy决策模型。前文解释共享 proxy state 和 baseline/decider，本篇继续追踪协议请求如何进入 policy、connector、MITM 或 relay；不展开 TLS 证书生成和 SOCKS5 wire protocol 的完整规范。

1. HTTP服务入口

HTTP service 把 CONNECT 和普通请求放入同一个 HTTP/1 pipeline：UpgradeLayer 匹配 CONNECT，其他请求进入 http_plain_proxy。两条分支共享 BindConnectionAttribution，所以 service 收到 request 时，state 已经可以代表对应 execution/environment。

源码位置：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()
}

2. CONNECT接受

CONNECT accept 阶段先解析 authority、检查 proxy enabled，再创建 NetworkPolicyRequest。当前实现把 NetworkRequestDisconnect 放入 request，并用 track_http_request 包住异步 policy evaluation；如果客户端在等待 decider 时断开，disconnect tracker 可以让策略请求停止等待。

源码位置：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"));
    }
}

Allow 只表示 domain policy 通过。接下来仍要根据 network mode、host MITM requirement 和 MITM state 判断是否能安全处理内层 HTTPS。

3. CONNECT分流

当前 http_connect_proxy 使用三个分支：Disabled 直接 opaque forwarding，Enabled 进入 MITM，DetectTls 先 peek TLS prefix，再选择 MITM 或 opaque forwarding。旧文使用的 Brokered 名称已经不存在，不能据此推断当前行为。

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

let result: Result<(), OpaqueError> = match connect_mitm_mode {
    ConnectMitmMode::Disabled => forward_connect_tunnel(upgraded).await,
    ConnectMitmMode::Enabled => mitm_connect_tunnel(upgraded).await,
    ConnectMitmMode::DetectTls => match mitm::peek_tls_prefix(upgraded).await {
        Ok((true, stream)) => mitm_connect_tunnel(stream).await,
        Ok((false, stream)) => forward_connect_tunnel(stream).await,
        Err(err) => Err(err),
    },
};
if let Err(err) = result {
    warn!("CONNECT tunnel error: {err}");
}

ConnectMitmMode 的选择发生在 accept 阶段：limited mode 强制 Enabled；host hook 或 TLS requirement 可以要求 MITM；brokered credential 场景则使用 DetectTls，避免对非 TLS 的 server-first opaque protocol 误做 TLS 解密。

4. Opaque CONNECT

opaque tunnel 不读取应用层 payload。它先根据 allow_upstream_proxy 决定 direct 或 HTTP upstream proxy，再通过 TlsConnectorLayer::tunnel 建立目标连接，最后把 upgraded client stream 与目标 stream 交给 StreamForwardService 双向转发。

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

let allow_upstream_proxy = match app_state.allow_upstream_proxy().await {
    Ok(allowed) => allowed,
    Err(err) => {
        error!("failed to read upstream proxy setting: {err}");
        false
    }
};
let proxy = if allow_upstream_proxy {
    proxy_for_connect(&authority)
} else {
    None
};

let mut extensions = upgraded.extensions().clone();
if let Some(proxy) = proxy {
    extensions.insert(proxy);
}

let req = TcpRequest::new_with_extensions(authority.clone(), extensions)
    .with_protocol(Protocol::HTTPS);
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);

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

info!("CONNECT upstream dial started (target={authority})");
let connect_started_at = Instant::now();
let EstablishedClientConnection { conn: target, .. } = match connector.connect(req).await {
    Ok(connection) => connection,
    Err(err) => {
        warn!(
            "CONNECT upstream dial failed (target={authority}, elapsed_ms={})",
            connect_started_at.elapsed().as_millis()
        );
        return Err(OpaqueError::from_boxed(err)
            .with_context(|| format!("establish CONNECT tunnel to {authority}")));
    }
};

let proxy_req = ProxyRequest {
    source: upgraded,
    target,
};
info!("CONNECT tunnel forwarding started (target={authority})");
let forward_started_at = Instant::now();
StreamForwardService::default()
    .serve(proxy_req)
    .await
    .map_err(|err| {
        warn!(
            "CONNECT tunnel forwarding failed (target={authority}, elapsed_ms={})",
            forward_started_at.elapsed().as_millis()
        );
        OpaqueError::from_boxed(err.into())
            .with_context(|| format!("forward CONNECT tunnel to {authority}"))
    })?;

目标连接失败会带有 establish CONNECT tunnel 上下文；转发失败则带有 forward CONNECT tunnel 上下文。两者都发生在 policy Allow 之后。

5. MITM CONNECT

MITM 分支要求 stream extension 中存在 MitmState，然后为 CONNECT 目标建立 TLS server/client 两侧。MITM 内层请求会再次经过 hook、method 和 local/private target 检查；因此 CONNECT 首次 Allow 并不意味着内层每个 HTTP request 都自动 Allow。

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

async fn mitm_connect_tunnel<S>(stream: S) -> Result<(), OpaqueError>
where
    S: Stream + Unpin + ExtensionsMut,
{
    let target = stream
        .extensions()
        .get::<ProxyTarget>()
        .map(|target| target.0.clone())
        .ok_or_else(|| OpaqueError::from_display("missing MITM authority"))?;
    let host = normalize_host(&target.host.to_string());
    let port = target.port;
    let mode = stream
        .extensions()
        .get::<NetworkMode>()
        .copied()
        .unwrap_or(NetworkMode::Full);
    if stream.extensions().get::<Arc<mitm::MitmState>>().is_none() {
        return Err(OpaqueError::from_display(format!(
            "cannot enable MITM without state (host={host}, port={port})"
        )));
    }
    info!("CONNECT MITM enabled (host={host}, port={port}, mode={mode:?})");
    mitm::mitm_stream(stream)
        .await
        .map_err(|err| OpaqueError::from_display(format!("MITM tunnel error: {err}")))
}

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

// CONNECT already handled allowlist/denylist + decider policy. Re-check local/private
// resolution here to defend against DNS rebinding between CONNECT and inner HTTPS requests.
if matches!(
    policy
        .app_state
        .host_blocked(&policy.target_host, policy.target_port)
        .await?,
    HostBlockDecision::Blocked(HostBlockReason::NotAllowedLocal)
) {
    let reason = HostBlockReason::NotAllowedLocal.as_str();
    let _ = policy
        .app_state
        .record_blocked(BlockedRequest::new(BlockedRequestArgs {
            host: policy.target_host.clone(),
            reason: reason.to_string(),
            client: client.clone(),
            method: Some(method.clone()),
            mode: Some(policy.mode),
            protocol: "https".to_string(),
            decision: None,
            source: None,
            port: Some(policy.target_port),
        }))
        .await;
    return Ok(MitmPolicyDecision::Block(blocked_text_response(reason)));
}

内层检查的意义是时间边界：DNS 解析结果可能在 CONNECT 和实际 HTTPS request 之间变化，MITM 不能只相信第一次 authority policy。

6. HTTP plain

普通 HTTP proxy 使用 absolute-form URI。handler 先解析目标 host/port，再验证 URI host 与 Host header 一致；method policy、proxy enabled、unix-socket 特殊路径和 domain policy 都通过后，才选择 upstream proxy/direct client 并移除 hop-by-hop headers。

源码位置：codex-rs/network-proxy/src/http_proxy.rs :: http_plain_proxy、validate_absolute_form_host_header

let method_allowed = match app_state
    .method_allowed(req.method().as_str())
    .await
    .map_err(|err| internal_error("failed to evaluate method policy", err))
{
    Ok(allowed) => allowed,
    Err(resp) => return Ok(resp),
};

let request_ctx = match RequestContext::try_from(&req) {
    Ok(ctx) => ctx,
    Err(err) => {
        warn!("missing host: {err}");
        return Ok(text_response(StatusCode::BAD_REQUEST, "missing host"));
    }
};
let authority = request_ctx.host_with_port();
let host = normalize_host(&authority.host.to_string());
let port = authority.port;
if let Err(reason) = validate_absolute_form_host_header(&req, &request_ctx) {
    return Ok(text_response(StatusCode::BAD_REQUEST, reason));
}

header mismatch 是 bad request，不是 upstream failure；method blocked、proxy disabled 和 domain denied 则各自生成对应的 policy details 与 blocked record。

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

fn remove_hop_by_hop_request_headers(headers: &mut HeaderMap) {
    while let Some(raw_connection) = headers.get(header::CONNECTION).cloned() {
        headers.remove(header::CONNECTION);
        for name in raw_connection.to_str().unwrap_or_default().split(',') {
            if let Ok(name) = HeaderName::from_bytes(name.trim().as_bytes()) {
                headers.remove(name);
            }
        }
    }
    headers.remove(header::PROXY_CONNECTION);
    headers.remove(header::KEEP_ALIVE);
    headers.remove(header::TE);
    headers.remove(header::TRAILER);
    headers.remove(header::TRANSFER_ENCODING);
    headers.remove(header::UPGRADE);
}

7. SOCKS5 TCP

SOCKS5 service 通过 TargetCheckedTcpConnector 进入 TCP handler。handler 先检查 state enabled、limited/full mode、目标 host，再调用 evaluate_host_policy；Allow 后 connector 才开始 upstream dial。HTTPS SOCKS target 还可能进入 MITM，非 HTTPS opaque target 则保留原始 stream。

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

let host = normalize_host(&req.authority.host.to_string());
let port = req.authority.port;
let target = req.authority.clone();
if host.is_empty() {
    return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid host").into());
}

源码位置：codex-rs/network-proxy/src/socks5.rs :: handle_socks5_tcp（limited mode guard）

let mode = match app_state.network_mode().await {
    Ok(mode) => mode,
    Err(err) => {
        error!("failed to evaluate method policy: {err}");
        return Err(io::Error::other("proxy error").into());
    }
};
// SOCKS5 only exposes host and port, so only the default HTTPS port is identifiable as a
// TLS stream that the HTTPS MITM path can safely terminate.
let socks5_tcp_target_is_https = port == 443;

源码继续在该条件下生成 PolicyDecisionDetails、记录 blocked request 并返回 policy error；这里不把后续分支拼进代码块，避免把不同阶段的片段误读为一段新的实现。

后续 handler 将 NetworkDecision::Deny 转成 SOCKS policy error，并把 Allow 交给 connector；NetworkDecision::Ask 不会被当成成功连接。

8. SOCKS5 UDP

UDP relay 不建立一个永久的“已授权目标连接”。inspect_socks5_udp 对每个 relay request 解析目标 IP/port，检查 proxy enabled 和 network mode，再构造 NetworkPolicyRequest。Deny 会返回 policy error；Allow 才返回携带 payload 的 RelayResponse。

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

let RelayRequest {
    server_address,
    payload,
    extensions,
    ..
} = request;

let host = normalize_host(&server_address.ip_addr.to_string());
let port = server_address.port;
if host.is_empty() {
    return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid host"));
}

let request = NetworkPolicyRequest::new(NetworkPolicyRequestArgs {
    protocol: NetworkProtocol::Socks5Udp,
    host: host.clone(),
    port,
    environment_id,
    client_addr: client.clone(),
    method: None,
    command: None,
    exec_policy_hint: None,
});

match evaluate_host_policy(&state, policy_decider.as_ref(), &request).await {
    Ok(NetworkDecision::Deny {
        reason,
        source,
        decision,
    }) => {
        let details = PolicyDecisionDetails {
            decision,
            reason: &reason,
            source,
            protocol: NetworkProtocol::Socks5Udp,
            host: &host,
            port,
        };
        Err(policy_denied_error(&reason, &details))
    }
    Ok(NetworkDecision::Allow) => Ok(RelayResponse {
        maybe_payload: Some(payload),
        extensions,
    }),
    Err(_) => Err(io::Error::other("proxy error")),
}

limited mode 下 UDP 会在 domain policy 前被 ModeGuard 阻断；因此 UDP policy event 的 source 可能是 ModeGuard、ProxyState 或 BaselinePolicy，取决于失败发生的位置。

9. 错误阶段

四类错误必须分开：

• 请求解析/Host mismatch：HTTP 400 或 SOCKS invalid target；
• enabled/method/mode/domain policy：带 reason/source 的 blocked response；
• upstream DNS/TCP/TLS：Allow 后的连接错误；
• tunnel/HTTP body/UDP relay 读写：连接建立后的 forwarding error。

10. 测试与边界

相关测试覆盖 limited mode 的 CONNECT/MITM requirement、brokered CONNECT 的 TLS prefix detection、absolute-form Host mismatch、denylisted CONNECT、SOCKS TCP/UDP policy、upstream proxy 选择和 attribution。测试中的本地 listener 只证明 fixture 的请求顺序与响应，不证明任意目标服务器或公网网络行为。

源码位置：codex-rs/network-proxy/src/http_proxy.rs :: brokered_connect_forwards_server_first_opaque_protocol_without_mitm、http_plain_proxy_rejects_absolute_uri_host_header_mismatch

let response = http_plain_proxy(
    Some(decider),
    /*environment_id*/ None,
    request,
)
.await
.expect("plain proxy response");
assert_eq!(response.status(), StatusCode::BAD_REQUEST);

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

assert_eq!(
    decision,
    NetworkDecision::Deny {
        reason: REASON_NOT_ALLOWED.to_string(),
        source: NetworkDecisionSource::Decider,
        decision: NetworkPolicyDecision::Ask,
    }
);

测试能证明：policy deny/ask 发生在 dial 前；Host mismatch 不会进入 upstream；DetectTls 对非 TLS stream 选择 opaque；MITM/connector/relay 失败保留各自错误阶段。不能外推到 DNS rebinding 防护在所有 resolver、TLS root store、企业代理或第三方网络设备中的最终效果。

在源码仓库中可运行：

cd codex-rs
cargo test -p codex-network-proxy --lib http_proxy -- --test-threads=1
cargo test -p codex-network-proxy --lib socks5 -- --test-threads=1
cargo test -p codex-network-proxy --lib upstream -- --test-threads=1

补充调用分流图：http_connect_accept 只负责接受 CONNECT，策略与连接阶段分别在后续函数中完成。

11. 阅读闭环

建议按 HTTP service → CONNECT accept → ConnectMitmMode → opaque/MITM → HTTP plain validation → SOCKS TCP/UDP inspector → target connector → upstream/forwarder → error response 阅读。读完后应能解释：为什么 policy 必须早于 dial；为什么 DetectTls 不能写成旧的 Brokered；为什么 MITM 需要再次检查 local/private target；为什么 UDP 是逐 datagram 检查；以及协议错误、policy block、upstream error 和 forwarding error 如何分别被消费者处理。

下一篇进入网络审批与规则持久化。