Coverage for node / src / stigmem_node / utility / net_util.py: 90%
66 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Outbound HTTP safety utilities — SSRF guard (H-SEC-1)."""
3from __future__ import annotations
5import ipaddress
6import socket
7from urllib.parse import urlparse
9# RFC 1918, loopback, link-local, and IPv6 equivalents.
10# Cloud IMDS (169.254.169.254) is covered by 169.254.0.0/16.
11# Most of these are also covered by the is_* classification flags below; they
12# are kept as an explicit, auditable denylist (defense in depth).
13_BLOCKED_NETS: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] = (
14 ipaddress.ip_network("127.0.0.0/8"),
15 ipaddress.ip_network("10.0.0.0/8"),
16 ipaddress.ip_network("172.16.0.0/12"),
17 ipaddress.ip_network("192.168.0.0/16"),
18 ipaddress.ip_network("169.254.0.0/16"),
19 ipaddress.ip_network("0.0.0.0/8"),
20 ipaddress.ip_network("100.64.0.0/10"), # RFC 6598 CGNAT — not flagged is_private
21 ipaddress.ip_network("::1/128"),
22 ipaddress.ip_network("fc00::/7"),
23 ipaddress.ip_network("fe80::/10"),
24)
26# NAT64 well-known prefix (RFC 6052): the low 32 bits embed an IPv4 address.
27_NAT64_WKP = ipaddress.ip_network("64:ff9b::/96")
30def _embedded_ipv4(
31 ip: ipaddress.IPv6Address,
32) -> ipaddress.IPv4Address | None:
33 """Return the IPv4 address embedded in an IPv4-mapped / 6to4 / NAT64 IPv6.
35 A blocked IPv4 (loopback, IMDS, RFC1918) can be smuggled past a v6-blind
36 check as ``::ffff:169.254.169.254``, ``2002:a9fe:a9fe::`` (6to4), or
37 ``64:ff9b::a9fe:a9fe`` (NAT64). Unwrap so the embedded v4 is classified.
38 """
39 if ip.ipv4_mapped is not None:
40 return ip.ipv4_mapped
41 if ip.sixtofour is not None: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true
42 return ip.sixtofour
43 if ip in _NAT64_WKP:
44 return ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)
45 return None
48def _ip_is_blocked(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
49 """Return True iff *ip* is unsafe to connect to (SSRF target).
51 Unwraps IPv4-in-IPv6 embeddings first (F-SSRF-3), then rejects any
52 private / loopback / link-local / reserved / multicast / unspecified
53 address (covers IMDS, RFC1918, CGNAT, etc.) via both the stdlib
54 classification flags and the explicit denylist.
55 """
56 if isinstance(ip, ipaddress.IPv6Address):
57 embedded = _embedded_ipv4(ip)
58 if embedded is not None:
59 ip = embedded
60 if (
61 ip.is_private
62 or ip.is_loopback
63 or ip.is_link_local
64 or ip.is_reserved
65 or ip.is_multicast
66 or ip.is_unspecified
67 ):
68 return True
69 return any(ip in net for net in _BLOCKED_NETS)
72def node_url_is_loopback(node_url: str) -> bool:
73 """Return True iff *node_url*'s host is a literal loopback host.
75 Shared by the startup bind-safety check and the federation approval-time
76 SSRF-skip gate so the loopback host set lives in exactly one place.
77 """
78 try:
79 parsed = urlparse(node_url)
80 host = (parsed.hostname or "").lower()
81 except ValueError:
82 return False
83 return host in {"localhost", "127.0.0.1", "::1"}
86def assert_safe_url(
87 url: str,
88 *,
89 allow_schemes: frozenset[str] = frozenset({"https"}),
90) -> None:
91 """Raise ValueError if *url* is unsafe to fetch.
93 Checks:
94 - scheme is in *allow_schemes*
95 - hostname resolves (DNS failure → ValueError)
96 - no resolved address falls in RFC 1918, loopback, or link-local ranges
98 Residual risk: DNS rebinding window between this check and the actual
99 connection. Callers MUST also set follow_redirects=False so redirects
100 cannot send the connection to a private address after validation.
101 """
102 parsed = urlparse(url)
103 if parsed.scheme not in allow_schemes:
104 raise ValueError(f"Disallowed URL scheme: {parsed.scheme!r}")
105 hostname = parsed.hostname or ""
106 if not hostname: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 raise ValueError(f"URL has no hostname: {url!r}")
108 try:
109 infos = socket.getaddrinfo(hostname, None)
110 except socket.gaierror as exc:
111 raise ValueError(f"Cannot resolve hostname {hostname!r}: {exc}") from exc
112 for info in infos:
113 ip = ipaddress.ip_address(info[4][0])
114 if _ip_is_blocked(ip):
115 raise ValueError(f"Blocked private/loopback address for {hostname!r}: {ip}")
118def resolve_pinned_address(
119 url: str,
120 *,
121 allow_schemes: frozenset[str] = frozenset({"https"}),
122) -> str:
123 """Resolve *url*'s hostname ONCE and return a single safe pinned IP literal.
125 Closes the DNS-rebinding TOCTOU (H9 / F-SSRF-1): ``assert_safe_url`` validates
126 the resolved addresses but then hands the *hostname* to the HTTP client, which
127 re-resolves at connect time — a TTL-0 rebind attacker can serve a public IP at
128 validation and a private/loopback/IMDS IP at connect. Callers must instead
129 connect to the *exact* IP this returns (preserving Host header + TLS SNI +
130 cert verification against the original hostname).
132 Semantics:
133 - scheme must be in *allow_schemes* (https-only by default)
134 - hostname must resolve (DNS failure → ValueError)
135 - if ANY resolved A/AAAA record is private/loopback/link-local/IMDS, the WHOLE
136 url is rejected (ValueError). A rebinder controls which record is served, so
137 we never cherry-pick a public record out of a mixed set.
138 - returns the first resolved IP (a bare literal, e.g. ``"203.0.113.7"`` or
139 ``"2001:db8::1"`` — caller is responsible for bracketing IPv6 in a URL).
141 All failure modes raise ValueError so the caller can fail closed.
142 """
143 parsed = urlparse(url)
144 if parsed.scheme not in allow_schemes:
145 raise ValueError(f"Disallowed URL scheme: {parsed.scheme!r}")
146 hostname = parsed.hostname or ""
147 if not hostname: 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 raise ValueError(f"URL has no hostname: {url!r}")
149 try:
150 infos = socket.getaddrinfo(hostname, None)
151 except socket.gaierror as exc:
152 raise ValueError(f"Cannot resolve hostname {hostname!r}: {exc}") from exc
154 pinned: str | None = None
155 for info in infos:
156 addr = str(info[4][0])
157 ip = ipaddress.ip_address(addr)
158 if _ip_is_blocked(ip):
159 # Reject the whole URL — the rebinder chooses which record is served.
160 raise ValueError(f"Blocked private/loopback address for {hostname!r}: {ip}")
161 if pinned is None:
162 pinned = addr
164 if pinned is None: 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true
165 raise ValueError(f"Hostname {hostname!r} resolved to no addresses")
166 return pinned