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 884581a..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: @@ -380,6 +393,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 +471,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_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 cbd7d8a..55c9c7b 100644 --- a/plugins/communication_protocols/http/tests/test_redirect_security.py +++ b/plugins/communication_protocols/http/tests/test_redirect_security.py @@ -958,3 +958,88 @@ 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 + # 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: + # 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] 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]