Files
Bastien Chanot b00e8ef442 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.
2026-07-17 19:58:59 +02:00

192 lines
7.7 KiB
Python

#!/usr/bin/env python3
"""Sitemap discovery -> normalized JSON. Stdlib only: no venv, no requests, no
auth. Gives STEP 9 COVERAGE the denominator it was told to report and never
had, and STEP 5 a real sampling frame instead of "5-15 key pages" chosen by
eye.
Deliberately NOT a security boundary. urllib fetches these URLs, so nothing
here reaches a shell and there is no injection surface to guard. The consumer
is different: seo-analyzer interpolates URLs into curl, so IT must run
lib/url-guard.sh at the point of use (same pattern as the sameAs check).
Duplicating the guard here would just add a second copy to drift. `_sane`
below is a cheap garbage filter, not that guard.
"""
import argparse, gzip, json, os
from urllib.parse import urlparse
MAX_URLS = 50000 # sitemaps.org caps one file at 50k
MAX_CHILDREN = 50 # sitemapindex fan-out cap: bound the work, report the cut
TIMEOUT = 20
def _mock(name):
d = os.environ.get("SEO_DATA_MOCK_DIR")
if not d:
return None
path = os.path.join(d, name)
if not os.path.exists(path):
return None
with open(path, "rb") as f:
return f.read()
def _fetch(url):
# SSRF + DNS-rebinding safe: resolve-then-pin, redirects re-validated.
# This is the single seam for ALL network egress — linkgraph/render_check/
# drift all call sitemap._fetch — so pinning here covers every verb.
import safe_fetch # sibling, lazy
raw = safe_fetch.safe_fetch(url, timeout=TIMEOUT, max_bytes=20 * 1024 * 1024)
if raw[:2] == b"\x1f\x8b": # sitemap.xml.gz is common
raw = gzip.decompress(raw)
return raw
class UnsafeXML(Exception):
"""A DTD reached the parser. Refused before parsing, not mitigated after."""
def _refuse_dtd(raw):
"""A sitemap NEVER has a DTD: sitemaps.org is <?xml?> then <urlset xmlns=>.
So refuse any doctype/entity outright, at the door.
This is the reason we do not pull in defusedxml. The stdlib parser is not
the problem for XXE — xml.etree.ElementTree does not expand external
entities, it raises on them — but it IS vulnerable to billion-laughs, where
a 1 KB document expands to gigabytes in RAM. The 20 MB read ceiling bounds
the input, not the expansion. Rejecting the construct beats depending on
the parser's internals, and keeps this module stdlib-only: no venv, same as
google_seo.py's mock/degrade paths. A sitemap with a DTD is not a sitemap
we want anyway.
"""
# Scan the WHOLE document, not a prefix. A security review (2026-07-17)
# showed a >4 KB leading comment pushed <!DOCTYPE past the old raw[:4096]
# window while ET.fromstring still parsed and EXPANDED the entities —
# billion-laughs reopened. A legitimate sitemap contains neither construct
# anywhere, so a full case-insensitive scan is correct; over ≤20 MB it is a
# single re.search, microseconds, no 20 MB uppercased copy.
import re # stdlib, lazy
if re.search(rb"(?i)<!\s*(DOCTYPE|ENTITY)", raw):
raise UnsafeXML("DTD in sitemap")
SITEMAP_NS = "{http://www.sitemaps.org/schemas/sitemap/0.9}"
def _is_page_loc(tag):
"""A PAGE <loc>: sitemaps.org namespace, or namespace-less.
NOT <image:loc> or <video:loc>. Those live in Google's extension
namespaces and name an ASSET inside a <url>, not a page of its own. An
endswith('}loc') test matches them too — that shipped, and a real site
caught it: 24 <url> + 3 <image:loc> came back as a count of 27, so the
COVERAGE denominator was 12.5% too high and img/logo.png was about to be
sampled and audited as a page.
"""
return tag == SITEMAP_NS + "loc" or tag == "loc"
def _locs(raw):
"""(page <loc> texts, is_sitemapindex).
Walks the DIRECT children of each <url>/<sitemap> rather than root.iter():
that alone excludes <image:image><image:loc>, and the namespace test above
is the second lock. XML comments iterate as elements with no children, so
they fall through harmlessly.
"""
import xml.etree.ElementTree as ET # stdlib, lazy
_refuse_dtd(raw)
root = ET.fromstring(raw)
is_index = root.tag.endswith("sitemapindex")
out = []
for entry in root: # <url> | <sitemap>
for child in entry: # direct children only
if _is_page_loc(child.tag):
text = (child.text or "").strip()
if text:
out.append(text)
break # one <loc> per entry
return out, is_index
def _sane(u):
"""Cheap garbage filter — NOT lib/url-guard.sh. Drops what could never be a
real page URL; the consumer still guards before curling."""
if not u or len(u) > 2048:
return False
if any(c in u for c in '\n\r\t "\'\\`$<>{}|^'):
return False
return urlparse(u).scheme in ("http", "https")
def _expand(children):
"""Fetch each child sitemap of an index. A child that fails is skipped and
counted, never fatal: one dead child must not lose the other 49."""
urls, ok, failed = [], 0, 0
for c in children:
raw = _mock("sitemap_child.xml")
if raw is None:
try:
raw = _fetch(c)
except Exception:
failed += 1
continue
try:
sub, _ = _locs(raw)
except Exception:
failed += 1
continue
urls.extend(sub)
ok += 1
return urls, ok, failed
def sitemap(url):
raw = _mock("sitemap.xml")
if raw is None:
try:
raw = _fetch(url)
except Exception:
return {"status": "degraded", "reason": "fetch_failed"}
try:
locs, is_index = _locs(raw)
except UnsafeXML:
# Distinct from parse_failed on purpose: this one is a finding, not a
# glitch. A sitemap carrying a DTD is either broken tooling or someone
# aiming a billion-laughs at the auditor.
return {"status": "degraded", "reason": "unsafe_xml_dtd"}
except Exception:
return {"status": "degraded", "reason": "parse_failed"}
out = {"status": "ok", "source": "sitemap", "index": is_index}
if is_index:
out["children_total"] = len(locs)
kids, ok, failed = _expand(locs[:MAX_CHILDREN])
out["children_read"], out["children_failed"] = ok, failed
if len(locs) > MAX_CHILDREN: # say what was cut
out["children_skipped"] = len(locs) - MAX_CHILDREN
locs = kids
seen, urls, dropped = set(), [], 0
for u in locs:
if not _sane(u):
dropped += 1
continue
if u in seen:
continue
seen.add(u)
urls.append(u)
if len(urls) > MAX_URLS:
out["truncated"] = len(urls) - MAX_URLS
urls = urls[:MAX_URLS]
out["count"], out["dropped"], out["urls"] = len(urls), dropped, urls
if not urls:
return {"status": "degraded", "reason": "no_urls"}
return out
def _cli():
try:
p = argparse.ArgumentParser()
p.add_argument("--url", required=True)
p.add_argument("--store", default=None) # accepted+ignored: uniform dispatch
args = p.parse_args()
print(json.dumps(sitemap(args.url), indent=2))
except SystemExit as e:
if e.code not in (0, None):
print(json.dumps({"status": "error", "reason": "bad_usage"}))
raise
except Exception:
# Same fail-open contract as google_seo.py: never a traceback, never
# empty stdout, exit 0 so the audit degrades instead of dying.
print(json.dumps({"status": "degraded", "reason": "unexpected_error"}))
if __name__ == "__main__":
_cli()