From 554705d1fd3ea0e841c98e3d73ba870b20c6eec5 Mon Sep 17 00:00:00 2001 From: Dizhan Xue Date: Wed, 14 Oct 2026 07:57:21 +0000 Subject: [PATCH] refactor: share loopback IP checks across URL and RPC policies Move canonical loopback IP classification into security.hosts and reuse it from the browser origin and URL plaintext policies. Keep each caller's local-name rules and the browser Origin/Host check in place. Cover canonical addresses, legacy spellings, local aliases and forwarded sign-ins so sharing the IP predicate preserves both policies. --- raven/rpc/transports/ws.py | 16 ++++++---------- raven/security/hosts.py | 7 +++++++ raven/security/urls.py | 9 ++------- tests/test_rpc_transport.py | 36 ++++++++++++++++++++++++++---------- tests/test_security_hosts.py | 36 ++++++++++++++++++++++++++++++++++++ tests/test_security_urls.py | 12 ++++++++++++ 6 files changed, 89 insertions(+), 27 deletions(-) diff --git a/raven/rpc/transports/ws.py b/raven/rpc/transports/ws.py index e8d680e58..9310683b8 100644 --- a/raven/rpc/transports/ws.py +++ b/raven/rpc/transports/ws.py @@ -16,7 +16,6 @@ from __future__ import annotations import asyncio -import ipaddress import json import secrets import time @@ -28,6 +27,8 @@ from aiohttp import WSCloseCode, WSMsgType, web from loguru import logger +from raven.security.hosts import is_loopback_ip + DEFAULT_PORT = 18792 _PORT_PROBE_SPAN = 20 # A sign-in nonce is a live credential until redeemed. The browser may be @@ -94,20 +95,15 @@ def cookie_name(port: int) -> str: def _loopback_origin(origin: str) -> bool: """Whether a browser origin names this machine: ``localhost`` or a loopback IP. - Both parses raise on input a client other than a browser can send: - ``urlsplit`` on an IPv6 bracket left open, ``ip_address`` on any name, which - is how the rebinding case lands here. Either one is a refusal, not an error. + Local aliases and trailing dots stay outside the browser sign-in policy, + even though URL policies may allow them. An unclosed IPv6 bracket is a + refusal too, rather than an error from ``urlsplit``. """ try: hostname = urlsplit(origin).hostname except ValueError: return False - if hostname == "localhost": - return True - try: - return hostname is not None and ipaddress.ip_address(hostname).is_loopback - except ValueError: - return False + return hostname is not None and (hostname == "localhost" or is_loopback_ip(hostname)) _COOKIE_MAX_AGE_S = 30 * 24 * 60 * 60 diff --git a/raven/security/hosts.py b/raven/security/hosts.py index 9a4c51789..0b5835399 100644 --- a/raven/security/hosts.py +++ b/raven/security/hosts.py @@ -83,6 +83,12 @@ def as_ip(host: str) -> Address | None: return None +def is_loopback_ip(host: str) -> bool: + """Whether a canonical IP literal denotes loopback, without resolving names.""" + addr = as_ip(host) + return addr is not None and addr.is_loopback + + def is_legacy_ip_form(host: str) -> bool: """Whether ``host`` is an address written in a form resolvers disagree on. @@ -220,6 +226,7 @@ def not_public(addr: Address) -> bool: "embedded_v4", "is_legacy_ip_form", "is_local_name", + "is_loopback_ip", "is_public", "legacy_ipv4", "mapped_host", diff --git a/raven/security/urls.py b/raven/security/urls.py index 1211092f1..b7cf85f1e 100644 --- a/raven/security/urls.py +++ b/raven/security/urls.py @@ -16,7 +16,7 @@ from typing import Any from urllib.parse import urlsplit -from raven.security.hosts import as_ip, is_legacy_ip_form, is_local_name, names_this_machine, not_public +from raven.security.hosts import as_ip, is_legacy_ip_form, is_local_name, is_loopback_ip, names_this_machine, not_public class HubTrustError(ValueError): @@ -26,12 +26,7 @@ class HubTrustError(ValueError): def _host_is_local(host: str) -> bool: """Whether a URL host may buy the plaintext exemption: one of the names that mean this machine outright, or a loopback address in canonical form.""" - if is_local_name(host): - return True - if is_legacy_ip_form(host): - return False - addr = as_ip(host) - return addr is not None and addr.is_loopback + return is_local_name(host) or is_loopback_ip(host) def _split(url: str, what: str) -> Any: diff --git a/tests/test_rpc_transport.py b/tests/test_rpc_transport.py index b055e0f3a..6bd940a16 100644 --- a/tests/test_rpc_transport.py +++ b/tests/test_rpc_transport.py @@ -287,16 +287,14 @@ async def test_a_page_reached_through_a_port_forward_opens_its_socket(forwarded) assert not ws.closed -async def test_a_page_reached_as_localhost_through_a_port_forward_signs_in(forwarded) -> None: - """A browser may reach the gateway as `localhost` rather than `127.0.0.1`, and - the two are different origins to it. Through a forward neither is on the bound - port's allowlist, so the forwarded rule has to know the name as well as the - loopback addresses.""" +@pytest.mark.parametrize("host", ["localhost", "LOCALHOST", "127.0.0.2", "[::1]"]) +async def test_a_page_reached_as_a_loopback_host_through_a_port_forward_signs_in(forwarded, host: str) -> None: + """Each loopback spelling must pass when Host and Origin name the same forwarded port.""" gateway, client, browser_port = forwarded resp = await client.post( "/auth/exchange", json={"nonce": gateway.mint_nonce()}, - headers={"Origin": f"http://localhost:{browser_port}", "Host": f"localhost:{browser_port}"}, + headers={"Origin": f"http://{host}:{browser_port}", "Host": f"{host}:{browser_port}"}, ) assert resp.status == 200, await resp.text() @@ -316,11 +314,29 @@ async def test_a_page_on_another_local_port_is_refused(gateway_client) -> None: assert resp.status == 403 -async def test_a_name_that_only_resolves_to_this_machine_is_refused(gateway_client) -> None: - """DNS rebinding: a site whose name answers 127.0.0.1 asks its own origin, so - its Origin and Host agree. Requiring a loopback host is what refuses it.""" +@pytest.mark.parametrize( + "host", + [ + "rebind.example", + "127.0.0.1.evil.example", + "evil.localhost", + "localhost.localdomain", + "localhost.", + "localhost.localdomain.", + "127.0.0.1.", + "10.1.2.3", + "8.8.8.8", + "127.1", + "2130706433", + "0x7f000001", + "0177.0.0.1", + "[::ffff:8.8.8.8]", + ], +) +async def test_matching_origin_and_host_do_not_allow_other_names_or_addresses(gateway_client, host: str) -> None: + """Matching headers cannot grant a name or address the browser sign-in policy refuses.""" gateway, client = gateway_client - rebound = f"rebind.example:{gateway.port}" + rebound = f"{host}:{gateway.port}" resp = await client.post( "/auth/exchange", json={"nonce": gateway.mint_nonce()}, diff --git a/tests/test_security_hosts.py b/tests/test_security_hosts.py index 5938680ef..7ae7f28cb 100644 --- a/tests/test_security_hosts.py +++ b/tests/test_security_hosts.py @@ -36,6 +36,42 @@ def test_a_canonical_address_is_not_legacy() -> None: assert hosts.as_ip("[::1]") == ipaddress.ip_address("::1") +@pytest.mark.parametrize( + ("host", "want"), + [ + ("127.0.0.1", True), + ("127.0.0.2", True), + ("127.255.255.255", True), + ("::1", True), + ("0:0:0:0:0:0:0:1", True), + ("[::1]", True), + ("126.255.255.255", False), + ("128.0.0.0", False), + ("8.8.8.8", False), + ("0.0.0.0", False), + ("10.0.0.1", False), + ("::", False), + ("::2", False), + ("fe80::1", False), + ("::ffff:8.8.8.8", False), + ("64:ff9b::7f00:1", False), + ("localhost", False), + ("localhost.localdomain", False), + ("localhost.", False), + ("evil.localhost", False), + ("127.0.0.1.evil.example", False), + ("127.0.0.1.", False), + ("127.1", False), + ("2130706433", False), + ("0x7f000001", False), + ("0177.0.0.1", False), + ("", False), + ], +) +def test_only_canonical_loopback_ips_are_loopback(host: str, want: bool) -> None: + assert hosts.is_loopback_ip(host) is want + + def test_fullwidth_stops_map_onto_the_address() -> None: assert hosts.mapped_host("169\uff0e254\uff0e169\uff0e254") == "169.254.169.254" diff --git a/tests/test_security_urls.py b/tests/test_security_urls.py index abf5dbef4..d3cddcd9d 100644 --- a/tests/test_security_urls.py +++ b/tests/test_security_urls.py @@ -10,7 +10,13 @@ [ "https://hub.evermind.ai", "http://localhost:8000", + "http://LOCALHOST:8000", + "http://localhost.:8000", + "http://localhost.localdomain:8000", + "http://LOCALHOST.LOCALDOMAIN.:8000", "http://127.0.0.1:8000/", + "http://127.0.0.2:8000/", + "http://127.255.255.255:8000/", "http://[::1]:8000", ], ) @@ -36,6 +42,12 @@ def test_a_hub_endpoint_may_be_https_or_local_plaintext(url: str) -> None: # resolvers say 10.0.0.1. Guessing is worse than refusing. "http://010.0.0.1", "http://2130706433", + "http://127.1", + "http://0x7f000001", + "http://127.0.0.1.", + "http://127.0.0.1.evil.example", + "http://[::ffff:8.8.8.8]", + "http://[::2]", ], ) def test_a_hub_endpoint_may_not_be_plaintext_remote_or_non_http(url: str) -> None: