http/websocket/gql: a redirect never enters loopback, and loopback is what the resolver says it is

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

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]

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

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):

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]