diff --git a/CHANGELOG.md b/CHANGELOG.md index 1dfaecb..f963825 100644 --- a/CHANGELOG.md +++ b/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. \ No newline at end of file diff --git a/src/egressweave/request_body_safety.py b/src/egressweave/request_body_safety.py index 1df62f5..cb945c1 100644 --- a/src/egressweave/request_body_safety.py +++ b/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 diff --git a/tests/test_request_body_empty_chunks.py b/tests/test_request_body_empty_chunks.py new file mode 100644 index 0000000..b75162e --- /dev/null +++ b/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