fix(fetch_url): block cloud-metadata via IPv4-mapped IPv6 in SSRF guard

The fetch_url SSRF guard guarantees cloud-metadata endpoints (e.g.
169.254.169.254) are ALWAYS blocked, even when CAI_FETCH_ALLOW_INTERNAL
is set for an authorised internal pentest. That guarantee could be
bypassed with the IPv4-mapped IPv6 form of the IMDS address, e.g.
http://[::ffff:169.254.169.254]/ (or its hex form ::ffff:a9fe:a9fe),
because the metadata block is a plain string comparison against the bare
IPv4 literal. With allow_internal=True the private/reserved check is
skipped, so nothing caught the mapped form and the request reached IMDS.

Unwrap IPv4-mapped IPv6 addresses to their embedded IPv4 before the
metadata and private-range checks in both the literal-IP and DNS-resolved
paths. This also hardens the private-range check on Python versions that
do not classify mapped addresses as private/link-local.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Mike German <mike@stepsventures.com>
This commit is contained in:
Mike German 2026-07-06 21:31:06 -04:00
parent 1c79507140
commit bb79146c68
2 changed files with 41 additions and 5 deletions

View File

@ -106,6 +106,22 @@ def _int_env(name: str, default: int) -> int:
return default
def _unwrap_mapped_ip(
ip: ipaddress.IPv4Address | ipaddress.IPv6Address,
) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
"""Return the embedded IPv4 address for IPv4-mapped IPv6 addresses.
``::ffff:169.254.169.254`` and ``::ffff:127.0.0.1`` route to the same
hosts as their bare IPv4 forms, but a plain string comparison against
``_METADATA_IPS`` misses them and some Python versions do not flag the
mapped form as private/link-local. Unwrapping to the embedded IPv4 makes
the metadata and private-range checks apply uniformly. Non-mapped
addresses are returned unchanged.
"""
mapped = getattr(ip, "ipv4_mapped", None)
return mapped if mapped is not None else ip
def _is_unsafe_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
return (
ip.is_private
@ -149,13 +165,22 @@ def _check_ssrf(url: str, *, allow_internal: bool) -> str:
# Literal IP path.
try:
ip_obj = ipaddress.ip_address(host)
except ValueError:
ip_obj = None # not a literal IP; fall through to FQDN resolution
if ip_obj is not None:
# Unwrap IPv4-mapped IPv6 (e.g. ::ffff:169.254.169.254) so the checks
# below cannot be bypassed with the mapped form of a blocked address.
ip_obj = _unwrap_mapped_ip(ip_obj)
# Cloud metadata is ALWAYS blocked, regardless of allow_internal. The
# string pre-check above only matches the bare IPv4 literal, so the
# mapped IPv6 form is caught here.
if str(ip_obj) in _METADATA_IPS:
raise _SSRFBlocked(f"host '{host}' is a cloud-metadata endpoint")
if not allow_internal and _is_unsafe_ip(ip_obj):
raise _SSRFBlocked(f"host '{host}' is a private/reserved address")
return host
except ValueError:
pass # fall through to FQDN resolution
# FQDN: resolve, validate every record, return the first safe IP.
try:
infos = socket.getaddrinfo(host, None)
@ -167,10 +192,10 @@ def _check_ssrf(url: str, *, allow_internal: bool) -> str:
raw_addr = info[4][0]
ip_str = str(raw_addr) # getaddrinfo returns int for AF_UNIX edge cases
try:
ip = ipaddress.ip_address(ip_str)
ip = _unwrap_mapped_ip(ipaddress.ip_address(ip_str))
except ValueError:
continue
if ip_str in _METADATA_IPS:
if str(ip) in _METADATA_IPS:
raise _SSRFBlocked(
f"host '{host}' resolves to cloud-metadata IP '{ip_str}'"
)

View File

@ -72,6 +72,17 @@ class TestSSRFGuard:
with pytest.raises(_SSRFBlocked):
_check_ssrf("http://metadata.google.internal/", allow_internal=True)
def test_blocks_ipv4_mapped_metadata_even_when_internal_allowed(self) -> None:
"""The IPv4-mapped IPv6 form of the IMDS IP must not slip past the
cloud-metadata guard, even with CAI_FETCH_ALLOW_INTERNAL=true."""
for host in ("[::ffff:169.254.169.254]", "[::ffff:a9fe:a9fe]"):
with pytest.raises(_SSRFBlocked):
_check_ssrf(f"http://{host}/latest/", allow_internal=True)
def test_blocks_ipv4_mapped_loopback(self) -> None:
with pytest.raises(_SSRFBlocked):
_check_ssrf("http://[::ffff:127.0.0.1]/", allow_internal=False)
def test_allows_public_ip_and_returns_it(self) -> None:
assert _check_ssrf("https://1.1.1.1/", allow_internal=False) == "1.1.1.1"