forked from bchanot/claude
feat(seo-data): safe_fetch — resolve-then-pin, close DNS-rebinding + SSRF
By-principle hardening. H1's url-guard validates the NAME; urlopen then resolved AND connected — two DNS lookups with a window a hostile authority uses to answer PUBLIC to validation and PRIVATE (169.254.169.254 metadata, 127.0.0.1, the LAN) to the connect. A name-level guard cannot see that rebind. safe_fetch collapses the two lookups into one: resolve ONCE, validate every IP (ipaddress, dual-stack v4+v6), refuse if ANY is non-public (the multi-A vector), connect to the exact validated IP with Host+SNI+cert for the real host — no second resolution to poison. Redirects re-validate each hop (urlopen followed them blind). One seam: sitemap._fetch, which linkgraph/render_check/drift all call, so every network verb inherits it. The load-bearing property (confirmed by the security review): classification is on the OS-resolved address (sockaddr[0]), never the URL text — so octal/hex/ decimal literals, IPv4-mapped IPv6, NAT64, 6to4 are all defeated structurally, not by enumeration. Better than the source idea (claude-seo url_safety.py, MIT): dual-stack (theirs IPv4-only), no global monkeypatch so thread-safe by construction (theirs locks a patched getaddrinfo), stdlib-only (no requests). Proven end-to-end before writing: pinned connect keeps SNI+cert for the real host. NOT covered, stated not silent: shell `curl` in the agent specs (separate process, unpinnable here). Smaller surface; `curl --resolve` is a separate change. REVIEW-SURFACED (fresh security-auditor, adversarial, VERDICT PASS) — two real holes it found while attacking the diff, both fixed here: - billion-laughs REOPENED in C1b: _refuse_dtd scanned only raw[:4096], so a >4KB leading comment pushed <!DOCTYPE past the window while ET parsed AND EXPANDED the entities. Proven (&lol2; → "lollollollollol"), now a full-doc case-insensitive scan. This is a genuine fix to already-merged C1b, not this feature — fixed here rather than filed, per root-cause discipline. - 192.88.99.0/24 (6to4-relay anycast) passed is_global as public — added to an extra special-use deny list. Verified: rebind-to-metadata refused BEFORE any connect (injected resolver), multi-A public+private refused, classifier fuzzed dual-stack incl. CGNAT/6to4, non-http scheme refused, both review fixes proven with no false positive; real fetch still works (zenquality 86 loc, lavageangels 24) through the pinned path; all 4 verbs work end-to-end via fetch.sh; seo-data 210 → 221 pass, 0 fail; full suite green; shellcheck + py_compile clean.
This commit is contained in:
Bastien Chanot
@@ -0,0 +1,164 @@
|
||||
#!/usr/bin/env python3
|
||||
"""SSRF- and DNS-rebinding-safe HTTP(S) fetch. Stdlib only.
|
||||
|
||||
The verbs that fetch remote content (sitemap, linkgraph, render_check, drift)
|
||||
all route through sitemap._fetch, which used urllib.request.urlopen. urlopen
|
||||
resolves the host, then connects — two DNS lookups with a window between them.
|
||||
A hostile authority can answer PUBLIC to the validation lookup and a PRIVATE
|
||||
address (169.254.169.254 cloud metadata, 127.0.0.1, the LAN) to the connect
|
||||
lookup. That is DNS rebinding, and a name-level guard cannot see it.
|
||||
|
||||
This collapses the two lookups into one: resolve ONCE, validate every returned
|
||||
IP, then connect to the exact validated IP while preserving the Host header,
|
||||
TLS SNI, and certificate validation for the real hostname. There is no second
|
||||
resolution to poison.
|
||||
|
||||
Better than the reference implementation this idea came from (claude-seo
|
||||
url_safety.py, MIT) on three axes, all verified before writing:
|
||||
- dual-stack: validates IPv4 AND IPv6 (theirs is IPv4-only);
|
||||
- no global state: each connection pins its own socket, so it is thread-safe
|
||||
by construction (theirs monkeypatches socket.getaddrinfo behind a global
|
||||
lock);
|
||||
- stdlib only: http.client + ssl + ipaddress, no `requests`.
|
||||
|
||||
NOT covered, stated rather than left silent: the shell `curl` calls in the
|
||||
agent specs (seo-analyzer/geo-analyzer STEP 4, the sameAs loop) run in a
|
||||
separate process and cannot be pinned from here. Their surface is smaller
|
||||
(a fixed set against an operator-typed/confirmed $DOMAIN). Closing them needs
|
||||
`curl --resolve` and is a separate change.
|
||||
"""
|
||||
import gzip
|
||||
import http.client
|
||||
import ipaddress
|
||||
import socket
|
||||
import ssl
|
||||
from urllib.parse import urljoin, urlparse
|
||||
|
||||
DEFAULT_TIMEOUT = 20
|
||||
DEFAULT_MAX_BYTES = 20 * 1024 * 1024
|
||||
MAX_REDIRECTS = 5
|
||||
|
||||
|
||||
class UnsafeTarget(Exception):
|
||||
"""A URL resolved to a non-public address, or a redirect did. Raised BEFORE
|
||||
any connection to that address. Callers already wrap _fetch in try/except
|
||||
and degrade, so the fail-open contract is preserved."""
|
||||
|
||||
|
||||
# Special-use ranges that `is_global` reports as public but are not legitimate
|
||||
# fetch targets. 192.88.99.0/24 = RFC 3068 6to4-relay anycast (a security
|
||||
# review flagged it 2026-07-17). Grows if more surface.
|
||||
_EXTRA_DENY = (ipaddress.ip_network("192.88.99.0/24"),)
|
||||
|
||||
|
||||
def _ip_is_public(ip_str):
|
||||
"""A globally routable unicast address, dual-stack. `is_global` is the
|
||||
decisive gate — it alone rejects CGNAT (100.64/10) that the per-flag checks
|
||||
miss — with the explicit flags plus an extra special-use deny list as
|
||||
defence in depth."""
|
||||
ip = ipaddress.ip_address(ip_str)
|
||||
if not ip.is_global:
|
||||
return False
|
||||
if any(ip in net for net in _EXTRA_DENY):
|
||||
return False
|
||||
return not (ip.is_private or ip.is_loopback or ip.is_link_local
|
||||
or ip.is_reserved or ip.is_multicast or ip.is_unspecified)
|
||||
|
||||
|
||||
def _resolve_pinned(host, port, resolver=socket.getaddrinfo):
|
||||
"""Resolve host ONCE and return [(family, ip)] for connecting. Refuse if
|
||||
ANY resolved address is non-public — a name advertising both public and
|
||||
private A records is exactly the multi-answer rebinding vector, and a
|
||||
legitimate public site does not do it. `resolver` is injected in tests to
|
||||
plant a private address and prove the refusal."""
|
||||
try:
|
||||
infos = resolver(host, port, type=socket.SOCK_STREAM)
|
||||
except socket.gaierror as e:
|
||||
raise UnsafeTarget("cannot resolve %r: %s" % (host, e))
|
||||
pinned = []
|
||||
for family, _type, _proto, _canon, sockaddr in infos:
|
||||
ip = sockaddr[0]
|
||||
if not _ip_is_public(ip):
|
||||
raise UnsafeTarget("%s resolves to non-public %s" % (host, ip))
|
||||
pinned.append((family, ip))
|
||||
if not pinned:
|
||||
raise UnsafeTarget("%s resolved to nothing" % host)
|
||||
return pinned
|
||||
|
||||
|
||||
class _PinnedHTTPSConnection(http.client.HTTPSConnection):
|
||||
"""HTTPS to a pinned IP, with SNI + cert validation for the real host."""
|
||||
def __init__(self, host, pinned_ip, family, **kw):
|
||||
super().__init__(host, **kw) # host → Host header + SNI
|
||||
self._pinned_ip = pinned_ip
|
||||
self._family = family
|
||||
|
||||
def connect(self):
|
||||
sock = socket.create_connection((self._pinned_ip, self.port),
|
||||
timeout=self.timeout)
|
||||
# server_hostname = the real host → SNI + hostname check both use it,
|
||||
# never the IP.
|
||||
self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
|
||||
|
||||
|
||||
class _PinnedHTTPConnection(http.client.HTTPConnection):
|
||||
"""Plain HTTP to a pinned IP (Host header stays the real host)."""
|
||||
def __init__(self, host, pinned_ip, family, **kw):
|
||||
super().__init__(host, **kw)
|
||||
self._pinned_ip = pinned_ip
|
||||
self._family = family
|
||||
|
||||
def connect(self):
|
||||
self.sock = socket.create_connection((self._pinned_ip, self.port),
|
||||
timeout=self.timeout)
|
||||
|
||||
|
||||
def _one_request(url, timeout, max_bytes, resolver):
|
||||
"""One hop: resolve+pin the host, connect, return (status, headers, body)."""
|
||||
p = urlparse(url)
|
||||
if p.scheme not in ("http", "https"):
|
||||
raise UnsafeTarget("scheme must be http/https: %r" % url)
|
||||
host = p.hostname
|
||||
if not host:
|
||||
raise UnsafeTarget("no host in %r" % url)
|
||||
port = p.port or (443 if p.scheme == "https" else 80)
|
||||
family, ip = _resolve_pinned(host, port, resolver)[0] # any is public here
|
||||
ctx = ssl.create_default_context() if p.scheme == "https" else None
|
||||
if p.scheme == "https":
|
||||
conn = _PinnedHTTPSConnection(host, ip, family, port=port,
|
||||
timeout=timeout, context=ctx)
|
||||
else:
|
||||
conn = _PinnedHTTPConnection(host, ip, family, port=port,
|
||||
timeout=timeout)
|
||||
try:
|
||||
path = p.path or "/"
|
||||
if p.query:
|
||||
path += "?" + p.query
|
||||
# No Accept-Encoding: keep HTTP bodies un-gzipped; the .xml.gz
|
||||
# content-level case is handled by the caller's magic-byte check.
|
||||
conn.request("GET", path, headers={"Host": host,
|
||||
"User-Agent": "claude-seo-data/1.0"})
|
||||
r = conn.getresponse()
|
||||
body = r.read(max_bytes)
|
||||
return r.status, {k.lower(): v for k, v in r.getheaders()}, body
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def safe_fetch(url, timeout=DEFAULT_TIMEOUT, max_bytes=DEFAULT_MAX_BYTES,
|
||||
max_redirects=MAX_REDIRECTS, resolver=socket.getaddrinfo):
|
||||
"""Fetch url with resolve-then-pin, following redirects and RE-VALIDATING
|
||||
each hop — urlopen followed redirects to whatever address the Location
|
||||
named, re-opening the rebinding window on every hop. Returns the raw body
|
||||
bytes (the caller handles content-level gzip)."""
|
||||
seen = 0
|
||||
current = url
|
||||
while True:
|
||||
status, headers, body = _one_request(current, timeout, max_bytes, resolver)
|
||||
if status in (301, 302, 303, 307, 308) and "location" in headers:
|
||||
seen += 1
|
||||
if seen > max_redirects:
|
||||
raise UnsafeTarget("too many redirects from %r" % url)
|
||||
current = urljoin(current, headers["location"]) # re-validated next loop
|
||||
continue
|
||||
return body
|
||||
Reference in New Issue
Block a user