Support bounded request bodies in AWS SigV4 signing (#56277)

## Why The network proxy previously required AWS signing requests to have an empty body or use `UNSIGNED-PAYLOAD`, preventing signing of nonempty payloads. ## What changed - Buffer and hash request bodies before signing, then forward the original bytes. - Add `max_body_bytes` with an 8 MiB default and a 16 MiB maximum, plus a 30-second body read timeout. - Limit concurrent buffering to eight requests with a five-second capacity wait. Retain reservations while forwarded body data remains alive. - Reject trailers and mismatched content lengths for buffered payloads while preserving streaming for explicitly unsigned payloads. ## Testing Add coverage for payload hashing and forwarding, body limits and timeouts, fragmented body collection, and buffer reservation lifetimes. Extend signing tests for S3 path and query canonicalization, header replacement, and secret exclusion from trace logs. GitOrigin-RevId: 48c18442fa90ae3efb1a4a99a168d07e1be2c496
2026-10-23 18:08:45 +00:00 · 2026-10-22 22:28:10 +00:00
parent ca87ef86e1
commit 129220bf05
6 changed files with 485 additions and 18 deletions
@@ -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 }
@@ -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<ProxySecretSource>,
    #[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<Request> {
+    pub(crate) async fn sign(&self, mut req: Request, guest: &GuestSignature) -> Result<Request> {
        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
+        })
    }
 }

@@ -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<Request> {
+async fn resign(signer: &AwsSigV4Signer, mut req: Request, headers: &str) -> Result<Request> {
    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<_>>(),
+        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::<std::io::Result<rama_core::bytes::Bytes>>();
+    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"));
 }
@@ -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<Arc<Semaphore>> =
+    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<OwnedSemaphorePermit>);
+
+pub(crate) async fn reserve(req: &mut Request) -> Result<Arc<OwnedSemaphorePermit>> {
+    reserve_in(req, &BUFFER_SLOTS).await
+}
+
+async fn reserve_in(
+    req: &mut Request,
+    slots: &Arc<Semaphore>,
+) -> Result<Arc<OwnedSemaphorePermit>> {
+    if let Some(permit) = req
+        .extensions()
+        .get::<BufferLease>()
+        .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<HeaderMap>)> {
+    let mut bytes = Vec::new();
+    let mut trailers: Option<HeaderMap> = 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<Arc<OwnedSemaphorePermit>>,
+}
+
+struct LeasedBytes {
+    bytes: Bytes,
+    _lease: Arc<OwnedSemaphorePermit>,
+}
+
+impl AsRef<[u8]> for LeasedBytes {
+    fn as_ref(&self) -> &[u8] {
+        &self.bytes
+    }
+}
+
+impl StreamingBody for RetainedBody {
+    type Data = Bytes;
+    type Error = <Body as StreamingBody>::Error;
+
+    fn poll_frame(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, 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::<BufferLease>()
+        .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;
@@ -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<AtomicUsize>,
+    }
+
+    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);
+}
@@ -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;