diff --git a/docs/oidc.md b/docs/oidc.md index f99aa409..4607ff12 100644 --- a/docs/oidc.md +++ b/docs/oidc.md @@ -124,11 +124,14 @@ allow_private_network = true or via `TURNSTONE_OIDC_ALLOW_PRIVATE_NETWORK=true` (the env var wins when both are set). -The opt-in admits private-range (RFC 1918), unique-local, CGNAT -(100.64/10 — tailnets), and loopback addresses. Link-local, multicast, -and reserved ranges stay refused even with the opt-in — cloud metadata -services (169.254.169.254) live there, and no legitimate IdP does. The -HTTPS requirement and the same-origin endpoint checks are unaffected. +The opt-in admits private-range (RFC 1918), unique-local, site-local, +CGNAT (100.64/10, where overlay VPNs commonly assign hosts), and +loopback addresses. Link-local, multicast, reserved ranges and known +cloud-metadata endpoints stay refused even with the opt-in — no +legitimate IdP lives there. An address is judged by what it actually +reaches, so an IPv6 transition address (NAT64, 6to4, Teredo) wrapping +an internal IPv4 is treated exactly as that IPv4 would be. The HTTPS +requirement and the same-origin endpoint checks are unaffected. This knob only affects the login-flow IdP configured here. OAuth endpoints advertised by remote MCP servers are untrusted input and are diff --git a/docs/tools.md b/docs/tools.md index cd0b2327..1dc353f5 100644 --- a/docs/tools.md +++ b/docs/tools.md @@ -327,7 +327,7 @@ Fetch a URL and extract specific information from it. | `url` | string | yes | The URL to fetch (must start with `http://` or `https://`). | | `question` | string | yes | What to extract or answer from the page content. | -- **What it does**: Fetches the URL, strips HTML to plain text, and uses the LLM to extract the answer to the question from the page content. Every redirect hop is SSRF-screened before it is requested. Private/internal addresses are refused by default; enable `tools.allow_private_network` (console Settings → Tools) to make them approvable for self-hosted setups whose services live on the local network — the approval prompt marks such requests, and a public site redirecting into private space is refused regardless. +- **What it does**: Fetches the URL, strips HTML to plain text, and uses the LLM to extract the answer to the question from the page content. Every redirect hop is SSRF-screened before it is requested. Private/internal addresses are refused by default; enable `tools.allow_private_network` (console Settings → Tools) to make them approvable for self-hosted setups whose services live on the local network — the approval prompt marks such requests, and a public site redirecting into private space is refused regardless. Cloud metadata endpoints and link-local, multicast and reserved addresses are refused even with the opt-in enabled, including as a redirect target from a private address you approved. An address is judged by what it actually reaches, so an IPv6 transition address (NAT64, 6to4, Teredo) wrapping an internal IPv4 is treated exactly as that IPv4 would be. - **Auto-approve**: No -- requires user confirmation (makes network requests). - **Agent availability**: `task_agent`. diff --git a/tests/test_doctor.py b/tests/test_doctor.py index 184affde..6ad59993 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -373,7 +373,9 @@ class TestCheckLlmBackendTool: fake = {"reachable": True, "available_models": ["m"], "error": None} with patch("turnstone.core.model_registry.probe_model_endpoint", return_value=fake): - out = _tool_check_llm_backend({"provider": "openai", "base_url": "http://x/v1"}) + # A resolvable host: the guard screens the RESOLVED address, and an + # unresolvable name is refused before the probe runs. + out = _tool_check_llm_backend({"provider": "openai", "base_url": "http://127.0.0.1/v1"}) assert "reachable" in out def test_rejects_metadata_base_url(self) -> None: @@ -388,6 +390,39 @@ class TestCheckLlmBackendTool: assert "Refused" in out probe.assert_not_called() + def test_rejects_metadata_reached_indirectly(self) -> None: + """The guard RESOLVES, so a name or a wrapper cannot launder the address. + + The previous ``host.startswith("169.254.")`` string test only ever saw + the literal, so any DNS name pointing at the metadata service — or any + IPv6 transition address wrapping it — walked straight through and the + response body came back to the model. + """ + import socket + + from turnstone.doctor import _tool_check_llm_backend + + literals = ["http://[::ffff:169.254.169.254]/v1", "http://[64:ff9b::a9fe:a9fe]/v1"] + for base_url in literals: + with patch("turnstone.core.model_registry.probe_model_endpoint") as probe: + out = _tool_check_llm_backend({"provider": "openai", "base_url": base_url}) + assert "Refused" in out, base_url + probe.assert_not_called() + + # A perfectly ordinary hostname whose DNS answer is the metadata address. + infos = [ + (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("169.254.169.254", 0)) + ] + with ( + patch("socket.getaddrinfo", return_value=infos), + patch("turnstone.core.model_registry.probe_model_endpoint") as probe, + ): + out = _tool_check_llm_backend( + {"provider": "openai", "base_url": "http://imds.attacker.example/v1"} + ) + assert "Refused" in out + probe.assert_not_called() + def test_rejects_non_http_base_url(self) -> None: from turnstone.doctor import _tool_check_llm_backend diff --git a/tests/test_ip_classify.py b/tests/test_ip_classify.py new file mode 100644 index 00000000..7b93a91b --- /dev/null +++ b/tests/test_ip_classify.py @@ -0,0 +1,623 @@ +"""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::`` 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) diff --git a/tests/test_oauth_ssrf.py b/tests/test_oauth_ssrf.py index 6eb9cf52..cf39a41f 100644 --- a/tests/test_oauth_ssrf.py +++ b/tests/test_oauth_ssrf.py @@ -104,7 +104,7 @@ class TestValidateUrlNoSSRF: assert parsed.hostname == "auth.corp.example.com" def test_allow_private_accepts_cgnat(self) -> None: - # 100.64/10 (RFC 6598, shared address space) — e.g. a tailnet-hosted IdP. + # 100.64/10 (RFC 6598, shared address space) — e.g. an overlay-VPN IdP. with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("100.64.0.7", 0))]): validate_url_no_ssrf("https://idp.tail.example", allow_http=False, allow_private=True) diff --git a/tests/test_oidc.py b/tests/test_oidc.py index f5235884..9ec68928 100644 --- a/tests/test_oidc.py +++ b/tests/test_oidc.py @@ -570,12 +570,18 @@ class TestValidateIssuerURL: validate_issuer_url("https://evil.example.com") def test_rejects_link_local(self): - """Hostnames resolving to link-local addresses are rejected.""" + """Hostnames resolving to link-local addresses are rejected. + + Refused as link-local rather than as merely non-public, and WITHOUT the + allow_private_network hint: that opt-in never admits link-local, so + offering it would send the operator to a dead end. + """ with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("169.254.169.254", 0))]), - pytest.raises(OIDCError, match="non-public address.*169.254.169.254"), + pytest.raises(OIDCError, match="link-local.*169.254.169.254") as exc_info, ): validate_issuer_url("https://metadata.internal") + assert "allow_private_network" not in str(exc_info.value) def test_rejects_unresolvable_hostname(self): """DNS resolution failure is rejected.""" @@ -681,11 +687,11 @@ class TestValidateDiscoveredEndpoint: trusted_endpoint_hosts=frozenset(), ) - def test_rejects_private_ip(self): - """Endpoint resolving to a private/link-local IP is rejected.""" + def test_rejects_link_local_ip(self): + """Endpoint resolving to a link-local IP is rejected, with no dead-end hint.""" with ( patch("socket.getaddrinfo", return_value=self._PRIVATE_ADDR), - pytest.raises(OIDCError, match="non-public address.*169.254.169.254"), + pytest.raises(OIDCError, match="link-local.*169.254.169.254") as exc_info, ): validate_discovered_endpoint( "https://idp.example.com/token", @@ -693,6 +699,26 @@ class TestValidateDiscoveredEndpoint: allow_http=False, trusted_endpoint_hosts=frozenset(), ) + assert "allow_private_network" not in str(exc_info.value) + + def test_rejects_private_ip(self): + """Endpoint resolving to a genuinely private IP is rejected — and IS hinted. + + ``_PRIVATE_ADDR`` is 169.254.169.254, which is link-local rather than + private, so this case covers what the name promises: an RFC 1918 address + the operator CAN reach by enabling the opt-in, and is told so. + """ + with ( + patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.7", 0))]), + pytest.raises(OIDCError, match="non-public address.*10.0.0.7") as exc_info, + ): + validate_discovered_endpoint( + "https://idp.example.com/token", + self._issuer(), + allow_http=False, + trusted_endpoint_hosts=frozenset(), + ) + assert "allow_private_network" in str(exc_info.value) def test_rejects_embedded_credentials(self): """Endpoint with userinfo (user:pass@host) rejected.""" diff --git a/tests/test_open_preview_tool.py b/tests/test_open_preview_tool.py index 4939cb05..a563a6b0 100644 --- a/tests/test_open_preview_tool.py +++ b/tests/test_open_preview_tool.py @@ -13,6 +13,8 @@ from __future__ import annotations from types import SimpleNamespace from unittest.mock import MagicMock +import pytest + from turnstone.core.session import ChatSession from turnstone.core.trajectory import Role, turn_from_dict, turn_to_dict @@ -40,6 +42,46 @@ class _RecordingUI: self.tool_results.append((call_id, name, output, kwargs)) +@pytest.fixture +def _no_network_screen(monkeypatch): + """Keep prepare-time SSRF screening off the network. + + Opt-in, NOT autouse: as a module-wide fixture it also stubbed the tests + whose whole point is the screen, so ``test_screen_public_url_passes`` + asserted on the stub and would have passed even if screen_url refused + every hostname. Request it only where the hostname is incidental. + + Screening fails closed on a resolution failure, so a test naming a + third-party host (``example.com``) would otherwise depend on live public + DNS and break on an isolated CI runner. IP literals are passed through to + the real screen — they resolve locally, and the tests that exercise the + private/never lanes are written with literals precisely so they exercise + the real classifier. + """ + import ipaddress + from urllib.parse import urlparse + + from turnstone.core import web + + real = web.screen_url + permissive = _screen_stub() + + def _screen(url): + try: + host = urlparse(url).hostname or "" + except ValueError: + return real(url) + if not host: + return real(url) # malformed URLs are the real screen's business + try: + ipaddress.ip_address(host) + except ValueError: + return permissive(url) + return real(url) + + monkeypatch.setattr("turnstone.core.session.screen_url", _screen) + + def _make_session(**kwargs): defaults = dict( client=MagicMock(), @@ -83,7 +125,7 @@ class TestPrepareOpenPreview: item = s._prepare_open_preview("c1", {"target": "a.txt", "kind": "hologram"}) assert "kind must be one of" in item["error"] - def test_url_target_needs_approval(self): + def test_url_target_needs_approval(self, _no_network_screen): s = _make_session() item = s._prepare_open_preview("c1", {"target": "https://example.com/x"}) assert item["needs_approval"] is True @@ -120,7 +162,7 @@ class TestPrepareOpenPreview: class TestExecOpenPreview: - def test_url_html_builds_web_descriptor(self, monkeypatch): + def test_url_html_builds_web_descriptor(self, _no_network_screen, monkeypatch): s = _make_session() body = b"Acme Pricingx" monkeypatch.setattr( @@ -143,7 +185,7 @@ class TestExecOpenPreview: results = s.ui.tool_results assert results and results[-1][3].get("preview") == descriptor - def test_url_userinfo_stripped_from_descriptor(self, monkeypatch): + def test_url_userinfo_stripped_from_descriptor(self, _no_network_screen, monkeypatch): s = _make_session() body = b"x" monkeypatch.setattr( @@ -157,7 +199,7 @@ class TestExecOpenPreview: assert "sekret" not in descriptor["title"] assert b"sekret" not in att.content # the injected - def test_redirect_into_private_space_blocked(self, monkeypatch): + def test_redirect_into_private_space_blocked(self, _no_network_screen, monkeypatch): s = _make_session() # The guarded fetch raises BEFORE requesting a private hop — the @@ -166,12 +208,16 @@ class TestExecOpenPreview: raise ValueError("Blocked: URL resolves to private/internal address (169.254.169.254)") monkeypatch.setattr("turnstone.core.session.fetch_with_ssrf_guard", _blocked) + # The prepare-time screen fails closed on an unresolvable host, and + # innocent.example does not resolve — stub it so this test exercises + # the executor's ValueError lane rather than the screen. + monkeypatch.setattr("turnstone.core.session.screen_url", _screen_stub()) item = s._prepare_open_preview("c1", {"target": "https://innocent.example/"}) _, msg = s._exec_open_preview(item) assert msg.startswith("Error: fetch failed: Blocked") assert "c1" not in s._tool_previews - def test_oversized_web_content_errors(self, monkeypatch): + def test_oversized_web_content_errors(self, _no_network_screen, monkeypatch): s = _make_session() big = b"" + b"x" * (4 * 1024 * 1024 + 16) + b"" monkeypatch.setattr( @@ -183,7 +229,7 @@ class TestExecOpenPreview: assert msg.startswith("Error:") assert "too large" in msg - def test_url_pdf_over_10mb_previews_to_kind_cap(self, monkeypatch): + def test_url_pdf_over_10mb_previews_to_kind_cap(self, _no_network_screen, monkeypatch): # Review finding (PR #800): a flat 10 MB URL pre-check rejected PDFs # the 32 MiB pdf kind cap allows — the fetch ceiling must track the # widest kind cap and leave the per-kind caps as the authority. @@ -497,11 +543,32 @@ class _FakeClient: return _FakeClient.table[url] +def _screen_stub(blocked=None): + """Stand in for the real per-hop screen, so tests need no DNS. + + Patches what the guard actually calls. Patching a function the guard has + stopped calling would leave the test green while screening nothing. + """ + from turnstone.core.ip_classify import AddressLane + from turnstone.core.web import UrlScreen + + table = blocked or {} + + def _screen(url): + err = table.get(url) + if err is None: + return UrlScreen(AddressLane.PUBLIC, None, False) + return UrlScreen(AddressLane.NEVER, err, False) + + return _screen + + class TestFetchWithSsrfGuard: - def _wire(self, monkeypatch, table): + def _wire(self, monkeypatch, table, blocked=None): _FakeClient.calls = [] _FakeClient.table = table monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient) + monkeypatch.setattr("turnstone.core.web.screen_url", _screen_stub(blocked)) def test_follows_public_redirect_chain(self, monkeypatch): from turnstone.core.web import fetch_with_ssrf_guard @@ -513,7 +580,6 @@ class TestFetchWithSsrfGuard: "https://b.example/x": _FakeHop(200, {}, body=b"landed"), }, ) - monkeypatch.setattr("turnstone.core.web.check_ssrf", lambda url: None) resp = fetch_with_ssrf_guard("https://a.example/", timeout=5) assert resp.status_code == 200 assert _FakeClient.calls == ["https://a.example/", "https://b.example/x"] @@ -528,9 +594,8 @@ class TestFetchWithSsrfGuard: { "https://a.example/": _FakeHop(302, {"location": "http://169.254.169.254/latest"}), }, + blocked={"http://169.254.169.254/latest": "Blocked: private"}, ) - blocked = {"http://169.254.169.254/latest": "Blocked: private"} - monkeypatch.setattr("turnstone.core.web.check_ssrf", lambda url: blocked.get(url)) with pytest.raises(ValueError, match="Blocked: private"): fetch_with_ssrf_guard("https://a.example/", timeout=5) # The load-bearing assertion: the private hop was NEVER requested. @@ -546,7 +611,6 @@ class TestFetchWithSsrfGuard: "https://a.example/moved": _FakeHop(200, {}), }, ) - monkeypatch.setattr("turnstone.core.web.check_ssrf", lambda url: None) resp = fetch_with_ssrf_guard("https://a.example/start", timeout=5) assert resp.status_code == 200 # The realized response carries the FINAL hop's URL — open_preview's @@ -562,7 +626,6 @@ class TestFetchWithSsrfGuard: monkeypatch, {"https://a.example/": _FakeHop(302, {"location": "https://a.example/"})}, ) - monkeypatch.setattr("turnstone.core.web.check_ssrf", lambda url: None) with pytest.raises(ValueError, match="redirects"): fetch_with_ssrf_guard("https://a.example/", timeout=5) @@ -575,7 +638,6 @@ class TestFetchWithSsrfGuard: monkeypatch, {"https://a.example/": _FakeHop(200, {}, body=[b"aaaa", b"bbbb", b"cccc"])}, ) - monkeypatch.setattr("turnstone.core.web.check_ssrf", lambda url: None) with pytest.raises(ValueError, match="fetch limit"): fetch_with_ssrf_guard("https://a.example/", timeout=5, max_bytes=10) @@ -593,7 +655,6 @@ class TestFetchWithSsrfGuard: "https://b.example/x": _FakeHop(200, {}, body=b"ok"), }, ) - monkeypatch.setattr("turnstone.core.web.check_ssrf", lambda url: None) resp = fetch_with_ssrf_guard("https://a.example/", timeout=5) assert resp.status_code == 200 assert resp.content == b"ok" @@ -615,7 +676,6 @@ class TestFetchWithSsrfGuard: ) }, ) - monkeypatch.setattr("turnstone.core.web.check_ssrf", lambda url: None) resp = fetch_with_ssrf_guard("https://a.example/", timeout=5) # iter_bytes() hands the guard content-DECODED bytes — a surviving # content-encoding would make .text try to gunzip plain text, and the @@ -717,13 +777,13 @@ class TestAllowPrivateNetwork: def test_screen_public_url_passes(self): from turnstone.core.session import _screen_tool_url - err, private = _screen_tool_url("https://example.com/x", False) + err, private, _block = _screen_tool_url("https://93.184.216.34/x", False) assert err is None and private is False def test_screen_private_blocked_with_discoverable_hint(self): from turnstone.core.session import _screen_tool_url - err, private = _screen_tool_url("http://10.0.0.7/grafana", False) + err, private, _block = _screen_tool_url("http://10.0.0.7/grafana", False) assert err is not None and private is False # The refusal teaches the knob (mirrors the oidc opt-in hint pattern). assert "tools.allow_private_network" in err @@ -732,13 +792,13 @@ class TestAllowPrivateNetwork: def test_screen_private_allowed_when_opted_in(self): from turnstone.core.session import _screen_tool_url - err, private = _screen_tool_url("http://10.0.0.7/grafana", True) + err, private, _block = _screen_tool_url("http://10.0.0.7/grafana", True) assert err is None and private is True def test_screen_invalid_url_never_hints(self): from turnstone.core.session import _screen_tool_url - err, private = _screen_tool_url("http://", True) + err, private, _block = _screen_tool_url("http://", True) assert err is not None and private is False assert "allow_private_network" not in err @@ -792,7 +852,13 @@ class TestAllowPrivateNetwork: s._exec_open_preview(item) assert seen["allow_private_origin"] is True - def test_guard_skips_hop_screen_for_private_origin(self, monkeypatch): + def test_guard_permits_private_hops_for_an_approved_private_origin(self, monkeypatch): + """Screening is not skipped for a private origin — it is WIDENED. + + The operator approved a private URL, so the PRIVATE lane is acceptable + on this chain; every hop is still classified. (These are IP literals, + so the real screen resolves them without touching the network.) + """ from turnstone.core.web import fetch_with_ssrf_guard _FakeClient.calls = [] @@ -801,15 +867,172 @@ class TestAllowPrivateNetwork: "http://10.0.0.8/b": _FakeHop(200, {}), } monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient) - - def _explode(url): - raise AssertionError("hop screening must be skipped for a private origin") - - monkeypatch.setattr("turnstone.core.web.check_ssrf", _explode) resp = fetch_with_ssrf_guard("http://10.0.0.7/a", timeout=5, allow_private_origin=True) assert resp.status_code == 200 assert _FakeClient.calls == ["http://10.0.0.7/a", "http://10.0.0.8/b"] + def test_private_hop_refused_without_the_private_origin_flag(self, monkeypatch): + import pytest + + from turnstone.core.web import fetch_with_ssrf_guard + + _FakeClient.calls = [] + _FakeClient.table = {"http://10.0.0.7/a": _FakeHop(200, {})} + monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient) + with pytest.raises(ValueError, match="private/internal"): + fetch_with_ssrf_guard("http://10.0.0.7/a", timeout=5) + assert _FakeClient.calls == [] + + def test_public_bounce_cannot_return_to_the_origin_host(self, monkeypatch): + """``private -> public -> back to the origin`` must not be re-admitted. + + An origin-host exemption (added to let a dual-stack host redirect to + itself) made this reachable: the permission was keyed on the hostname + and never cleared, so a public hop could send the fetcher back to the + approved host at a path of its choosing. Mixed-record origins are now + refused before the fetch instead, so the guard needs no exemption. + """ + import socket + + import pytest + + from turnstone.core.web import fetch_with_ssrf_guard + + _FakeClient.calls = [] + _FakeClient.table = { + "http://home.example/": _FakeHop(302, {"location": "http://attacker.example/"}), + "http://attacker.example/": _FakeHop(302, {"location": "http://home.example/admin"}), + "http://home.example/admin": _FakeHop(200, {}), + } + monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient) + + def _resolve(host, port=None, *a, **kw): + if host == "home.example": + return [ + (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("10.0.0.5", 0)), + ( + socket.AF_INET, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("93.184.216.34", 0), + ), + ] + return [ + (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("93.184.216.34", 0)) + ] + + monkeypatch.setattr("socket.getaddrinfo", _resolve) + with pytest.raises(ValueError, match="private/internal"): + fetch_with_ssrf_guard("http://home.example/", timeout=5, allow_private_origin=True) + assert _FakeClient.calls == ["http://home.example/", "http://attacker.example/"] + + def test_dual_stack_origin_cannot_redirect_to_another_private_host(self, monkeypatch): + """The approval covers that host, not the rest of the network.""" + import socket + + import pytest + + from turnstone.core.web import fetch_with_ssrf_guard + + _FakeClient.calls = [] + _FakeClient.table = { + "http://grafana.home.arpa/": _FakeHop(302, {"location": "http://10.0.0.1/admin"}), + "http://10.0.0.1/admin": _FakeHop(200, {}), + } + monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient) + infos = [ + (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("10.0.0.5", 0)), + (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("93.184.216.34", 0)), + ] + monkeypatch.setattr("socket.getaddrinfo", lambda *a, **kw: infos) + with pytest.raises(ValueError, match="private/internal"): + fetch_with_ssrf_guard("http://grafana.home.arpa/", timeout=5, allow_private_origin=True) + assert _FakeClient.calls == ["http://grafana.home.arpa/"] + + def test_mixed_record_hop_revokes_the_private_permission(self, monkeypatch): + """A hop that merely CONTAINS a private record must not keep the permission. + + The revocation used to key on the folded lane, so a hop resolving to + both a private and an attacker-controlled public record folded to + PRIVATE, was fetched, and did NOT revoke — letting the attacker's + server steer the next hop back into private space. + """ + import pytest + + from turnstone.core import web + from turnstone.core.ip_classify import AddressLane + from turnstone.core.web import UrlScreen, fetch_with_ssrf_guard + + _FakeClient.calls = [] + _FakeClient.table = { + "http://10.0.0.7/a": _FakeHop(302, {"location": "http://mixed.example/b"}), + "http://mixed.example/b": _FakeHop(302, {"location": "http://10.0.0.1/admin"}), + "http://10.0.0.1/admin": _FakeHop(200, {}), + } + monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient) + + real = web.screen_url + + def _screen(url): + if "mixed.example" in url: + # Worst lane PRIVATE, but not wholly private. + return UrlScreen(AddressLane.PRIVATE, "Blocked: private/internal", False) + return real(url) + + monkeypatch.setattr("turnstone.core.web.screen_url", _screen) + with pytest.raises(ValueError, match="private/internal"): + fetch_with_ssrf_guard("http://10.0.0.7/a", timeout=5, allow_private_origin=True) + assert _FakeClient.calls == ["http://10.0.0.7/a", "http://mixed.example/b"] + + def test_public_bounce_revokes_the_private_permission(self, monkeypatch): + """``private -> public -> private`` must not reach the final hop. + + The operator approved their own hosts, not whatever a public site + picks next. Without this, a LAN page serving attacker-authored content + could steer the fetcher into internal endpoints of the attacker's + choosing and hand back the response. + """ + import pytest + + from turnstone.core.web import fetch_with_ssrf_guard + + _FakeClient.calls = [] + _FakeClient.table = { + "http://10.0.0.7/wiki": _FakeHop(302, {"location": "http://93.184.216.34/"}), + "http://93.184.216.34/": _FakeHop(302, {"location": "http://10.0.0.1/admin"}), + "http://10.0.0.1/admin": _FakeHop(200, {}), + } + monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient) + with pytest.raises(ValueError, match="private/internal"): + fetch_with_ssrf_guard("http://10.0.0.7/wiki", timeout=5, allow_private_origin=True) + assert _FakeClient.calls == ["http://10.0.0.7/wiki", "http://93.184.216.34/"] + + def test_guard_still_blocks_metadata_hop_from_a_private_origin(self, monkeypatch): + """Approving a private origin says "this is my network" — not "and the IMDS". + + The PRIVATE lane is not re-screened for an approved private origin (see + the test above), but the NEVER lane is absolute: a LAN host that is + compromised, or simply serving attacker-authored content, must not be + able to bounce the fetcher into the cloud metadata endpoint. + """ + import pytest + + from turnstone.core.web import fetch_with_ssrf_guard + + _FakeClient.calls = [] + _FakeClient.table = { + "http://10.0.0.7/a": _FakeHop( + 302, {"location": "http://169.254.169.254/latest/meta-data/"} + ), + "http://169.254.169.254/latest/meta-data/": _FakeHop(200, {}), + } + monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient) + + with pytest.raises(ValueError, match="link-local"): + fetch_with_ssrf_guard("http://10.0.0.7/a", timeout=5, allow_private_origin=True) + assert _FakeClient.calls == ["http://10.0.0.7/a"], "metadata hop was never issued" + def test_registry_entry_shape(self): from turnstone.core.settings_registry import SETTINGS diff --git a/tests/test_service_auth_boundary.py b/tests/test_service_auth_boundary.py index 7ada00d0..0f3dc02c 100644 --- a/tests/test_service_auth_boundary.py +++ b/tests/test_service_auth_boundary.py @@ -134,7 +134,7 @@ class TestVerifyCollectorServiceScope: app = _scope_probe_app( services=[ - {"service_id": "node-1", "url": "http://node-1:8001"}, + {"service_id": "node-1", "url": "http://127.0.0.1:8001"}, ] ) @@ -159,7 +159,7 @@ class TestVerifyCollectorServiceScope: the drift at boot rather than chasing empty-dashboard reports.""" from turnstone.console.server import _verify_collector_service_scope - app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://node-1:8001"}]) + app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://127.0.0.1:8001"}]) def handler(_req: httpx.Request) -> httpx.Response: return httpx.Response(403, text='{"error":"service scope required"}') @@ -183,7 +183,7 @@ class TestVerifyCollectorServiceScope: class as 403 — refuse to serve.""" from turnstone.console.server import _verify_collector_service_scope - app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://node-1:8001"}]) + app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://127.0.0.1:8001"}]) def handler(_req: httpx.Request) -> httpx.Response: return httpx.Response(401, text="unauthorized") @@ -212,7 +212,7 @@ class TestVerifyCollectorServiceScope: Leave ``collector_scope_error`` empty and log a warning.""" from turnstone.console.server import _verify_collector_service_scope - app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://node-1:8001"}]) + app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://127.0.0.1:8001"}]) def handler(_req: httpx.Request) -> httpx.Response: raise httpx.ConnectError("connection refused") diff --git a/turnstone/channels/_formatter.py b/turnstone/channels/_formatter.py index 0fd0e433..c9046434 100644 --- a/turnstone/channels/_formatter.py +++ b/turnstone/channels/_formatter.py @@ -214,27 +214,27 @@ def try_parse_media(output: str) -> dict[str, Any] | None: _BLOCKED_HOSTNAMES = frozenset({"localhost", "metadata.google.internal"}) -# Cloud-metadata deny-list applied *before* the `is_private` allowance so -# ULA-hosted vendor metadata endpoints don't slip through the "private IPs -# are fine, we trust the LAN" exception. IPv4 169.254.169.254 is caught -# by `is_link_local`; IPv6 ULA metadata (AWS Nitro IMDS at fd00:ec2::254, -# ECS task metadata at fd00:ec2::23) is `is_private` and needs explicit -# blocking. Add new vendor prefixes here as they're published. -_BLOCKED_IP_NETWORKS: tuple[str, ...] = ( - "fd00:ec2::/32", # AWS Nitro IMDS / ECS task metadata over IPv6 -) - async def _is_safe_image_url(url: str) -> bool: """Validate that *url* uses http(s), has no embedded credentials, and does - not target loopback, link-local (incl. cloud metadata 169.254.169.254), - or reserved ranges — even after DNS resolution. + not target loopback or anything in the never-allowed lane — even after + DNS resolution. Resolves the hostname and checks every returned address so a DNS rebinding attack cannot swap a safe-looking public IP for an internal one between validation and fetch. Private/LAN IPs are - still allowed (media servers typically live on the local network), - so only loopback + link-local + multicast + reserved are rejected. + still allowed (media servers typically live on the local network); + what is rejected is loopback — the bot's own host is not a media + server — plus the shared never-allowed lane, which covers link-local + (including cloud metadata at 169.254.169.254), multicast, unspecified, + reserved, and known vendor metadata prefixes such as AWS Nitro IMDS + over IPv6. That lane lives in :mod:`turnstone.core.ip_classify` so + this guard cannot drift from the OAuth and fetch-tool guards again. + + Note the classifier judges a transition address by the IPv4 it + REACHES, replacing the wrapper rather than adding to it, so + ``::ffff:192.168.0.6`` is the LAN address it routes to (allowed here) + rather than the reserved wrapper it looks like. NOTE: there is a residual TOCTOU gap because httpx resolves the hostname again when it actually issues the GET. A 0-TTL rebinding @@ -244,10 +244,15 @@ async def _is_safe_image_url(url: str) -> bool: backfill pass. """ import asyncio - import ipaddress - import socket from urllib.parse import urlparse + from turnstone.core.ip_classify import ( + AddressLane, + ResolutionError, + effective_addresses, + resolve_and_classify, + ) + try: parsed = urlparse(url) except Exception: # noqa: BLE001 @@ -264,37 +269,22 @@ async def _is_safe_image_url(url: str) -> bool: # Collect candidate IPs: either an IP literal in the URL, or every # A/AAAA record the resolver returns for a hostname. - candidates: list[str] = [] try: - ipaddress.ip_address(hostname) - candidates.append(hostname) - except ValueError: - try: - infos = await asyncio.to_thread(socket.getaddrinfo, hostname, None, socket.AF_UNSPEC) - except socket.gaierror: - return False - # Strip IPv6 zone IDs (e.g. ``fe80::1%eth0``) before parsing — - # ipaddress.ip_address would raise on them and we'd drop the host - # on unrelated metadata. - candidates = [str(info[4][0]).partition("%")[0] for info in infos] - if not candidates: - return False + classified = await asyncio.to_thread(resolve_and_classify, hostname) + except ResolutionError: + return False - blocked_networks = [ipaddress.ip_network(cidr) for cidr in _BLOCKED_IP_NETWORKS] - for raw in candidates: - try: - ip = ipaddress.ip_address(raw) - except ValueError: + for lane, ip in classified: + # A transition address is judged by the IPv4 it routes to *instead of* + # by the wrapper — the classifier REPLACES the wrapper, it does not add + # to it. 2002:a9fe:a9fe:: is ordinary global unicast to ``ipaddress`` + # but reaches 169.254.169.254 through a 6to4 relay. + if lane is AddressLane.NEVER: return False - if ( - ip.is_loopback - or ip.is_link_local - or ip.is_multicast - or ip.is_reserved - or ip.is_unspecified - ): - return False - if any(ip in net for net in blocked_networks): + # Loopback is the one class this guard refuses that the shared NEVER + # lane does not: PRIVATE is allowed here (media servers live on the + # LAN), but the bot's own host is not a media server. + if any(item.is_loopback for item in effective_addresses(ip)): return False return True diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 6c0f2492..f59f3473 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -4811,15 +4811,91 @@ def _probe_candidate_url(services: list[dict[str, Any]] | None) -> tuple[str, st if parsed.scheme not in _PROBE_ALLOWED_SCHEMES: continue host = (parsed.hostname or "").lower() - # 169.254.0.0/16 is the AWS / GCP instance metadata range; - # an http target there would turn a compromised registry into - # an SSRF to IMDS. Loopback is retained for single-box dev. - if host.startswith("169.254."): + # Cheap, DNS-free rejection of an entry that NAMES a bad address. A + # poisoned registry pointing straight at cloud metadata is refused here + # without touching the resolver, so selection stays pure and console + # startup does no I/O; a hostname that RESOLVES somewhere bad is caught + # by _probe_url_is_safe, off the event loop, before the request. + if not host or _names_never_allowed_address(host): continue return raw_url, nid return "", "" +def _names_never_allowed_address(host: str) -> bool: + """True when *host* is an IP literal in the never-allowed lane. No DNS.""" + import ipaddress + + from turnstone.core.ip_classify import AddressLane, classify_address + + try: + addr = ipaddress.ip_address(host) + except ValueError: + return False # a name, not a literal — resolved later by the caller + return classify_address(addr) is AddressLane.NEVER + + +def _probe_url_is_safe(url: str) -> bool: + """True when *url* resolves entirely outside the never-allowed lane. + + Blocking: the caller runs it off the event loop under a deadline. Uses the + same screen as the fetch tools rather than a private copy, so a new lane or + a new metadata prefix reaches this guard automatically. + + Fails CLOSED. The probe sends ``Authorization: Bearer ``, + so a registry entry that SERVFAILs here and resolves at request time must + not be probed. Note this bounds the DAMAGE, not the audience: any + resolvable public host in the registry still receives that token, which is + a property of the registry being trusted input, not of this check. + """ + from turnstone.core.ip_classify import AddressLane + from turnstone.core.web import screen_url + + return screen_url(url).lane is not AddressLane.NEVER + + +_PROBE_RESOLVE_TIMEOUT_SECONDS = 2.0 + + +async def _first_safe_candidate(services: list[dict[str, Any]] | None) -> tuple[str, str, int]: + """Return the first registry entry that passes the safety screen. + + Returns ``(url, node_id, refused)``. ``refused`` counts entries that were + selectable but screened out, so the caller can tell "registry is malformed" + (an operator-actionable alarm) from "the entries are fine, none is reachable + right now" — logging the former for the latter sends operators to audit a + healthy registry. + + Walks the WHOLE registry rather than a fixed prefix: capping the walk meant + a cluster whose first few entries were briefly unresolvable skipped the boot + check entirely and still raised the malformed alarm. + + Each resolution runs off the event loop under a deadline, because this is + awaited before the console lifespan yields and ``getaddrinfo`` has no + timeout of its own. A deadline bounds the AWAIT, not the work — the thread + stays parked until the resolver gives up — so a timeout stops the walk + rather than starting another one, keeping at most one thread parked on the + shared executor. + """ + remaining = list(services or []) + refused = 0 + while True: + url, nid = _probe_candidate_url(remaining) + if not url: + return "", "", refused + remaining = [s for s in remaining if s.get("service_id") != nid] + try: + async with asyncio.timeout(_PROBE_RESOLVE_TIMEOUT_SECONDS): + safe = await asyncio.to_thread(_probe_url_is_safe, url) + except TimeoutError: + log.warning("collector_scope_probe.resolver_timeout node=%s", nid) + return "", "", refused + 1 + if safe: + return url, nid, refused + refused += 1 + log.warning("collector_scope_probe.candidate_refused node=%s", nid) + + async def _verify_collector_service_scope(app: Starlette, client: httpx.AsyncClient) -> None: """Probe one upstream node to confirm the collector token's scopes. @@ -4860,13 +4936,19 @@ async def _verify_collector_service_scope(app: Starlette, client: httpx.AsyncCli exc_info=True, ) return - probe_url, probe_node = _probe_candidate_url(services) + probe_url, probe_node, refused = await _first_safe_candidate(services) if not probe_url: - # Distinguish "registry empty" (normal pre-discovery) from - # "registry populated but every entry malformed" (operator- - # actionable drift) so the two aren't both logged as INFO - # silent-skips. - if services: + # Three distinct states, three distinct log lines: an empty registry is + # normal pre-discovery, entries that screened out are a reachability + # problem, and entries that could not even be selected are the + # operator-actionable drift the malformed alarm is for. + if refused: + log.warning( + "collector_scope_probe.no_reachable_candidate count=%d refused=%d", + len(services or []), + refused, + ) + elif services: log.warning( "collector_scope_probe.registry_malformed count=%d", len(services), diff --git a/turnstone/core/ip_classify.py b/turnstone/core/ip_classify.py new file mode 100644 index 00000000..717575de --- /dev/null +++ b/turnstone/core/ip_classify.py @@ -0,0 +1,300 @@ +"""Shared address classification for the tree's SSRF guards. + +Five guards screen outbound URLs, and their *policies* differ on purpose: +:mod:`turnstone.core.oauth_ssrf` demands a globally routable endpoint unless the +operator opted in, :func:`turnstone.core.web.screen_url` (the ``web_fetch`` / +``open_preview`` tools) honours ``tools.allow_private_network``, and +``turnstone.channels._formatter._is_safe_image_url`` allows the LAN outright +because media servers live there. + +What they must not differ on is *which lane an address falls in*. This module +answers that with a single function, :func:`classify_address`, returning exactly +one :class:`AddressLane`. Guards branch on the lane; they never re-derive it +from raw ``ipaddress`` predicates and never order two overlapping tests +themselves. Overlapping predicates are the trap here: several addresses are +simultaneously globally routable *and* metadata-reaching, so any design that +asks a caller to test two booleans makes the verdict depend on which one it +happens to check first. One function, one lane, disjoint by construction. + +Two normalizations feed the classification: + +*IPv6 transition addresses* carry an IPv4 destination in their low bits, and +``ipaddress`` classifies the wrapper rather than the destination: +``64:ff9b::a9fe:a9fe`` is ``is_global`` because ``64:ff9b::/96`` is global +unicast, even though a NAT64 gateway routes it to 169.254.169.254. The wrapper +is *replaced* by what it routes to — never merely added alongside, since +``64:ff9b::/96`` and ``::/96`` are themselves ``is_reserved`` and judging the +wrapper would refuse every NAT64 address including the ones DNS64 synthesizes +for ordinary IPv4-only websites. + +*Vendor metadata prefixes* that the stdlib has no opinion about (AWS Nitro IMDS +over IPv6 at ``fd00:ec2::/32``) are ULA — ``is_private`` and nothing else — so +without an explicit entry they would land in the operator-approvable lane and a +home-lab opt-in would expose instance credentials. + +Residual, deliberately not addressed: a NAT64 Network-Specific Prefix +(RFC 6052 §3.2) is built from the operator's own global prefix, so it is +indistinguishable from ordinary global unicast and carries no marker any +address-based check could key on. +""" + +from __future__ import annotations + +import enum +import ipaddress +import socket +from typing import TypeAlias + +IPAddress: TypeAlias = ipaddress.IPv4Address | ipaddress.IPv6Address + + +class AddressLane(enum.IntEnum): + """Which policy lane an address falls in. Ordered by severity. + + Ordering lets a caller fold several resolved addresses with ``max()``: a + hostname is only as safe as the worst address it resolves to. + """ + + PUBLIC = 0 + """Globally routable. Every guard allows it.""" + + PRIVATE = 1 + """LAN, ULA, CGNAT, loopback. Refused unless the operator opted in.""" + + NEVER = 2 + """Refused regardless of any opt-in — link-local (cloud metadata at + 169.254.169.254 is the canonical SSRF target), multicast, unspecified, + reserved, and known vendor metadata prefixes. No legitimate IdP, home-lab + dashboard, or media server lives in these.""" + + +# Translation prefixes carrying an embedded IPv4: +# 64:ff9b::/96 RFC 6052 §3.1 well-known prefix. A compliant NAT64 gateway +# MUST NOT use it for non-global IPv4, but a misconfigured +# one will — so decode and re-classify rather than trusting +# the RFC to hold. +# 64:ff9b:1::/48 RFC 8215 local-use translation prefix. +# ::/96 RFC 4291 §2.5.5.1 IPv4-compatible (deprecated). ``::`` and +# ``::1`` sit inside it but are the unspecified and loopback +# addresses, not wrappers — the floor below excludes them +# along with every other embedding no host would route. +_LOWEST_ROUTABLE_V4 = ipaddress.IPv4Address("1.0.0.0") + +# Metadata endpoints the stdlib does not flag. The 169.254.169.254 most vendors +# use is caught by ``is_link_local``; these are not. Add new vendor prefixes +# here — this is the single list, shared by every guard. +_VENDOR_METADATA: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] = ( + ipaddress.IPv6Network("fd00:ec2::/32"), # AWS Nitro IMDS / ECS task metadata + # Alibaba's sits in CGNAT, so it would land in the operator-approvable + # lane; Azure's and Oracle's are ordinary global unicast, so without an + # entry they are reachable with NO opt-in at all — a worse position than + # the RFC 1918 host beside them. + ipaddress.IPv4Network("100.100.100.200/32"), # Alibaba Cloud ECS metadata + ipaddress.IPv4Network("168.63.129.16/32"), # Azure host agent / wire server + ipaddress.IPv4Network("192.0.0.192/32"), # Oracle Cloud metadata +) + +# RFC 6052 §2.2 embeds the IPv4 at a position that depends on the translation +# prefix length, and requires bits 64-71 (the "u" octet) to be zero. The +# well-known prefix is /96 by definition (RFC 6052 §3.1); the local-use prefix +# is a /48 that a gateway may subnet at any RFC 6052 length, and the address +# does not say which. Every length is therefore decoded and the WORST result +# classified: guessing one layout would let a mis-decode of an internal target +# read as an unrelated public address. Only lengths at or below the prefix are +# possible — RFC 8215 defines the local-use prefix as a /48, so a /32 or /40 +# "layout" would overlap the prefix bits and decode pure garbage. +_RFC6052_LOCAL_USE_LENGTHS = (48, 56, 64, 96) +_TRANSLATION_PREFIXES: tuple[tuple[ipaddress.IPv6Network, tuple[int, ...]], ...] = ( + (ipaddress.IPv6Network("64:ff9b::/96"), (96,)), + (ipaddress.IPv6Network("64:ff9b:1::/48"), _RFC6052_LOCAL_USE_LENGTHS), + (ipaddress.IPv6Network("::/96"), (96,)), +) + + +def _rfc6052_ipv4(addr: ipaddress.IPv6Address, prefix_len: int) -> ipaddress.IPv4Address | None: + """Decode the IPv4 an RFC 6052 translation address carries, or None. + + The 32 IPv4 bits start at *prefix_len* and skip bits 64-71, which the RFC + reserves as a zero "u" octet — a non-zero value there means this is not a + translation address of that length, so the layout does not apply. + """ + value = int(addr) + if prefix_len == 96: + return ipaddress.IPv4Address(value & 0xFFFFFFFF) + if (value >> 56) & 0xFF: + return None + high_bits = min(64 - prefix_len, 32) + low_bits = 32 - high_bits + high = (value >> (128 - prefix_len - high_bits)) & ((1 << high_bits) - 1) + low = ((value >> (56 - low_bits)) & ((1 << low_bits) - 1)) if low_bits else 0 + return ipaddress.IPv4Address((high << low_bits) | low) + + +BLOCKED_HOSTNAMES = frozenset({"metadata.google.internal"}) +"""Metadata endpoints reached by NAME rather than by address. + +Resolution would classify these correctly on most clouds, but the name is +stable while the address it answers with is not, so the guards refuse it +outright. Shared for the same reason the prefixes are: a per-caller copy is +how they drifted before. +""" + + +def embedded_ipv4(addr: IPAddress) -> tuple[ipaddress.IPv4Address, ...]: + """Return the IPv4 addresses *addr* routes through, if it is a wrapper. + + Empty for a plain address. Teredo yields both the server and the client + address, since either being unsafe makes the wrapper unsafe. + """ + if not isinstance(addr, ipaddress.IPv6Address): + return () + if addr.ipv4_mapped is not None: + return (addr.ipv4_mapped,) + if addr.sixtofour is not None: + return (addr.sixtofour,) + teredo = addr.teredo + if teredo is not None: + return teredo + # ``ipaddress`` exposes no accessor for the translation prefixes, so decode + # by hand per RFC 6052 §2.2. + for network, lengths in _TRANSLATION_PREFIXES: + if addr not in network: + continue + decoded = tuple( + item + for item in (_rfc6052_ipv4(addr, length) for length in lengths) + if item is not None and item >= _LOWEST_ROUTABLE_V4 + ) + # Several layouts can decode at once; the worst-lane fold in + # ``classify_address`` then judges the address by all of them. + return tuple(dict.fromkeys(decoded)) + return () + + +def effective_addresses(addr: IPAddress) -> tuple[IPAddress, ...]: + """Return the addresses that decide how *addr* should be classified. + + For a transition wrapper that is the IPv4 it routes to; the wrapper's own + classification describes the transition prefix, not the destination. For + anything else it is the address itself. + """ + return embedded_ipv4(addr) or (addr,) + + +def _classify_one(addr: IPAddress) -> AddressLane: + """Classify a single already-unwrapped address into exactly one lane.""" + # Loopback first: ``::1`` is inside ``::/8``, which CPython lists as + # reserved, so a bare ``is_reserved`` test would drag IPv6 loopback into + # NEVER and lock a self-hoster out of their own dev server — while the + # identical 127.0.0.1 stayed approvable. + if addr.is_loopback: + return AddressLane.PRIVATE + if ( + addr.is_link_local + or addr.is_multicast + or addr.is_unspecified + or addr.is_reserved + or any(addr in net for net in _VENDOR_METADATA) + ): + return AddressLane.NEVER + # Deprecated IPv6 site-local (RFC 3879). CPython reports is_global True and + # is_private False for fec0::/10, so without this it would read as ordinary + # public unicast and skip the opt-in entirely — legacy and embedded gear + # still carries it. + if isinstance(addr, ipaddress.IPv6Address) and addr.is_site_local: + return AddressLane.PRIVATE + if addr.is_global: + return AddressLane.PUBLIC + return AddressLane.PRIVATE + + +def classify_address(addr: IPAddress) -> AddressLane: + """Return the single lane *addr* falls in, after unwrapping transitions. + + A wrapper is only as safe as what it routes to, so the worst lane among the + effective addresses wins. + """ + return max(_classify_one(item) for item in effective_addresses(addr)) + + +def reaches_only_loopback(addr: IPAddress) -> bool: + """True when every address *addr* routes to is loopback. + + One spelling of the predicate, because two spellings is how the cleartext + gate and the localhost development lane ended up disagreeing: a Teredo + address carries two IPv4s, so ``any`` and ``all`` give opposite answers for + a wrapper whose server half is loopback and whose client half is not. + """ + return all(item.is_loopback for item in effective_addresses(addr)) + + +class ResolutionError(Exception): + """A hostname could not be resolved to any address the guards can judge.""" + + +def resolve_and_classify(hostname: str, port: int = 0) -> list[tuple[AddressLane, IPAddress]]: + """Resolve *hostname* and classify every address it answers with. + + The single resolution path for every guard in the tree. Each guard used to + hand-roll ``getaddrinfo`` → parse → classify, and the copies had already + drifted on the one thing the loop must get right: which failures are + caught. ``getaddrinfo`` raises ``UnicodeError`` (a ``ValueError``, not an + ``OSError``) out of the IDNA encoder for an over-long label, so a guard + catching only ``socket.gaierror`` lets that escape into its caller. + + Raises :class:`ResolutionError` when the name yields nothing usable — + including an empty answer list, which a caller looping over results would + otherwise treat as "no objections found". + """ + try: + infos = socket.getaddrinfo(hostname, port or None, proto=socket.IPPROTO_TCP) + except (OSError, ValueError) as exc: + raise ResolutionError(f"hostname cannot be resolved ({hostname})") from exc + + classified: list[tuple[AddressLane, IPAddress]] = [] + for info in infos: + raw = str(info[4][0]) + try: + addr = parse_resolved_address(raw) + except ValueError as exc: + raise ResolutionError(f"unable to parse resolved address ({raw})") from exc + classified.append((classify_address(addr), addr)) + if not classified: + raise ResolutionError(f"hostname cannot be resolved ({hostname})") + return classified + + +def describe_address(addr: IPAddress) -> str: + """Render *addr* for an operator-facing refusal, naming what it reaches.""" + embedded = embedded_ipv4(addr) + if not embedded: + return str(addr) + return f"{addr} (routes to {', '.join(str(item) for item in embedded)})" + + +def parse_resolved_address(raw: str) -> IPAddress: + """Parse an address as returned by ``getaddrinfo``, dropping any zone id. + + ``getaddrinfo`` renders a link-local result as ``fe80::1%eth0``. The stdlib + parses that fine, but a scoped address does not compare equal to its + unscoped form, so keeping the zone would make every address the guards + resolve depend on which interface answered — and would leak the interface + name into operator-facing refusal text. The classification is identical + either way, so the zone is dropped at the single point every guard parses. + """ + return ipaddress.ip_address(raw.partition("%")[0]) + + +__all__ = [ + "BLOCKED_HOSTNAMES", + "AddressLane", + "IPAddress", + "ResolutionError", + "classify_address", + "describe_address", + "effective_addresses", + "embedded_ipv4", + "parse_resolved_address", + "reaches_only_loopback", + "resolve_and_classify", +] diff --git a/turnstone/core/oauth_ssrf.py b/turnstone/core/oauth_ssrf.py index 8e091f41..071bb829 100644 --- a/turnstone/core/oauth_ssrf.py +++ b/turnstone/core/oauth_ssrf.py @@ -21,10 +21,16 @@ transport is a future hardening step. from __future__ import annotations import asyncio -import ipaddress -import socket import urllib.parse +from turnstone.core.ip_classify import ( + AddressLane, + ResolutionError, + describe_address, + reaches_only_loopback, + resolve_and_classify, +) + # --------------------------------------------------------------------------- # Trusted-host allowlist for well-known multi-origin IdPs / authorization # servers whose discovery documents legitimately reference endpoints on @@ -134,46 +140,84 @@ def validate_url_no_ssrf( IdP lives in those ranges. Non-public rejections raise the :class:`OAuthSSRFPrivateAddressError` subclass so callers that *have* an opt-in can point the operator at it. - """ - parsed = urllib.parse.urlparse(url) - hostname = parsed.hostname + Both knobs judge an IPv6 transition address (NAT64, 6to4, Teredo, + IPv4-mapped, IPv4-compatible) by the IPv4 address it routes to rather + than by the wrapper — see :mod:`turnstone.core.ip_classify`. A NAT64 + address embedding 192.168.1.5 is therefore treated exactly as + 192.168.1.5 would be: allowed under ``allow_private``, refused + without it. + + ``http://`` on a localhost hostname additionally requires every resolved + address to *be* loopback. ``*.localhost`` is ordinary DNS, so a hostile + authority can point it anywhere; the name alone is not evidence, and + accepting it would put an OIDC token exchange — ``client_secret`` and + authorization code included — on the wire in the clear. + """ + try: + parsed = urllib.parse.urlparse(url) + hostname = parsed.hostname + _ = parsed.port # parsed lazily; raises for an out-of-range value + except ValueError as exc: + # Callers catch OAuthSSRFError only, and the URL can come from a + # hostile MCP server's discovery document — a bare ValueError here + # would escape discovery instead of becoming a clean refusal. + raise OAuthSSRFError(f"endpoint URL is malformed: {url}") from exc + if not hostname: raise OAuthSSRFError(f"endpoint URL has no hostname: {url}") if parsed.username or parsed.password: raise OAuthSSRFError("endpoint URL must not contain embedded credentials (userinfo)") + # Cleartext is permitted only for a genuine loopback dev server. The + # hostname is not evidence of that — ``*.localhost`` is ordinary DNS a + # hostile authority answers however it likes — so the verdict is deferred + # until resolution proves loopback below. Deciding it here on the name + # alone would send an OIDC token exchange, client_secret and all, in the + # clear to whatever address that name happens to return. + http_pending_loopback_proof = False if parsed.scheme != "https": if allow_http and parsed.scheme == "http" and is_localhost(hostname): - pass + http_pending_loopback_proof = True else: raise OAuthSSRFError(f"endpoint URL must use HTTPS (got {parsed.scheme}://): {url}") + # ``resolve_and_classify`` raises rather than returning an empty list, so the + # deferred cleartext verdict below cannot be granted by default when a name + # answers with nothing. try: - addr_infos = socket.getaddrinfo(hostname, None, proto=socket.IPPROTO_TCP) - except socket.gaierror as exc: - raise OAuthSSRFError(f"endpoint hostname cannot be resolved: {hostname}") from exc + classified = resolve_and_classify(hostname) + except ResolutionError as exc: + raise OAuthSSRFError(f"endpoint {exc}: {url}") from exc - for _family, _type, _proto, _canonname, sockaddr in addr_infos: - try: - addr = ipaddress.ip_address(sockaddr[0]) - except ValueError as exc: + for lane, addr in classified: + # Ordered so the refusal names the real problem. Deciding the scheme + # first reports a metadata address as merely "must use HTTPS", and an + # operator acting on that re-registers the same endpoint over TLS + # without ever learning it pointed at the instance metadata service. + if lane is AddressLane.NEVER: raise OAuthSSRFError( - f"endpoint hostname resolved to invalid IP {sockaddr[0]!r}: {hostname}" - ) from exc - if addr.is_global or is_localhost(hostname): + f"endpoint URL resolves to a link-local/multicast/unspecified/" + f"reserved/metadata address ({describe_address(addr)}), refused " + f"even with private addresses allowed: {url}" + ) + # Cleartext must reach loopback and nothing else, on every record. + if http_pending_loopback_proof and not reaches_only_loopback(addr): + raise OAuthSSRFError( + f"endpoint URL must use HTTPS: {hostname} does not resolve to " + f"loopback ({describe_address(addr)}): {url}" + ) + if lane is AddressLane.PUBLIC or allow_private: continue - if allow_private: - if addr.is_link_local or addr.is_multicast or addr.is_unspecified or addr.is_reserved: - raise OAuthSSRFError( - f"endpoint URL resolves to a link-local/multicast/" - f"unspecified/reserved address ({addr}), refused even " - f"with private addresses allowed: {url}" - ) + # The localhost development lane, gated on the RESOLVED address rather + # than the hostname — and on the same predicate as the cleartext gate, + # since two spellings of "reaches loopback" disagree for a Teredo + # wrapper whose server half is loopback and whose client half is not. + if is_localhost(hostname) and reaches_only_loopback(addr): continue raise OAuthSSRFPrivateAddressError( - f"endpoint URL resolves to non-public address ({addr}): {url}" + f"endpoint URL resolves to non-public address ({describe_address(addr)}): {url}" ) return parsed diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 69d3c387..66310ac7 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -62,6 +62,7 @@ from turnstone.core.background_shells import ( from turnstone.core.config import get_searxng_engines, get_searxng_url, get_workspace_dir from turnstone.core.deadline import StreamAbortRef from turnstone.core.edit import find_occurrences, pick_nearest +from turnstone.core.ip_classify import AddressLane from turnstone.core.log import get_logger from turnstone.core.lowering import ( TIMEOUT_OUTCOME_CLAUSE, @@ -237,7 +238,7 @@ from turnstone.core.trajectory import ( turns_from_dicts, ) from turnstone.core.watch import WATCH_REMINDER_OPTIONAL_KEYS -from turnstone.core.web import check_ssrf, fetch_with_ssrf_guard, strip_html +from turnstone.core.web import fetch_with_ssrf_guard, screen_url, strip_html from turnstone.core.workstream import ( INTERJECTION_CAP_CHARS, PENDING_SENDS_MAX, @@ -2047,34 +2048,66 @@ def _notify_auth_headers() -> dict[str, str]: return header -def _screen_tool_url(url: str, allow_private_network: bool) -> tuple[str | None, bool]: +def _screen_tool_url(url: str, allow_private_network: bool) -> tuple[str | None, bool, bool]: """SSRF-screen a tool's target URL under the operator's private-network opt-in. ``allow_private_network`` is the live ``tools.allow_private_network`` setting (admin Settings → Tools; DB-backed, hot-toggleable — the caller - reads it per prepare). Returns ``(error, private_origin)``. ``error`` is - the rejection text (``None`` = proceed); a private-address rejection names - the setting so a self-hosted operator learns the knob from the refusal - itself. ``private_origin`` is True when the target NAMES a private - address the operator opted into: the approval header tags it, and the - guarded fetch skips per-hop screening for that chain — the gate approved a - private URL, so its redirects are the operator's own network. Public - origins never set it, keeping the public→private redirect bounce blocked - regardless of the opt-in. + reads it per prepare). Returns ``(error, private_origin, private_block)``. + + ``error`` is the rejection text, or ``None`` to proceed. A private-address + rejection names the setting so a self-hosted operator learns the knob from + the refusal itself. + + ``private_block`` means the target is in the PRIVATE lane — the lane the + opt-in governs — so callers can label the request without re-deriving that + from the refusal wording. It is about the LANE, not about the outcome: it + is True both when the opt-in cleared the target and when the absence of the + opt-in refused it. Callers use it for the "(private network)" header. + + ``private_origin`` is True when the operator opted in and EVERY address the + target resolves to is private. A hostname answering with both a private + and a public record is refused instead, naming the remedy: the connection + may land on either record, so approving it would not describe where the + fetch actually goes. Public origins never set it, keeping the + public→private redirect bounce blocked regardless of the opt-in. + + Because a granted chain therefore starts wholly private, + ``fetch_with_ssrf_guard`` never has to make an exception for the origin — + it revokes private-hop permission after any hop that is not wholly private + and needs no notion of an "approved host". """ - ssrf_err = check_ssrf(url) - if not ssrf_err: - return None, False - is_private_block = "private/internal address" in ssrf_err + screen = screen_url(url) + if screen.error is None: + return None, False, False + # The LANE is the contract, not the wording. Substring-matching the refusal + # text made a docstring rewrap able to silently open the opt-in lane for + # cloud metadata, with nothing failing at the edit site. + is_private_block = screen.lane is AddressLane.PRIVATE + ssrf_err = screen.error if is_private_block and allow_private_network: - return None, True + if screen.all_private: + return None, True, True + # A hostname answering with BOTH a private and a public address cannot + # be approved as a private-network target: the connection may land on + # either, so the approval would not describe where the fetch goes, and + # granting the chain private-hop permission on that basis let an + # attacker-controlled public record steer the fetcher into the network. + # Refused with the remedy, rather than silently narrowed. + return ( + f"Error: {ssrf_err}. That hostname also resolves to a public address, so it" + " cannot be approved as a private-network target — point the tool at the" + " LAN address directly.", + False, + True, + ) hint = "" if is_private_block: hint = ( " Enable 'tools.allow_private_network' in the console" " (Settings → Tools) to allow fetching private-network addresses." ) - return f"Error: {ssrf_err}.{hint}", False + return f"Error: {ssrf_err}.{hint}", False, is_private_block def _tool_turn_meta( @@ -14897,13 +14930,15 @@ class ChatSession: # SSRF screen \u2014 a NAMED private address is approvable under the # tools.allow_private_network opt-in (the header tags it so the # operator approves it as what it is). - screen_err, private_origin = _screen_tool_url(url, self._allow_private_network()) + screen_err, private_origin, private_block = _screen_tool_url( + url, self._allow_private_network() + ) if screen_err: return { "call_id": call_id, "func_name": "web_fetch", "header": "\u2717 web_fetch: blocked (private network)" - if "private/internal" in screen_err + if private_block else "\u2717 web_fetch: blocked", "preview": f" {url}", "needs_approval": False, @@ -14969,13 +15004,15 @@ class ChatSession: # Same opt-in lane as web_fetch: a named private address is # approvable under tools.allow_private_network, tagged so the # operator approves it as what it is. - screen_err, private_origin = _screen_tool_url(target, self._allow_private_network()) + screen_err, private_origin, private_block = _screen_tool_url( + target, self._allow_private_network() + ) if screen_err: return { "call_id": call_id, "func_name": "open_preview", "header": "✗ open_preview: blocked (private network)" - if "private/internal" in screen_err + if private_block else "✗ open_preview: blocked", "preview": f" {target}", "needs_approval": False, diff --git a/turnstone/core/settings_registry.py b/turnstone/core/settings_registry.py index 6ba9585b..959031e8 100644 --- a/turnstone/core/settings_registry.py +++ b/turnstone/core/settings_registry.py @@ -235,7 +235,10 @@ def _build_registry() -> dict[str, SettingDef]: "can be approved instead of being refused outright — the approval prompt " "marks it as a private-network request. A public site that redirects into " "your private network is still refused either way: that address never " - "appeared in the approval prompt, so it is never fetched.", + "appeared in the approval prompt, so it is never fetched. Cloud metadata " + "endpoints and link-local, multicast and reserved addresses stay refused " + "even with this on, including as a redirect target from a private address " + "you approved — no legitimate service of yours lives there.", ), SettingDef( "tools.search", diff --git a/turnstone/core/web.py b/turnstone/core/web.py index a8927fd3..b3170bd2 100644 --- a/turnstone/core/web.py +++ b/turnstone/core/web.py @@ -1,13 +1,20 @@ """Web utilities — HTML stripping, SSRF protection, and the guarded fetch.""" -import ipaddress +import dataclasses import re -import socket from html import unescape as _html_unescape from urllib.parse import urlparse import httpx +from turnstone.core.ip_classify import ( + BLOCKED_HOSTNAMES, + AddressLane, + ResolutionError, + describe_address, + resolve_and_classify, +) + _RE_INVISIBLE = re.compile( r"<(script|style|template|noscript)\b[^>]*>.*?", re.DOTALL | re.IGNORECASE, @@ -92,36 +99,86 @@ def strip_html(html: str) -> str: return text.strip() -def check_ssrf(url: str) -> str | None: - """Return error string if URL resolves to a private/link-local address, else None. +@dataclasses.dataclass(frozen=True) +class UrlScreen: + """The verdict on one URL: which lane, why, and whether it is wholly private.""" - Checks both IPv4 and IPv6 addresses via getaddrinfo to prevent bypasses - using IPv6 loopback (``::1``), link-local (``fe80::``), or unique-local - (``fd00::``/``fc00::``) addresses. + lane: AddressLane + error: str | None + all_private: bool + """True when EVERY resolved address is private. + + ``lane`` is the worst address, which is the right basis for refusing. It is + the wrong basis for treating a chain as "inside the operator's network": a + hostname with both a private and a public A record folds to PRIVATE, but the + connection may land on the public record, so the approval the operator gave + does not describe where the fetch actually goes. + """ + + +def screen_url(url: str) -> UrlScreen: + """Classify every address *url* resolves to and return the worst lane. + + The test is "globally routable", not "not in a private range". Those are + not complements: CGNAT (100.64.0.0/10, RFC 6598 shared address space — + where overlay VPNs commonly assign internal hosts) is neither private nor + global, so a denylist let it through. IPv6 transition addresses are judged + by the IPv4 they route to rather than by the wrapper (see + :mod:`turnstone.core.ip_classify`). + + Every resolved address is classified and the *worst* lane wins. Returning + on the first offending address would let a hostname whose first A record is + merely private mask a second record that is link-local: the caller would + see the approvable lane, and under the operator opt-in the whole hostname — + including the record it never looked at — would be fetched. + + Fails closed. A resolution failure is a refusal, not a pass: the fetch that + follows resolves again, so an authority that answers the guard's query with + SERVFAIL and the fetch's query with an internal address would otherwise + turn the guard off for that hop. Malformed URLs are refusals too — every + exception path returns a verdict rather than raising, because the callers + screen model-supplied URLs and one of them prepares tools outside any + ``try``. """ try: parsed = urlparse(url) hostname = parsed.hostname - if not hostname: - return "Invalid URL: no hostname" - # Resolve all address families (IPv4 + IPv6) - results = socket.getaddrinfo(hostname, parsed.port or 80, proto=socket.IPPROTO_TCP) - for _family, _type, _proto, _canonname, sockaddr in results: - addr = str(sockaddr[0]) - # Strip IPv6 zone/scope identifier (e.g. "fe80::1%lo0") - addr_clean = addr.split("%", 1)[0] if "%" in addr else addr - try: - ip = ipaddress.ip_address(addr_clean) - except ValueError: - return f"Blocked: unable to parse resolved address ({addr})" - # Normalize IPv4-mapped IPv6 (e.g. ::ffff:127.0.0.1) - if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: - ip = ip.ipv4_mapped - if ip.is_private or ip.is_loopback or ip.is_link_local: - return f"Blocked: URL resolves to private/internal address ({addr})" - except (socket.gaierror, OSError): - pass # DNS failure — let the actual fetch handle it - return None + # Touched, not used: ``urlsplit.port`` parses lazily and raises for an + # out-of-range value, which must become a refusal rather than escape. + # It is not passed to resolution — a numeric service does not change + # which addresses come back, and classification looks only at those. + _ = parsed.port + except ValueError: + return UrlScreen(AddressLane.NEVER, f"Blocked: malformed URL ({url})", False) + if not hostname: + return UrlScreen(AddressLane.NEVER, "Invalid URL: no hostname", False) + if hostname.lower() in BLOCKED_HOSTNAMES: + return UrlScreen( + AddressLane.NEVER, f"Blocked: URL names a metadata host ({hostname})", False + ) + + try: + classified = resolve_and_classify(hostname) + except ResolutionError as exc: + return UrlScreen(AddressLane.NEVER, f"Blocked: {exc}", False) + + worst, offender = max(classified, key=lambda item: item[0]) + all_private = all(lane is AddressLane.PRIVATE for lane, _ in classified) + + if worst is AddressLane.PUBLIC: + return UrlScreen(AddressLane.PUBLIC, None, False) + if worst is AddressLane.NEVER: + return UrlScreen( + worst, + "Blocked: URL resolves to a link-local/multicast/unspecified/" + f"reserved/metadata address ({describe_address(offender)})", + all_private, + ) + return UrlScreen( + worst, + f"Blocked: URL resolves to private/internal address ({describe_address(offender)})", + all_private, + ) FETCH_BYTE_CEILING = 32 * 1024 * 1024 @@ -148,13 +205,19 @@ def fetch_with_ssrf_guard( executing the private-network request even if the response is later discarded. Here each hop's URL is screened BEFORE its request is issued. - ``allow_private_origin`` is the ``[tools] allow_private_network`` lane: - the CALLER sets it only when the operator opted in AND the ORIGINAL - target itself named a private address — the approval gate then saw and - approved that private URL, so its redirect chain is the operator's own - network and hop screening is skipped. A public origin never sets it, - so a public site bouncing the fetcher into private space stays blocked - regardless of the opt-in: that hop was never shown to the approval gate. + EVERY hop is screened, in every mode. ``allow_private_origin`` widens + which lanes are acceptable, it does not turn screening off: the caller + sets it only when the operator opted in AND the original target itself + named a private address, so the approval gate saw and approved that + private URL. + + The permission is also revoked the moment the chain leaves that network. + Once any hop resolves PUBLIC, private hops are refused for the rest of the + chain — the operator approved their own hosts, not whatever a public site + picks next, so ``private -> public -> private`` cannot be used to steer the + fetcher into internal endpoints of an attacker's choosing. The NEVER lane + is absolute throughout: approving a private origin says "this is my + network", which is not a claim about the cloud metadata endpoint. The body is streamed under a *max_bytes* budget rather than buffered blind — ``client.get()`` would read an unbounded body into memory before @@ -173,16 +236,29 @@ def fetch_with_ssrf_guard( stays the caller's call. """ current = url + private_allowed = allow_private_origin with httpx.Client( headers={"User-Agent": user_agent}, timeout=timeout, follow_redirects=False, ) as client: for _hop in range(max_redirects + 1): - if not allow_private_origin: - ssrf_err = check_ssrf(current) - if ssrf_err: - raise ValueError(ssrf_err) + screen = screen_url(current) + if screen.lane is AddressLane.NEVER: + raise ValueError(screen.error) + if screen.lane is AddressLane.PRIVATE and not private_allowed: + raise ValueError(screen.error) + if not screen.all_private: + # The chain can no longer be shown to be inside the operator's + # network, so private hops stop being allowed from here on. + # There is deliberately no exemption for the origin host: an + # earlier attempt to keep one let a public hop steer the fetcher + # back into the approved host at a path of its choosing, and + # made the grant re-entrant across same-host redirects with + # fresh DNS each time. The caller refuses a mixed-record origin + # outright instead, so a chain that gets here wholly private + # stays that way or ends. + private_allowed = False with client.stream("GET", current) as resp: if resp.status_code in _REDIRECT_STATUSES: location = resp.headers.get("location") diff --git a/turnstone/doctor.py b/turnstone/doctor.py index f9f88b1b..b8427603 100644 --- a/turnstone/doctor.py +++ b/turnstone/doctor.py @@ -239,20 +239,37 @@ def _reject_option(value: str) -> str | None: def _assert_safe_http_url(url: str) -> None: """Reject a model-supplied URL that isn't safe to fetch from this host. - Restricts the scheme to http/https (no ``file://``/``ftp://``) and blocks the - cloud link-local metadata range (``169.254.0.0/16`` / ``metadata.google.internal``) + Restricts the scheme to http/https (no ``file://``/``ftp://``) and refuses + the NEVER lane — cloud metadata and link-local/multicast/reserved space — so an LLM-driven probe can't be steered at the instance-metadata service. - Loopback and private cluster IPs stay allowed — probing - ``http://localhost:PORT/health`` and private node URLs is the job. Raises - ``ValueError`` when the URL is unsafe. Shared by every tool that fetches a - model-supplied URL (``http_health``, ``node_health``, ``check_llm_backend``). + Loopback and private cluster IPs stay allowed: probing + ``http://localhost:PORT/health`` and private node URLs is the job, so this + guard accepts the PRIVATE lane where the fetch tools gate it behind an + operator opt-in. Raises ``ValueError`` when the URL is unsafe. Shared by + every tool that fetches a model-supplied URL (``http_health``, + ``node_health``, ``check_llm_backend``). + + Classification goes through :mod:`turnstone.core.ip_classify`, which + RESOLVES the hostname. The previous ``host.startswith("169.254.")`` string + test never resolved, so any DNS name pointing at the metadata service — or + any IPv6 transition address wrapping it, e.g. + ``::ffff:169.254.169.254`` — walked straight through. """ - parts = urllib.parse.urlsplit(url) + from turnstone.core.ip_classify import AddressLane + from turnstone.core.web import screen_url + + try: + parts = urllib.parse.urlsplit(url) + except ValueError as exc: + raise ValueError(f"refusing malformed URL: {url!r}") from exc if parts.scheme not in ("http", "https"): raise ValueError(f"refusing non-http(s) URL: {url!r}") - host = (parts.hostname or "").lower() - if host.startswith("169.254.") or host == "metadata.google.internal": - raise ValueError(f"refusing link-local/metadata host: {host!r}") + # One screen, shared with the fetch tools: parse, resolve, classify, fold. + # This guard accepts the PRIVATE lane where they gate it behind an operator + # opt-in — probing a private node URL is the job — so only NEVER refuses. + screen = screen_url(url) + if screen.lane is AddressLane.NEVER: + raise ValueError(f"refusing unsafe URL: {screen.error}") def _http_get_json(url: str, timeout: float = 5.0) -> Any: