openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge commit '9329dccfa7' into HEAD

Browse Source
This commit is contained in:
Winston Howes
2026-06-16 12:44:03 -07:00
parent 248ed486d2 9329dccfa7
commit c9cb2cdcf4
1624 changed files with 117329 additions and 37308 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/network-proxy/Cargo.toml
-1
View File
@@ -14,7 +14,6 @@ workspace = true
[dependencies]
anyhow = { workspace = true }
async-trait = { workspace = true }
base64 = { workspace = true }
clap = { workspace = true, features = ["derive"] }
chrono = { workspace = true }
codex-rs/network-proxy/README.md
+5 -4
View File
@@ -107,9 +107,9 @@ When a request is blocked, the proxy responds with `403` and includes:
- `blocked-by-method-policy`
- `blocked-by-policy`
In "limited" mode, only `GET`, `HEAD`, and `OPTIONS` are allowed. HTTPS `CONNECT` requests require
MITM to enforce limited-mode method policy; otherwise they are blocked. SOCKS5 remains blocked in
limited mode.
In "limited" mode, only `GET`, `HEAD`, and `OPTIONS` are allowed. HTTPS `CONNECT` requests and
HTTPS SOCKS5 TCP targets on `:443` require MITM to enforce limited-mode method policy; otherwise
they are blocked. SOCKS5 UDP and non-HTTPS SOCKS5 TCP remain blocked in limited mode.
Websocket clients typically tunnel `wss://` through HTTPS `CONNECT`; those CONNECT targets still go
through the same host allowlist/denylist checks.
@@ -215,7 +215,8 @@ what it can reasonably guarantee.
allowlisted (best-effort DNS lookup).
- Limited mode enforcement:
- only `GET`, `HEAD`, and `OPTIONS` are allowed
- HTTPS `CONNECT` remains a tunnel; limited-mode method enforcement does not apply to HTTPS
- HTTPS `CONNECT` requests and HTTPS SOCKS5 TCP targets on `:443` require MITM so the proxy can
enforce limited-mode method policy; SOCKS5 UDP and non-HTTPS SOCKS5 TCP remain blocked
- Listener safety defaults:
- the HTTP proxy listener clamps non-loopback binds unless explicitly enabled via
`dangerously_allow_non_loopback_proxy`
codex-rs/network-proxy/src/config.rs
+1 -1
View File
@@ -276,7 +276,7 @@ impl NetworkProxySettings {
pub enum NetworkMode {
/// Limited (read-only) access: only GET/HEAD/OPTIONS are allowed for HTTP. HTTPS CONNECT is
/// blocked unless MITM is enabled so the proxy can enforce method policy on inner requests.
/// SOCKS5 remains blocked in limited mode.
/// SOCKS5 UDP and non-HTTPS SOCKS5 TCP remain blocked in limited mode.
Limited,
/// Full network access: all HTTP methods are allowed. HTTPS CONNECTs are tunneled directly.
/// MITM hooks do not currently make full mode enter MITM.
codex-rs/network-proxy/src/lib.rs
+3
View File
@@ -37,6 +37,7 @@ pub use mitm_hook::MitmHookMatchConfig;
pub use network_policy::NetworkDecision;
pub use network_policy::NetworkDecisionSource;
pub use network_policy::NetworkPolicyDecider;
pub use network_policy::NetworkPolicyDeciderFuture;
pub use network_policy::NetworkPolicyDecision;
pub use network_policy::NetworkPolicyRequest;
pub use network_policy::NetworkPolicyRequestArgs;
@@ -63,7 +64,9 @@ pub use proxy::proxy_url_env_value;
pub use runtime::BlockedRequest;
pub use runtime::BlockedRequestArgs;
pub use runtime::BlockedRequestObserver;
pub use runtime::BlockedRequestObserverFuture;
pub use runtime::ConfigReloader;
pub use runtime::ConfigReloaderFuture;
pub use runtime::ConfigState;
pub use runtime::NetworkProxyState;
pub use state::NetworkProxyAuditMetadata;
codex-rs/network-proxy/src/mitm.rs
+18 -8
View File
@@ -21,10 +21,12 @@ use rama_core::Layer;
use rama_core::Service;
use rama_core::bytes::Bytes;
use rama_core::error::BoxError;
use rama_core::extensions::ExtensionsMut;
use rama_core::extensions::ExtensionsRef;
use rama_core::futures::stream::Stream;
use rama_core::futures::stream::Stream as FuturesStream;
use rama_core::rt::Executor;
use rama_core::service::service_fn;
use rama_core::stream::Stream;
use rama_http::Body;
use rama_http::BodyDataStream;
use rama_http::HeaderMap;
@@ -135,17 +137,25 @@ impl MitmState {
/// Terminate the upgraded CONNECT stream with a generated leaf cert and proxy inner HTTPS traffic.
pub(crate) async fn mitm_tunnel(upgraded: Upgraded) -> Result<()> {
let mitm = upgraded
mitm_stream(upgraded).await
}
/// Terminate a raw client stream with a generated leaf cert and proxy inner HTTPS traffic.
pub(crate) async fn mitm_stream<S>(stream: S) -> Result<()>
where
S: Stream + Unpin + ExtensionsMut,
{
let mitm = stream
.extensions()
.get::<Arc<MitmState>>()
.cloned()
.context("missing MITM state")?;
let app_state = upgraded
let app_state = stream
.extensions()
.get::<Arc<NetworkProxyState>>()
.cloned()
.context("missing app state")?;
let target = upgraded
let target = stream
.extensions()
.get::<ProxyTarget>()
.context("missing proxy target")?
@@ -154,7 +164,7 @@ pub(crate) async fn mitm_tunnel(upgraded: Upgraded) -> Result<()> {
let target_host = normalize_host(&target.host.to_string());
let target_port = target.port;
let acceptor_data = mitm.tls_acceptor_data_for_host(&target_host)?;
let mode = upgraded
let mode = stream
.extensions()
.get::<NetworkMode>()
.copied()
@@ -169,7 +179,7 @@ pub(crate) async fn mitm_tunnel(upgraded: Upgraded) -> Result<()> {
mitm,
});
let executor = upgraded
let executor = stream
.extensions()
.get::<Executor>()
.cloned()
@@ -194,7 +204,7 @@ pub(crate) async fn mitm_tunnel(upgraded: Upgraded) -> Result<()> {
.into_layer(http_service);
https_service
.serve(upgraded)
.serve(stream)
.await
.map_err(|err| anyhow!("MITM serve error: {err}"))?;
Ok(())
@@ -456,7 +466,7 @@ struct InspectStream<T> {
max_body_bytes: usize,
}
impl<T: BodyLoggable> Stream for InspectStream<T> {
impl<T: BodyLoggable> FuturesStream for InspectStream<T> {
type Item = Result<Bytes, BoxError>;
fn poll_next(self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<Option<Self::Item>> {
codex-rs/network-proxy/src/network_policy.rs
+15 -15
View File
@@ -3,10 +3,10 @@ use crate::runtime::HostBlockDecision;
use crate::runtime::HostBlockReason;
use crate::state::NetworkProxyState;
use anyhow::Result;
use async_trait::async_trait;
use chrono::SecondsFormat;
use chrono::Utc;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
const AUDIT_TARGET: &str = "codex_otel.network_proxy";
@@ -263,26 +263,26 @@ fn audit_timestamp() -> String {
/// If `command` or `exec_policy_hint` is provided, callers can map exec-policy
/// approvals to network access (e.g., allow all requests for commands matching
/// approved prefixes like `curl *`).
#[async_trait]
pub trait NetworkPolicyDecider: Send + Sync + 'static {
async fn decide(&self, req: NetworkPolicyRequest) -> NetworkDecision;
fn decide(&self, req: NetworkPolicyRequest) -> NetworkPolicyDeciderFuture<'_>;
}
#[async_trait]
pub type NetworkPolicyDeciderFuture<'a> =
Pin<Box<dyn Future<Output = NetworkDecision> + Send + 'a>>;
impl<D: NetworkPolicyDecider + ?Sized> NetworkPolicyDecider for Arc<D> {
async fn decide(&self, req: NetworkPolicyRequest) -> NetworkDecision {
(**self).decide(req).await
fn decide(&self, req: NetworkPolicyRequest) -> NetworkPolicyDeciderFuture<'_> {
Box::pin(async move { (**self).decide(req).await })
}
}
#[async_trait]
impl<F, Fut> NetworkPolicyDecider for F
where
F: Fn(NetworkPolicyRequest) -> Fut + Send + Sync + 'static,
Fut: Future<Output = NetworkDecision> + Send,
Fut: Future<Output = NetworkDecision> + Send + 'static,
{
async fn decide(&self, req: NetworkPolicyRequest) -> NetworkDecision {
(self)(req).await
fn decide(&self, req: NetworkPolicyRequest) -> NetworkPolicyDeciderFuture<'_> {
Box::pin((self)(req))
}
}
@@ -541,6 +541,7 @@ mod tests {
use crate::reasons::REASON_NOT_ALLOWED;
use crate::reasons::REASON_NOT_ALLOWED_LOCAL;
use crate::runtime::ConfigReloader;
use crate::runtime::ConfigReloaderFuture;
use crate::runtime::ConfigState;
use crate::runtime::NetworkProxyAuditMetadata;
use crate::state::NetworkProxyConstraints;
@@ -560,14 +561,13 @@ mod tests {
state: ConfigState,
}
#[async_trait]
impl ConfigReloader for StaticReloader {
async fn maybe_reload(&self) -> anyhow::Result<Option<ConfigState>> {
Ok(None)
fn maybe_reload(&self) -> ConfigReloaderFuture<'_, Option<ConfigState>> {
Box::pin(async { Ok(None) })
}
async fn reload_now(&self) -> anyhow::Result<ConfigState> {
Ok(self.state.clone())
fn reload_now(&self) -> ConfigReloaderFuture<'_, ConfigState> {
Box::pin(async { Ok(self.state.clone()) })
}
fn source_label(&self) -> String {
codex-rs/network-proxy/src/runtime.rs
+17 -18
View File
@@ -20,7 +20,6 @@ use crate::state::build_config_state;
use crate::state::validate_policy_against_constraints;
use anyhow::Context;
use anyhow::Result;
use async_trait::async_trait;
use codex_utils_absolute_path::AbsolutePathBuf;
use globset::GlobSet;
use serde::Serialize;
@@ -30,6 +29,7 @@ use std::future::Future;
use std::net::IpAddr;
use std::net::SocketAddr;
use std::path::Path;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use time::OffsetDateTime;
@@ -168,38 +168,38 @@ pub struct ConfigState {
pub blocked_total: u64,
}
#[async_trait]
pub trait ConfigReloader: Send + Sync {
/// Human-readable description of where config is loaded from, for logs.
fn source_label(&self) -> String;
/// Return a freshly loaded state if a reload is needed; otherwise, return `None`.
async fn maybe_reload(&self) -> Result<Option<ConfigState>>;
fn maybe_reload(&self) -> ConfigReloaderFuture<'_, Option<ConfigState>>;
/// Force a reload, regardless of whether a change was detected.
async fn reload_now(&self) -> Result<ConfigState>;
fn reload_now(&self) -> ConfigReloaderFuture<'_, ConfigState>;
}
#[async_trait]
pub type ConfigReloaderFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T>> + Send + 'a>>;
pub trait BlockedRequestObserver: Send + Sync + 'static {
async fn on_blocked_request(&self, request: BlockedRequest);
fn on_blocked_request(&self, request: BlockedRequest) -> BlockedRequestObserverFuture<'_>;
}
#[async_trait]
pub type BlockedRequestObserverFuture<'a> = Pin<Box<dyn Future<Output = ()> + Send + 'a>>;
impl<O: BlockedRequestObserver + ?Sized> BlockedRequestObserver for Arc<O> {
async fn on_blocked_request(&self, request: BlockedRequest) {
(**self).on_blocked_request(request).await
fn on_blocked_request(&self, request: BlockedRequest) -> BlockedRequestObserverFuture<'_> {
Box::pin(async move { (**self).on_blocked_request(request).await })
}
}
#[async_trait]
impl<F, Fut> BlockedRequestObserver for F
where
F: Fn(BlockedRequest) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send,
Fut: Future<Output = ()> + Send + 'static,
{
async fn on_blocked_request(&self, request: BlockedRequest) {
(self)(request).await
fn on_blocked_request(&self, request: BlockedRequest) -> BlockedRequestObserverFuture<'_> {
Box::pin((self)(request))
}
}
@@ -891,18 +891,17 @@ pub(crate) fn network_proxy_state_for_policy(
struct NoopReloader;
#[cfg(test)]
#[async_trait]
impl ConfigReloader for NoopReloader {
fn source_label(&self) -> String {
"test config state".to_string()
}
async fn maybe_reload(&self) -> Result<Option<ConfigState>> {
Ok(None)
fn maybe_reload(&self) -> ConfigReloaderFuture<'_, Option<ConfigState>> {
Box::pin(async { Ok(None) })
}
async fn reload_now(&self) -> Result<ConfigState> {
Err(anyhow::anyhow!("force reload is not supported in tests"))
fn reload_now(&self) -> ConfigReloaderFuture<'_, ConfigState> {
Box::pin(async { Err(anyhow::anyhow!("force reload is not supported in tests")) })
}
}
codex-rs/network-proxy/src/socks5.rs
+400 -121
View File
@@ -1,5 +1,6 @@
use crate::config::NetworkMode;
use crate::connect_policy::TargetCheckedTcpConnector;
use crate::mitm;
use crate::network_policy::BlockDecisionAuditEventArgs;
use crate::network_policy::NetworkDecision;
use crate::network_policy::NetworkDecisionSource;
@@ -24,10 +25,17 @@ use anyhow::Result;
use rama_core::Layer;
use rama_core::Service;
use rama_core::error::BoxError;
use rama_core::extensions::Extensions;
use rama_core::extensions::ExtensionsMut;
use rama_core::extensions::ExtensionsRef;
use rama_core::layer::AddInputExtensionLayer;
use rama_core::service::service_fn;
use rama_net::address::HostWithPort;
use rama_net::client::EstablishedClientConnection;
use rama_net::proxy::ProxyRequest;
use rama_net::proxy::ProxyTarget;
use rama_net::proxy::StreamForwardService;
use rama_net::stream::Socket;
use rama_net::stream::SocketInfo;
use rama_socks5::Socks5Acceptor;
use rama_socks5::server::DefaultConnector;
@@ -40,8 +48,14 @@ use rama_tcp::server::TcpListener;
use std::io;
use std::net::SocketAddr;
use std::net::TcpListener as StdTcpListener;
use std::pin::Pin;
use std::sync::Arc;
use std::task::Context as TaskContext;
use std::task::Poll;
use std::time::Instant;
use tokio::io::AsyncRead;
use tokio::io::AsyncWrite;
use tokio::io::ReadBuf;
use tracing::error;
use tracing::info;
use tracing::warn;
@@ -88,7 +102,9 @@ async fn run_socks5_with_listener(
match state.network_mode().await {
Ok(NetworkMode::Limited) => {
info!("SOCKS5 is blocked in limited mode; set mode=\"full\" to allow SOCKS5");
info!(
"SOCKS5 UDP and non-HTTPS SOCKS5 TCP are blocked in limited mode; HTTPS SOCKS5 TCP requires MITM inspection"
);
}
Ok(NetworkMode::Full) => {}
Err(err) => {
@@ -106,7 +122,10 @@ async fn run_socks5_with_listener(
}
});
let socks_connector = DefaultConnector::default().with_connector(policy_tcp_connector);
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);
if enable_socks5_udp {
@@ -135,7 +154,7 @@ async fn handle_socks5_tcp(
req: TcpRequest,
tcp_connector: TargetCheckedTcpConnector,
policy_decider: Option<Arc<dyn NetworkPolicyDecider>>,
) -> Result<EstablishedClientConnection<TcpStream, TcpRequest>, BoxError> {
) -> Result<EstablishedClientConnection<Socks5TcpConnection, TcpRequest>, BoxError> {
let app_state = req
.extensions()
.get::<Arc<NetworkProxyState>>()
@@ -144,6 +163,7 @@ async fn 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());
}
@@ -196,94 +216,52 @@ async fn handle_socks5_tcp(
}
}
match app_state.network_mode().await {
Ok(NetworkMode::Limited) => {
emit_socks_block_decision_audit_event(
&app_state,
NetworkDecisionSource::ModeGuard,
REASON_METHOD_NOT_ALLOWED,
NetworkProtocol::Socks5Tcp,
host.as_str(),
port,
client.as_deref(),
);
let details = PolicyDecisionDetails {
decision: NetworkPolicyDecision::Deny,
reason: REASON_METHOD_NOT_ALLOWED,
source: NetworkDecisionSource::ModeGuard,
protocol: NetworkProtocol::Socks5Tcp,
host: &host,
port,
};
let _ = app_state
.record_blocked(BlockedRequest::new(BlockedRequestArgs {
host: host.clone(),
reason: REASON_METHOD_NOT_ALLOWED.to_string(),
client: client.clone(),
method: None,
mode: Some(NetworkMode::Limited),
protocol: "socks5".to_string(),
decision: Some(details.decision.as_str().to_string()),
source: Some(details.source.as_str().to_string()),
port: Some(port),
}))
.await;
let client = client.as_deref().unwrap_or_default();
warn!(
"SOCKS blocked by method policy (client={client}, host={host}, mode=limited, allowed_methods=GET, HEAD, OPTIONS)"
);
return Err(policy_denied_error(REASON_METHOD_NOT_ALLOWED, &details).into());
}
Ok(NetworkMode::Full) => {}
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());
}
}
match app_state.host_has_mitm_hooks(&host).await {
Ok(true) => {
emit_socks_block_decision_audit_event(
&app_state,
NetworkDecisionSource::ModeGuard,
REASON_MITM_REQUIRED,
NetworkProtocol::Socks5Tcp,
host.as_str(),
port,
client.as_deref(),
);
let details = PolicyDecisionDetails {
decision: NetworkPolicyDecision::Deny,
reason: REASON_MITM_REQUIRED,
source: NetworkDecisionSource::ModeGuard,
protocol: NetworkProtocol::Socks5Tcp,
host: &host,
port,
};
let _ = app_state
.record_blocked(BlockedRequest::new(BlockedRequestArgs {
host: host.clone(),
reason: REASON_MITM_REQUIRED.to_string(),
client: client.clone(),
method: None,
mode: Some(NetworkMode::Full),
protocol: "socks5".to_string(),
decision: Some(details.decision.as_str().to_string()),
source: Some(details.source.as_str().to_string()),
port: Some(port),
}))
.await;
let client = client.as_deref().unwrap_or_default();
warn!(
"SOCKS blocked; MITM required to enforce HTTPS policy (client={client}, host={host}, mode=full)"
);
return Err(policy_denied_error(REASON_MITM_REQUIRED, &details).into());
}
Ok(false) => {}
Err(err) => {
error!("failed to inspect MITM hooks for {host}: {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;
if mode == NetworkMode::Limited && !socks5_tcp_target_is_https {
emit_socks_block_decision_audit_event(
&app_state,
NetworkDecisionSource::ModeGuard,
REASON_METHOD_NOT_ALLOWED,
NetworkProtocol::Socks5Tcp,
host.as_str(),
port,
client.as_deref(),
);
let details = PolicyDecisionDetails {
decision: NetworkPolicyDecision::Deny,
reason: REASON_METHOD_NOT_ALLOWED,
source: NetworkDecisionSource::ModeGuard,
protocol: NetworkProtocol::Socks5Tcp,
host: &host,
port,
};
let _ = app_state
.record_blocked(BlockedRequest::new(BlockedRequestArgs {
host: host.clone(),
reason: REASON_METHOD_NOT_ALLOWED.to_string(),
client: client.clone(),
method: None,
mode: Some(NetworkMode::Limited),
protocol: "socks5".to_string(),
decision: Some(details.decision.as_str().to_string()),
source: Some(details.source.as_str().to_string()),
port: Some(port),
}))
.await;
let client = client.as_deref().unwrap_or_default();
warn!(
"SOCKS blocked; limited mode only supports HTTPS MITM (client={client}, host={host}, port={port})"
);
return Err(policy_denied_error(REASON_METHOD_NOT_ALLOWED, &details).into());
}
let request = NetworkPolicyRequest::new(NetworkPolicyRequestArgs {
@@ -337,9 +315,85 @@ async fn handle_socks5_tcp(
}
}
let host_has_mitm_hooks = match app_state.host_has_mitm_hooks(&host).await {
Ok(has_hooks) => has_hooks,
Err(err) => {
error!("failed to inspect MITM hooks for {host}: {err}");
return Err(io::Error::other("proxy error").into());
}
};
let mitm_state = match app_state.mitm_state().await {
Ok(state) => state,
Err(err) => {
error!("failed to load MITM state: {err}");
return Err(io::Error::other("proxy error").into());
}
};
let socks_needs_mitm =
socks5_tcp_target_is_https && (mode == NetworkMode::Limited || host_has_mitm_hooks);
if (host_has_mitm_hooks && !socks5_tcp_target_is_https)
|| (socks_needs_mitm && mitm_state.is_none())
{
emit_socks_block_decision_audit_event(
&app_state,
NetworkDecisionSource::ModeGuard,
REASON_MITM_REQUIRED,
NetworkProtocol::Socks5Tcp,
host.as_str(),
port,
client.as_deref(),
);
let details = PolicyDecisionDetails {
decision: NetworkPolicyDecision::Deny,
reason: REASON_MITM_REQUIRED,
source: NetworkDecisionSource::ModeGuard,
protocol: NetworkProtocol::Socks5Tcp,
host: &host,
port,
};
let _ = app_state
.record_blocked(BlockedRequest::new(BlockedRequestArgs {
host: host.clone(),
reason: REASON_MITM_REQUIRED.to_string(),
client: client.clone(),
method: None,
mode: Some(mode),
protocol: "socks5".to_string(),
decision: Some(details.decision.as_str().to_string()),
source: Some(details.source.as_str().to_string()),
port: Some(port),
}))
.await;
let client = client.as_deref().unwrap_or_default();
warn!(
"SOCKS blocked; MITM required to enforce HTTPS policy (client={client}, host={host}, mode={mode:?}, hooked_host={host_has_mitm_hooks}, https_target={socks5_tcp_target_is_https})"
);
return Err(policy_denied_error(REASON_MITM_REQUIRED, &details).into());
}
if socks_needs_mitm && let Some(mitm_state) = mitm_state {
let client = client.as_deref().unwrap_or_default();
info!("SOCKS MITM enabled (client={client}, host={host}, port={port}, mode={mode:?})");
return Ok(EstablishedClientConnection {
input: req,
conn: Socks5TcpConnection::Mitm {
target,
mode,
mitm: mitm_state,
extensions: Extensions::new(),
},
});
}
info!("SOCKS upstream dial started (host={host}, port={port})");
let connect_started_at = Instant::now();
let result = tcp_connector.serve(req).await;
let result = tcp_connector.serve(req).await.map(|connection| {
let EstablishedClientConnection { input, conn } = connection;
EstablishedClientConnection {
input,
conn: Socks5TcpConnection::Direct(conn),
}
});
match &result {
Ok(_) => info!(
"SOCKS upstream dial established (host={host}, port={port}, elapsed_ms={})",
@@ -353,6 +407,113 @@ async fn handle_socks5_tcp(
result
}
/// 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(Debug)]
enum Socks5TcpConnection {
Direct(TcpStream),
Mitm {
target: HostWithPort,
mode: NetworkMode,
mitm: Arc<mitm::MitmState>,
extensions: Extensions,
},
}
impl AsyncRead for Socks5TcpConnection {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut TaskContext<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
match self.get_mut() {
Self::Direct(stream) => Pin::new(stream).poll_read(cx, buf),
Self::Mitm { .. } => Poll::Ready(Ok(())),
}
}
}
impl AsyncWrite for Socks5TcpConnection {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut TaskContext<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
match self.get_mut() {
Self::Direct(stream) => Pin::new(stream).poll_write(cx, buf),
Self::Mitm { .. } => Poll::Ready(Ok(buf.len())),
}
}
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::Mitm { .. } => Poll::Ready(Ok(())),
}
}
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::Mitm { .. } => Poll::Ready(Ok(())),
}
}
}
impl Socket for Socks5TcpConnection {
fn local_addr(&self) -> io::Result<SocketAddr> {
match self {
Self::Direct(stream) => stream.local_addr(),
Self::Mitm { .. } => Ok(SocketAddr::from(([0, 0, 0, 0], 0))),
}
}
fn peer_addr(&self) -> io::Result<SocketAddr> {
match self {
Self::Direct(stream) => stream.peer_addr(),
Self::Mitm { .. } => Ok(SocketAddr::from(([0, 0, 0, 0], 0))),
}
}
}
impl ExtensionsRef for Socks5TcpConnection {
fn extensions(&self) -> &Extensions {
match self {
Self::Direct(stream) => stream.extensions(),
Self::Mitm { extensions, .. } => extensions,
}
}
}
impl ExtensionsMut for Socks5TcpConnection {
fn extensions_mut(&mut self) -> &mut Extensions {
match self {
Self::Direct(stream) => stream.extensions_mut(),
Self::Mitm { extensions, .. } => extensions,
}
}
}
async fn proxy_socks5_tcp(
request: ProxyRequest<TcpStream, Socks5TcpConnection>,
) -> Result<(), BoxError> {
let ProxyRequest { mut source, target } = request;
match target {
Socks5TcpConnection::Direct(target) => StreamForwardService::default()
.serve(ProxyRequest { source, target })
.await
.map_err(Into::into),
Socks5TcpConnection::Mitm {
target, mode, mitm, ..
} => {
source.extensions_mut().insert(ProxyTarget(target));
source.extensions_mut().insert(mode);
source.extensions_mut().insert(mitm);
mitm::mitm_stream(source).await.map_err(Into::into)
}
}
}
async fn inspect_socks5_udp(
request: RelayRequest,
state: Arc<NetworkProxyState>,
@@ -553,11 +714,10 @@ mod tests {
use crate::network_policy::test_support::capture_events;
use crate::network_policy::test_support::find_event_by_name;
use crate::runtime::ConfigReloader;
use crate::runtime::ConfigReloaderFuture;
use crate::runtime::ConfigState;
use crate::runtime::network_proxy_state_for_policy;
use crate::state::NetworkProxyConstraints;
use crate::state::build_config_state;
use async_trait::async_trait;
use pretty_assertions::assert_eq;
use rama_core::extensions::Extensions;
use rama_core::extensions::ExtensionsMut;
@@ -567,20 +727,24 @@ mod tests {
use std::net::IpAddr;
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::sync::Mutex;
// Managed MITM CA files live under the shared test CODEX_HOME, so MITM-enabled config state
// must be materialized one test at a time.
static MITM_CONFIG_STATE_LOCK: Mutex<()> = Mutex::new(());
#[derive(Clone)]
struct StaticReloader {
state: ConfigState,
}
#[async_trait]
impl ConfigReloader for StaticReloader {
async fn maybe_reload(&self) -> anyhow::Result<Option<ConfigState>> {
Ok(None)
fn maybe_reload(&self) -> ConfigReloaderFuture<'_, Option<ConfigState>> {
Box::pin(async { Ok(None) })
}
async fn reload_now(&self) -> anyhow::Result<ConfigState> {
Ok(self.state.clone())
fn reload_now(&self) -> ConfigReloaderFuture<'_, ConfigState> {
Box::pin(async { Ok(self.state.clone()) })
}
fn source_label(&self) -> String {
@@ -590,6 +754,10 @@ mod tests {
fn state_for_settings(network: NetworkProxySettings) -> Arc<NetworkProxyState> {
let config = NetworkProxyConfig { network };
let _mitm_config_state_guard = config
.network
.mitm
.then(|| MITM_CONFIG_STATE_LOCK.lock().unwrap());
let state = build_config_state(config, NetworkProxyConstraints::default()).unwrap();
let reloader = Arc::new(StaticReloader {
state: state.clone(),
@@ -639,43 +807,56 @@ mod tests {
}
#[tokio::test(flavor = "current_thread")]
async fn handle_socks5_tcp_blocks_hooked_host_in_full_mode() {
let state = Arc::new(network_proxy_state_for_policy(NetworkProxySettings {
async fn handle_socks5_tcp_uses_mitm_in_limited_mode() {
let mut settings = NetworkProxySettings {
enabled: true,
mode: NetworkMode::Full,
mode: NetworkMode::Limited,
mitm: true,
mitm_hooks: vec![MitmHookConfig {
host: "api.github.com".to_string(),
matcher: MitmHookMatchConfig {
methods: vec!["GET".to_string()],
path_prefixes: vec!["/".to_string()],
..MitmHookMatchConfig::default()
},
..MitmHookConfig::default()
}],
..NetworkProxySettings::default()
}));
};
settings.set_allowed_domains(vec!["example.com".to_string()]);
let state = state_for_settings(settings);
let mut request =
TcpRequest::new(HostWithPort::try_from("api.github.com:443").expect("valid authority"));
TcpRequest::new(HostWithPort::try_from("example.com:443").expect("valid authority"));
request.extensions_mut().insert(state.clone());
let result = handle_socks5_tcp(
request,
TargetCheckedTcpConnector::new(state),
/*policy_decider*/ None,
)
.await
.expect("limited-mode HTTPS should use MITM");
assert!(matches!(result.conn, Socks5TcpConnection::Mitm { .. }));
}
#[tokio::test(flavor = "current_thread")]
async fn handle_socks5_tcp_blocks_non_https_in_limited_mode() {
let mut settings = NetworkProxySettings {
enabled: true,
mode: NetworkMode::Limited,
..NetworkProxySettings::default()
};
settings.set_allowed_domains(vec!["example.com".to_string()]);
let state = state_for_settings(settings);
let mut request =
TcpRequest::new(HostWithPort::try_from("example.com:80").expect("valid authority"));
request.extensions_mut().insert(state.clone());
let (result, events) = capture_events(|| async {
handle_socks5_tcp(
request,
TargetCheckedTcpConnector::new(state.clone()),
TargetCheckedTcpConnector::new(state),
/*policy_decider*/ None,
)
.await
})
.await;
assert!(result.is_err(), "hooked host should require MITM");
let blocked = state.drain_blocked().await.unwrap();
assert_eq!(blocked.len(), 1);
assert_eq!(blocked[0].reason, REASON_MITM_REQUIRED);
assert_eq!(blocked[0].host, "api.github.com");
assert_eq!(blocked[0].port, Some(443));
assert_eq!(blocked[0].protocol, "socks5");
assert!(
result.is_err(),
"limited-mode non-HTTPS SOCKS should be denied"
);
let event = find_event_by_name(&events, POLICY_DECISION_EVENT_NAME)
.expect("expected policy decision event");
@@ -684,18 +865,116 @@ mod tests {
assert_eq!(event.field("network.policy.source"), Some("mode_guard"));
assert_eq!(
event.field("network.policy.reason"),
Some(REASON_MITM_REQUIRED)
Some(REASON_METHOD_NOT_ALLOWED)
);
assert_eq!(
event.field("network.transport.protocol"),
Some("socks5_tcp")
);
assert_eq!(event.field("server.address"), Some("api.github.com"));
assert_eq!(event.field("server.port"), Some("443"));
assert_eq!(event.field("server.address"), Some("example.com"));
assert_eq!(event.field("server.port"), Some("80"));
assert_eq!(event.field("http.request.method"), Some("none"));
assert_eq!(event.field("client.address"), Some("unknown"));
}
#[tokio::test(flavor = "current_thread")]
async fn handle_socks5_tcp_blocks_limited_mode_without_mitm_state() {
let mut settings = NetworkProxySettings {
enabled: true,
mode: NetworkMode::Limited,
..NetworkProxySettings::default()
};
settings.set_allowed_domains(vec!["example.com".to_string()]);
let state = state_for_settings(settings);
let mut request =
TcpRequest::new(HostWithPort::try_from("example.com:443").expect("valid authority"));
request.extensions_mut().insert(state.clone());
let err = handle_socks5_tcp(
request,
TargetCheckedTcpConnector::new(state),
/*policy_decider*/ None,
)
.await
.expect_err("limited-mode HTTPS requires MITM");
assert!(
format!("{err:?}").contains("MITM required"),
"unexpected error: {err:?}"
);
}
#[tokio::test(flavor = "current_thread")]
async fn handle_socks5_tcp_uses_mitm_for_hooked_host_in_full_mode() {
let mut settings = NetworkProxySettings {
enabled: true,
mode: NetworkMode::Full,
mitm: true,
mitm_hooks: vec![MitmHookConfig {
host: "api.github.com".to_string(),
matcher: MitmHookMatchConfig {
methods: vec!["POST".to_string()],
path_prefixes: vec!["/repos/openai/".to_string()],
..MitmHookMatchConfig::default()
},
..MitmHookConfig::default()
}],
..NetworkProxySettings::default()
};
settings.set_allowed_domains(vec!["api.github.com".to_string()]);
let state = state_for_settings(settings);
let mut request =
TcpRequest::new(HostWithPort::try_from("api.github.com:443").expect("valid authority"));
request.extensions_mut().insert(state.clone());
let result = handle_socks5_tcp(
request,
TargetCheckedTcpConnector::new(state),
/*policy_decider*/ None,
)
.await
.expect("hooked HTTPS should use MITM");
assert!(matches!(result.conn, Socks5TcpConnection::Mitm { .. }));
}
#[tokio::test(flavor = "current_thread")]
async fn handle_socks5_tcp_blocks_hooked_non_https_host_in_full_mode() {
let mut settings = NetworkProxySettings {
enabled: true,
mode: NetworkMode::Full,
mitm: true,
mitm_hooks: vec![MitmHookConfig {
host: "api.github.com".to_string(),
matcher: MitmHookMatchConfig {
methods: vec!["POST".to_string()],
path_prefixes: vec!["/repos/openai/".to_string()],
..MitmHookMatchConfig::default()
},
..MitmHookConfig::default()
}],
..NetworkProxySettings::default()
};
settings.set_allowed_domains(vec!["api.github.com".to_string()]);
let state = state_for_settings(settings);
let mut request =
TcpRequest::new(HostWithPort::try_from("api.github.com:80").expect("valid authority"));
request.extensions_mut().insert(state.clone());
let err = handle_socks5_tcp(
request,
TargetCheckedTcpConnector::new(state),
/*policy_decider*/ None,
)
.await
.expect_err("hooked non-HTTPS SOCKS should require MITM");
assert!(
format!("{err:?}").contains("MITM required"),
"unexpected error: {err:?}"
);
}
#[tokio::test(flavor = "current_thread")]
async fn inspect_socks5_udp_emits_block_decision_for_mode_guard_deny() {
let state = state_for_settings(NetworkProxySettings {