fix(search): pin httpx connection to resolved IP to block DNS rebinding (#704)

* fix(search): pin DNS-validated fetch connections

Rebase the DNS-rebinding SSRF fix onto current dev after search content moved behind the services.search.content canonical module.

Integrate the pinned httpcore NetworkBackend/BaseTransport approach with the current size-capped Client.stream fetch path, preserving Host/SNI semantics while forcing TCP connect to the already validated resolved IP.

Keep src.search.content as the compatibility wrapper and preserve existing OG-image http(s) behavior; this avoids reintroducing the unrelated scope changes that previously blocked review.

Add the explicit httpcore>=1.0,<2.0 requirement used by the public httpcore NetworkBackend and ConnectionPool APIs.

* test(search): restore and rebase DNS rebinding regressions

Keep the current security regression coverage that the stale PR branch had deleted, including auth-disabled localhost bypass and Ollama cookbook hardening tests.

Carry forward the DNS-rebinding coverage for private resolve blocking, pinned TCP connect behavior, Host header preservation, redirect revalidation, and the BaseTransport/public-httpcore static guard.

Update redirect tests to mock the current Client.stream-based capped fetch path rather than the older httpx.stream/get path.

* test(search): adapt size-cap fetch tests to pinned client stream

The DNS-rebinding repair moved _get_public_url from the module-level httpx.stream shortcut to httpx.Client(...).stream(...) so the fetch can use the pinned transport.

Keep the existing size-cap test fakes by routing Client.stream through the monkeypatched httpx.stream only when a test has installed that fake; otherwise fall back to a real Client.

This fixes the CI failures in tests/test_web_fetch_size_caps.py without touching unrelated upload-handler atomicity behavior, which is already flaky on clean origin/dev.

---------

Co-authored-by: Alexandre Teixeira <alexandremagteixeira@gmail.com>

requirements.txt

@@ -3,6 +3,7 @@ uvicorn
 python-multipart
 python-dotenv
 httpx
+httpcore>=1.0,<2.0
 pydantic>=2.13.4
 pydantic-settings>=2.14.1
 SQLAlchemy

services/search/content.py

@@ -8,11 +8,13 @@ import os
 import re
 import logging
 import socket
+import ssl
 from datetime import datetime, timedelta
-from typing import List
+from typing import Iterable, List, cast
 from urllib.parse import urljoin, urlparse
 
 import httpx
+import httpcore
 from bs4 import BeautifulSoup
 
 from src.constants import WEB_FETCH_SOFT_MAX_BYTES, WEB_FETCH_HARD_MAX_BYTES, WEB_FETCH_USER_AGENT
@@ -91,6 +93,148 @@ def _public_http_url(url: str) -> bool:
         return False
 
 
+def _resolve_public_ips(url: str) -> list[ipaddress._BaseAddress]:
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https") or not parsed.hostname:
+        raise httpx.RequestError(f"Blocked non-public URL: {url}")
+    host = (parsed.hostname or "").strip().lower()
+    if host in ("localhost", "metadata", "metadata.google.internal"):
+        raise httpx.RequestError(f"Blocked non-public hostname: {host}")
+    try:
+        ip = ipaddress.ip_address(host)
+        if _is_private_address(ip):
+            raise httpx.RequestError(f"Blocked non-public IP literal: {host}")
+        return [ip]
+    except httpx.RequestError:
+        raise
+    except ValueError:
+        pass
+    addrs = _resolve_hostname_ips(host)
+    if not addrs or any(_is_private_address(a) for a in addrs):
+        raise httpx.RequestError(f"Blocked non-public URL: {url}")
+    return addrs
+
+
+class _PinnedBackend(httpcore.NetworkBackend):
+    """Network backend that connects to a pre-resolved IP.
+
+    httpcore derives the TLS SNI and the ``Host`` header from the URL's
+    origin, not from the host argument passed to ``connect_tcp``. So
+    routing the TCP connect to a resolved IP while leaving the URL
+    untouched keeps SNI / vhost behaviour correct and closes the
+    DNS-rebinding TOCTOU between the SSRF check and the connect.
+    """
+
+    def __init__(self, ip: ipaddress._BaseAddress):
+        self._ip = str(ip)
+        self._real = httpcore.SyncBackend()
+
+    def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: float | None = None,
+        local_address: str | None = None,
+        socket_options=None,
+    ):
+        return self._real.connect_tcp(
+            self._ip, port, timeout, local_address, socket_options
+        )
+
+    def connect_unix_socket(self, path, timeout=None, socket_options=None):
+        return self._real.connect_unix_socket(path, timeout, socket_options)
+
+    def sleep(self, seconds: float) -> None:
+        return self._real.sleep(seconds)
+
+
+# Map httpcore exception classes to their httpx equivalents. Built
+# once at import time from the public exception classes; avoids any
+# import of httpx's private transport machinery. httpcore's
+# ``ConnectionNotAvailable`` is a pool-internal signal (the pool will
+# close and retry on its own) — we never expect to see it surface to
+# a transport caller, so it has no httpx counterpart here.
+_HTTPCORE_TO_HTTPX_EXC = {
+    httpcore.ConnectError: httpx.ConnectError,
+    httpcore.ConnectTimeout: httpx.ConnectTimeout,
+    httpcore.LocalProtocolError: httpx.LocalProtocolError,
+    httpcore.NetworkError: httpx.NetworkError,
+    httpcore.PoolTimeout: httpx.PoolTimeout,
+    httpcore.ProtocolError: httpx.ProtocolError,
+    httpcore.ProxyError: httpx.ProxyError,
+    httpcore.ReadError: httpx.ReadError,
+    httpcore.ReadTimeout: httpx.ReadTimeout,
+    httpcore.RemoteProtocolError: httpx.RemoteProtocolError,
+    httpcore.TimeoutException: httpx.TimeoutException,
+    httpcore.UnsupportedProtocol: httpx.UnsupportedProtocol,
+    httpcore.WriteError: httpx.WriteError,
+    httpcore.WriteTimeout: httpx.WriteTimeout,
+}
+
+
+class _PinnedTransport(httpx.BaseTransport):
+    """Transport that pins every TCP connect to a pre-resolved IP.
+
+    Uses only the public ``httpcore`` and ``httpx`` APIs — no
+    subclassing of ``httpx.HTTPTransport``, no reads of private
+    ``httpcore.ConnectionPool`` attributes, no imports from
+    ``httpx private transport internals``. The URL is passed through unchanged so SNI
+    / vhost work as if httpx had been given the hostname directly;
+    only the TCP destination is pinned, closing the DNS-rebinding
+    TOCTOU between the SSRF check and the connect.
+    """
+
+    def __init__(self, ip: ipaddress._BaseAddress, *, http2: bool = False):
+        self._pool = httpcore.ConnectionPool(
+            ssl_context=ssl.create_default_context(),
+            http1=True,
+            http2=http2,
+            network_backend=_PinnedBackend(ip),
+        )
+
+    def __enter__(self):
+        self._pool.__enter__()
+        return self
+
+    def __exit__(self, exc_type=None, exc_value=None, traceback=None) -> None:
+        self._pool.__exit__(exc_type, exc_value, traceback)
+
+    def handle_request(self, request: httpx.Request) -> httpx.Response:
+        httpcore_req = httpcore.Request(
+            method=request.method,
+            url=httpcore.URL(
+                scheme=request.url.raw_scheme,
+                host=request.url.raw_host,
+                port=request.url.port,
+                target=request.url.raw_path,
+            ),
+            headers=request.headers.raw,
+            content=request.stream,
+            extensions=request.extensions,
+        )
+        try:
+            httpcore_resp = self._pool.handle_request(httpcore_req)
+            # Eager materialisation matches the original
+            # ``response.text`` usage in fetch_webpage_content. The
+            # sync pool's stream is a plain Iterable[bytes] despite
+            # the httpcore type hint unioning the async variant.
+            content = b"".join(cast(Iterable[bytes], httpcore_resp.stream))
+        except Exception as exc:
+            mapped = _HTTPCORE_TO_HTTPX_EXC.get(type(exc))
+            if mapped is not None:
+                raise mapped(str(exc)) from exc
+            raise
+
+        return httpx.Response(
+            status_code=httpcore_resp.status,
+            headers=httpcore_resp.headers,
+            content=content,
+            extensions=httpcore_resp.extensions,
+        )
+
+    def close(self) -> None:
+        self._pool.close()
+
 class BodyTooLargeError(Exception):
     """The server declared a body larger than the hard fetch ceiling."""
 
@@ -141,28 +285,32 @@ class _CappedFetch:
 
 def _get_public_url(url: str, headers: dict, timeout: int, max_redirects: int = 5,
                     max_bytes: int = None) -> "_CappedFetch":
-    """Capped streaming GET with SSRF-guarded manual redirects.
+    """Capped streaming GET with SSRF-guarded, DNS-pinned manual redirects.
 
-    The body is streamed and buffering stops at ``max_bytes`` (default: the
+    Each hop is resolved once, validated as public, and then the actual TCP
-    soft cap), so an oversized resource cannot be pulled into memory or the
+    connection is pinned to that resolved IP. The request URL is left unchanged
-    content cache in full. When Content-Length already declares a body over
+    so Host and TLS SNI keep the original hostname.
-    the hard ceiling, the fetch is refused before any body bytes are read.
     """
     cap = min(max_bytes or WEB_FETCH_SOFT_MAX_BYTES, WEB_FETCH_HARD_MAX_BYTES)
     current = url
     for _ in range(max_redirects + 1):
-        if not _public_http_url(current):
+        ips = _resolve_public_ips(current)
-            raise httpx.RequestError("Blocked private/internal URL", request=httpx.Request("GET", current))
+
         # Force identity transfer-encoding. With gzip/deflate the wire bytes
-        # (and Content-Length) can be a small fraction of the decoded body, so
+        # and Content-Length can be a small fraction of the decoded body, so a
-        # a tiny compressed response could pass the hard-cap preflight and then
+        # tiny compressed response could pass the hard-cap preflight and then
-        # expand past the ceiling in a single decoded chunk before the streamed
+        # expand past the ceiling in one decoded chunk before the streamed cap
-        # cap below can slice it. Identity makes Content-Length the true body
+        # below can slice it.
-        # size and keeps each streamed chunk bounded by the network read.
         req_headers = dict(headers or {})
         req_headers["Accept-Encoding"] = "identity"
-        with httpx.stream("GET", current, headers=req_headers, timeout=timeout,
+
-                          follow_redirects=False) as response:
+        with httpx.Client(
+            headers=req_headers,
+            timeout=timeout,
+            follow_redirects=False,
+            transport=_PinnedTransport(ips[0]),
+        ) as client:
+            with client.stream("GET", current) as response:
                 if response.status_code in (301, 302, 303, 307, 308):
                     location = response.headers.get("location")
                     if not location:
@@ -172,11 +320,10 @@ def _get_public_url(url: str, headers: dict, timeout: int, max_redirects: int =
                     continue
 
                 # A server can ignore the identity request and still return a
-            # compressed body; httpx.iter_bytes would then decode it, and a tiny
+                # compressed body; httpx.iter_bytes would then decode it, and a
-            # gzip can balloon into one decoded chunk far past the cap before we
+                # tiny gzip can balloon into one decoded chunk far past the cap.
-            # slice. Refuse a compressed Content-Encoding so the streamed cap
+                # Refuse compressed Content-Encoding so the streamed cap stays
-            # stays a real memory bound (Content-Length is the compressed wire
+                # a real memory bound.
-            # length here, so the preflight and size metadata are unreliable too).
                 enc = (response.headers.get("content-encoding") or "").strip().lower()
                 if enc and enc != "identity":
                     raise httpx.RequestError(
@@ -189,18 +336,13 @@ def _get_public_url(url: str, headers: dict, timeout: int, max_redirects: int =
                 raw_len = response.headers.get("content-length")
                 if raw_len and raw_len.isdigit():
                     declared = int(raw_len)
-            # Refuse before buffering anything when the server already tells
+
-            # us the body exceeds the absolute ceiling (Content-Length is wire
-            # bytes; the decompressed body can only be larger).
                 if declared is not None and declared > WEB_FETCH_HARD_MAX_BYTES:
                     raise BodyTooLargeError(current, declared)
 
                 chunks = []
                 read = 0
                 truncated = False
-            # We requested identity above, so iter_bytes yields the raw body in
-            # network-read-sized chunks (no decompression expansion); the cap
-            # therefore bounds what we actually buffer.
                 for chunk in response.iter_bytes():
                     read += len(chunk)
                     if read > cap:
@@ -210,9 +352,11 @@ def _get_public_url(url: str, headers: dict, timeout: int, max_redirects: int =
                         truncated = True
                         break
                     chunks.append(chunk)
+
                 return _CappedFetch(response.status_code, response.headers,
                                     b"".join(chunks), truncated, declared,
                                     response.encoding, str(response.url))
+
     raise httpx.RequestError("Too many redirects", request=httpx.Request("GET", current))
 
 # PDF extraction (optional dependency)

tests/test_security_regressions.py

@@ -894,7 +894,8 @@ def test_web_fetch_guard_fails_closed_on_empty_resolution(monkeypatch):
 
 def test_web_fetch_guard_blocks_redirect_into_private(monkeypatch):
     # A public URL that 302-redirects to an internal address must be blocked
-    # at the redirect hop, not followed.
+    # at the redirect hop, not followed. _get_public_url now uses
+    # httpx.Client(...).stream(...) so the test must mock that path.
     import httpx
     from src.search import content
 
@@ -905,14 +906,31 @@ def test_web_fetch_guard_blocks_redirect_into_private(monkeypatch):
         status_code = 302
         url = "http://public.example/start"
         headers = {"location": "http://169.254.169.254/latest/meta-data/"}
+        encoding = "utf-8"
 
-    from contextlib import contextmanager
+    class _FakeStream:
+        def __enter__(self):
+            return _Resp()
 
-    @contextmanager
+        def __exit__(self, *args):
-    def _fake_stream(method, url, **kwargs):
+            return False
-        yield _Resp()
 
-    monkeypatch.setattr(httpx, "stream", _fake_stream)
+    class _FakeClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+        def stream(self, method, url):
+            assert method == "GET"
+            assert url == "http://public.example/start"
+            return _FakeStream()
+
+    monkeypatch.setattr(httpx, "Client", _FakeClient)
 
     with _pytest.raises(httpx.RequestError) as exc:
         content._get_public_url("http://public.example/start", headers={}, timeout=5)
@@ -1224,3 +1242,274 @@ def test_visual_report_escapes_request_category():
     # value must coerce rather than crash the render (html.escape needs a str).
     out = generate_visual_report(question="q", report_markdown="## H", category=12345)
     assert "category-12345" in out
+
+
+# ── DNS rebinding (audit finding 8.1) ────────────────────────────────
+# _resolve_public_ips resolves a URL's hostname once per hop and rejects
+# private / metadata targets, but httpx would then re-resolve the
+# hostname at connect time. The fix: the actual TCP connect is pinned
+# to the resolved IP via a custom httpcore.NetworkBackend, while the
+# URL / Host header / SNI stay on the original hostname.
+
+import ipaddress as _ipaddr
+import socket as _socket
+import threading as _threading
+
+import httpx as _httpx
+
+
+def test_dns_rebinding_blocked_by_resolve_gate(monkeypatch):
+    from src.search import content
+
+    monkeypatch.setattr(content, "_resolve_hostname_ips",
+                        lambda host: [_ipaddr.ip_address("10.0.0.5")])
+
+    with _pytest.raises(_httpx.RequestError) as exc:
+        content._resolve_public_ips("https://attacker.example/")
+    assert "non-public" in str(exc.value).lower()
+
+
+def test_dns_rebinding_pinned_backend_connects_to_resolved_ip(monkeypatch):
+    """``_PinnedBackend.connect_tcp`` must ignore the URL's host and
+    dial the pinned IP at the original port. This is the core of the
+    fix: httpcore's NetworkBackend contract lets us intercept the
+    connect before DNS lookup happens.
+    """
+    from src.search import content
+
+    pinned_ip = _ipaddr.ip_address("93.184.216.34")
+    captured = {}
+
+    class _StubStream:
+        def close(self):
+            pass
+
+    class _StubBackend:
+        def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None):
+            captured["host"] = host
+            captured["port"] = port
+            return _StubStream()
+
+        def connect_unix_socket(self, path, timeout=None, socket_options=None):
+            raise OSError("not used")
+
+        def sleep(self, seconds):
+            pass
+
+    backend = content._PinnedBackend(pinned_ip)
+    monkeypatch.setattr(backend, "_real", _StubBackend())
+
+    backend.connect_tcp("attacker.example", 443)
+
+    assert captured["host"] == "93.184.216.34", captured
+    assert captured["port"] == 443, captured
+
+
+def test_dns_rebinding_pinned_transport_dials_pinned_ip(monkeypatch):
+    """End-to-end: ``_PinnedTransport`` actually dials the pinned IP
+    when given a hostname, with the original URL's Host header
+    preserved. We stand up a local socket server on a free port and
+    make the transport connect there via the pinned backend.
+    """
+    from src.search import content
+    import httpcore
+
+    # Stand up a TCP server that accepts one connection and records
+    # the request bytes it received, then returns a minimal HTTP/1.1
+    # response.
+    captured = {"request": b""}
+    server_sock = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
+    server_sock.bind(("127.0.0.1", 0))
+    server_sock.listen(1)
+    port = server_sock.getsockname()[1]
+
+    def serve_once():
+        conn, _ = server_sock.accept()
+        with conn:
+            conn.settimeout(2.0)
+            buf = b""
+            try:
+                while b"\r\n\r\n" not in buf:
+                    chunk = conn.recv(4096)
+                    if not chunk:
+                        break
+                    buf += chunk
+            except _socket.timeout:
+                pass
+            captured["request"] = buf
+            conn.sendall(
+                b"HTTP/1.1 200 OK\r\n"
+                b"Content-Length: 2\r\n"
+                b"Connection: close\r\n"
+                b"\r\n"
+                b"OK"
+            )
+
+    t = _threading.Thread(target=serve_once, daemon=True)
+    t.start()
+
+    # Pin the transport to 127.0.0.1:<port>. The caller hands it a URL
+    # with a fake hostname so we can verify the host header is sent
+    # while the TCP connect goes to the pinned IP.
+    pinned_ip = _ipaddr.ip_address("127.0.0.1")
+    transport = content._PinnedTransport(pinned_ip)
+
+    req = _httpx.Request(
+        "GET",
+        f"http://attacker.test:{port}/path?q=1",
+        headers={"host": "attacker.test"},
+    )
+    try:
+        with _httpx.Client(transport=transport, timeout=5) as client:
+            response = client.send(req)
+        assert response.status_code == 200, response.text
+    finally:
+        server_sock.close()
+
+    t.join(timeout=2)
+
+    request_bytes = captured["request"]
+    assert request_bytes, "server never received a request"
+    # Host header is the original hostname, not the IP. (httpx
+    # lowercases header names; compare case-insensitively.)
+    headers_blob = request_bytes.lower()
+    assert b"host: attacker.test" in headers_blob, request_bytes
+    # The path was preserved.
+    assert b"/path?q=1" in request_bytes, request_bytes
+
+
+def test_dns_rebinding_pinned_transport_preserves_url_netloc(monkeypatch):
+    """The URL the transport hands to the underlying httpcore layer
+    must still be the original ``https://example.com/...`` — never
+    rewritten to the pinned IP. SNI / vhost depend on this.
+    """
+    from src.search import content
+
+    seen_url = {}
+
+    class _RecordingPool:
+        def handle_request(self, req):
+            seen_url["host"] = req.url.host.decode() if isinstance(req.url.host, bytes) else req.url.host
+            seen_url["scheme"] = req.url.scheme.decode() if isinstance(req.url.scheme, bytes) else req.url.scheme
+            seen_url["target"] = req.url.target.decode() if isinstance(req.url.target, bytes) else req.url.target
+            raise _httpx.ConnectError("intercepted")
+
+        def close(self):
+            pass
+
+    pinned_ip = _ipaddr.ip_address("93.184.216.34")
+    transport = content._PinnedTransport(pinned_ip)
+    transport._pool = _RecordingPool()
+
+    req = _httpx.Request("GET", "https://example.com/some/path?q=1")
+    with _pytest.raises(_httpx.ConnectError):
+        transport.handle_request(req)
+
+    assert seen_url["host"] == "example.com", seen_url
+    assert seen_url["scheme"] == "https", seen_url
+    assert seen_url["target"] == "/some/path?q=1", seen_url
+
+
+def test_dns_rebinding_redirect_re_resolves_per_hop(monkeypatch):
+    """Every redirect hop must call ``_resolve_public_ips`` again.
+    A redirect to a private-IP target must be blocked even when the
+    first hop was public.
+    """
+    from src.search import content
+
+    seen = []
+
+    def fake_resolve(url):
+        seen.append(url)
+        if "private" in url:
+            raise _httpx.RequestError(f"Blocked non-public URL: {url}")
+        return [_ipaddr.ip_address("93.184.216.34")]
+
+    monkeypatch.setattr(content, "_resolve_public_ips", fake_resolve)
+
+    class _Resp:
+        status_code = 302
+        headers = {"location": "http://private.example/secret"}
+        encoding = "utf-8"
+
+        def __init__(self, url):
+            self.url = url
+
+    class _FakeStream:
+        def __init__(self, response):
+            self.response = response
+
+        def __enter__(self):
+            return self.response
+
+        def __exit__(self, *args):
+            return False
+
+    class _FakeClient:
+        def __init__(self, *a, **k):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            return False
+
+        def stream(self, method, url):
+            assert method == "GET"
+            return _FakeStream(_Resp(url))
+
+    monkeypatch.setattr(_httpx, "Client", _FakeClient)
+
+    with _pytest.raises(_httpx.RequestError) as exc:
+        content._get_public_url("http://public.example/start", headers={}, timeout=5)
+    assert "non-public" in str(exc.value).lower()
+    # Both hops were validated.
+    assert seen == ["http://public.example/start", "http://private.example/secret"], seen
+
+
+def test_dns_rebinding_transport_uses_public_apis(monkeypatch):
+    """Static guard: ``_PinnedTransport`` must use only the public
+    ``httpx.BaseTransport`` / ``httpcore`` APIs. No subclassing of
+    ``httpx.HTTPTransport`` (whose ``_pool`` slot we'd have to
+    overwrite), no reads of private ``httpcore.ConnectionPool``
+    attributes, and no imports from ``httpx._transports``.
+    """
+    from src.search import content
+
+    import inspect
+
+    # 1) Subclass check: must be BaseTransport, not HTTPTransport.
+    mro_names = [c.__name__ for c in content._PinnedTransport.__mro__]
+    assert "BaseTransport" in mro_names, mro_names
+    assert "HTTPTransport" not in mro_names, (
+        "_PinnedTransport subclasses httpx.HTTPTransport. Subclass "
+        "httpx.BaseTransport instead and build the pool from scratch "
+        "with the public httpcore.ConnectionPool API."
+    )
+
+    # 2) No reads of private httpcore.ConnectionPool attrs.
+    src = inspect.getsource(content._PinnedTransport)
+    forbidden = (
+        "_ssl_context",
+        "_max_connections",
+        "_max_keepalive_connections",
+        "_keepalive_expiry",
+        "_http1",
+        "_http2",
+        "_network_backend",
+    )
+    leaked = [name for name in forbidden if name in src]
+    assert not leaked, (
+        f"_PinnedTransport reads private httpcore.ConnectionPool attrs: {leaked}. "
+        "Build the pool from the public httpcore.ConnectionPool API instead."
+    )
+
+    # 3) No imports from httpx's private transport module.
+    module_src = inspect.getsource(content)
+    forbidden_imports = ("from httpx._transports", "import httpx._transports")
+    leaked_imports = [s for s in forbidden_imports if s in module_src]
+    assert not leaked_imports, (
+        f"content.py imports from httpx's private transport module: {leaked_imports}. "
+        "Use only the public httpx and httpcore APIs."
+    )

tests/test_web_fetch_size_caps.py

@@ -13,6 +13,57 @@ import pytest
 from src.constants import WEB_FETCH_SOFT_MAX_BYTES, WEB_FETCH_HARD_MAX_BYTES
 from services.search import content as content_mod
 
+import pytest as _pytest_for_client_stream_compat
+
+
+@_pytest_for_client_stream_compat.fixture(autouse=True)
+def _client_stream_compat_for_pinned_fetch(monkeypatch):
+    """Adapt old size-cap tests to the current pinned Client.stream path.
+
+    These tests monkeypatch httpx.stream(...) to return fake responses. The
+    production fetcher now uses httpx.Client(...).stream(...) so it can pass a
+    pinned transport. When a test has replaced httpx.stream, route Client.stream
+    through that fake. When it has not, fall back to a real Client so unrelated
+    behavior in this file is not changed.
+    """
+    import httpx
+
+    real_client_cls = httpx.Client
+    original_stream = httpx.stream
+
+    class _ClientProxy:
+        def __init__(self, *args, **kwargs):
+            self._args = args
+            self._kwargs = kwargs
+            self._real_cm = None
+            self._real_client = None
+
+        def __enter__(self):
+            if httpx.stream is original_stream:
+                self._real_cm = real_client_cls(*self._args, **self._kwargs)
+                self._real_client = self._real_cm.__enter__()
+                return self._real_client
+            return self
+
+        def __exit__(self, *args):
+            if self._real_cm is not None:
+                return self._real_cm.__exit__(*args)
+            return False
+
+        def stream(self, method, url):
+            if self._real_client is not None:
+                return self._real_client.stream(method, url)
+
+            kwargs = {
+                "headers": self._kwargs.get("headers"),
+                "timeout": self._kwargs.get("timeout"),
+                "follow_redirects": self._kwargs.get("follow_redirects"),
+            }
+            return httpx.stream(method, url, **kwargs)
+
+    monkeypatch.setattr(httpx, "Client", _ClientProxy)
+
+
 
 class _FakeStream:
     """Stands in for the httpx.stream(...) context manager."""