diff --git a/requirements.txt b/requirements.txt index 493cb5206..be5f5d450 100644 --- a/requirements.txt +++ b/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 diff --git a/services/search/content.py b/services/search/content.py index 49d050a4f..05aa23753 100644 --- a/services/search/content.py +++ b/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,78 +285,78 @@ 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 - soft cap), so an oversized resource cannot be pulled into memory or the - content cache in full. When Content-Length already declares a body over - the hard ceiling, the fetch is refused before any body bytes are read. + Each hop is resolved once, validated as public, and then the actual TCP + connection is pinned to that resolved IP. The request URL is left unchanged + so Host and TLS SNI keep the original hostname. """ 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): - raise httpx.RequestError("Blocked private/internal URL", request=httpx.Request("GET", current)) + ips = _resolve_public_ips(current) + # Force identity transfer-encoding. With gzip/deflate the wire bytes - # (and Content-Length) can be a small fraction of the decoded body, so - # a tiny compressed response could pass the hard-cap preflight and then - # expand past the ceiling in a single decoded chunk before the streamed - # cap below can slice it. Identity makes Content-Length the true body - # size and keeps each streamed chunk bounded by the network read. + # and Content-Length can be a small fraction of the decoded body, so a + # tiny compressed response could pass the hard-cap preflight and then + # expand past the ceiling in one decoded chunk before the streamed cap + # below can slice it. 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: - if response.status_code in (301, 302, 303, 307, 308): - location = response.headers.get("location") - if not location: - return _CappedFetch(response.status_code, response.headers, b"", - False, None, response.encoding, str(response.url)) - current = urljoin(str(response.url), location) - continue - # A server can ignore the identity request and still return a - # compressed body; httpx.iter_bytes would then decode it, and a tiny - # gzip can balloon into one decoded chunk far past the cap before we - # slice. Refuse a compressed Content-Encoding so the streamed cap - # stays a real memory bound (Content-Length is the compressed wire - # 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( - f"Refusing compressed response (Content-Encoding: {enc}) after " - "requesting identity: cannot bound decoded body size", - request=httpx.Request("GET", current), - ) + 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: + return _CappedFetch(response.status_code, response.headers, b"", + False, None, response.encoding, str(response.url)) + current = urljoin(str(response.url), location) + continue - declared = None - 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) + # A server can ignore the identity request and still return a + # compressed body; httpx.iter_bytes would then decode it, and a + # tiny gzip can balloon into one decoded chunk far past the cap. + # Refuse compressed Content-Encoding so the streamed cap stays + # a real memory bound. + enc = (response.headers.get("content-encoding") or "").strip().lower() + if enc and enc != "identity": + raise httpx.RequestError( + f"Refusing compressed response (Content-Encoding: {enc}) after " + "requesting identity: cannot bound decoded body size", + request=httpx.Request("GET", current), + ) + + declared = None + raw_len = response.headers.get("content-length") + if raw_len and raw_len.isdigit(): + declared = int(raw_len) + + if declared is not None and declared > WEB_FETCH_HARD_MAX_BYTES: + raise BodyTooLargeError(current, declared) + + chunks = [] + read = 0 + truncated = False + for chunk in response.iter_bytes(): + read += len(chunk) + if read > cap: + keep = cap - (read - len(chunk)) + if keep > 0: + chunks.append(chunk[:keep]) + truncated = True + break + chunks.append(chunk) + + return _CappedFetch(response.status_code, response.headers, + b"".join(chunks), truncated, declared, + response.encoding, str(response.url)) - 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: - keep = cap - (read - len(chunk)) - if keep > 0: - chunks.append(chunk[:keep]) - 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) diff --git a/tests/test_security_regressions.py b/tests/test_security_regressions.py index b8f398017..f6a05383d 100644 --- a/tests/test_security_regressions.py +++ b/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 _fake_stream(method, url, **kwargs): - yield _Resp() + def __exit__(self, *args): + return False - 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:. 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." + ) diff --git a/tests/test_web_fetch_size_caps.py b/tests/test_web_fetch_size_caps.py index 19320c6c2..a3cfa64ed 100644 --- a/tests/test_web_fetch_size_caps.py +++ b/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."""