diff --git a/codex-rs/network-proxy/Cargo.toml b/codex-rs/network-proxy/Cargo.toml index df3c80240b..0706660b95 100644 --- a/codex-rs/network-proxy/Cargo.toml +++ b/codex-rs/network-proxy/Cargo.toml @@ -57,6 +57,7 @@ sha2 = { workspace = true } [dev-dependencies] pretty_assertions = { workspace = true } tempfile = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } [target.'cfg(target_family = "unix")'.dependencies] libc = { workspace = true } diff --git a/codex-rs/network-proxy/src/aws_sigv4.rs b/codex-rs/network-proxy/src/aws_sigv4.rs index 4fc2cd656b..717a3fccc5 100644 --- a/codex-rs/network-proxy/src/aws_sigv4.rs +++ b/codex-rs/network-proxy/src/aws_sigv4.rs @@ -14,6 +14,7 @@ use aws_sigv4::http_request::UriPathNormalizationMode; use aws_sigv4::http_request::sign; use aws_sigv4::sign::v4; use codex_utils_absolute_path::AbsolutePathBuf; +use rama_core::bytes::Bytes; use rama_core::extensions::ExtensionsMut; use rama_core::extensions::ExtensionsRef; use rama_http::Body; @@ -28,6 +29,7 @@ use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; use sha2::Digest as _; +use std::time::Duration; use std::time::SystemTime; /// A credential source in the trusted proxy process, never in the executor environment. @@ -77,11 +79,16 @@ pub struct AwsSigV4Config { pub session_token: Option, #[serde(default = "https_port")] pub port: u16, + #[serde(default = "default_max_body_bytes")] + pub max_body_bytes: usize, } pub(crate) fn https_port() -> u16 { 443 } +pub(crate) fn default_max_body_bytes() -> usize { + 8 * 1024 * 1024 +} #[derive(Clone, PartialEq, Eq)] pub(crate) struct AwsSigV4Signer { @@ -118,6 +125,10 @@ impl AwsSigV4Config { } } ensure!(self.port != 0, "invalid AWS signing port"); + ensure!( + self.max_body_bytes <= 16 * 1024 * 1024, + "AWS signing body limit exceeds 16 MiB" + ); self.access_key_id.validate()?; self.secret_access_key.validate()?; if let Some(token) = &self.session_token { @@ -333,7 +344,7 @@ impl AwsSigV4Signer { })) } - pub(crate) fn sign(&self, req: Request, guest: &GuestSignature) -> Result { + pub(crate) async fn sign(&self, mut req: Request, guest: &GuestSignature) -> Result { ensure!( req.uri().scheme_str() == Some("https") && req.uri().port_u16().unwrap_or(443) == self.config.port @@ -341,6 +352,7 @@ impl AwsSigV4Signer { "AWS signing requires the configured HTTPS origin and supported framing" ); let had_payload_hash = req.headers().contains_key("x-amz-content-sha256"); + let mut permit = None; let (mut parts, body, payload_hash) = if guest.unsigned_payload { let (parts, body) = req.into_parts(); ensure!( @@ -357,14 +369,26 @@ impl AwsSigV4Signer { ); (parts, body, "UNSIGNED-PAYLOAD".to_string()) } else { - let (parts, body) = req.into_parts(); + let (bytes, trailers) = if req.body().is_end_stream() { + (Bytes::new(), None) + } else { + permit = Some(crate::body_budget::reserve(&mut req).await?); + let body = std::mem::take(req.body_mut()); + tokio::time::timeout( + Duration::from_secs(30), + crate::body_budget::collect_body(body, self.config.max_body_bytes), + ) + .await + .map_err(|_| anyhow!("AWS signing body read timed out"))?? + }; + let (parts, _) = req.into_parts(); ensure!( - body.is_end_stream() - && crate::body_budget::content_length_matches(&parts.headers, /*size*/ 0), - "AWS signing requires an empty or explicitly unsigned payload" + trailers.is_none() + && crate::body_budget::content_length_matches(&parts.headers, bytes.len()), + "AWS signing requires unambiguous payload framing" ); - let digest = format!("{:x}", sha2::Sha256::digest([])); - (parts, Body::empty(), digest) + let digest = format!("{:x}", sha2::Sha256::digest(&bytes)); + (parts, Body::from(bytes), digest) }; if ["connection", "proxy-connection", "upgrade", "te"] .iter() @@ -497,7 +521,12 @@ impl AwsSigV4Signer { ); parts.headers.insert(header.name(), value); } - Ok(Request::from_parts(parts, body)) + let req = Request::from_parts(parts, body); + Ok(if permit.is_some() { + crate::body_budget::retain(req) + } else { + req + }) } } diff --git a/codex-rs/network-proxy/src/aws_sigv4_tests.rs b/codex-rs/network-proxy/src/aws_sigv4_tests.rs index 94fe412ead..fcde5d2389 100644 --- a/codex-rs/network-proxy/src/aws_sigv4_tests.rs +++ b/codex-rs/network-proxy/src/aws_sigv4_tests.rs @@ -13,6 +13,7 @@ fn config(service: &str) -> AwsSigV4Config { secret_access_key: ProxySecretSource::EnvVar("SECRET_KEY".into()), session_token: None, port: 443, + max_body_bytes: 1024, } } @@ -35,6 +36,12 @@ fn signer(config: AwsSigV4Config) -> AwsSigV4Signer { .unwrap() } +fn time() -> SystemTime { + chrono::DateTime::parse_from_rfc3339("2015-08-30T12:36:00Z") + .unwrap() + .into() +} + pub(crate) fn guest_authorization(service: &str, signed_headers: &str) -> String { format!( "AWS4-HMAC-SHA256 Credential=MASKEDACCESSKEY/20150830/us-east-1/{service}/aws4_request, SignedHeaders={signed_headers}, Signature={:064x}", @@ -42,18 +49,20 @@ pub(crate) fn guest_authorization(service: &str, signed_headers: &str) -> String ) } -fn resign(signer: &AwsSigV4Signer, mut req: Request, headers: &str) -> Result { +async fn resign(signer: &AwsSigV4Signer, mut req: Request, headers: &str) -> Result { let auth = guest_authorization(&signer.config.service, headers); req.headers_mut() .insert("authorization", HeaderValue::from_str(&auth)?); req.headers_mut() .insert("x-amz-date", HeaderValue::from_static("20150830T123600Z")); let guest = signer.plan(&req)?.expect("selected guest credential"); - signer.sign(req, &guest) + signer.sign(req, &guest).await } #[test] fn matches_aws_published_iam_signature_vector() { + use rama_core::futures::FutureExt as _; + // AWS Signature Version 4 ListUsers example, 2015-08-30. let request = Request::builder() .method("GET") @@ -70,6 +79,8 @@ fn matches_aws_published_iam_signature_vector() { request, "content-type;host;x-amz-date", ) + .now_or_never() + .expect("an empty request must sign without waiting for buffer capacity") .unwrap(); assert_eq!( signed.headers()["authorization"], @@ -78,17 +89,65 @@ fn matches_aws_published_iam_signature_vector() { } #[tokio::test] -async fn streams_explicitly_unsigned_s3_uploads() { +async fn signs_actual_payload_and_replaces_untrusted_signing_headers() { let mut config = config("s3"); config.session_token = Some(ProxySecretSource::EnvVar("SESSION_TOKEN".into())); let signer = signer(config); + let mut cookie = HeaderValue::from_static("b=2"); + cookie.set_sensitive(true); + let request = Request::builder() + .method("PUT") + .version(rama_http::Version::HTTP_2) + .uri("https://bucket.s3.amazonaws.com/a%20b?tag=b&tag=a&empty=") + .header("host", "bucket.s3.amazonaws.com") + .header("cookie", "a=1") + .header("cookie", cookie) + .header("authorization", "guest-signature") + .header("x-amz-date", "19700101T000000Z") + .header("x-amz-security-token", "guest-token") + .header("x-amz-content-sha256", "untrusted-guest-hash") + .body(Body::from("hello")) + .unwrap(); + let signed = resign( + &signer, + request, + "cookie;host;x-amz-content-sha256;x-amz-date", + ) + .await + .unwrap(); + assert_eq!( + signed.headers()["x-amz-content-sha256"], + "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" + ); + assert_eq!(signed.headers()["x-amz-date"], "20150830T123600Z"); + assert_eq!( + signed.headers()["x-amz-security-token"], + "secret-session-token" + ); + assert!(signed.headers()["x-amz-security-token"].is_sensitive()); + assert!(signed.headers()["authorization"].is_sensitive()); + assert_eq!( + signed + .headers() + .get_all("cookie") + .iter() + .collect::>(), + vec![&HeaderValue::from_static("a=1; b=2")] + ); + assert!(signed.headers()["cookie"].is_sensitive()); + let forwarded = signed.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(forwarded.as_ref(), b"hello"); + assert!(!format!("{signer:?}").contains("EXAMPLEKEY")); + assert!(!format!("{signer:?}").contains("AKIDEXAMPLE")); + + // An SDK's unsigned S3 upload streams through without consuming the literal-body budget. let upload = Request::builder() .uri("https://bucket.s3.amazonaws.com/upload") .header("x-amz-content-sha256", "UNSIGNED-PAYLOAD") .header("x-unsigned-metadata", "forward only") .body(Body::from(vec![b'x'; 1025])) .unwrap(); - let upload = resign(&signer, upload, "host;x-amz-date").unwrap(); + let upload = resign(&signer, upload, "host;x-amz-date").await.unwrap(); assert_eq!(upload.headers()["x-amz-content-sha256"], "UNSIGNED-PAYLOAD"); let authorization = upload.headers()["authorization"].to_str().unwrap(); assert!(authorization.contains("SignedHeaders=host;x-amz-date;x-amz-security-token,")); @@ -99,8 +158,57 @@ async fn streams_explicitly_unsigned_s3_uploads() { ); } -#[test] -fn rejects_wrong_transport_scope_streaming_and_nonempty_bodies() { +#[tokio::test] +async fn canonicalizes_query_order_without_changing_encoded_s3_path() { + use sha2::Digest; + + for service in ["s3", "s3-object-lambda", "s3-outposts"] { + let signer = signer(config(service)); + let payload_hash = format!("{:x}", sha2::Sha256::digest(b"")); + let signed_headers = "host;x-amz-content-sha256;x-amz-date"; + // Build the canonical request independently of SDK request canonicalization. + // The encoded space and repeated slash must remain part of the object key. + let canonical_request = format!( + "GET\n/a%20b//key\na=one&a=two&space=a%20b&z=last\nhost:bucket.s3.amazonaws.com\nx-amz-content-sha256:{payload_hash}\nx-amz-date:20150830T123600Z\n\n{signed_headers}\n{payload_hash}" + ); + let scope = format!("20150830/us-east-1/{service}/aws4_request"); + let string_to_sign = format!( + "AWS4-HMAC-SHA256\n20150830T123600Z\n{scope}\n{:x}", + sha2::Sha256::digest(canonical_request.as_bytes()) + ); + let key = aws_sigv4::sign::v4::generate_signing_key( + "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY", + time(), + "us-east-1", + service, + ); + let signature = aws_sigv4::sign::v4::calculate_signature(key, string_to_sign.as_bytes()); + let authorization = format!( + "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/{scope}, SignedHeaders={signed_headers}, Signature={signature}" + ); + for query in [ + "z=last&a=two&a=one&space=a%20b", + "a=one&space=a%20b&a=two&z=last", + ] { + let uri = format!("https://bucket.s3.amazonaws.com/a%20b//key?{query}"); + let request = Request::builder() + .uri(&uri) + .header("x-amz-content-sha256", payload_hash.as_str()) + .body(Body::empty()) + .unwrap(); + let signed = resign(&signer, request, signed_headers).await.unwrap(); + assert_eq!(signed.uri().to_string(), uri); + assert_eq!( + signed.headers()["authorization"], + authorization, + "{service}" + ); + } + } +} + +#[tokio::test] +async fn rejects_wrong_transport_port_streaming_presigning_and_body_overflow() { let signer = signer(config("s3")); let valid = "https://bucket.s3.amazonaws.com/key"; for uri in [ @@ -109,7 +217,25 @@ fn rejects_wrong_transport_scope_streaming_and_nonempty_bodies() { "https://bucket.s3.amazonaws.com/key?X-Amz-Signature=guest", ] { let req = Request::builder().uri(uri).body(Body::empty()).unwrap(); - assert!(resign(&signer, req, "host;x-amz-date").is_err(), "{uri}"); + assert!( + resign(&signer, req, "host;x-amz-date").await.is_err(), + "{uri}" + ); + } + for (name, value) in [ + ("content-encoding", "aws-chunked"), + ("x-amz-content-sha256", "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"), + ("content-length", "1"), + ] { + let req = Request::builder() + .uri(valid) + .header(name, value) + .body(Body::from("abc")) + .unwrap(); + assert!( + resign(&signer, req, "host;x-amz-date").await.is_err(), + "{name}" + ); } let auth = guest_authorization("s3", "host;x-amz-date"); for value in [ @@ -136,6 +262,73 @@ fn rejects_wrong_transport_scope_streaming_and_nonempty_bodies() { Body::empty().with_trailer_headers(HeaderMap::new()), ] { let req = Request::builder().uri(valid).body(body).unwrap(); - assert!(resign(&signer, req, "host;x-amz-date").is_err()); + assert!(resign(&signer, req, "host;x-amz-date").await.is_err()); } + let stalled = rama_core::futures::stream::pending::>(); + let mut req = Request::builder() + .uri(valid) + .body(Body::from_stream(stalled)) + .unwrap(); + // Other tests use the process-wide pool on real clocks. + let _permit = crate::body_budget::reserve(&mut req).await.unwrap(); + tokio::time::pause(); + let start = tokio::time::Instant::now(); + let signing = resign(&signer, req, "host;x-amz-date"); + let result = tokio::time::timeout(Duration::from_secs(31), signing) + .await + .expect("signer must enforce its own deadline"); + let error = result.unwrap_err(); + assert_eq!(error.to_string(), "AWS signing body read timed out"); + assert!(start.elapsed() >= Duration::from_secs(30)); +} + +#[test] +fn credentials_are_required_and_configured_limits_are_bounded() { + assert!( + config("s3") + .compile(&|_| None, &|_| panic!("unexpected file")) + .is_err() + ); + let mut config = config("s3"); + config.max_body_bytes = 16 * 1024 * 1024 + 1; + assert_eq!( + config.validate().unwrap_err().to_string(), + "AWS signing body limit exceeds 16 MiB" + ); + config.port = 0; + assert_eq!( + config.validate().unwrap_err().to_string(), + "invalid AWS signing port" + ); +} + +#[tokio::test] +async fn sdk_trace_does_not_log_canonical_secret_headers() { + use tracing::instrument::WithSubscriber; + let capture = tempfile::NamedTempFile::new().unwrap(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_max_level(tracing::Level::TRACE) + .with_writer(std::sync::Arc::new(capture.reopen().unwrap())) + .finish(); + let request = Request::builder() + .uri("https://example.com/") + .header("x-api-key", "secret-header-value") + .body(Body::empty()) + .unwrap(); + let mut config = config("execute-api"); + config.session_token = Some(ProxySecretSource::EnvVar("SESSION_TOKEN".into())); + async { + tracing::trace!("proxy audit remains enabled"); + resign(&signer(config), request, "host;x-amz-date;x-api-key") + .await + .unwrap(); + } + .with_subscriber(subscriber) + .await; + let logs = std::fs::read_to_string(capture.path()).unwrap(); + assert!(logs.contains("proxy audit remains enabled")); + assert!(!logs.contains("secret-header-value")); + assert!(!logs.contains("secret-session-token")); + assert!(!logs.contains("EXAMPLEKEY")); } diff --git a/codex-rs/network-proxy/src/body_budget.rs b/codex-rs/network-proxy/src/body_budget.rs index fb90b1fafa..f00e148ac2 100644 --- a/codex-rs/network-proxy/src/body_budget.rs +++ b/codex-rs/network-proxy/src/body_budget.rs @@ -1,6 +1,60 @@ -//! Validate original HTTP framing and the declared length for AWS signing. +//! Shared leases bound buffers across quota checks, secret substitution, and signing. +//! Eight requests each reserve space for a 16 MiB input and 16 MiB transformed output. +use anyhow::Result; +use anyhow::anyhow; +use rama_core::bytes::Bytes; +use rama_core::extensions::ExtensionsMut; +use rama_core::extensions::ExtensionsRef; +use rama_http::Body; use rama_http::HeaderMap; +use rama_http::Request; +use rama_http::body::Frame; +use rama_http::body::SizeHint; +use rama_http::body::StreamingBody; +use rama_http::body::util::BodyExt; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::LazyLock; +use std::sync::Weak; +use std::task::Context; +use std::task::Poll; +use std::time::Duration; +use tokio::sync::OwnedSemaphorePermit; +use tokio::sync::Semaphore; + +const BUFFER_SLOT_COUNT: usize = 8; +static BUFFER_SLOTS: LazyLock> = + LazyLock::new(|| Arc::new(Semaphore::new(BUFFER_SLOT_COUNT))); + +// Rama clones request metadata through response headers. Keep ownership in the transform or body. +#[derive(Debug, Clone)] +struct BufferLease(Weak); + +pub(crate) async fn reserve(req: &mut Request) -> Result> { + reserve_in(req, &BUFFER_SLOTS).await +} + +async fn reserve_in( + req: &mut Request, + slots: &Arc, +) -> Result> { + if let Some(permit) = req + .extensions() + .get::() + .and_then(|lease| lease.0.upgrade()) + { + return Ok(permit); + } + let permit = tokio::time::timeout(Duration::from_secs(5), slots.clone().acquire_owned()) + .await + .map_err(|_| anyhow!("request buffer capacity unavailable"))? + .map_err(|_| anyhow!("request buffer budget closed"))?; + let permit = Arc::new(permit); + req.extensions_mut() + .insert(BufferLease(Arc::downgrade(&permit))); + Ok(permit) +} pub(crate) fn is_amazon_event_stream(headers: &HeaderMap) -> bool { headers.get_all("content-type").iter().any(|value| { @@ -43,3 +97,99 @@ pub(crate) fn content_length_matches(headers: &HeaderMap, size: usize) -> bool { } } } + +/// Copy frames as they arrive so fragmented bodies cannot retain per-frame allocations. +pub(crate) async fn collect_body( + mut body: Body, + max_bytes: usize, +) -> Result<(Bytes, Option)> { + let mut bytes = Vec::new(); + let mut trailers: Option = None; + while let Some(frame) = body.frame().await { + let frame = frame.map_err(|_| anyhow!("cannot read request body"))?; + if trailers.is_some() { + return Err(anyhow!("unexpected request frame after trailers")); + } + match frame.into_data() { + Ok(data) => { + if data.len() > max_bytes - bytes.len() { + return Err(anyhow!("request body exceeds configured limit")); + } + let len = bytes.len() + data.len(); + if len > bytes.capacity() { + // Grow geometrically without reserving beyond the configured byte limit. + let capacity = max_bytes.min(bytes.capacity().saturating_mul(2).max(len)); + bytes + .try_reserve_exact(capacity - bytes.len()) + .map_err(|_| anyhow!("request body buffer allocation failed"))?; + } + bytes.extend_from_slice(&data); + } + Err(frame) => trailers = frame.into_trailers().ok(), + } + } + Ok((Bytes::from(bytes), trailers)) +} + +struct RetainedBody { + body: Body, + lease: Option>, +} + +struct LeasedBytes { + bytes: Bytes, + _lease: Arc, +} + +impl AsRef<[u8]> for LeasedBytes { + fn as_ref(&self) -> &[u8] { + &self.bytes + } +} + +impl StreamingBody for RetainedBody { + type Data = Bytes; + type Error = ::Error; + + fn poll_frame( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + match Pin::new(&mut self.body).poll_frame(cx) { + Poll::Ready(Some(Ok(frame))) => { + let lease = self.lease.clone(); + Poll::Ready(Some(Ok(frame.map_data(|bytes| match lease { + Some(lease) => Bytes::from_owner(LeasedBytes { + bytes, + _lease: lease, + }), + None => bytes, + })))) + } + other => other, + } + } + + fn is_end_stream(&self) -> bool { + self.body.is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.body.size_hint() + } +} + +pub(crate) fn retain(req: Request) -> Request { + let (parts, body) = req.into_parts(); + let lease = parts + .extensions + .get::() + .and_then(|lease| lease.0.upgrade()); + // Keep exact sizing, and retain the lease in data frames queued after the body is dropped. + let body = Body::new(RetainedBody { body, lease }); + Request::from_parts(parts, body) +} + +#[cfg(test)] +#[path = "body_budget_tests.rs"] +mod tests; diff --git a/codex-rs/network-proxy/src/body_budget_tests.rs b/codex-rs/network-proxy/src/body_budget_tests.rs new file mode 100644 index 0000000000..1336742ed1 --- /dev/null +++ b/codex-rs/network-proxy/src/body_budget_tests.rs @@ -0,0 +1,94 @@ +//! Reservations survive request metadata disposal and are reused across transform stages. + +use super::*; +use pretty_assertions::assert_eq; +use rama_core::futures::stream; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +#[tokio::test] +async fn fragmented_body_collection_releases_chunks_and_enforces_the_byte_limit() { + struct Chunk { + bytes: [u8; 1], + live: Arc, + } + + impl AsRef<[u8]> for Chunk { + fn as_ref(&self) -> &[u8] { + &self.bytes + } + } + + impl Drop for Chunk { + fn drop(&mut self) { + self.live.fetch_sub(1, Ordering::SeqCst); + } + } + + for max_bytes in [1024, 1023] { + let live = Arc::new(AtomicUsize::new(0)); + let chunks = (0..1024).map({ + let live = live.clone(); + move |_| { + assert_eq!( + live.fetch_add(1, Ordering::SeqCst), + 0, + "the collector retained a previous chunk" + ); + Ok::<_, std::io::Error>(Bytes::from_owner(Chunk { + bytes: *b"x", + live: live.clone(), + })) + } + }); + let result = collect_body(Body::from_stream(stream::iter(chunks)), max_bytes).await; + assert_eq!(live.load(Ordering::SeqCst), 0); + if max_bytes == 1024 { + assert_eq!(result.unwrap(), (Bytes::from(vec![b'x'; 1024]), None)); + } else { + assert!(result.is_err()); + } + } +} + +#[tokio::test] +async fn body_keeps_buffer_lease_and_nested_stages_reuse_it() { + let mut request = Request::new(Body::from("payload")); + let slots = Arc::new(Semaphore::new(BUFFER_SLOT_COUNT)); + let permit = reserve_in(&mut request, &slots).await.unwrap(); + let lease = Arc::downgrade(&permit); + let nested = reserve_in(&mut request, &slots).await.unwrap(); + assert!(Arc::ptr_eq(&permit, &nested)); + let request = retain(request); + drop(nested); + drop(permit); + let metadata = request.extensions().clone(); + let (parts, mut body) = request.into_parts(); + assert_eq!(body.size_hint().exact(), Some(7)); + drop(parts); + assert!(lease.upgrade().is_some()); + let frame = body.frame().await.expect("body frame").expect("body data"); + drop(body); + assert!(lease.upgrade().is_some()); + let bytes = frame.into_data().expect("data frame"); + assert_eq!(bytes.as_ref(), b"payload"); + let queued = bytes.clone(); + drop(bytes); + assert!(lease.upgrade().is_some()); + let mut held = Vec::new(); + for _ in 1..8 { + let permit = slots + .clone() + .try_acquire_owned() + .expect("eight buffer slots"); + held.push(permit); + } + let mut ninth = Request::new(Body::empty()); + let reservation = reserve_in(&mut ninth, &slots); + tokio::pin!(reservation); + assert!(rama_core::futures::poll!(&mut reservation).is_pending()); + drop(queued); + assert!(lease.upgrade().is_none()); + let _permit = reservation.await.expect("released buffer slot"); + drop(metadata); +} diff --git a/codex-rs/network-proxy/src/mitm.rs b/codex-rs/network-proxy/src/mitm.rs index 9d0a475187..7c21a85a56 100644 --- a/codex-rs/network-proxy/src/mitm.rs +++ b/codex-rs/network-proxy/src/mitm.rs @@ -370,7 +370,7 @@ async fn forward_request(req: Request, request_ctx: &MitmRequestContext) -> Resu let mut upstream_req = Request::from_parts(parts, body); if let Some((signer, guest)) = signing { - let Ok(signed) = signer.sign(upstream_req, &guest) else { + let Ok(signed) = signer.sign(upstream_req, &guest).await else { return Ok(request_ctx.policy.deny_hook(&method, client).await); }; upstream_req = signed;