Files
turnstone/tests/test_ip_classify.py
Patrick Buckley f4fd7e1f67 fix(security): classify outbound addresses by what they reach (GHSA-wm4f-79pw-pfr9) (#1003)
* fix(security): classify outbound addresses by what they reach (GHSA-wm4f-79pw-pfr9)

Five guards screened outbound URLs and each hand-rolled its own address
normalization and policy tests, so each had a different hole. An IPv6
transition address carries an IPv4 destination in its low bits and
`ipaddress` classifies the wrapper, not the destination: 64:ff9b::a9fe:a9fe
reports is_global because 64:ff9b::/96 is global unicast, while a NAT64
gateway routes it to the cloud metadata endpoint. CGNAT (100.64.0.0/10) is
neither is_private nor is_global, so a denylist built on is_private missed
it with no gateway involved at all.

Add turnstone/core/ip_classify.py as the single classifier. One function
returns exactly one policy lane — PUBLIC, PRIVATE (operator-approvable) or
NEVER — and every guard branches on the lane rather than re-deriving it.
Two overlapping booleans would make a verdict depend on which one a caller
tested first; several addresses are simultaneously globally routable and
metadata-reaching.

- Decode transition addresses per RFC 6052 §2.2 (NAT64 well-known and
  local-use prefixes, 6to4, Teredo, IPv4-mapped, IPv4-compatible) and judge
  them by the IPv4 they reach. The local-use prefix does not say which
  layout its gateway uses, so every length it can carry is decoded and the
  worst result classified.
- Share hostname resolution too. The five copies had already drifted on
  which failures they caught, and getaddrinfo raises UnicodeError — not an
  OSError — from the IDNA encoder.
- Resolution failure is a refusal, not a pass: the fetch resolves again, so
  an authority answering the guard with SERVFAIL and the fetch with an
  internal address would otherwise switch the guard off for that hop.
- Screen every redirect hop in every mode. allow_private_origin widens which
  lanes are acceptable rather than turning screening off, and the permission
  is revoked after any hop that is not wholly private.
- Cleartext http is allowed only for a hostname that RESOLVES to loopback.
  *.localhost is ordinary DNS, and trusting the name put an OIDC token
  exchange on the wire in the clear.
- Screen doctor and console-probe URLs through the classifier. Both used a
  host.startswith("169.254.") string test that never resolved, so any DNS
  name pointing at the metadata service passed and its body was returned to
  the model.
- Add known vendor metadata prefixes the stdlib does not flag, and place
  deprecated IPv6 site-local outside the public lane.

The operator's private-network opt-in still admits the whole home lab,
including IPv6 loopback, CGNAT and split-horizon hosts. Metadata,
link-local, multicast, unspecified and reserved addresses stay refused
regardless of the opt-in, including as a redirect target from an approved
private origin — the settings help and docs now say so.

Reported by @tonghuaroot.

* fix(security): close Azure/Oracle metadata gap and restore dual-stack origins

Review follow-ups on the address-classification rework.

Azure's host-agent endpoint (168.63.129.16) and Oracle Cloud's metadata
endpoint (192.0.0.192) sit in ordinary unicast space, so the stdlib reported
them as globally routable and both classified PUBLIC — reachable with no
opt-in at all, a worse position than the RFC 1918 host beside them, and
directly contradicting the "metadata stays refused even with the opt-in"
guarantee the settings help and docs now advertise. Both join the shared
vendor list.

Revoking the private-hop permission on the ORIGIN hop broke the case
`_screen_tool_url` deliberately admits: a dual-stack or split-horizon
home-lab host answering with both a LAN and a public record was approved,
then refused on its own `302 /login` — one hop was all it ever got. Track
the approved HOST instead, so redirects that stay on it remain covered while
a redirect to any other private host is still refused once the chain is no
longer wholly private.

Also:

- Try several registry candidates for the collector-scope probe instead of
  abandoning it when the first is unresolvable, which also stopped a healthy
  registry from logging as malformed.
- Bound the probe's name resolution with an explicit timeout matching the
  2s the httpx connect deadline used to provide; it runs before the console
  lifespan yields and getaddrinfo has no timeout of its own.
- Route doctor and the console probe through `web.screen_url` rather than
  keeping a third and fourth copy of parse/resolve/classify/fold, which had
  already diverged on default port and empty-hostname wording. An empty
  hostname no longer reports as a cloud-metadata refusal.
- Give `screen_url` a scheme-aware default port.
- Stop doubling the word "hostname" in the OAuth resolution refusal.
- Correct the `_screen_tool_url` docstring: it described `private_origin` as
  requiring every record to be private, which the mixed-record decision
  reversed, and `private_block` as a property of a refusal when it reports
  the lane on the success path too.
- Make the preview tests' screening stub opt-in rather than autouse — as a
  module-wide fixture it also stubbed the tests whose subject IS the screen,
  so one of them would have passed even if screening refused everything.
  Verified the module now passes with all name resolution blocked.

* fix(security): refuse mixed-record private origins instead of exempting them

The previous commit let an approved private origin redirect to itself by
exempting its hostname from the chain-wide revocation. That exemption was
wrong three ways: it was captured once and never cleared, so a public hop
could steer the fetcher back into the approved host at a path of its
choosing — reopening the private -> public -> private bypass; it was
re-entrant across same-host redirects with fresh DNS each time, so a
self-redirecting host could walk arbitrary internal addresses; and it
matched on bare hostname, so it spanned every port on the approved box.
All three were reproduced against the parent commit, which refuses them.

Delete the exemption rather than repair it. The case it existed for — a
dual-stack host answering with both a LAN and a public record — is now
refused where it is actually decidable, in `_screen_tool_url`, with the
remedy in the message: point the tool at the LAN address directly. A
granted chain therefore always starts wholly private, so the fetch guard
needs no notion of an approved host and stays one unconditional rule.

That the accommodation could not be expressed safely in the guard is the
signal: the connection may land on either record, so approving such a host
never described where the fetch would go.

Also from the same review:

- Walk the whole service registry for a collector-scope probe candidate
  instead of the first three, and split the outcome into three log lines,
  so entries that are merely unreachable stop raising the malformed-registry
  alarm and skipping the boot check cluster-wide.
- Stop the candidate walk on a resolver timeout. `asyncio.timeout` bounds
  the await, not the work, so continuing left one parked thread per timed-out
  candidate on the shared executor.
- Move the metadata-hostname denylist into `ip_classify` and enforce it in
  `screen_url`, so doctor and the console probe inherit it instead of each
  keeping a copy.
- Drop the scheme-aware default port: a numeric service does not change
  which addresses resolution returns, and classification reads only those.
  `parsed.port` is still touched so an out-of-range value refuses.
- Correct the vendor-metadata comment, which generalized a claim true of
  Azure's and Oracle's addresses to Alibaba's CGNAT one.
- Rename a test class that was still named for the rule it no longer tests.
2026-08-11 02:18:03 -07:00

624 lines
26 KiB
Python

"""Cross-guard regression tests for SSRF address classification.
Three guards screen outbound URLs — :func:`turnstone.core.oauth_ssrf.validate_url_no_ssrf`
(OAuth/OIDC endpoints), :func:`turnstone.core.web.screen_url` (the ``web_fetch`` /
``open_preview`` tools) and ``turnstone.channels._formatter._is_safe_image_url``
(inline images). Each once hand-rolled its own normalization and its own policy
tests, so each had a different hole: NAT64 and IPv4-compatible walked through the
first two, 6to4 walked through the third, and CGNAT walked through the denylist.
These tests pin what no single guard's own file would catch:
1. A transition address is judged by the IPv4 it routes to, in EVERY guard, for
EVERY wrapper form.
2. Every address lands in exactly ONE lane, and each guard honours the lane
rather than re-deriving it. A previous revision exposed two overlapping
booleans, so lane assignment depended on which one a caller tested first —
``64:ff9b:1::a9fe:a9fe`` was both "public" and "never allowed", and the OAuth
guard accepted it while the web guard refused it.
3. The operator's ``allow_private_network`` opt-in still admits the whole home
lab (v4 and v6), and still never admits the NEVER lane.
4. Unwrapping did not become a blanket rejection of transition prefixes: DNS64
on an IPv6-only node synthesizes ``64:ff9b::<public-v4>`` for every IPv4-only
website, so refusing the prefix would black-hole ordinary browsing.
"""
from __future__ import annotations
import asyncio
import ipaddress
import socket
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, patch
import pytest
from turnstone.channels._formatter import _is_safe_image_url
from turnstone.core.ip_classify import (
AddressLane,
classify_address,
describe_address,
effective_addresses,
embedded_ipv4,
parse_resolved_address,
)
from turnstone.core.oauth_ssrf import OAuthSSRFError, validate_url_no_ssrf
from turnstone.core.web import screen_url
if TYPE_CHECKING:
from collections.abc import Callable
from contextlib import AbstractContextManager
# ---------------------------------------------------------------------------
# Wrapper constructors — build each transition form around an IPv4 target.
# ---------------------------------------------------------------------------
def _plain(v4: str) -> str:
return v4
def _nat64_wkp(v4: str) -> str:
"""RFC 6052 §3.1 well-known prefix: IPv4 in the low 32 bits of 64:ff9b::/96."""
base = int(ipaddress.IPv6Address("64:ff9b::"))
return str(ipaddress.IPv6Address(base | int(ipaddress.IPv4Address(v4))))
def _nat64_local_use(v4: str) -> str:
"""RFC 8215 local-use translation prefix 64:ff9b:1::/48."""
base = int(ipaddress.IPv6Address("64:ff9b:1::"))
return str(ipaddress.IPv6Address(base | int(ipaddress.IPv4Address(v4))))
def _sixtofour(v4: str) -> str:
"""RFC 3056: IPv4 in bits 16-47 of 2002::/16."""
base = int(ipaddress.IPv6Address("2002::"))
return str(ipaddress.IPv6Address(base | (int(ipaddress.IPv4Address(v4)) << 80)))
def _ipv4_compatible(v4: str) -> str:
"""RFC 4291 §2.5.5.1 (deprecated): IPv4 in the low 32 bits of ::/96."""
return str(ipaddress.IPv6Address(int(ipaddress.IPv4Address(v4))))
def _ipv4_mapped(v4: str) -> str:
return str(ipaddress.IPv6Address("::ffff:" + v4))
_TEREDO_SERVER = "8.8.8.8"
def _teredo(v4: str) -> str:
"""RFC 4380: 2001:0::/32, server in bits 32-63, obfuscated client in the low 32.
The target goes in the CLIENT field over a public server, so the wrapper's
lane is the target's lane — either field being unsafe must condemn it.
"""
value = int(ipaddress.IPv6Address("2001::"))
value |= int(ipaddress.IPv4Address(_TEREDO_SERVER)) << 64
value |= (~int(ipaddress.IPv4Address(v4))) & 0xFFFFFFFF
return str(ipaddress.IPv6Address(value))
WRAPPERS: dict[str, Callable[[str], str]] = {
"plain": _plain,
"nat64-wkp": _nat64_wkp,
"nat64-local-use": _nat64_local_use,
"6to4": _sixtofour,
"ipv4-compatible": _ipv4_compatible,
"ipv4-mapped": _ipv4_mapped,
"teredo": _teredo,
}
WRAPPER_IDS = sorted(WRAPPERS)
PUBLIC_CAPABLE_WRAPPER_IDS = WRAPPER_IDS
METADATA = "169.254.169.254"
PUBLIC = "93.184.216.34"
# Everything an operator may opt in to reaching.
PRIVATE_TARGETS = ["127.0.0.1", "192.168.1.5", "10.0.0.1", "100.64.0.1"]
PRIVATE_V6_TARGETS = ["::1", "fd00::1", "fc00::1"]
# Everything refused regardless of the opt-in.
NEVER_TARGETS = [METADATA, "224.0.0.1", "0.0.0.0", "240.0.0.1", "255.255.255.255"]
NEVER_V6_TARGETS = ["fe80::1", "::", "ff02::1", "fd00:ec2::254", "fd00:ec2::23"]
# ---------------------------------------------------------------------------
# Guard runners.
# ---------------------------------------------------------------------------
def _resolving_to(*addrs: str) -> AbstractContextManager[MagicMock]:
infos: list[tuple[int, int, int, str, tuple[Any, ...]]] = []
for a in addrs:
parsed = ipaddress.ip_address(a)
if parsed.version == 6:
infos.append(
(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (a, 0, 0, 0))
)
else:
infos.append((socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (a, 0)))
return patch("socket.getaddrinfo", return_value=infos)
def _oauth_allows(addr: str, *, allow_private: bool = False) -> bool:
with _resolving_to(addr):
try:
validate_url_no_ssrf(
"https://idp.example.com/x", allow_http=False, allow_private=allow_private
)
except OAuthSSRFError:
return False
return True
def _web_lane(*addrs: str) -> AddressLane:
with _resolving_to(*addrs):
return screen_url("https://site.example/x").lane
def _image_allows(addr: str) -> bool:
async def _run() -> bool:
with _resolving_to(addr):
return await _is_safe_image_url("https://cdn.example/i.png")
return asyncio.run(_run())
def _screen_tool(*addrs: str, allow_private_network: bool) -> tuple[str | None, bool, bool]:
"""Drive the REAL consumer of the lane contract, not just the guard."""
from turnstone.core.session import _screen_tool_url
with _resolving_to(*addrs):
return _screen_tool_url("http://target.example/x", allow_private_network)
# ---------------------------------------------------------------------------
# Normalization.
# ---------------------------------------------------------------------------
class TestEmbeddedIPv4:
def test_unwraps_every_transition_form(self) -> None:
for name, wrap in WRAPPERS.items():
if name in ("plain", "teredo"):
continue
addr = ipaddress.ip_address(wrap(PUBLIC))
assert embedded_ipv4(addr) == (ipaddress.IPv4Address(PUBLIC),), name
def test_plain_addresses_are_not_wrappers(self) -> None:
for a in ("93.184.216.34", "2606:4700:4700::1111", "fd00::1", "fe80::1"):
assert embedded_ipv4(ipaddress.ip_address(a)) == (), a
def test_loopback_and_unspecified_are_not_ipv4_compatible_wrappers(self) -> None:
"""``::`` and ``::1`` sit inside ``::/96`` but are not IPv4 wrappers."""
assert embedded_ipv4(ipaddress.ip_address("::1")) == ()
assert embedded_ipv4(ipaddress.ip_address("::")) == ()
def test_floor_declines_to_unwrap_unroutable_embeddings(self) -> None:
"""The 0.0.0.0/8 floor applies inside real wrapper prefixes too.
Without this, ``64:ff9b::1`` would unwrap to 0.0.0.1 and be judged on a
meaningless address; with it, the wrapper itself is classified.
"""
for a in ("64:ff9b::1", "::5", "64:ff9b:1::1"):
assert embedded_ipv4(ipaddress.ip_address(a)) == (), a
assert classify_address(ipaddress.ip_address(a)) is AddressLane.NEVER, a
def test_teredo_yields_server_and_client(self) -> None:
addr = ipaddress.ip_address(_teredo(METADATA))
assert embedded_ipv4(addr) == (
ipaddress.IPv4Address(_TEREDO_SERVER),
ipaddress.IPv4Address(METADATA),
)
def test_zone_identifier_is_stripped(self) -> None:
assert parse_resolved_address("fe80::1%eth0") == ipaddress.ip_address("fe80::1")
assert parse_resolved_address("10.0.0.1") == ipaddress.ip_address("10.0.0.1")
class TestEffectiveAddresses:
def test_wrapper_is_replaced_not_augmented(self) -> None:
"""The wrapper's own class describes the prefix, not the destination.
``64:ff9b::/96`` and ``::/96`` are both ``is_reserved``; consulting the
wrapper as well would refuse every NAT64 address, including the ones
DNS64 synthesizes for ordinary IPv4-only websites.
"""
addr = ipaddress.ip_address(_nat64_wkp(PUBLIC))
assert addr.is_reserved, "precondition: the WKP is reserved"
assert effective_addresses(addr) == (ipaddress.IPv4Address(PUBLIC),)
assert classify_address(addr) is AddressLane.PUBLIC
def test_plain_address_is_its_own_effective_address(self) -> None:
addr = ipaddress.ip_address(PUBLIC)
assert effective_addresses(addr) == (addr,)
class TestDescribeAddress:
def test_names_what_a_wrapper_reaches(self) -> None:
assert METADATA in describe_address(ipaddress.ip_address(_nat64_wkp(METADATA)))
def test_plain_address_renders_bare(self) -> None:
assert describe_address(ipaddress.ip_address(PUBLIC)) == PUBLIC
# ---------------------------------------------------------------------------
# Lanes are disjoint and total — the property the two-boolean design broke.
# ---------------------------------------------------------------------------
class TestLaneAssignment:
@pytest.mark.parametrize("target", NEVER_TARGETS + NEVER_V6_TARGETS)
def test_never_lane(self, target: str) -> None:
assert classify_address(ipaddress.ip_address(target)) is AddressLane.NEVER
@pytest.mark.parametrize("target", PRIVATE_TARGETS + PRIVATE_V6_TARGETS)
def test_private_lane(self, target: str) -> None:
assert classify_address(ipaddress.ip_address(target)) is AddressLane.PRIVATE
@pytest.mark.parametrize("target", [PUBLIC, "2606:4700:4700::1111", "1.1.1.1"])
def test_public_lane(self, target: str) -> None:
assert classify_address(ipaddress.ip_address(target)) is AddressLane.PUBLIC
def test_ipv6_loopback_shares_the_ipv4_loopback_lane(self) -> None:
"""``::1`` is inside ``::/8`` which CPython lists as reserved.
Folding ``is_reserved`` in blindly put IPv6 loopback in NEVER while
127.0.0.1 stayed approvable, so ``http://localhost:8080/`` succeeded or
failed depending on getaddrinfo ordering on a dual-stack host.
"""
assert ipaddress.ip_address("::1").is_reserved, "precondition"
assert classify_address(ipaddress.ip_address("::1")) is AddressLane.PRIVATE
assert classify_address(ipaddress.ip_address("127.0.0.1")) is AddressLane.PRIVATE
def test_vendor_metadata_is_never_despite_being_ula(self) -> None:
addr = ipaddress.ip_address("fd00:ec2::254")
assert addr.is_private and not addr.is_reserved, "precondition: only ULA"
assert classify_address(addr) is AddressLane.NEVER
# ---------------------------------------------------------------------------
# Cross-guard matrix.
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("wrapper", WRAPPER_IDS)
class TestMetadataIsNeverReachable:
def test_oauth_guard_blocks(self, wrapper: str) -> None:
assert not _oauth_allows(WRAPPERS[wrapper](METADATA))
def test_oauth_guard_blocks_even_with_private_opt_in(self, wrapper: str) -> None:
assert not _oauth_allows(WRAPPERS[wrapper](METADATA), allow_private=True)
def test_web_guard_assigns_the_never_lane(self, wrapper: str) -> None:
assert _web_lane(WRAPPERS[wrapper](METADATA)) is AddressLane.NEVER
def test_tool_screen_refuses_even_with_the_opt_in(self, wrapper: str) -> None:
err, private_origin, _block = _screen_tool(
WRAPPERS[wrapper](METADATA), allow_private_network=True
)
assert err is not None
assert private_origin is False
def test_image_guard_blocks(self, wrapper: str) -> None:
assert not _image_allows(WRAPPERS[wrapper](METADATA))
@pytest.mark.parametrize("wrapper", WRAPPER_IDS)
@pytest.mark.parametrize("target", PRIVATE_TARGETS)
class TestPrivateTargetsAreOperatorGated:
def test_oauth_guard_blocks_by_default(self, wrapper: str, target: str) -> None:
assert not _oauth_allows(WRAPPERS[wrapper](target))
def test_oauth_guard_allows_under_opt_in(self, wrapper: str, target: str) -> None:
"""The home-lab lane must survive unwrapping, wrapped or bare."""
assert _oauth_allows(WRAPPERS[wrapper](target), allow_private=True)
def test_web_guard_assigns_the_private_lane(self, wrapper: str, target: str) -> None:
assert _web_lane(WRAPPERS[wrapper](target)) is AddressLane.PRIVATE
def test_tool_screen_offers_the_opt_in(self, wrapper: str, target: str) -> None:
addr = WRAPPERS[wrapper](target)
err, _, _block = _screen_tool(addr, allow_private_network=False)
assert err is not None, "refused without the opt-in"
err, private_origin, _block = _screen_tool(addr, allow_private_network=True)
assert err is None and private_origin is True, "admitted with it"
@pytest.mark.parametrize("target", PRIVATE_V6_TARGETS)
class TestIPv6HomeLabIsReachableUnderTheOptIn:
def test_oauth_guard(self, target: str) -> None:
assert not _oauth_allows(target)
assert _oauth_allows(target, allow_private=True)
def test_tool_screen(self, target: str) -> None:
err, private_origin, _block = _screen_tool(target, allow_private_network=True)
assert err is None and private_origin is True
@pytest.mark.parametrize("target", NEVER_V6_TARGETS)
class TestIPv6NeverTargetsStayRefused:
def test_oauth_guard_even_with_opt_in(self, target: str) -> None:
assert not _oauth_allows(target, allow_private=True)
def test_tool_screen_even_with_opt_in(self, target: str) -> None:
err, private_origin, _block = _screen_tool(target, allow_private_network=True)
assert err is not None and private_origin is False
@pytest.mark.parametrize("wrapper", PUBLIC_CAPABLE_WRAPPER_IDS)
class TestPublicTargetsStayReachable:
"""False-negative guard: unwrapping must not black-hole legitimate traffic."""
def test_oauth_guard_allows(self, wrapper: str) -> None:
assert _oauth_allows(WRAPPERS[wrapper](PUBLIC))
def test_web_guard_allows(self, wrapper: str) -> None:
assert _web_lane(WRAPPERS[wrapper](PUBLIC)) is AddressLane.PUBLIC
def test_image_guard_allows(self, wrapper: str) -> None:
assert _image_allows(WRAPPERS[wrapper](PUBLIC))
def test_tool_screen_allows_without_marking_it_private(self, wrapper: str) -> None:
err, private_origin, _block = _screen_tool(
WRAPPERS[wrapper](PUBLIC), allow_private_network=True
)
assert err is None and private_origin is False
def _nat64_local_use_48(v4: str) -> str:
"""RFC 6052 §2.2 /48 layout: IPv4 across bits 48-63 and 72-87, u octet zero."""
value = int(ipaddress.IPv6Address("64:ff9b:1::"))
packed = int(ipaddress.IPv4Address(v4))
value |= (packed >> 16) << 64
value |= (packed & 0xFFFF) << 40
return str(ipaddress.IPv6Address(value))
class TestLocalUseNAT64:
"""RFC 8215 64:ff9b:1::/48 — a fixed prefix, not the §3.2 NSP residual.
RFC 6052 §2.2 puts the embedded IPv4 at a position that depends on the
prefix length, and deployments carve /96s out of this /48, so both layouts
are decoded. Getting this wrong in either direction is costly: too strict
and an IPv6-only node cannot browse at all, too loose and a NAT64 gateway
translates the guard's blessing straight to the metadata service.
"""
def test_96_layout_wrapping_metadata_is_never(self) -> None:
assert _web_lane(_nat64_local_use(METADATA)) is AddressLane.NEVER
assert not _oauth_allows(_nat64_local_use(METADATA), allow_private=True)
def test_48_layout_wrapping_metadata_is_never(self) -> None:
assert _web_lane(_nat64_local_use_48(METADATA)) is AddressLane.NEVER
assert not _oauth_allows(_nat64_local_use_48(METADATA), allow_private=True)
def test_96_layout_wrapping_public_stays_public(self) -> None:
"""A DNS64 node must reach IPv4-only sites without any opt-in."""
addr = _nat64_local_use(PUBLIC)
assert _web_lane(addr) is AddressLane.PUBLIC
assert _oauth_allows(addr)
def test_48_layout_wrapping_public_stays_public(self) -> None:
addr = _nat64_local_use_48(PUBLIC)
# The address does not say which RFC 6052 length its gateway uses, so
# every possible layout is decoded and the worst wins. The correct
# decode must be among them; the others are judged too.
assert ipaddress.IPv4Address(PUBLIC) in embedded_ipv4(ipaddress.ip_address(addr))
assert _web_lane(addr) is AddressLane.PUBLIC
assert _oauth_allows(addr)
def test_64_layout_wrapping_an_internal_host_is_not_public(self) -> None:
"""A layout the decoder does not try would mis-read as an unrelated public IPv4.
64:ff9b:1:100:a:0:700:0 carries 10.0.0.7 at the /64 layout. Decoding
only /48 and /96 yielded 1.0.10.0 and 7.0.0.0 — both global — and
promoted an internal target to PUBLIC, below even what HEAD refused.
"""
addr = "64:ff9b:1:100:a:0:700:0"
assert ipaddress.IPv4Address("10.0.0.7") in embedded_ipv4(ipaddress.ip_address(addr))
assert _web_lane(addr) is AddressLane.PRIVATE
assert not _oauth_allows(addr)
def test_non_zero_u_octet_is_not_a_48_layout(self) -> None:
"""RFC 6052 reserves bits 64-71; a non-zero value there rules the layout out."""
value = int(ipaddress.IPv6Address(_nat64_local_use_48(PUBLIC))) | (0xFF << 56)
assert _rfc6052_rejects_u_octet(str(ipaddress.IPv6Address(value)))
def _rfc6052_rejects_u_octet(addr: str) -> bool:
from turnstone.core.ip_classify import _rfc6052_ipv4
return _rfc6052_ipv4(ipaddress.IPv6Address(addr), 48) is None
class TestMultipleResolvedAddresses:
"""A hostname is only as safe as the WORST address it resolves to."""
def test_private_record_does_not_mask_a_later_never_record(self) -> None:
assert _web_lane("10.0.0.7", METADATA) is AddressLane.NEVER
def test_order_does_not_matter(self) -> None:
assert _web_lane(METADATA, "10.0.0.7") is AddressLane.NEVER
def test_public_record_does_not_mask_a_private_one(self) -> None:
assert _web_lane(PUBLIC, "10.0.0.7") is AddressLane.PRIVATE
def test_opt_in_does_not_admit_a_masked_never_record(self) -> None:
"""The exploit the worst-lane fold exists to prevent."""
err, private_origin, _block = _screen_tool("10.0.0.7", METADATA, allow_private_network=True)
assert err is not None
assert private_origin is False
class TestResolutionFailureFailsClosed:
"""A guard that cannot resolve must refuse, not pass.
The fetch resolves again, so an authority answering the guard's query with
SERVFAIL and the fetch's query with an internal address would otherwise
switch the guard off for that hop.
"""
def test_web_guard_refuses(self) -> None:
with patch("socket.getaddrinfo", side_effect=socket.gaierror("SERVFAIL")):
screen = screen_url("http://evil.example/x")
assert screen.lane is AddressLane.NEVER
assert screen.error is not None
def test_tool_screen_refuses_even_with_the_opt_in(self) -> None:
from turnstone.core.session import _screen_tool_url
with patch("socket.getaddrinfo", side_effect=socket.gaierror("SERVFAIL")):
err, private_origin, _block = _screen_tool_url("http://evil.example/x", True)
assert err is not None and private_origin is False
class TestMixedRecordOrigins:
"""A host answering with both a private and a public record is refused.
The worst-lane fold is right for refusing and wrong for deciding that a
chain sits inside the operator's network: the connection may land on the
public record, so the approval would describe somewhere the fetch is not.
Granting the chain private-hop permission on that basis is what let a
public record steer the fetcher back into the network.
"""
def test_dual_record_host_is_refused_with_a_remedy(self) -> None:
"""Refused, and told what to do instead.
Admitting it and relying on the fetch guard to contain the chain was
tried and failed: the containment needed an origin exemption, and that
exemption let a public record steer the fetcher back into the network.
Refusing here is what lets the guard stay a single unconditional rule.
"""
err, private_origin, block = _screen_tool("10.0.0.1", "1.2.3.4", allow_private_network=True)
assert err is not None
assert private_origin is False
assert block is True
assert "LAN address directly" in err
def test_wholly_private_host_still_qualifies(self) -> None:
err, private_origin, _block = _screen_tool(
"10.0.0.1", "192.168.1.5", allow_private_network=True
)
assert err is None and private_origin is True
class TestSiteLocalAndVendorMetadata:
@pytest.mark.parametrize(
"addr,who",
[
("fd00:ec2::254", "AWS Nitro IMDS over IPv6"),
("100.100.100.200", "Alibaba Cloud ECS"),
("168.63.129.16", "Azure host agent / wire server"),
("192.0.0.192", "Oracle Cloud"),
],
)
def test_vendor_metadata_is_never(self, addr: str, who: str) -> None:
"""These sit in ordinary unicast space, so the stdlib calls them routable.
Without an explicit entry they are reachable with NO opt-in at all —
a worse position than the RFC 1918 host next to them — and the docs
and settings help promise the opposite.
"""
assert classify_address(ipaddress.ip_address(addr)) is AddressLane.NEVER, who
err, private_origin, _block = _screen_tool(addr, allow_private_network=True)
assert err is not None and private_origin is False
def test_deprecated_site_local_is_not_public(self) -> None:
"""CPython reports fec0::/10 as is_global, so it needs an explicit rule."""
assert ipaddress.ip_address("fec0::1").is_global, "precondition"
assert classify_address(ipaddress.ip_address("fec0::1")) is AddressLane.PRIVATE
assert not _oauth_allows("fec0::1")
assert _oauth_allows("fec0::1", allow_private=True)
class TestCleartextRequiresProvenLoopback:
"""``http://`` is for a real local dev server, and only that.
A hostname is not proof: accepting one would put an OIDC token exchange —
client_secret and authorization code included — on the wire in the clear.
"""
def test_localhost_name_resolving_public_is_refused(self) -> None:
with _resolving_to(PUBLIC), pytest.raises(OAuthSSRFError, match="HTTPS"):
validate_url_no_ssrf("http://evil.localhost/token", allow_http=True)
def test_localhost_name_resolving_private_is_refused(self) -> None:
with _resolving_to("10.0.0.1"), pytest.raises(OAuthSSRFError):
validate_url_no_ssrf("http://evil.localhost/token", allow_http=True)
def test_genuine_loopback_is_allowed(self) -> None:
for addr in ("127.0.0.1", "::1"):
with _resolving_to(addr):
validate_url_no_ssrf("http://localhost:8080/x", allow_http=True)
def test_transition_wrapper_of_loopback_counts_as_loopback(self) -> None:
"""``::7f00:1`` reaches 127.0.0.1, so the classifier and this lane must agree."""
assert not ipaddress.ip_address("::7f00:1").is_loopback, "precondition"
with _resolving_to("::7f00:1"):
validate_url_no_ssrf("https://localhost/x", allow_http=False)
class TestLocalhostNameIsNotEvidence:
"""``*.localhost`` is ordinary DNS a hostile authority answers at will.
The dev lane is gated on the RESOLVED address being loopback, not on the
hostname. A PRIVATE target is what discriminates here: a NEVER target is
refused by the lane check regardless, so it cannot detect the name being
trusted on its own.
"""
def _validate(self, hostname: str, addr: str, *, allow_http: bool = False) -> bool:
with _resolving_to(addr):
try:
validate_url_no_ssrf(
f"http://{hostname}/x" if allow_http else f"https://{hostname}/x",
allow_http=allow_http,
)
except OAuthSSRFError:
return False
return True
def test_localhost_name_resolving_to_a_private_address_is_refused(self) -> None:
assert not self._validate("evil.localhost", "10.0.0.1")
def test_localhost_name_resolving_to_metadata_is_refused(self) -> None:
assert not self._validate("evil.localhost", METADATA)
def test_genuine_loopback_still_works(self) -> None:
"""The dev lane must survive: real localhost resolves to loopback."""
assert self._validate("localhost", "127.0.0.1", allow_http=True)
assert self._validate("localhost", "::1", allow_http=True)
class TestMalformedInput:
def test_out_of_range_port_returns_instead_of_raising(self) -> None:
"""``urlsplit.port`` parses lazily and raises; the contract is str|None."""
assert screen_url("http://example.com:99999/x").error is not None
def test_missing_hostname(self) -> None:
assert screen_url("http:///nohost").error is not None
class TestCGNATIsNotPublic:
"""RFC 6598 shared address space is neither ``is_private`` nor ``is_global``.
A denylist built on ``is_private`` missed it entirely, which let
``web_fetch`` reach hosts on an overlay VPN — a common place to find
100.64.0.0/10 — with no transition gateway involved at all.
"""
def test_stdlib_classifies_cgnat_as_neither(self) -> None:
addr = ipaddress.ip_address("100.64.0.1")
assert not addr.is_private, "precondition: a denylist on is_private misses CGNAT"
assert not addr.is_global
def test_cgnat_is_operator_gated(self) -> None:
assert classify_address(ipaddress.ip_address("100.64.0.1")) is AddressLane.PRIVATE
assert not _oauth_allows("100.64.0.1")
assert _oauth_allows("100.64.0.1", allow_private=True)