security(response): require exact response stream bytes

CHANGELOG.md

@@ -88,6 +88,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   disable the recurring loop.
 
 ### Security
+- Require exact built-in `bytes` from synchronous and asynchronous dependency
+  response streams before body-length accounting. Polymorphic chunks can no
+  longer under-report their payload through subclass-defined `__len__`; rejected
+  streams close deterministically and callers retain the generic non-leaking
+  policy denial.
 - 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 +434,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/response_safety.py

@@ -6,8 +6,11 @@
 capacity. EgressWeave bounds decoded response-header fields and their name/value
 bytes, requests identity coding, rejects a body-bearing response that nevertheless
 applies a content coding, rejects unsafe declared lengths before caller-visible
-delivery, and counts every transfer-decoded body byte while it is consumed. The
-identity-coding invariant prevents decompression expansion outside the byte budget.
+delivery, and counts every transfer-decoded body byte while it is consumed. Only
+exact built-in ``bytes`` chunks are accepted before that accounting, so a private
+dependency shape cannot under-report body size through polymorphic length behavior.
+The identity-coding invariant prevents decompression expansion outside the byte
+budget.
 """
 
 from __future__ import annotations
@@ -228,7 +231,7 @@ def _enforce_declared_response_size(
 
 
 class _BoundedSyncResponseStream(httpx.SyncByteStream):
-    """Count identity-coded sync response bytes and close on first overrun."""
+    """Count exact identity-coded sync response bytes and close on denial."""
 
     def __init__(
         self, stream: httpx.SyncByteStream, max_response_bytes: int
@@ -238,9 +241,12 @@ def __init__(
         self._max_response_bytes = max_response_bytes
 
     def __iter__(self) -> Iterator[bytes]:
-        """Yield chunks until the next complete chunk exceeds the budget."""
+        """Yield exact chunks until the next complete chunk exceeds the budget."""
         consumed_bytes = 0
         for chunk in self._stream:
+            if type(chunk) is not bytes:
+                _close_sync_response_after_policy_denial(self._stream)
+                raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
             consumed_bytes += len(chunk)
             if consumed_bytes > self._max_response_bytes:
                 try:
@@ -255,7 +261,7 @@ def close(self) -> None:
 
 
 class _BoundedAsyncResponseStream(httpx.AsyncByteStream):
-    """Count identity-coded async response bytes and close on first overrun."""
+    """Count exact identity-coded async response bytes and close on denial."""
 
     def __init__(
         self, stream: httpx.AsyncByteStream, max_response_bytes: int
@@ -265,9 +271,12 @@ def __init__(
         self._max_response_bytes = max_response_bytes
 
     async def __aiter__(self) -> AsyncIterator[bytes]:
-        """Yield chunks until the next complete chunk exceeds the budget."""
+        """Yield exact chunks until the next complete chunk exceeds the budget."""
         consumed_bytes = 0
         async for chunk in self._stream:
+            if type(chunk) is not bytes:
+                await _close_async_response_after_policy_denial(self._stream)
+                raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
             consumed_bytes += len(chunk)
             if consumed_bytes > self._max_response_bytes:
                 try:

tests/test_response_stream_exact_bytes.py

@@ -0,0 +1,105 @@
+"""Regressions for exact response-stream chunk accounting."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Iterator
+
+import httpx
+import pytest
+
+from egressweave import EgressNotAllowedError
+from egressweave.response_safety import (
+    _BoundedAsyncResponseStream,
+    _BoundedSyncResponseStream,
+)
+
+
+class _LyingBytes(bytes):
+    """Expose bytes while hiding their real size from polymorphic accounting."""
+
+    def __len__(self) -> int:
+        """Return a false zero length if the wrapper trusts subclass behavior."""
+        return 0
+
+
+class _SyncResponseSource(httpx.SyncByteStream):
+    """Yield one configured response chunk and record deterministic cleanup."""
+
+    def __init__(self, chunk: object) -> None:
+        """Store the dependency-controlled response chunk."""
+        self.chunk = chunk
+        self.closed = False
+
+    def __iter__(self) -> Iterator[bytes]:
+        """Yield the configured object despite the static bytes contract."""
+        yield self.chunk  # type: ignore[misc]
+
+    def close(self) -> None:
+        """Record that the rejected dependency stream was released."""
+        self.closed = True
+
+
+class _AsyncResponseSource(httpx.AsyncByteStream):
+    """Yield one configured async response chunk and record cleanup."""
+
+    def __init__(self, chunk: object) -> None:
+        """Store the dependency-controlled response chunk."""
+        self.chunk = chunk
+        self.closed = False
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        """Yield the configured object despite the static bytes contract."""
+        yield self.chunk  # type: ignore[misc]
+
+    async def aclose(self) -> None:
+        """Record that the rejected dependency stream was released."""
+        self.closed = True
+
+
+def _assert_clean_denial(error: EgressNotAllowedError) -> None:
+    """Require the stable non-leaking response policy denial."""
+    assert str(error) == "egress URL is not allowed"
+    assert error.__cause__ is None
+    assert error.__context__ is None
+
+
+def test_sync_response_rejects_bytes_subclass_before_length_accounting() -> None:
+    """Prevent a response bytes subclass from under-reporting its body size."""
+    source = _SyncResponseSource(_LyingBytes(b"0123456789"))
+    stream = _BoundedSyncResponseStream(source, max_response_bytes=4)
+
+    with pytest.raises(EgressNotAllowedError) as caught:
+        next(iter(stream))
+
+    _assert_clean_denial(caught.value)
+    assert source.closed is True
+
+
+def test_sync_response_preserves_exact_bytes_within_budget() -> None:
+    """Keep ordinary exact byte chunks available when they satisfy the budget."""
+    source = _SyncResponseSource(b"0123")
+    stream = _BoundedSyncResponseStream(source, max_response_bytes=4)
+
+    assert list(stream) == [b"0123"]
+    assert source.closed is False
+
+
+async def test_async_response_rejects_bytes_subclass_before_length_accounting() -> None:
+    """Apply exact response-chunk accounting to the asynchronous boundary."""
+    source = _AsyncResponseSource(_LyingBytes(b"0123456789"))
+    stream = _BoundedAsyncResponseStream(source, max_response_bytes=4)
+
+    with pytest.raises(EgressNotAllowedError) as caught:
+        await anext(stream.__aiter__())
+
+    _assert_clean_denial(caught.value)
+    assert source.closed is True
+
+
+async def test_async_response_preserves_exact_bytes_within_budget() -> None:
+    """Keep ordinary asynchronous byte chunks available within the finite limit."""
+    source = _AsyncResponseSource(b"0123")
+    stream = _BoundedAsyncResponseStream(source, max_response_bytes=4)
+
+    assert [chunk async for chunk in stream] == [b"0123"]
+    assert source.closed is False