diff --git a/src/url_safety.py b/src/url_safety.py index c8bf86327..e0f66cfeb 100644 --- a/src/url_safety.py +++ b/src/url_safety.py @@ -16,6 +16,12 @@ break the primary use case. What it *always* rejects: For exposed multi-tenant deployments, set ``EMBEDDING_BLOCK_PRIVATE_IPS=true`` to additionally reject all private and loopback targets (full SSRF lockdown). + +On a DNS64/NAT64 network an IPv4-only host resolves to the RFC 6052 Well-Known +Prefix ``64:ff9b::/96``. Such an address is decoded to the IPv4 destination the +translator will actually contact, and that destination is then judged under the +strict policy — so the prefix reaches public IPv4 but never tunnels to loopback, +private, shared or link-local space. """ import ipaddress @@ -33,6 +39,39 @@ ALLOWED_SCHEMES = ("http", "https") # versions for other special ranges. _SHARED_ADDRESS_SPACE_V4 = ipaddress.ip_network("100.64.0.0/10") +# RFC 6052 §2.1 Well-Known Prefix for IPv4/IPv6 address translation (NAT64). +# An address inside exactly this /96 is not a destination in its own right: the +# low 32 bits carry the IPv4 address the translator will actually contact. On a +# DNS64/NAT64 network every public IPv4-only host resolves this way, so judging +# the outer IPv6 (which CPython reports as ``is_reserved``) would reject the +# whole public internet while telling us nothing about the real target. +# +# RFC 6052 §3.1 allows the Well-Known Prefix to represent *only* globally +# routable IPv4. The embedded destination is therefore always evaluated under +# the strict policy, whatever ``block_private`` the caller passed: the prefix +# must never become a path to loopback, private, shared, link-local, multicast, +# unspecified or otherwise non-global space. +# +# Deliberately exact. Network-specific prefixes carry locally assigned meaning +# and are NOT decoded here — notably 64:ff9b:1::/48 (RFC 8215), which this /96 +# membership test excludes and which stays rejected as reserved. +_NAT64_WELL_KNOWN_PREFIX_V6 = ipaddress.ip_network("64:ff9b::/96") + + +def _nat64_well_known_embedded_ipv4( + ip: ipaddress._BaseAddress, +) -> Optional[ipaddress.IPv4Address]: + """Return the IPv4 target embedded in an RFC 6052 Well-Known-Prefix address. + + ``None`` when ``ip`` is not inside ``64:ff9b::/96``, i.e. when no IPv4 + destination may be inferred from it. + """ + if not isinstance(ip, ipaddress.IPv6Address): + return None + if ip not in _NAT64_WELL_KNOWN_PREFIX_V6: + return None + return ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF) + def _default_resolver(host: str) -> List[str]: """Resolve a hostname to the list of IP strings it maps to (A + AAAA).""" @@ -44,6 +83,15 @@ def _classify(ip: ipaddress._BaseAddress, *, block_private: bool) -> Optional[st # IPv4-mapped IPv6 (e.g. ::ffff:169.254.169.254) — judge the embedded v4. if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: ip = ip.ipv4_mapped + else: + # RFC 6052 Well-Known Prefix — judge the IPv4 destination the NAT64 + # translator will contact, always under the strict policy. + translated = _nat64_well_known_embedded_ipv4(ip) + if translated is not None: + reason = _classify(translated, block_private=True) + if reason: + return f"NAT64 translated destination blocked: {reason}" + return None if ip.is_link_local: return f"link-local address blocked (SSRF metadata risk): {ip}" if ip.is_multicast or ip.is_reserved or ip.is_unspecified: diff --git a/tests/test_review_20260923_fixes.py b/tests/test_review_20260923_fixes.py index 9c1256c25..6d6cb1200 100644 --- a/tests/test_review_20260923_fixes.py +++ b/tests/test_review_20260923_fixes.py @@ -187,6 +187,79 @@ def test_redirect_limit_is_bounded(monkeypatch): assert len(called_urls) == search_core.MAX_SCHOLARLY_REDIRECTS + 1 +def test_redirect_does_not_replay_query_parameters_to_the_new_destination(monkeypatch): + """The original query must not be re-sent to a redirect target. + + The params belong to the endpoint that was asked for. Replaying them across + a redirect would hand the search terms to whatever host the redirect names. + """ + calls = [] + + def fake_get(url, **kwargs): + params = kwargs.get("params") + calls.append((url, params)) + # Mirror real httpx: response.url carries the query that was sent. + req = httpx.Request("GET", httpx.URL(url, params=params or {})) + if len(calls) == 1: + return httpx.Response( + 302, headers={"Location": "https://redirect.openalex.test/v2"}, request=req + ) + return httpx.Response(200, json={"results": []}, request=req) + + monkeypatch.setattr(httpx, "get", fake_get) + monkeypatch.setattr("src.url_safety.check_outbound_url", lambda url, **kw: (True, "")) + + result = search_core._scholarly_api_get( + OPENALEX_API_URL, {"search": "confidential-title"} + ) + + assert result is not None + assert len(calls) == 2 + second_url, second_params = calls[1] + assert second_url == "https://redirect.openalex.test/v2" + assert second_params is None + assert "confidential-title" not in second_url + + +def test_relative_redirect_resolves_against_the_responding_url(monkeypatch): + """A relative Location resolves against the URL that answered, not the origin.""" + calls = [] + + def fake_get(url, **kwargs): + calls.append(url) + req = httpx.Request("GET", httpx.URL(url, params=kwargs.get("params") or {})) + if len(calls) == 1: + return httpx.Response( + 302, headers={"Location": "https://mirror.arxiv.test/api/v1/query"}, request=req + ) + if len(calls) == 2: + # Relative hop: must resolve against mirror.arxiv.test, not arxiv. + return httpx.Response(302, headers={"Location": "../v2/query"}, request=req) + return httpx.Response(200, text="", request=req) + + monkeypatch.setattr(httpx, "get", fake_get) + monkeypatch.setattr("src.url_safety.check_outbound_url", lambda url, **kw: (True, "")) + + result = search_core._scholarly_api_get(ARXIV_API_URL, {"search_query": "x"}) + + assert result is not None + assert calls[2] == "https://mirror.arxiv.test/api/v2/query" + + +def test_http_error_status_degrades_to_no_results(monkeypatch): + """A 500 from a scholarly API must degrade to no results, not raise.""" + + def fake_get(url, **kwargs): + req = httpx.Request("GET", url) + return httpx.Response(500, text="upstream exploded", request=req) + + monkeypatch.setattr(httpx, "get", fake_get) + monkeypatch.setattr("src.url_safety.check_outbound_url", lambda url, **kw: (True, "")) + + assert search_core._arxiv_title_results("Attention Is All You Need") == [] + assert search_core._openalex_title_results("Attention Is All You Need") == [] + + def test_exhausted_budget_skips_the_request(monkeypatch): """Once the chain's budget is spent, later hops are skipped, not retried.""" calls = [] @@ -398,3 +471,52 @@ def test_ordinary_draft_is_unaffected(): from routes.editor_draft_routes import _dump_payload assert _dump_payload({"layers": []}) == '{"layers":[]}' + + +def test_route_class_guard_applies_only_to_body_bearing_methods(): + """The route class must not change GET/DELETE semantics. + + ``EditorDraftRoute`` is attached to the whole editor-draft router, so the + bodyless routes run through it too. They must reach their handler + untouched — the early ceiling belongs to the methods that carry a draft. + """ + from fastapi import APIRouter + + from routes.editor_draft_routes import EditorDraftRoute + + router = APIRouter(route_class=EditorDraftRoute) + + @router.get("/probe") + async def _get_probe(): + return {"reached": "get"} + + @router.delete("/probe") + async def _delete_probe(): + return {"reached": "delete"} + + @router.post("/probe") + async def _post_probe(): + return {"reached": "post"} + + app = FastAPI() + app.include_router(router) + client = TestClient(app) + + oversized = { + "content-type": "application/json", + "content-length": str(EDITOR_DRAFT_MAX_BYTES + 1), + } + + # Bodyless methods are unaffected even when a bogus huge length is declared. + for method, expected in (("GET", "get"), ("DELETE", "delete")): + response = client.request(method, "/probe", content=b"{}", headers=oversized) + assert response.status_code == 200, (method, response.text) + assert response.json() == {"reached": expected} + + # The same declaration on the body-bearing method is refused. + assert client.post("/probe", content=b"{}", headers=oversized).status_code == 413 + + # ...and an ordinary POST still reaches the handler. + ordinary = client.post("/probe", json={"layers": []}) + assert ordinary.status_code == 200 + assert ordinary.json() == {"reached": "post"} diff --git a/tests/test_service_search_provider_guards.py b/tests/test_service_search_provider_guards.py index 29b1b36c4..d7735ec32 100644 --- a/tests/test_service_search_provider_guards.py +++ b/tests/test_service_search_provider_guards.py @@ -4,6 +4,7 @@ The old src.search provider path aliases this module; these tests pin the behavior at the single implementation point. """ +import ipaddress import sys import pytest @@ -11,6 +12,34 @@ from services.search import core from services.search import providers +@pytest.fixture(autouse=True) +def _deterministic_outbound_dns(monkeypatch): + """Keep the outbound-URL policy running, but take it off live DNS. + + ``_scholarly_api_get`` validates each destination through + ``check_outbound_url`` before it reaches the mocked ``httpx`` transport, and + that check resolves the hostname. Without this stub these tests depend on + real DNS: on a DNS64/NAT64 network an IPv4-only host such as + export.arxiv.org resolves to ``64:ff9b::``, so the lookup is judged on a + synthesised address that has nothing to do with the title-resolution, + fallback, request-parameter and ordering behaviour asserted here. + + Only name resolution is replaced. The policy itself still runs in full, and + a literal-IP host still resolves to itself, so a test that points at a + prohibited address is still genuinely rejected. This is deliberately not a + stand-in for SSRF/NAT64 coverage, which lives in tests/test_url_safety.py + and tests/test_review_20260923_fixes.py. + """ + + def _resolve(host: str): + try: + return [str(ipaddress.ip_address(host))] + except ValueError: + return ["93.184.216.34"] # public, policy-clean + + monkeypatch.setattr("src.url_safety._default_resolver", _resolve) + + def test_html_transport_fallback_preserves_query_constraints(monkeypatch): seen = [] class Response: diff --git a/tests/test_url_safety.py b/tests/test_url_safety.py index d82d38878..efe0c7bb8 100644 --- a/tests/test_url_safety.py +++ b/tests/test_url_safety.py @@ -115,3 +115,146 @@ def test_resolver_skips_invalid_values_but_accepts_public_ip(): assert ok is True assert reason == "ok" + + +# --- RFC 6052 Well-Known Prefix (DNS64 / NAT64) --------------------------- +# +# On a DNS64/NAT64 network an IPv4-only host resolves to 64:ff9b::. The +# effective destination is the embedded IPv4 address, so that is what the +# policy must judge. RFC 6052 §3.1 permits the Well-Known Prefix to represent +# only globally-routable IPv4, so the embedded target is always held to the +# strict policy — the prefix must never tunnel past the SSRF guard. + + +def _nat64(v4: str) -> str: + """Render the RFC 6052 Well-Known-Prefix form of an IPv4 address.""" + import ipaddress + + packed = int(ipaddress.IPv4Address(v4)) + return str(ipaddress.IPv6Address(int(ipaddress.IPv6Address("64:ff9b::")) | packed)) + + +def test_nat64_well_known_prefix_allows_public_ipv4_destination(): + # The reproduced failure: export.arxiv.org resolves to 64:ff9b::924b:5b2a + # under DNS64. The embedded 146.75.91.42 is public, so the URL is allowed. + res = _resolver({"export.arxiv.org": [_nat64("146.75.91.42")]}) + ok, reason = check_outbound_url("https://export.arxiv.org/api/query", resolver=res) + assert ok is True, reason + # ...including under full lockdown, where scholarly lookups actually run. + ok, reason = check_outbound_url( + "https://export.arxiv.org/api/query", block_private=True, resolver=res + ) + assert ok is True, reason + + +def test_nat64_well_known_prefix_blocks_embedded_loopback(): + res = _resolver({"evil.example": [_nat64("127.0.0.1")]}) + for strict in (False, True): + ok, reason = check_outbound_url( + "http://evil.example/", block_private=strict, resolver=res + ) + assert ok is False, strict + assert "NAT64" in reason and "127.0.0.1" in reason + + +def test_nat64_well_known_prefix_blocks_embedded_metadata_address(): + # The headline SSRF vector must not become reachable through DNS64. + res = _resolver({"evil.example": [_nat64("169.254.169.254")]}) + for strict in (False, True): + ok, reason = check_outbound_url( + "http://evil.example/latest/meta-data/", block_private=strict, resolver=res + ) + assert ok is False, strict + assert "link-local" in reason and "169.254.169.254" in reason + + +def test_nat64_well_known_prefix_blocks_embedded_private_ipv4_under_strict(): + res = _resolver({"evil.example": [_nat64("192.168.1.50")]}) + ok, reason = check_outbound_url( + "http://evil.example/", block_private=True, resolver=res + ) + assert ok is False + assert "NAT64" in reason and "192.168.1.50" in reason + + +def test_nat64_well_known_prefix_blocks_embedded_shared_space_under_strict(): + # RFC 6598 shared/CGNAT space is not globally routable, so the Well-Known + # Prefix may not represent it. + res = _resolver({"evil.example": [_nat64("100.64.0.1")]}) + ok, reason = check_outbound_url( + "http://evil.example/", block_private=True, resolver=res + ) + assert ok is False + assert "NAT64" in reason and "100.64.0.1" in reason + + +def test_nat64_well_known_prefix_blocks_embedded_non_global_ipv4(): + # Multicast, unspecified and other non-global space must not be reachable + # through the Well-Known Prefix, whatever block_private says. + for v4 in ("224.0.0.1", "0.0.0.0", "198.18.0.1", "192.0.2.10", "240.0.0.1"): + res = _resolver({"evil.example": [_nat64(v4)]}) + for strict in (False, True): + ok, reason = check_outbound_url( + "http://evil.example/", block_private=strict, resolver=res + ) + assert ok is False, (v4, strict) + assert v4 in reason, (v4, reason) + + +def test_network_specific_nat64_prefixes_are_not_decoded(): + # RFC 8215 64:ff9b:1::/48 and arbitrary network-specific prefixes carry + # locally assigned semantics, so no embedded IPv4 may be inferred from them. + # 64:ff9b:1::8ef:5b2a would decode to the *public* 8.239.91.42; it stays + # rejected on the outer IPv6 address, which proves no decoding happened. + for addr in ("64:ff9b:1::8ef:5b2a", "64:ff9b:1::7f00:1"): + res = _resolver({"svc.example": [addr]}) + ok, reason = check_outbound_url("http://svc.example/", resolver=res) + assert ok is False, addr + assert "NAT64" not in reason, addr + assert addr in reason, addr + + # A documentation-range prefix is judged as an ordinary IPv6 address under + # the existing policy (local-first by default, blocked under lockdown) and + # is never reinterpreted as carrying an IPv4 destination. + res = _resolver({"svc.example": ["2001:db8:1::8ef:5b2a"]}) + ok, reason = check_outbound_url("http://svc.example/", resolver=res) + assert ok is True, reason + ok, reason = check_outbound_url( + "http://svc.example/", block_private=True, resolver=res + ) + assert ok is False + assert "NAT64" not in reason and "8.239.91.42" not in reason + + +def test_ordinary_ipv6_behaviour_is_unchanged(): + # A global unicast IPv6 host is allowed in both modes. + GLOBAL6 = _resolver({"v6.example": ["2606:2800:220:1:248:1893:25c8:1946"]}) + assert check_outbound_url("https://v6.example/", resolver=GLOBAL6)[0] is True + assert ( + check_outbound_url("https://v6.example/", block_private=True, resolver=GLOBAL6)[0] + is True + ) + + # ULA is local-first allowed by default, blocked under lockdown. + ULA = _resolver({"ula.example": ["fd00::1"]}) + assert check_outbound_url("http://ula.example/", resolver=ULA)[0] is True + ok, reason = check_outbound_url( + "http://ula.example/", block_private=True, resolver=ULA + ) + assert ok is False and "private" in reason + + # fe80::/10 is always blocked. + LL6 = _resolver({"ll.example": ["fe80::1"]}) + ok, reason = check_outbound_url("http://ll.example/", resolver=LL6) + assert ok is False and "link-local" in reason + + # Pre-existing behaviour, unchanged here: CPython reports ::1 as + # is_reserved, so IPv6 loopback is rejected in both modes (unlike 127.0.0.1, + # which the local-first default allows). + LOOP6 = _resolver({"loop6.example": ["::1"]}) + for strict in (False, True): + ok, reason = check_outbound_url( + "http://loop6.example/", block_private=strict, resolver=LOOP6 + ) + assert ok is False, strict + assert "NAT64" not in reason