From 6c4af034e6c90d991f95617bf2ea72250c86a325 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 23 Jun 2026 13:48:19 +0000 Subject: [PATCH] Harden outbound URL fetches against SSRF (resolve + per-redirect-hop check) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `read_url` live-agent tool, `speak --url`, and the podcast-feed probe all fetched URLs through a guard that only string-matched the literal host (`risk._LOCAL_HOST`). That missed DNS-based bypasses (a public hostname that resolves to 127.0.0.1/169.254.169.254), alternate IP spellings (decimal/hex IPv4, IPv4-mapped IPv6), and — critically — redirects: the fetch followed 30x hops with no re-check, so a public URL could redirect to the cloud-metadata endpoint and the body came back to the model. Add `core/ssrf.py`: resolve the host via getaddrinfo and refuse any private/loopback/link-local/reserved/multicast IP via `ipaddress`, enforced on the initial URL and on every redirect hop. `core/webpage._fetch` and `app/transcribe/feed._fetch` now follow redirects manually and call the guard each hop. Also cap `webpage` response bodies at 10 MB so a hostile URL can't exhaust memory (the feed fetch already capped). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01AWsVSeWJjTXE6bsG4e1J3V --- aai_cli/AGENTS.md | 6 ++- aai_cli/app/transcribe/feed.py | 52 ++++++++++++------- aai_cli/core/ssrf.py | 91 ++++++++++++++++++++++++++++++++ aai_cli/core/webpage.py | 73 +++++++++++++++++++++----- tests/test_ssrf.py | 78 ++++++++++++++++++++++++++++ tests/test_transcribe_feed.py | 74 ++++++++++++++++++++++++-- tests/test_webpage.py | 95 ++++++++++++++++++++++++++++++++-- 7 files changed, 425 insertions(+), 44 deletions(-) create mode 100644 aai_cli/core/ssrf.py create mode 100644 tests/test_ssrf.py diff --git a/aai_cli/AGENTS.md b/aai_cli/AGENTS.md index 5a0e3d8d..34e5bd6d 100644 --- a/aai_cli/AGENTS.md +++ b/aai_cli/AGENTS.md @@ -34,8 +34,10 @@ between layers is enforced — higher may import lower, never the reverse: `config_builder`, `keyring_store`, `environments`, `env`, `errors`, `llm`, `telemetry`, `debuglog`, `remotefs`, `sync_stt`, `signals`, `ws`, `youtube`, `wer`, `argscan`, `jsonshape`, `timeparse`, `microphone`, `procs`, `stdio`, - `choices`. Contract 4 also forbids `rich` here, so - "no Rich below the UI layer" is structural. + `choices`, `ssrf` (the outbound-fetch SSRF guard: resolves a URL's host and + refuses private/loopback/link-local IPs, re-checked on every redirect hop — + shared by `webpage` and `app.transcribe.feed`). Contract 4 also forbids `rich` + here, so "no Rich below the UI layer" is structural. Three things sit *beside* the stack, intentionally unlisted in the layers contract: diff --git a/aai_cli/app/transcribe/feed.py b/aai_cli/app/transcribe/feed.py index eb5a7dd1..191e1ef8 100644 --- a/aai_cli/app/transcribe/feed.py +++ b/aai_cli/app/transcribe/feed.py @@ -20,11 +20,11 @@ from __future__ import annotations from pathlib import PurePosixPath -from urllib.parse import urlsplit +from urllib.parse import urljoin, urlsplit from pydantic import BaseModel, Field -from aai_cli.core import youtube +from aai_cli.core import ssrf, youtube # A feed lives at an extensionless URL (e.g. feeds.simplecast.com/) or a feed # document (.xml/.rss/.atom). Every other path — .mp3, .txt, .pdf — is never a feed, @@ -98,26 +98,38 @@ def _episode_urls(body: str) -> list[str] | None: def _fetch(url: str) -> str | None: """Up to ``_MAX_FEED_BYTES`` of `url` decoded as text, or ``None`` on any failure - or when the response is obviously binary media (audio/video/image).""" + or when the response is obviously binary media (audio/video/image). + + Redirects are followed manually so the SSRF guard runs on every hop; a + non-public host (directly or via redirect) reads as ``None`` — the URL falls + through to the single-source path, which the API fetches server-side. + """ import httpx2 as httpx chunks: list[bytes] = [] + current = url try: - with ( - httpx.Client(timeout=_FETCH_TIMEOUT_SECONDS, follow_redirects=True) as client, - client.stream("GET", url) as response, - ): - if not response.is_success: - return None - content_type = response.headers.get("content-type", "").lower() - if content_type.startswith(("audio/", "video/", "image/")): - return None - total = 0 - for chunk in response.iter_bytes(): - chunks.append(chunk) - total += len(chunk) - if total >= _MAX_FEED_BYTES: - break - except (httpx.HTTPError, OSError): + with httpx.Client(timeout=_FETCH_TIMEOUT_SECONDS, follow_redirects=False) as client: + for _ in range(ssrf.MAX_REDIRECTS + 1): # pragma: no mutate -- hop cap; +-1 equivalent + ssrf.assert_public_url(current) + with client.stream("GET", current) as response: + if response.status_code in ssrf.REDIRECT_STATUS: + location = response.headers.get("location") + if location: + current = urljoin(current, location) + continue + if not response.is_success: + return None + content_type = response.headers.get("content-type", "").lower() + if content_type.startswith(("audio/", "video/", "image/")): + return None + total = 0 + for chunk in response.iter_bytes(): + chunks.append(chunk) + total += len(chunk) + if total >= _MAX_FEED_BYTES: + break + return b"".join(chunks).decode("utf-8", "replace") + return None + except (httpx.HTTPError, OSError, ssrf.BlockedURLError): return None - return b"".join(chunks).decode("utf-8", "replace") diff --git a/aai_cli/core/ssrf.py b/aai_cli/core/ssrf.py new file mode 100644 index 00000000..e03619aa --- /dev/null +++ b/aai_cli/core/ssrf.py @@ -0,0 +1,91 @@ +"""SSRF guard for outbound URL fetches. + +The CLI fetches URLs that originate from untrusted sources — a page the live +agent's ``read_url`` tool is steered to by web content it just read, a +``speak --url`` argument, a podcast feed URL. A bare string check (``startswith +"127."``) is not enough: a hostname can *resolve* to a private address, a public +URL can *redirect* to one, and ``127.0.0.1`` has many literal spellings +(decimal ``2130706433``, hex, IPv4-mapped IPv6). So the guard resolves the host +and inspects the resulting IPs, and callers re-run it on **every redirect hop**. + +A residual gap is DNS rebinding (the name resolves to a public IP here, then to a +private one when httpx connects); closing it fully needs IP-pinned connections, +which is out of scope — this raises the bar from "trivially bypassable" to +"resolves and is checked", which is the meaningful win for a single-user CLI. + +Stdlib-only (``socket``/``ipaddress``) plus :mod:`aai_cli.core.errors`, so both +``core.webpage`` and ``app.transcribe.feed`` import it without pulling httpx onto +their startup path. +""" + +from __future__ import annotations + +import ipaddress +import socket +from urllib.parse import urlsplit + +from aai_cli.core.errors import UsageError + +# Cap how many redirect hops a caller follows before giving up — also the loop +# bound that stops a redirect cycle from spinning forever. +MAX_REDIRECTS = 5 # pragma: no mutate -- tuning knob; a +-1 shift is behaviorally equivalent +# The HTTP status codes that carry a Location to follow (callers re-validate each). +REDIRECT_STATUS = frozenset({301, 302, 303, 307, 308}) # pragma: no mutate -- standard codes + + +class BlockedURLError(UsageError): + """A URL was refused because it isn't an http(s) address that resolves to a + public host (the SSRF guard). A ``UsageError`` so it renders as a clean exit-2 + message and callers that already catch ``UsageError`` handle it uniformly.""" + + +def _resolve_host(host: str) -> list[str]: + """The IP strings ``host`` resolves to (split out so tests can stub DNS). + + Uses ``getaddrinfo``, which resolves the same way httpx's connection will — + including libc's acceptance of non-dotted-decimal IPv4 literals — so the IPs + we inspect are the IPs that would actually be connected to. Each sockaddr's + address is normalized through ``ip_address`` to a canonical string. + """ + infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP) + return [ipaddress.ip_address(info[4][0]).compressed for info in infos] + + +def _is_blocked_ip(ip: str) -> bool: + """True when ``ip`` is a non-public address an outbound fetch must never reach. + + ``is_private`` already covers loopback, link-local (incl. the + ``169.254.169.254`` cloud-metadata address), unique-local, reserved, and the + RFC 1918 ranges for both families; multicast is the one internal class it + doesn't, so it is checked explicitly. + """ + addr = ipaddress.ip_address(ip) + # Normalize an IPv4-mapped IPv6 address (``::ffff:a.b.c.d``) to its v4 form so the + # v4 rules apply regardless of how the Python version classifies the mapped form. + mapped = addr.ipv4_mapped if isinstance(addr, ipaddress.IPv6Address) else None + if mapped is not None: # pragma: no mutate -- cross-version v4-mapped normalization + addr = mapped # pragma: no mutate + return addr.is_private or addr.is_multicast + + +def assert_public_url(url: str) -> None: + """Raise :class:`BlockedURLError` unless ``url`` is an http(s) URL whose host + resolves only to public addresses. + + Call this for the initial URL *and* for every redirect target, since a public + URL can 30x-redirect to an internal one. + """ + parts = urlsplit(url) + if parts.scheme not in ("http", "https"): + raise BlockedURLError( + f"Refused to fetch a non-web URL: {url}", + suggestion="Only http(s) URLs are fetched.", + ) + host = parts.hostname + if not host: + raise BlockedURLError(f"Refused to fetch a URL with no host: {url}") + if any(_is_blocked_ip(ip) for ip in _resolve_host(host)): + raise BlockedURLError( + f"Refused to fetch a private or internal address: {url}", + suggestion="The URL resolves to a loopback, private, link-local, or internal address.", + ) diff --git a/aai_cli/core/webpage.py b/aai_cli/core/webpage.py index f711ff70..23ddda33 100644 --- a/aai_cli/core/webpage.py +++ b/aai_cli/core/webpage.py @@ -13,13 +13,18 @@ from __future__ import annotations from dataclasses import dataclass +from urllib.parse import urljoin import httpx2 as httpx +from aai_cli.core import ssrf from aai_cli.core.errors import APIError, UsageError # A page fetch shouldn't hang a TTS run; cap it. _TIMEOUT = 30.0 # pragma: no mutate -- request timeout; nothing observable to assert +# Bound the body so a hostile or huge URL can't exhaust memory; 10 MB is far past any +# article or PDF we'd narrate, and the cap is what makes the fetch memory-safe. +_MAX_BYTES = 10 * 1024 * 1024 # pragma: no mutate -- tuning knob, not behavior # Browser-like UA: some sites serve a stub or block page to unknown clients. _USER_AGENT = "Mozilla/5.0 (compatible; assembly-cli; +https://www.assemblyai.com)" # Every PDF begins with this signature; the robust signal when a server mislabels @@ -27,6 +32,16 @@ _PDF_MAGIC = b"%PDF-" +@dataclass(frozen=True) +class _Page: + """A fetched resource: size-capped raw bytes (for the PDF path), its + content-type, and the bytes decoded with the response's own charset (for HTML).""" + + content: bytes + content_type: str + text: str + + @dataclass(frozen=True) class Article: """The readable content extracted from a web page or PDF.""" @@ -49,40 +64,70 @@ def fetch_article(url: str) -> Article: f"Not a web page URL: {url}", suggestion="Pass an http(s) URL, e.g. assembly speak --url https://example.com/post.", ) - response = _fetch(url) - content_type = response.headers.get("content-type", "").lower() - data = response.content - if _is_pdf(data, content_type): - text, title = _extract_pdf(data) + page = _fetch(url) + if _is_pdf(page.content, page.content_type): + text, title = _extract_pdf(page.content) empty_hint = ( "The PDF may be scanned or image-only — there's no text layer to read " "(that needs OCR, which speak doesn't do)." ) else: - text, title = _extract(response.text) + text, title = _extract(page.text) empty_hint = "The page may be paywalled, JavaScript-rendered, or not an article." if not text: raise UsageError(f"Couldn't find readable text at {url}.", suggestion=empty_hint) return Article(text=text, title=title, url=url) -def _fetch(url: str) -> httpx.Response: - """GET ``url``, mapping any network/HTTP failure to APIError. +def _fetch(url: str) -> _Page: + """GET ``url`` as a size-capped, SSRF-checked :class:`_Page`. - Returns the fully-read response so the caller can read it as text (HTML) or - bytes (PDF) depending on the content type. + Redirects are followed manually so the SSRF guard runs on **every hop** (a + public URL can redirect to an internal one). A network/HTTP failure maps to + APIError; a non-public host raises :class:`ssrf.BlockedURLError`. """ + current = url try: with httpx.Client( timeout=_TIMEOUT, - follow_redirects=True, + follow_redirects=False, headers={"User-Agent": _USER_AGENT}, ) as client: - response = client.get(url) - response.raise_for_status() - return response + for _ in range(ssrf.MAX_REDIRECTS + 1): # pragma: no mutate -- hop cap; +-1 equivalent + ssrf.assert_public_url(current) + with client.stream("GET", current) as response: + if response.status_code in ssrf.REDIRECT_STATUS: + location = response.headers.get("location") + if location: + current = urljoin(current, location) + continue + response.raise_for_status() + return _read_page(response) + raise APIError(f"Too many redirects fetching {url}.") except httpx.HTTPError as exc: raise APIError(f"Couldn't fetch {url}: {exc}") from exc + except OSError as exc: + raise APIError(f"Couldn't resolve {url}: {exc}") from exc + + +def _read_page(response: httpx.Response) -> _Page: + """Read at most ``_MAX_BYTES`` of ``response`` so a huge body can't exhaust memory.""" + chunks: list[bytes] = [] + # The running total + early break bound *memory* (stop reading a huge stream); the + # output is bounded by the slice below, which is what the test asserts — so the + # counter itself is not output-observable and can't be killed by a mutation test. + total = 0 # pragma: no mutate -- memory-bound counter; output is sliced below + for chunk in response.iter_bytes(): + chunks.append(chunk) + total += len(chunk) # pragma: no mutate -- memory-bound counter; output is sliced below + if total >= _MAX_BYTES: # pragma: no mutate -- memory bound; can't be asserted by OOM + break + content = b"".join(chunks)[:_MAX_BYTES] + content_type = response.headers.get("content-type", "").lower() + # Decode with the charset httpx parsed from the Content-Type header (the same + # source response.text would use), falling back to UTF-8 for an unlabeled page. + text = content.decode(response.charset_encoding or "utf-8", errors="replace") + return _Page(content=content, content_type=content_type, text=text) def _is_pdf(data: bytes, content_type: str) -> bool: diff --git a/tests/test_ssrf.py b/tests/test_ssrf.py new file mode 100644 index 00000000..3e49f124 --- /dev/null +++ b/tests/test_ssrf.py @@ -0,0 +1,78 @@ +"""The SSRF guard (`aai_cli/core/ssrf.py`): IP classification and URL validation. + +DNS is stubbed (`_resolve_host`) so these stay hermetic under the socket-blocked +suite while still exercising the real `ipaddress`-based classification. +""" + +from __future__ import annotations + +import pytest + +from aai_cli.core import ssrf +from aai_cli.core.errors import UsageError + + +@pytest.mark.parametrize( + "ip", + [ + "127.0.0.1", # loopback + "10.0.0.1", # RFC 1918 private + "192.168.1.1", # RFC 1918 private + "172.16.0.1", # RFC 1918 private + "169.254.169.254", # link-local — the cloud-metadata address + "0.0.0.1", # the unspecified 0.0.0.0/8 range + "224.0.0.1", # multicast + "::1", # IPv6 loopback + "fd00::1", # IPv6 unique-local + "fe80::1", # IPv6 link-local + "::ffff:169.254.169.254", # IPv4-mapped IPv6 of the metadata address + "::ffff:127.0.0.1", # IPv4-mapped IPv6 loopback + ], +) +def test_internal_ips_are_blocked(ip): + assert ssrf._is_blocked_ip(ip) is True + + +@pytest.mark.parametrize("ip", ["93.184.216.34", "8.8.8.8", "2606:4700:4700::1111"]) +def test_public_ips_are_allowed(ip): + assert ssrf._is_blocked_ip(ip) is False + + +@pytest.mark.allow_hosts(["127.0.0.1"]) +def test_resolve_host_returns_canonical_ips(): + # A numeric literal resolves locally (no DNS), exercising the real getaddrinfo + # path the rest of the suite stubs out. + assert ssrf._resolve_host("127.0.0.1") == ["127.0.0.1"] + + +def test_assert_public_url_allows_public_host(monkeypatch): + monkeypatch.setattr(ssrf, "_resolve_host", lambda host: ["93.184.216.34"]) + # Returns without raising for a host that resolves to a public address. + ssrf.assert_public_url("https://example.com/page") + + +def test_assert_public_url_blocks_host_resolving_internal(monkeypatch): + # A perfectly ordinary-looking hostname that resolves to an internal IP is the + # DNS-based bypass the string regex missed. + monkeypatch.setattr(ssrf, "_resolve_host", lambda host: ["169.254.169.254"]) + with pytest.raises(ssrf.BlockedURLError): + ssrf.assert_public_url("http://metadata.example.com/") + + +def test_assert_public_url_rejects_non_http_scheme(monkeypatch): + monkeypatch.setattr(ssrf, "_resolve_host", lambda host: ["93.184.216.34"]) + with pytest.raises(ssrf.BlockedURLError): + ssrf.assert_public_url("file:///etc/passwd") + + +def test_assert_public_url_rejects_missing_host(monkeypatch): + # Even if resolution would pass, a URL with no host is refused before resolving. + monkeypatch.setattr(ssrf, "_resolve_host", lambda host: ["93.184.216.34"]) + with pytest.raises(ssrf.BlockedURLError): + ssrf.assert_public_url("http:///just-a-path") + + +def test_blocked_url_error_is_a_usage_error(): + # So it renders as a clean exit-2 message and existing `except UsageError` + # handlers (e.g. the read_url tool) catch it uniformly. + assert issubclass(ssrf.BlockedURLError, UsageError) diff --git a/tests/test_transcribe_feed.py b/tests/test_transcribe_feed.py index 0dd2e40f..7ce77822 100644 --- a/tests/test_transcribe_feed.py +++ b/tests/test_transcribe_feed.py @@ -14,11 +14,19 @@ from aai_cli.app.transcribe import feed from aai_cli.app.transcribe import sources as transcribe_sources -from aai_cli.core import config +from aai_cli.core import config, ssrf from aai_cli.main import app runner = CliRunner() + +@pytest.fixture(autouse=True) +def _public_dns(monkeypatch): + # The SSRF guard resolves the feed host; stub DNS to a public IP so the + # socket-blocked suite stays hermetic. Guard-specific tests override this. + monkeypatch.setattr(ssrf, "_resolve_host", lambda host: ["93.184.216.34"]) + + _TRANSCRIBE = "aai_cli.app.transcribe.run.client.transcribe" # A minimal but realistic RSS 2.0 podcast feed: two episodes, newest first, with @@ -143,10 +151,14 @@ def test_feed_episode_urls_returns_none_for_non_feed_body(monkeypatch): class _FakeStream: - def __init__(self, *, status=200, content_type="application/rss+xml", chunks=(b"",)): + def __init__( + self, *, status=200, content_type="application/rss+xml", chunks=(b"",), location=None + ): self.status_code = status self.is_success = 200 <= status < 300 # mirror httpx.Response.is_success self.headers = {"content-type": content_type} + if location is not None: + self.headers["location"] = location self._chunks = chunks def __enter__(self): @@ -184,11 +196,40 @@ def factory(**kwargs): return captured +class _SequenceClient: + """A client whose successive `stream` calls return the given streams in order, + so a redirect hop and its destination can be distinguished.""" + + def __init__(self, streams): + self._streams = list(streams) + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def stream(self, method, url): + return self._streams.pop(0) + + +def test_fetch_follows_redirect_to_final_body(monkeypatch): + # A 301 to a public CDN is followed to the real feed body — proves the status is + # recognized as a redirect (not parsed as the body itself). + streams = [ + _FakeStream(status=301, location="https://cdn.example.com/final"), + _FakeStream(chunks=(b"real",)), + ] + monkeypatch.setattr(httpx, "Client", lambda **kwargs: _SequenceClient(streams)) + assert feed._fetch("https://feeds.example.com/show") == "real" + + def test_fetch_returns_decoded_body(monkeypatch): captured = _patch_client(monkeypatch, _FakeStream(chunks=(b"", b""))) assert feed._fetch("https://feeds.example.com/show") == "" - # Feeds commonly 301/302 to a CDN, so redirects must be followed. - assert captured["follow_redirects"] is True + # Redirects are followed manually (per-hop SSRF check), so the client itself + # must not auto-follow. + assert captured["follow_redirects"] is False assert captured["timeout"] == feed._FETCH_TIMEOUT_SECONDS @@ -224,6 +265,31 @@ def _raise(**kwargs): assert feed._fetch("https://feeds.example.com/show") is None +def test_fetch_returns_none_for_internal_host(monkeypatch): + # A feed URL whose host resolves to an internal address is refused (None) so the + # local fetch never reaches it — it falls through to the API's server-side fetch. + monkeypatch.setattr(ssrf, "_resolve_host", lambda host: ["169.254.169.254"]) + _patch_client(monkeypatch, _FakeStream()) + assert feed._fetch("https://feeds.internal.example/show") is None + + +def test_fetch_returns_none_on_redirect_to_internal_host(monkeypatch): + # A public feed URL that redirects to an internal address is caught on the hop. + monkeypatch.setattr( + ssrf, + "_resolve_host", + lambda host: ["169.254.169.254"] if "internal" in host else ["93.184.216.34"], + ) + _patch_client(monkeypatch, _FakeStream(status=301, location="http://internal.host/meta")) + assert feed._fetch("https://feeds.example.com/show") is None + + +def test_fetch_returns_none_on_redirect_loop(monkeypatch): + # A redirect that never resolves is bounded by the hop cap and gives up (None). + _patch_client(monkeypatch, _FakeStream(status=301, location="https://feeds.example.com/again")) + assert feed._fetch("https://feeds.example.com/show") is None + + # --- expand_sources seam ------------------------------------------------------ diff --git a/tests/test_webpage.py b/tests/test_webpage.py index 2c18c46c..52345a9e 100644 --- a/tests/test_webpage.py +++ b/tests/test_webpage.py @@ -5,9 +5,19 @@ import httpx2 as httpx import pytest -from aai_cli.core import webpage +from aai_cli.core import ssrf, webpage from aai_cli.core.errors import APIError, UsageError +# A public IP every host resolves to by default, so the SSRF guard passes without +# real DNS (the suite is socket-blocked). Tests exercising the guard override this. +_PUBLIC_IP = "93.184.216.34" + + +@pytest.fixture(autouse=True) +def _public_dns(monkeypatch): + monkeypatch.setattr(ssrf, "_resolve_host", lambda host: [_PUBLIC_IP]) + + # An article wrapped in the usual page chrome: nav, a comment thread, and a # . The extractor should keep the body and drop the rest. ARTICLE_HTML = """<!DOCTYPE html><html><head><title>The Real Headline @@ -53,13 +63,13 @@ def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, text="ok") _client_returning(monkeypatch, handler) - assert webpage._fetch("https://example.com/post").text == "ok" + assert webpage._fetch("https://example.com/post").content == b"ok" # The browser-like UA is sent so sites don't serve a stub/block page. assert "assembly-cli" in seen["ua"] def test_fetch_follows_redirects(monkeypatch): - # A 301 must be followed to the final 200; without follow_redirects the + # A 301 must be followed to the final 200; without following redirects the # client would return the empty 301 body instead of the article. def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/start": @@ -67,7 +77,84 @@ def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, text="final body") _client_returning(monkeypatch, handler) - assert webpage._fetch("https://example.com/start").text == "final body" + assert webpage._fetch("https://example.com/start").content == b"final body" + + +def test_fetch_blocks_internal_host(monkeypatch): + # A host that resolves to a link-local (cloud-metadata) address is refused + # outright — the SSRF guard, before any request is sent. + monkeypatch.setattr(ssrf, "_resolve_host", lambda host: ["169.254.169.254"]) + _client_returning(monkeypatch, lambda request: httpx.Response(200, text="secret")) + with pytest.raises(ssrf.BlockedURLError): + webpage._fetch("http://metadata.internal/latest/meta-data/") + + +def test_fetch_blocks_redirect_to_internal_host(monkeypatch): + # The headline bypass: a public URL that 30x-redirects to an internal address + # must be caught on the redirect hop, not just the input URL. + monkeypatch.setattr( + ssrf, + "_resolve_host", + lambda host: ["169.254.169.254"] if "internal" in host else [_PUBLIC_IP], + ) + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.host == "example.com": + return httpx.Response(301, headers={"Location": "http://internal.host/meta"}) + return httpx.Response(200, text="cloud credentials") + + _client_returning(monkeypatch, handler) + with pytest.raises(ssrf.BlockedURLError): + webpage._fetch("https://example.com/start") + + +def test_fetch_caps_oversize_body(monkeypatch): + # A body larger than the cap is truncated so a hostile URL can't exhaust memory. + monkeypatch.setattr(webpage, "_MAX_BYTES", 10) + _client_returning( + monkeypatch, + lambda request: httpx.Response( + 200, content=b"x" * 100, headers={"content-type": "text/html"} + ), + ) + assert webpage._fetch("https://example.com/big").content == b"x" * 10 + + +def test_fetch_decodes_with_the_declared_charset(monkeypatch): + # A non-UTF-8 page is decoded with its declared charset, not assumed UTF-8. + _client_returning( + monkeypatch, + lambda request: httpx.Response( + 200, + content="café".encode("latin-1"), + headers={"content-type": "text/html; charset=iso-8859-1"}, + ), + ) + page = webpage._fetch("https://example.com/latin1") + # Decoded as UTF-8 the é byte would turn into a replacement character instead. + assert page.text == "café" + + +def test_fetch_too_many_redirects_is_api_error(monkeypatch): + # A redirect loop is bounded; exceeding the hop cap is an APIError, not a hang. + _client_returning( + monkeypatch, + lambda request: httpx.Response(302, headers={"Location": "https://example.com/again"}), + ) + with pytest.raises(APIError) as exc: + webpage._fetch("https://example.com/loop") + assert "redirect" in exc.value.message.lower() + + +def test_fetch_dns_failure_is_api_error(monkeypatch): + # A name that doesn't resolve surfaces as an APIError, not an uncaught OSError. + def _boom(host): + raise OSError("name or service not known") + + monkeypatch.setattr(ssrf, "_resolve_host", _boom) + with pytest.raises(APIError) as exc: + webpage._fetch("https://does-not-resolve.example/") + assert "does-not-resolve.example" in exc.value.message def test_fetch_non_2xx_becomes_api_error(monkeypatch):