test(security): reproduce response-driven cookie persistence

src/egressweave/cookie_safety.py

@@ -0,0 +1,18 @@
+"""Cookie-state controls shared by EgressWeave HTTPX client builders."""
+
+from __future__ import annotations
+
+from http.cookiejar import Cookie, CookieJar, DefaultCookiePolicy
+
+
+class _RejectResponseCookiePolicy(DefaultCookiePolicy):
+    """Reject response-provided cookies while allowing explicit caller state."""
+
+    def set_ok(self, cookie: Cookie, request: object) -> bool:
+        """Refuse automatic response-driven cookie persistence."""
+        return False
+
+
+def _new_explicit_cookie_jar() -> CookieJar:
+    """Return a cookie jar that accepts direct caller writes but no responses."""
+    return CookieJar(policy=_RejectResponseCookiePolicy())

src/egressweave/sync_transport.py

@@ -19,6 +19,7 @@
 import httpx
 from httpx._transports.default import ResponseStream, map_httpcore_exceptions
 
+from egressweave.cookie_safety import _new_explicit_cookie_jar
 from egressweave.policy import EgressPolicy, _normalize_host
 from egressweave.request_body_safety import (
     _BoundedSyncRequestStream,
@@ -294,8 +295,9 @@ def close(self) -> None:
 
 
 def _build_sync_httpx_client(transport: httpx.BaseTransport) -> httpx.Client:
-    """Build a client without HTTPX's ambient hop-by-hop connection header."""
+    """Build a client with safe ambient headers and caller-owned cookie state."""
     client = httpx.Client(
+        cookies=_new_explicit_cookie_jar(),
         follow_redirects=False,
         trust_env=False,
         transport=transport,
@@ -318,6 +320,8 @@ def build_egress_sync_client(
     :class:`~egressweave.validation.EgressNotAllowedError`. The builder removes
     HTTPX's ambient ``Connection`` header so ordinary caller requests reach the
     transport's strict hop-by-hop-header policy without weakening that policy.
+    Response-provided cookies remain visible to callers but are not persisted as
+    ambient request state; callers may still supply cookie state explicitly.
     Exact outbound targets are limited by ``policy.max_request_target_bytes``.
     Final outbound fields are limited by ``policy.max_request_header_fields`` and
     ``policy.max_request_header_bytes``. Request bodies are limited to
@@ -353,14 +357,15 @@ def build_pinned_https_client(
     The supplied result is revalidated without another DNS lookup. HTTPX's
     ambient ``Connection`` header is removed at construction while any caller-
     supplied hop-by-hop field remains subject to the transport's fail-closed
-    request-header policy. Every connection is pinned to its addresses, any
-    forged result or authority change is rejected before network I/O, every
-    exact outbound target and request header section is bounded after trusted
-    rewriting, every request body is constrained by ``policy.max_request_bytes``
-    and exact declared framing, every request phase is capped by
-    ``policy.request_timeout_policy``, response metadata is bounded by the finite
-    header policy, and every identity-coded response body is constrained by
-    ``policy.max_response_bytes``.
+    request-header policy. Response cookies are observable but never promoted to
+    ambient later-request state; explicit caller-owned cookies remain available.
+    Every connection is pinned to its addresses, any forged result or authority
+    change is rejected before network I/O, every exact outbound target and
+    request header section is bounded after trusted rewriting, every request body
+    is constrained by ``policy.max_request_bytes`` and exact declared framing,
+    every request phase is capped by ``policy.request_timeout_policy``, response
+    metadata is bounded by the finite header policy, and every identity-coded
+    response body is constrained by ``policy.max_response_bytes``.
     """
     return _build_sync_httpx_client(
         _PinnedEgressTransport(

src/egressweave/transport.py

@@ -29,6 +29,7 @@
 from httpcore._backends.auto import AutoBackend
 from httpx._transports.default import AsyncResponseStream, map_httpcore_exceptions
 
+from egressweave.cookie_safety import _new_explicit_cookie_jar
 from egressweave.policy import EgressPolicy, _normalize_host
 from egressweave.request_body_safety import (
     _BoundedAsyncRequestStream,
@@ -447,8 +448,9 @@ async def aclose(self) -> None:
 
 
 def _build_async_httpx_client(transport: httpx.AsyncBaseTransport) -> httpx.AsyncClient:
-    """Build a client without HTTPX's ambient hop-by-hop connection header."""
+    """Build a client with safe ambient headers and caller-owned cookie state."""
     client = httpx.AsyncClient(
+        cookies=_new_explicit_cookie_jar(),
         follow_redirects=False,
         trust_env=False,
         transport=transport,
@@ -467,7 +469,9 @@ async def build_egress_http_client(
 
     Empty or absent URLs return a deny-all client. HTTPX's ambient ``Connection``
     header is removed at construction while caller-supplied hop-by-hop fields
-    remain subject to the transport's strict request-header policy. Exact outbound
+    remain subject to the transport's strict request-header policy. Response-
+    provided cookies remain visible but are not promoted to ambient later-request
+    state; callers may still provide cookie state explicitly. Exact outbound
     targets are limited by ``policy.max_request_target_bytes``. Final outbound
     fields are limited by ``policy.max_request_header_fields`` and
     ``policy.max_request_header_bytes``. Request bodies are limited to
@@ -502,7 +506,8 @@ def build_pinned_https_async_client(
 
     HTTPX's ambient ``Connection`` header is removed at construction while any
     caller-supplied hop-by-hop field remains subject to the fail-closed transport
-    policy.
+    policy. Response cookies are observable but never promoted to ambient later-
+    request state; explicit caller-owned cookies remain available.
     """
     return _build_async_httpx_client(
         _PinnedEgressAsyncTransport(

tests/test_response_cookie_state.py

@@ -0,0 +1,145 @@
+"""Security contracts for response-driven cookie state in pinned clients."""
+
+from __future__ import annotations
+
+import httpcore
+import pytest
+
+from egressweave import (
+    EgressNotAllowedError,
+    EgressPolicy,
+    build_pinned_https_async_client,
+    build_pinned_https_client,
+    validate_egress_url_details,
+)
+from egressweave import validation as v
+
+POLICY = EgressPolicy.from_hosts("api.openai.com")
+_SERVER_COOKIE = b"session=server; Domain=.openai.com; Path=/"
+
+
+def _validated_result(monkeypatch):
+    """Return one deterministic public validated authority without public DNS."""
+
+    def fake_getaddrinfo(host, port, type=None):
+        return [(2, 1, 6, "", ("93.184.216.34", port))]
+
+    monkeypatch.setattr(v.socket, "getaddrinfo", fake_getaddrinfo)
+    validated = validate_egress_url_details(
+        "https://api.openai.com", policy=POLICY
+    )
+    assert validated is not None
+    return validated
+
+
+def _header_value(request, name: bytes) -> bytes | None:
+    """Return one recorded HTTP core request header by lowercase name."""
+    return {key.lower(): value for key, value in request.headers}.get(name.lower())
+
+
+class _SyncCookiePool:
+    """Return one response cookie and record every synchronous outbound request."""
+
+    def __init__(self) -> None:
+        self.requests = []
+        self.closed = False
+
+    def handle_request(self, request):
+        self.requests.append(request)
+        headers = [(b"set-cookie", _SERVER_COOKIE)] if len(self.requests) == 1 else []
+        return httpcore.Response(204, headers=headers, content=b"")
+
+    def close(self) -> None:
+        self.closed = True
+
+
+class _AsyncCookiePool:
+    """Return one response cookie and record every asynchronous outbound request."""
+
+    def __init__(self) -> None:
+        self.requests = []
+        self.closed = False
+
+    async def handle_async_request(self, request):
+        self.requests.append(request)
+        headers = [(b"set-cookie", _SERVER_COOKIE)] if len(self.requests) == 1 else []
+
+        async def empty_content():
+            if False:  # pragma: no cover - required async-iterator shape only
+                yield b""
+
+        return httpcore.Response(204, headers=headers, content=empty_content())
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+def test_sync_client_does_not_replay_response_cookie(monkeypatch) -> None:
+    """A response cookie must stay visible without becoming later request state."""
+    validated = _validated_result(monkeypatch)
+    pool = _SyncCookiePool()
+
+    with build_pinned_https_client(validated, policy=POLICY) as client:
+        client._transport._pool = pool
+        first = client.get("https://api.openai.com/first")
+        assert first.headers["set-cookie"].startswith("session=server")
+
+        client.get("https://api.openai.com/second")
+
+    assert len(pool.requests) == 2
+    assert _header_value(pool.requests[1], b"cookie") is None
+    assert pool.closed is True
+
+
+async def test_async_client_does_not_replay_response_cookie(monkeypatch) -> None:
+    """Async response cookies must not silently become later request metadata."""
+    validated = _validated_result(monkeypatch)
+    pool = _AsyncCookiePool()
+
+    async with build_pinned_https_async_client(validated, policy=POLICY) as client:
+        client._transport._pool = pool
+        first = await client.get("https://api.openai.com/first")
+        assert first.headers["set-cookie"].startswith("session=server")
+
+        await client.get("https://api.openai.com/second")
+
+    assert len(pool.requests) == 2
+    assert _header_value(pool.requests[1], b"cookie") is None
+    assert pool.closed is True
+
+
+def test_sync_client_preserves_explicit_host_owned_cookie_state(monkeypatch) -> None:
+    """Callers may still opt into explicit same-authority cookie credentials."""
+    validated = _validated_result(monkeypatch)
+    pool = _SyncCookiePool()
+
+    with build_pinned_https_client(validated, policy=POLICY) as client:
+        client._transport._pool = pool
+        client.cookies.set(
+            "host_session",
+            "caller",
+            domain="api.openai.com",
+            path="/",
+        )
+        client.get("https://api.openai.com/explicit")
+
+    cookie = _header_value(pool.requests[0], b"cookie")
+    assert cookie is not None
+    assert b"host_session=caller" in cookie
+
+
+def test_sync_client_keeps_exact_authority_denial_after_response_cookie(monkeypatch) -> None:
+    """Cookie-domain metadata cannot widen the validated transport authority."""
+    validated = _validated_result(monkeypatch)
+    pool = _SyncCookiePool()
+
+    with build_pinned_https_client(validated, policy=POLICY) as client:
+        client._transport._pool = pool
+        client.get("https://api.openai.com/first")
+
+        with pytest.raises(
+            EgressNotAllowedError, match="^egress URL is not allowed$"
+        ):
+            client.get("https://other.openai.com/second")
+
+    assert len(pool.requests) == 1