Add SSRF guard for outbound URL fetches

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:

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/<id>) 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")

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.",
+        )

aai_cli/core/webpage.py

@@ -13,20 +13,35 @@
 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
 # the Content-Type (e.g. application/octet-stream) or the URL has no .pdf suffix.
 _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:

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)