security(request): reconstruct zero-progress guard on protected main

CHANGELOG.md

@@ -88,6 +88,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   disable the recurring loop.
 
 ### Security
+- Reject repeated exact empty `bytes` chunks from synchronous and asynchronous
+  outbound request streams after permitting one compatibility empty chunk without
+  dispatch. A valid empty body can therefore retain HTTPX's canonical one-empty-
+  chunk encoding, while a second zero-progress chunk closes the source
+  deterministically and receives the generic non-leaking denial instead of
+  consuming CPU outside byte and write timeout budgets.
 - Harden release publication evidence with validated integrating-PR identity,
   cross-repository required-workflow source checks, and Strix check-run
   annotations without adding an elevated release credential.
@@ -429,4 +435,4 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   gap (CWE-350), with redirects and environment proxies disabled.
 - `EgressNotAllowedError` (a `ValueError` subclass) and `ValidatedEgressURL`.
 - 35 tests covering URL rejection, address classification, the `allow_local`
-  container case, DNS-to-private rejection, and transport pinning.
+  container case, DNS-to-private rejection, and transport pinning.

src/egressweave/request_body_safety.py

@@ -8,11 +8,12 @@
 present, actual stream consumption must also equal that declaration exactly.
 Each bounded request stream is single-consumption so an exhausted or replayable
 source cannot be retried under stale framing or a reset allowance. Only exact
-built-in ``bytes`` chunks are accepted before length accounting, preventing a
-subclass or arbitrary object from executing attacker-controlled conversion or
-length behavior at this trust boundary. The stream that would cross any boundary
-is closed before the invalid chunk can be sent, while callers continue to
-receive EgressWeave's generic non-leaking denial error.
+built-in ``bytes`` chunks are accepted before length accounting. At most one
+empty chunk is consumed without dispatch so HTTPX's canonical empty-body stream
+remains compatible while a repeated zero-progress source fails closed instead
+of spinning outside the byte and write-timeout budgets. The stream that would
+cross any boundary is closed before the invalid chunk can be sent, while callers
+continue to receive EgressWeave's generic non-leaking denial error.
 """
 
 from __future__ import annotations
@@ -109,16 +110,23 @@ def __init__(
         self._iteration_started = False
 
     def __iter__(self) -> Iterator[bytes]:
-        """Yield exact bytes only after cumulative and framing-limit checks."""
+        """Yield exact bytes after bounded-progress and framing checks."""
         if self._iteration_started:
             _close_sync_request_after_policy_denial(self._stream)
             raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
         self._iteration_started = True
+        empty_chunk_seen = False
 
         for chunk in self._stream:
             if type(chunk) is not bytes:
                 _close_sync_request_after_policy_denial(self._stream)
                 raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
+            if not chunk:
+                if empty_chunk_seen:
+                    _close_sync_request_after_policy_denial(self._stream)
+                    raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
+                empty_chunk_seen = True
+                continue
             self._consumed_bytes += len(chunk)
             exceeds_declared_length = (
                 self._declared_request_bytes is not None
@@ -185,16 +193,23 @@ def __init__(
         self._iteration_started = False
 
     async def __aiter__(self) -> AsyncIterator[bytes]:
-        """Yield exact async bytes after cumulative and framing-limit checks."""
+        """Yield exact async bytes after bounded-progress and framing checks."""
         if self._iteration_started:
             await _close_async_request_after_policy_denial(self._stream)
             raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
         self._iteration_started = True
+        empty_chunk_seen = False
 
         async for chunk in self._stream:
             if type(chunk) is not bytes:
                 await _close_async_request_after_policy_denial(self._stream)
                 raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
+            if not chunk:
+                if empty_chunk_seen:
+                    await _close_async_request_after_policy_denial(self._stream)
+                    raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
+                empty_chunk_seen = True
+                continue
             self._consumed_bytes += len(chunk)
             exceeds_declared_length = (
                 self._declared_request_bytes is not None

tests/test_request_body_empty_chunks.py

@@ -0,0 +1,112 @@
+"""Regressions for bounded zero-progress outbound request streams."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Iterator
+from pathlib import Path
+
+import httpx
+import pytest
+
+from egressweave import EgressNotAllowedError
+from egressweave.request_body_safety import (
+    _BoundedAsyncRequestStream,
+    _BoundedSyncRequestStream,
+)
+
+
+class _RepeatedEmptySyncChunkStream(httpx.SyncByteStream):
+    """Yield repeated empty chunks and record fail-closed cleanup."""
+
+    def __init__(self) -> None:
+        """Initialize the source closure marker."""
+        self.closed = False
+
+    def __iter__(self) -> Iterator[bytes]:
+        """Model an unbounded no-write source with two observable iterations."""
+        yield b""
+        yield b""
+
+    def close(self) -> None:
+        """Record source closure after policy denial."""
+        self.closed = True
+
+
+class _RepeatedEmptyAsyncChunkStream(httpx.AsyncByteStream):
+    """Yield repeated empty async chunks and record cleanup."""
+
+    def __init__(self) -> None:
+        """Initialize the asynchronous source closure marker."""
+        self.closed = False
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        """Model an async no-write source with two observable iterations."""
+        yield b""
+        yield b""
+
+    async def aclose(self) -> None:
+        """Record source closure after policy denial."""
+        self.closed = True
+
+
+def _assert_clean_denial(error: EgressNotAllowedError) -> None:
+    """Require the stable non-leaking outbound policy denial."""
+    assert str(error) == "egress URL is not allowed"
+    assert error.__cause__ is None
+    assert error.__context__ is None
+
+
+def test_sync_request_rejects_repeated_empty_chunks_before_unbounded_spin() -> None:
+    """Stop a synchronous source after its second zero-progress chunk."""
+    source = _RepeatedEmptySyncChunkStream()
+    stream = _BoundedSyncRequestStream(source, max_request_bytes=1)
+
+    with pytest.raises(EgressNotAllowedError) as caught:
+        list(stream)
+
+    _assert_clean_denial(caught.value)
+    assert source.closed is True
+
+
+async def test_async_request_rejects_repeated_empty_chunks_before_unbounded_spin() -> None:
+    """Stop an asynchronous source after its second zero-progress chunk."""
+    source = _RepeatedEmptyAsyncChunkStream()
+    stream = _BoundedAsyncRequestStream(source, max_request_bytes=1)
+
+    with pytest.raises(EgressNotAllowedError) as caught:
+        _ = [chunk async for chunk in stream]
+
+    _assert_clean_denial(caught.value)
+    assert source.closed is True
+
+
+def test_sync_request_preserves_httpx_empty_body_encoding() -> None:
+    """Consume HTTPX's one canonical empty chunk without dispatching it."""
+    source = httpx.ByteStream(b"")
+    stream = _BoundedSyncRequestStream(
+        source,
+        max_request_bytes=1,
+        declared_request_bytes=0,
+    )
+
+    assert list(stream) == []
+
+
+async def test_async_request_preserves_httpx_empty_body_encoding() -> None:
+    """Consume HTTPX's async canonical empty chunk without dispatching it."""
+    source = httpx.ByteStream(b"")
+    stream = _BoundedAsyncRequestStream(
+        source,
+        max_request_bytes=1,
+        declared_request_bytes=0,
+    )
+
+    assert [chunk async for chunk in stream] == []
+
+
+def test_changelog_describes_repeated_zero_progress_boundary() -> None:
+    """Keep release history aligned with the one-empty-chunk compatibility rule."""
+    changelog = Path("CHANGELOG.md").read_text(encoding="utf-8")
+
+    assert "Reject repeated exact empty `bytes` chunks" in changelog
+    assert "Reject exact empty `bytes` chunks" not in changelog