From dc1270333b4c022725071f4a00ed6433b18e9d3e Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Wed, 16 Sep 2026 14:02:58 +0200 Subject: [PATCH 1/2] http: a redirect never enters loopback from a non-loopback origin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit safe_request_with_redirects re-validated every hop with ensure_secure_url, which accepts plain HTTP to loopback for local development. So a remote discovery or tool URL could 302 the client into http://127.0.0.1:... — an SSRF into the agent's own services, issued with the manual's headers — and, because reject_remote_loopback_tool_urls judges by the final URL, whatever loopback served then counted as locally discovered and could declare loopback tool URLs. The existing 'redirect to loopback is allowed' test only ever covered loopback-to-loopback. The helper now refuses any hop whose target is loopback when the URL being redirected is not: the loopback allowance is for requests the caller addressed to loopback, never for ones a remote server steers there. That also makes the final-URL rule sound by construction — a final loopback URL now means the chain started on loopback and never left it (loopback may still leave, and loses the local-dev exemption when it does). Documented on the helper. Tests: a remote origin cannot redirect into loopback (canonical, hostname, HTTPS and IPv6 loopback forms) and the loopback request is never issued; a chain that left loopback cannot be sent back. Scripted session, since every test server here lives on loopback. Mutation- checked: dropping the rule fails exactly those five. http suite 243/243. Raised by cubic on utcp-specification #66, from the sentence in the HTTP protocol page that described this behaviour. Co-Authored-By: Claude Fable 5.1 --- .../http/src/utcp_http/_security.py | 17 +++++ .../http/tests/test_redirect_security.py | 76 +++++++++++++++++++ 2 files changed, 93 insertions(+) diff --git a/plugins/communication_protocols/http/src/utcp_http/_security.py b/plugins/communication_protocols/http/src/utcp_http/_security.py index 884581a..796264e 100644 --- a/plugins/communication_protocols/http/src/utcp_http/_security.py +++ b/plugins/communication_protocols/http/src/utcp_http/_security.py @@ -380,6 +380,14 @@ async def safe_request_with_redirects( target against the current URL and runs ``ensure_secure_url`` on it before issuing the next hop. Rejection raises and the redirect chain is aborted with the connection released. + * Never follows a redirect INTO loopback from a non-loopback URL. + The loopback allowance in ``ensure_secure_url`` exists for + requests the caller addressed to loopback (local development); + a remote server must not be able to steer a request at the + agent's own services, nor make a loopback-served manual look + locally discovered (``reject_remote_loopback_tool_urls`` judges + by the final URL, which this rule keeps honest: a final loopback + URL means the chain started on loopback and never left it). * Caps the chain at ``max_redirects`` hops. Exceeding that raises ``RuntimeError``. * Mirrors RFC 7231 method semantics: 303 forces ``GET`` and drops @@ -450,6 +458,15 @@ async def safe_request_with_redirects( ensure_secure_url( next_url, context=f"{context} (redirect target)" ) + if is_loopback_url(next_url) and not is_loopback_url(current_url): + raise ValueError( + f"Security error during {context} (redirect target): " + f"{current_url!r} redirected to the loopback address " + f"{next_url!r}. A redirect is never followed into " + "loopback from a non-loopback origin: the loopback " + "allowance is for requests addressed to loopback by " + "the caller, not for ones a remote server steers there." + ) except Exception: response.release() raise diff --git a/plugins/communication_protocols/http/tests/test_redirect_security.py b/plugins/communication_protocols/http/tests/test_redirect_security.py index cbd7d8a..cff3cea 100644 --- a/plugins/communication_protocols/http/tests/test_redirect_security.py +++ b/plugins/communication_protocols/http/tests/test_redirect_security.py @@ -958,3 +958,79 @@ def test_legitimate_https_token_url_accepted(self) -> None: ) manual = converter.convert() assert len(manual.tools) == 1 + + +# --------------------------------------------------------------------------- +# A redirect never enters loopback from a non-loopback origin. +# --------------------------------------------------------------------------- + + +class _FakeResponse: + def __init__(self, status: int, headers: dict, url: str): + self.status = status + self.headers = headers + self.url = url + self.released = False + + def release(self) -> None: + self.released = True + + +class _ScriptedSession: + """Answers each URL from a script and records every request it was asked + to issue -- the test servers all live on loopback, so a genuinely remote + origin has to be scripted.""" + + def __init__(self, script: dict): + self.script = script + self.requested: list = [] + + async def request(self, method: str, url: str, **kwargs): + self.requested.append(url) + return self.script[url] + + +class TestRedirectNeverEntersLoopback: + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "loopback_target", + [ + "http://127.0.0.1:9200/_cat/indices", # canonical loopback + "http://localhost:9200/_cat/indices", # loopback hostname + "https://127.0.0.1:8443/admin", # HTTPS does not make loopback remote + "http://[::1]:9200/_cat/indices", # IPv6 loopback + ], + ) + async def test_remote_origin_cannot_redirect_into_loopback(self, loopback_target) -> None: + # The remote hop is secure (HTTPS) and the loopback hop would pass + # ensure_secure_url on its own -- only the direction is illegal. + remote = "https://attacker.example/manual" + session = _ScriptedSession({ + remote: _FakeResponse(302, {"Location": loopback_target}, remote), + }) + + with pytest.raises(ValueError, match="never followed into loopback"): + async with safe_request_with_redirects(session, "GET", remote, context="manual discovery"): + pass + + # The request to the agent's own service was never issued. + assert session.requested == [remote] + + @pytest.mark.asyncio + async def test_loopback_may_leave_but_a_chain_that_left_cannot_come_back(self) -> None: + # loopback -> remote is fine (the remote hop is validated like any + # other and the caller loses the local-dev exemption); the remote's + # attempt to send us back to loopback is refused. + local = "http://127.0.0.1:8765/manual" + remote = "https://mirror.example/manual" + session = _ScriptedSession({ + local: _FakeResponse(302, {"Location": remote}, local), + remote: _FakeResponse(302, {"Location": "http://127.0.0.1:9200/secret"}, remote), + }) + + with pytest.raises(ValueError, match="never followed into loopback"): + async with safe_request_with_redirects(session, "GET", local, context="manual discovery"): + pass + + assert session.requested == [local, remote] From 0cde10194a4fc9f231156ebb224ad4d36404208f Mon Sep 17 00:00:00 2001 From: Razvan Radulescu <43811028+h3xxit@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:23:09 +0200 Subject: [PATCH 2/2] http/websocket/gql: loopback is classified the way the resolver resolves it cubic on #109: the redirect guard asked is_loopback_url, which classifies through Python's ipaddress -- and ipaddress rejects the spellings the OS resolver happily routes to 127.0.0.1: the shorthand 127.1, the integer 2130706433, octal 0177.0.0.1, hex 0x7f000001, and the absolute-name forms 127.0.0.1. / localhost. with a trailing dot. Over HTTPS every one of them passed ensure_secure_url as "HTTPS anywhere" and slipped past the new guard, and past the remote-manual rule and the OpenAPI converter's servers[0] check for the same reason. One classifier now answers every loopback question in utcp_http -- _is_loopback_host: the plain-HTTP allowance, is_loopback_url, the redirect guard, the remote-manual rule and the converter all go through it, so they cannot disagree about what loopback is. It strips a trailing dot, accepts the canonical names, wildcard and any 127/8 or IPv4-mapped form as before, and when ipaddress rejects a numeric literal it asks the resolver's own parser (inet_aton) and classifies the result. utcp_websocket and utcp_gql carry mirrors of this module by design ("keep in sync"); both were missing the redirect-into-loopback rule from the first commit as well as the spellings. Both now carry both. Tests: the redirect guard and the remote-manual rule are parametrized over the new spellings (http); each mirror gets the spellings, the lookalikes that must stay non-loopback, and the remote-origin redirect. Mutation-checked in all three modules: dropping the resolver path fails exactly the spelling tests. http 255/255, websocket 52/52, gql 37/37. Co-Authored-By: Claude Fable 5.1 --- .../gql/src/utcp_gql/_security.py | 48 ++++++++++-- .../gql/tests/test_gql_security.py | 60 +++++++++++++++ .../http/src/utcp_http/_security.py | 77 +++++++++++-------- .../tests/test_loopback_manual_security.py | 7 ++ .../http/tests/test_redirect_security.py | 9 +++ .../websocket/src/utcp_websocket/_security.py | 46 +++++++++-- .../tests/test_websocket_security.py | 60 +++++++++++++++ 7 files changed, 263 insertions(+), 44 deletions(-) diff --git a/plugins/communication_protocols/gql/src/utcp_gql/_security.py b/plugins/communication_protocols/gql/src/utcp_gql/_security.py index 98a4c53..8b54b18 100644 --- a/plugins/communication_protocols/gql/src/utcp_gql/_security.py +++ b/plugins/communication_protocols/gql/src/utcp_gql/_security.py @@ -11,6 +11,7 @@ from __future__ import annotations import re +import socket from contextlib import asynccontextmanager from ipaddress import IPv6Address, ip_address from typing import Any, AsyncIterator, Dict, Optional @@ -67,22 +68,42 @@ def is_secure_url(url: str) -> bool: return False +# An IPv4 literal in any of the spellings the resolver accepts: dotted +# quads, fewer than four parts (``127.1``), a single integer +# (``2130706433``), octal (``0177.0.0.1``) or hex (``0x7f000001``) parts. +_RESOLVER_IPV4_LITERAL = re.compile(r"^[0-9a-fx.]+$") + + def _ip_is_loopback_like(host: str) -> bool: - """Mirror of ``utcp_http._security._ip_is_loopback_like``. See that - module for the full rationale -- covers 127.0.0.0/8, ::1, 0.0.0.0, - ::, and IPv4-mapped IPv6 loopback addresses. + """Mirror of ``utcp_http._security._is_loopback_host``. See that module + for the full rationale: True if ``host`` names the machine running the + agent, in any spelling the resolver routes locally -- ``localhost`` / + ``127.0.0.1`` / ``::1``, a trailing-dot absolute name, ``0.0.0.0`` / + ``::``, any 127.0.0.0/8 address, IPv4-mapped IPv6 forms, and the + shorthand/integer/octal/hex IPv4 spellings ``inet_aton`` accepts + (``127.1``, ``2130706433``, ``0177.0.0.1``, ``0x7f000001``) that + ``ipaddress`` rejects. """ - if host in {"0.0.0.0", "::"}: + host = host.rstrip(".") + if not host: + return False + if host in _LOOPBACK_HOSTNAMES or host in {"0.0.0.0", "::"}: return True try: addr = ip_address(host) except ValueError: - return False - if addr.is_loopback: + # Not a form ``ipaddress`` understands; ask the resolver's parser. + if not _RESOLVER_IPV4_LITERAL.match(host): + return False + try: + addr = ip_address(socket.inet_ntoa(socket.inet_aton(host))) + except (OSError, ValueError): + return False + if addr.is_loopback or addr.is_unspecified: return True if isinstance(addr, IPv6Address): mapped = addr.ipv4_mapped - if mapped is not None and mapped.is_loopback: + if mapped is not None and (mapped.is_loopback or mapped.is_unspecified): return True return False @@ -340,6 +361,19 @@ async def safe_request_with_redirects( ensure_secure_url( next_url, context=f"{context} (redirect target)" ) + # Mirror of utcp_http: a redirect never enters loopback from + # a non-loopback URL. The loopback allowance is for requests + # the caller addressed to loopback, not for ones a remote + # server steers there. + if is_loopback_url(next_url) and not is_loopback_url(current_url): + raise ValueError( + f"Security error during {context} (redirect target): " + f"{current_url!r} redirected to the loopback address " + f"{next_url!r}. A redirect is never followed into " + "loopback from a non-loopback origin: the loopback " + "allowance is for requests addressed to loopback by " + "the caller, not for ones a remote server steers there." + ) except Exception: response.release() raise diff --git a/plugins/communication_protocols/gql/tests/test_gql_security.py b/plugins/communication_protocols/gql/tests/test_gql_security.py index da93008..dc54e78 100644 --- a/plugins/communication_protocols/gql/tests/test_gql_security.py +++ b/plugins/communication_protocols/gql/tests/test_gql_security.py @@ -109,3 +109,63 @@ async def test_plain_http_non_loopback_token_url_rejected(self) -> None: ) with pytest.raises(ValueError, match="OAuth2 token URL"): await proto._handle_oauth2(auth) + + +# --------------------------------------------------------------------------- +# Mirror of utcp_http: loopback in every spelling the resolver accepts, and a +# redirect never enters loopback from a non-loopback origin. +# --------------------------------------------------------------------------- + +import pytest as _pytest +from utcp_gql._security import is_loopback_url as _is_loopback_url, safe_request_with_redirects as _safe_request_with_redirects + + +@_pytest.mark.parametrize( + "url", + [ + "https://127.1/x", # shorthand: inet_aton fills the middle octets + "https://2130706433/x", # 127.0.0.1 as a single integer + "https://0177.0.0.1/x", # octal + "https://0x7f000001/x", # hex + "https://127.0.0.1./x", # absolute-name form (trailing dot) + "https://localhost./x", + "https://0.0.0.0/x", # wildcard routes to the local host + "https://[::ffff:127.0.0.1]/x", + ], +) +def test_loopback_is_recognised_in_every_resolver_spelling(url): + assert _is_loopback_url(url) + + +@_pytest.mark.parametrize("url", ["https://localhost.evil.com/x", "https://127.0.0.1.attacker.example/x", "https://10.1/x"]) +def test_lookalikes_and_other_networks_are_not_loopback(url): + assert not _is_loopback_url(url) + + +class _FakeResponse: + def __init__(self, status, headers, url): + self.status, self.headers, self.url = status, headers, url + + def release(self): + pass + + +class _ScriptedSession: + def __init__(self, script): + self.script, self.requested = script, [] + + async def request(self, method, url, **kwargs): + self.requested.append(url) + return self.script[url] + + +@_pytest.mark.asyncio +@_pytest.mark.parametrize("loopback_target", ["http://127.0.0.1:9200/x", "https://127.1/x", "https://localhost./x"]) +async def test_remote_origin_cannot_redirect_into_loopback(loopback_target): + remote = "https://attacker.example/manual" + session = _ScriptedSession({remote: _FakeResponse(302, {"Location": loopback_target}, remote)}) + with _pytest.raises(ValueError, match="never followed into loopback"): + async with _safe_request_with_redirects(session, "GET", remote, context="manual discovery"): + pass + # The request to the agent's own service was never issued. + assert session.requested == [remote] diff --git a/plugins/communication_protocols/http/src/utcp_http/_security.py b/plugins/communication_protocols/http/src/utcp_http/_security.py index 796264e..41816a6 100644 --- a/plugins/communication_protocols/http/src/utcp_http/_security.py +++ b/plugins/communication_protocols/http/src/utcp_http/_security.py @@ -10,6 +10,7 @@ from __future__ import annotations import re +import socket from contextlib import asynccontextmanager from ipaddress import IPv6Address, ip_address from typing import Any, AsyncIterator, Dict, Optional @@ -55,49 +56,64 @@ def is_secure_url(url: str) -> bool: return True # http:// is only allowed for loopback. - if host in _LOOPBACK_HOSTNAMES: - return True + return _is_loopback_host(host) - # Catch any other literal loopback IP that urlparse normalised - # (e.g. ``http://127.000.000.001``). - try: - return ip_address(host).is_loopback - except ValueError: - return False +# An IPv4 literal in any of the spellings the resolver accepts: dotted +# quads, fewer than four parts (``127.1``), a single integer +# (``2130706433``), octal (``0177.0.0.1``) or hex (``0x7f000001``) parts. +_RESOLVER_IPV4_LITERAL = re.compile(r"^[0-9a-fx.]+$") -def _ip_is_loopback_like(host: str) -> bool: - """Return True if ``host`` is an IP literal that the local kernel will - route to the host running the agent. + +def _is_loopback_host(host: str) -> bool: + """Return True if ``host`` names the machine running the agent. + + ``host`` is a hostname as ``urlparse`` returns it (lowercase, no + brackets). One classifier for every loopback decision -- the plain-HTTP + allowance, the remote-manual rule, the redirect guard, the OpenAPI + converter -- so they cannot disagree about what loopback is. Wider than Python's stdlib ``ip_address(...).is_loopback`` because we - must also defend against: - - * ``0.0.0.0`` -- on Linux a TCP connect to 0.0.0.0 lands on 127.0.0.1. - * ``::`` -- the IPv6 equivalent of ``0.0.0.0``. - * IPv4-mapped IPv6 forms of any 127.0.0.0/8 address (e.g. - ``::ffff:127.0.0.1``, ``::ffff:127.0.0.2``) -- ``ipaddress`` does - not treat these as loopback per RFC 4291, but the dual-stack - socket layer routes them to the v4 loopback. - - Used by the OpenAPI converter to detect attacker-controlled - ``servers[0].url`` values that point at the agent's own loopback - interface (the GHSA-39j6-4867-gg4w SSRF pattern). Hostname-based, - never prefix-based. + must classify every spelling the *resolver* will route locally: + + * ``localhost`` and the canonical ``127.0.0.1`` / ``::1``. + * A trailing dot (``localhost.``, ``127.0.0.1.``): the absolute-name + form, which resolves the same. + * ``0.0.0.0`` -- on Linux a TCP connect to 0.0.0.0 lands on 127.0.0.1; + ``::`` is the IPv6 equivalent. + * Any 127.0.0.0/8 address. + * IPv4-mapped IPv6 forms of those (``::ffff:127.0.0.1``) -- not + loopback per RFC 4291, but the dual-stack socket layer routes them + to the v4 loopback. + * The shorthand, integer, octal and hex IPv4 spellings ``inet_aton`` + accepts (``127.1``, ``2130706433``, ``0177.0.0.1``, ``0x7f000001``) + -- ``ipaddress`` rejects them, the OS resolver does not. + + Hostname-based, never prefix-based: ``localhost.evil.com`` and + ``127.0.0.1.attacker.example`` are not loopback. """ - if host in {"0.0.0.0", "::"}: + host = host.rstrip(".") + if not host: + return False + if host in _LOOPBACK_HOSTNAMES or host in {"0.0.0.0", "::"}: return True try: addr = ip_address(host) except ValueError: - return False - if addr.is_loopback: + # Not a form ``ipaddress`` understands; ask the resolver's parser. + if not _RESOLVER_IPV4_LITERAL.match(host): + return False + try: + addr = ip_address(socket.inet_ntoa(socket.inet_aton(host))) + except (OSError, ValueError): + return False + if addr.is_loopback or addr.is_unspecified: return True # IPv4-mapped IPv6 loopback (``::ffff:127.0.0.1`` etc.) -- the # ``ipv4_mapped`` accessor surfaces the embedded v4 address. if isinstance(addr, IPv6Address): mapped = addr.ipv4_mapped - if mapped is not None and mapped.is_loopback: + if mapped is not None and (mapped.is_loopback or mapped.is_unspecified): return True return False @@ -124,10 +140,7 @@ def is_loopback_url(url: str) -> bool: if not host: return False - if host in _LOOPBACK_HOSTNAMES: - return True - - return _ip_is_loopback_like(host) + return _is_loopback_host(host) def ensure_secure_url(url: str, *, context: Optional[str] = None) -> None: diff --git a/plugins/communication_protocols/http/tests/test_loopback_manual_security.py b/plugins/communication_protocols/http/tests/test_loopback_manual_security.py index 3ce2da8..8533a69 100644 --- a/plugins/communication_protocols/http/tests/test_loopback_manual_security.py +++ b/plugins/communication_protocols/http/tests/test_loopback_manual_security.py @@ -34,6 +34,13 @@ def _manual(url: str) -> UtcpManual: "http://127.0.0.2:9200/secret", # 127.0.0.0/8, slips a naive "127.0.0.1" check "http://0.0.0.0:9200/secret", # wildcard, routes to the local host "http://[::ffff:127.0.0.1]/secret", # IPv4-mapped IPv6 loopback + # Spellings the resolver accepts but Python's ipaddress rejects + "https://127.1/secret", # shorthand: inet_aton fills the middle octets + "https://2130706433/secret", # 127.0.0.1 as a single integer + "https://0177.0.0.1/secret", # octal first octet + "https://0x7f000001/secret", # hex + "https://127.0.0.1./secret", # absolute-name form (trailing dot) + "https://localhost./secret", # same, for the hostname ], ) def test_remote_manual_with_loopback_tool_url_is_rejected(tool_url): diff --git a/plugins/communication_protocols/http/tests/test_redirect_security.py b/plugins/communication_protocols/http/tests/test_redirect_security.py index cff3cea..55c9c7b 100644 --- a/plugins/communication_protocols/http/tests/test_redirect_security.py +++ b/plugins/communication_protocols/http/tests/test_redirect_security.py @@ -1000,6 +1000,15 @@ class TestRedirectNeverEntersLoopback: "http://localhost:9200/_cat/indices", # loopback hostname "https://127.0.0.1:8443/admin", # HTTPS does not make loopback remote "http://[::1]:9200/_cat/indices", # IPv6 loopback + # Spellings the resolver routes to loopback but Python's ipaddress + # rejects -- over HTTPS, so only the loopback classification stands + # between them and being "HTTPS anywhere" + "https://127.1/admin", # shorthand + "https://2130706433/admin", # single integer + "https://0177.0.0.1/admin", # octal + "https://0x7f000001/admin", # hex + "https://127.0.0.1./admin", # absolute-name form + "https://localhost./admin", ], ) async def test_remote_origin_cannot_redirect_into_loopback(self, loopback_target) -> None: diff --git a/plugins/communication_protocols/websocket/src/utcp_websocket/_security.py b/plugins/communication_protocols/websocket/src/utcp_websocket/_security.py index 74714f0..1bed3f2 100644 --- a/plugins/communication_protocols/websocket/src/utcp_websocket/_security.py +++ b/plugins/communication_protocols/websocket/src/utcp_websocket/_security.py @@ -16,6 +16,7 @@ from __future__ import annotations import re +import socket from contextlib import asynccontextmanager from ipaddress import IPv6Address, ip_address from typing import Any, AsyncIterator, Dict, Optional @@ -25,19 +26,41 @@ _LOOPBACK_HOSTNAMES = frozenset({"localhost", "127.0.0.1", "::1", "[::1]"}) +# An IPv4 literal in any of the spellings the resolver accepts: dotted +# quads, fewer than four parts (``127.1``), a single integer +# (``2130706433``), octal (``0177.0.0.1``) or hex (``0x7f000001``) parts. +_RESOLVER_IPV4_LITERAL = re.compile(r"^[0-9a-fx.]+$") + + def _ip_is_loopback_like(host: str) -> bool: - """Mirror of ``utcp_http._security._ip_is_loopback_like``.""" - if host in {"0.0.0.0", "::"}: + """Mirror of ``utcp_http._security._is_loopback_host``: True if ``host`` + names the machine running the agent, in any spelling the resolver routes + locally -- ``localhost``/``127.0.0.1``/``::1``, a trailing-dot absolute + name, ``0.0.0.0``/``::``, any 127.0.0.0/8 address, IPv4-mapped IPv6 forms, + and the shorthand/integer/octal/hex IPv4 spellings ``inet_aton`` accepts + (``127.1``, ``2130706433``, ``0177.0.0.1``, ``0x7f000001``) that + ``ipaddress`` rejects. + """ + host = host.rstrip(".") + if not host: + return False + if host in _LOOPBACK_HOSTNAMES or host in {"0.0.0.0", "::"}: return True try: addr = ip_address(host) except ValueError: - return False - if addr.is_loopback: + # Not a form ``ipaddress`` understands; ask the resolver's parser. + if not _RESOLVER_IPV4_LITERAL.match(host): + return False + try: + addr = ip_address(socket.inet_ntoa(socket.inet_aton(host))) + except (OSError, ValueError): + return False + if addr.is_loopback or addr.is_unspecified: return True if isinstance(addr, IPv6Address): mapped = addr.ipv4_mapped - if mapped is not None and mapped.is_loopback: + if mapped is not None and (mapped.is_loopback or mapped.is_unspecified): return True return False @@ -390,6 +413,19 @@ async def safe_request_with_redirects( ensure_secure_url( next_url, context=f"{context} (redirect target)" ) + # Mirror of utcp_http: a redirect never enters loopback from + # a non-loopback URL. The loopback allowance is for requests + # the caller addressed to loopback, not for ones a remote + # server steers there. + if is_loopback_url(next_url) and not is_loopback_url(current_url): + raise ValueError( + f"Security error during {context} (redirect target): " + f"{current_url!r} redirected to the loopback address " + f"{next_url!r}. A redirect is never followed into " + "loopback from a non-loopback origin: the loopback " + "allowance is for requests addressed to loopback by " + "the caller, not for ones a remote server steers there." + ) except Exception: response.release() raise diff --git a/plugins/communication_protocols/websocket/tests/test_websocket_security.py b/plugins/communication_protocols/websocket/tests/test_websocket_security.py index c77840c..9e2ee0c 100644 --- a/plugins/communication_protocols/websocket/tests/test_websocket_security.py +++ b/plugins/communication_protocols/websocket/tests/test_websocket_security.py @@ -248,3 +248,63 @@ def test_internal_rejected(self) -> None: assert is_secure_url("http://169.254.169.254/token") is False with pytest.raises(ValueError): ensure_secure_url("http://169.254.169.254/token") + + +# --------------------------------------------------------------------------- +# Mirror of utcp_http: loopback in every spelling the resolver accepts, and a +# redirect never enters loopback from a non-loopback origin. +# --------------------------------------------------------------------------- + +import pytest as _pytest +from utcp_websocket._security import is_loopback_url as _is_loopback_url, safe_request_with_redirects as _safe_request_with_redirects + + +@_pytest.mark.parametrize( + "url", + [ + "https://127.1/x", # shorthand: inet_aton fills the middle octets + "https://2130706433/x", # 127.0.0.1 as a single integer + "https://0177.0.0.1/x", # octal + "https://0x7f000001/x", # hex + "https://127.0.0.1./x", # absolute-name form (trailing dot) + "https://localhost./x", + "https://0.0.0.0/x", # wildcard routes to the local host + "https://[::ffff:127.0.0.1]/x", + ], +) +def test_loopback_is_recognised_in_every_resolver_spelling(url): + assert _is_loopback_url(url) + + +@_pytest.mark.parametrize("url", ["https://localhost.evil.com/x", "https://127.0.0.1.attacker.example/x", "https://10.1/x"]) +def test_lookalikes_and_other_networks_are_not_loopback(url): + assert not _is_loopback_url(url) + + +class _FakeResponse: + def __init__(self, status, headers, url): + self.status, self.headers, self.url = status, headers, url + + def release(self): + pass + + +class _ScriptedSession: + def __init__(self, script): + self.script, self.requested = script, [] + + async def request(self, method, url, **kwargs): + self.requested.append(url) + return self.script[url] + + +@_pytest.mark.asyncio +@_pytest.mark.parametrize("loopback_target", ["http://127.0.0.1:9200/x", "https://127.1/x", "https://localhost./x"]) +async def test_remote_origin_cannot_redirect_into_loopback(loopback_target): + remote = "https://attacker.example/manual" + session = _ScriptedSession({remote: _FakeResponse(302, {"Location": loopback_target}, remote)}) + with _pytest.raises(ValueError, match="never followed into loopback"): + async with _safe_request_with_redirects(session, "GET", remote, context="manual discovery"): + pass + # The request to the agent's own service was never issued. + assert session.requested == [remote]