fix(network): honor loopback policy for ipv6 localhost

src/url_safety.py

@@ -12,7 +12,7 @@ break the primary use case. What it *always* rejects:
   - a non-HTTP(S) scheme (``file://``, ``gopher://``, ``ftp://`` …), and
   - the link-local range (``169.254.0.0/16`` / ``fe80::/10``), i.e. the cloud
     instance-metadata SSRF credential-exfil vector — nobody serves embeddings
-    there — plus multicast / reserved / unspecified addresses.
+    there, plus multicast / non-loopback reserved / unspecified addresses.
 
 For exposed multi-tenant deployments, set ``EMBEDDING_BLOCK_PRIVATE_IPS=true`` to
 additionally reject all private and loopback targets (full SSRF lockdown).
@@ -94,11 +94,15 @@ def _classify(ip: ipaddress._BaseAddress, *, block_private: bool) -> Optional[st
             return None
     if ip.is_link_local:
         return f"link-local address blocked (SSRF metadata risk): {ip}"
+    # IPv6 loopback is also reserved; its policy must match IPv4 loopback.
+    if ip.is_loopback:
+        if block_private:
+            return f"private/shared/loopback address blocked: {ip}"
+        return None
     if ip.is_multicast or ip.is_reserved or ip.is_unspecified:
         return f"disallowed address: {ip}"
     if block_private and (
         ip.is_private
-        or ip.is_loopback
         or (isinstance(ip, ipaddress.IPv4Address) and ip in _SHARED_ADDRESS_SPACE_V4)
     ):
         return f"private/shared/loopback address blocked: {ip}"

tests/test_codeql_python_residual_closure.py

@@ -227,6 +227,52 @@ def test_smtp_connects_to_the_checked_address_only(rebinding_dns, tls):
     assert rebinding_dns == ["rebind.test"]
 
 
+@pytest.mark.parametrize("protocol", ["imap", "smtp"])
+@pytest.mark.parametrize("tls", [False, True])
+@pytest.mark.parametrize("ipv6_first", [False, True])
+def test_mail_connects_to_dual_stack_localhost(monkeypatch, protocol, tls, ipv6_first):
+    from routes.email.email_helpers import (
+        _PolicyIMAP4, _PolicyIMAP4_SSL, _PolicySMTP, _PolicySMTP_SSL,
+    )
+
+    real = socket.getaddrinfo
+    lookups = []
+    ips = ["::1", "127.0.0.1"] if ipv6_first else ["127.0.0.1", "::1"]
+
+    def resolve(host, port, *args, **kwargs):
+        if host != "localhost":
+            return real(host, port, *args, **kwargs)
+        lookups.append(host)
+        assert lookups == ["localhost"]
+        return [info for ip in ips for info in real(ip, port, *args, **kwargs)]
+
+    monkeypatch.setattr(socket, "getaddrinfo", resolve)
+    server = _LineServer(protocol)
+    context = _RecordingTLS()
+    try:
+        if protocol == "imap":
+            cls = _PolicyIMAP4_SSL if tls else _PolicyIMAP4
+            kwargs = {"ssl_context": context} if tls else {}
+        else:
+            cls = _PolicySMTP_SSL if tls else _PolicySMTP
+            kwargs = {"context": context} if tls else {}
+        conn = cls("localhost", server.port, block_private=False, timeout=5, **kwargs)
+        try:
+            assert conn.sock.getpeername()[0] == "127.0.0.1"
+            if tls:
+                assert context.server_hostname == "localhost"
+            if protocol == "smtp":
+                assert conn.ehlo()[0] == 250
+        finally:
+            if protocol == "imap":
+                conn.logout()
+            else:
+                conn.quit()
+    finally:
+        server.close()
+    assert lookups == ["localhost"]
+
+
 def test_any_denied_answer_in_a_mixed_resolution_blocks_the_connection():
     from src.url_safety import OutboundAddressBlocked, connect_outbound_tcp
 
@@ -245,9 +291,11 @@ def test_any_denied_answer_in_a_mixed_resolution_blocks_the_connection():
 _ALWAYS_DENIED = [
     "169.254.169.254", "fe80::1", "0.0.0.0", "::", "224.0.0.1", "240.0.0.1",
     "::ffff:169.254.169.254", "64:ff9b::a9fe:a9fe", "64:ff9b::a00:5",
-    "::1",  # IPv6 loopback sits in ::/8, which the existing policy treats as reserved
+    "ff02::1", "::2", "100::1", "::ffff:0.0.0.0", "::ffff:224.0.0.1",
+    "::ffff:240.0.0.1", "64:ff9b::7f00:1", "64:ff9b::6440:1",
+    "64:ff9b::f000:1", "64:ff9b:1::7f00:1",
 ]
-_PRIVATE = ["127.0.0.1", "10.0.0.5", "172.16.0.1", "192.168.1.10", "100.64.0.1", "fd00::1"]
+_PRIVATE = ["127.0.0.1", "::1", "::ffff:127.0.0.1", "10.0.0.5", "172.16.0.1", "192.168.1.10", "100.64.0.1", "fd00::1"]
 _PUBLIC = ["93.184.216.34", "2606:2800:220:1::1"]
 
 
@@ -300,6 +348,31 @@ def test_private_ranges_follow_the_principal_policy(fake_sockets, ip):
     assert fake_sockets == [ip]
     with pytest.raises(OutboundAddressBlocked):
         connect_outbound_tcp("h", 993, block_private=True, resolver=_answer(ip))
+    assert fake_sockets == [ip]
+
+
+@pytest.mark.parametrize("block_private", [False, True])
+@pytest.mark.parametrize("ipv6_first", [False, True])
+def test_dual_stack_localhost_tcp_resolves_once(fake_sockets, block_private, ipv6_first):
+    from src.url_safety import OutboundAddressBlocked, connect_outbound_tcp
+
+    lookups = []
+    ips = ["::1", "127.0.0.1"] if ipv6_first else ["127.0.0.1", "::1"]
+
+    def resolve(host, port, *args):
+        lookups.append(host)
+        assert lookups == ["localhost"]
+        return [info for ip in ips for info in _answer(ip)(host, port, *args)]
+
+    if block_private:
+        with pytest.raises(OutboundAddressBlocked, match="loopback address blocked"):
+            connect_outbound_tcp("localhost", 993, block_private=True, resolver=resolve)
+        assert fake_sockets == []
+    else:
+        sock = connect_outbound_tcp("localhost", 993, block_private=False, resolver=resolve)
+        sock.close()
+        assert fake_sockets == [ips[0]]
+    assert lookups == ["localhost"]
 
 
 @pytest.mark.parametrize("ip", _PUBLIC)
@@ -339,7 +412,8 @@ def test_single_user_mode_allows_lan_mail_servers(monkeypatch):
     assert _mail_private_blocked("") is False
 
 
-def test_non_admin_cannot_probe_loopback_through_the_connection_test(monkeypatch):
+@pytest.mark.parametrize("host", ["127.0.0.1", "::1", "localhost"])
+def test_non_admin_cannot_probe_loopback_through_the_connection_test(monkeypatch, host):
     import routes.email_routes as email_routes
     import src.tool_security as tool_security
 
@@ -350,7 +424,7 @@ def test_non_admin_cannot_probe_loopback_through_the_connection_test(monkeypatch
 
     class _Request:
         async def json(self):
-            return {"imap_host": "127.0.0.1", "imap_port": 6379, "imap_user": "u", "imap_password": "p",
+            return {"imap_host": host, "imap_port": 6379, "imap_user": "u", "imap_password": "p",
                     "smtp_host": "10.0.0.5", "smtp_port": 25}
 
     result = asyncio.run(endpoint(req=_Request(), owner="bob"))

tests/test_url_safety.py

@@ -3,6 +3,8 @@
 A stub resolver is injected so the tests never touch real DNS.
 """
 
+import pytest
+
 from src.url_safety import check_outbound_url
 
 
@@ -64,6 +66,39 @@ def test_strict_mode_blocks_private_and_loopback():
     assert ok is False and "private" in reason
 
 
+@pytest.mark.parametrize("block_private", [False, True])
+@pytest.mark.parametrize("host,ips", [
+    ("127.0.0.1", ["127.0.0.1"]),
+    ("[::1]", ["::1"]),
+    ("[::ffff:127.0.0.1]", ["::ffff:127.0.0.1"]),
+    ("localhost", ["127.0.0.1", "::1"]),
+    ("localhost", ["::1", "127.0.0.1"]),
+])
+def test_loopback_urls_follow_private_policy(host, ips, block_private):
+    ok, reason = check_outbound_url(
+        f"http://{host}:8080", block_private=block_private,
+        resolver=_resolver({host.strip("[]"): ips}),
+    )
+    assert ok is (not block_private), reason
+    if block_private:
+        assert "loopback address blocked" in reason
+
+
+@pytest.mark.parametrize("block_private", [False, True])
+@pytest.mark.parametrize("ip", [
+    "169.254.169.254", "fe80::1", "0.0.0.0", "::", "224.0.0.1", "ff02::1",
+    "240.0.0.1", "::2", "100::1", "::ffff:169.254.169.254",
+    "::ffff:0.0.0.0", "::ffff:224.0.0.1", "::ffff:240.0.0.1",
+])
+def test_special_ranges_stay_blocked_alongside_loopback(ip, block_private):
+    ok, reason = check_outbound_url(
+        "http://localhost:8080", block_private=block_private,
+        resolver=_resolver({"localhost": [ip, "127.0.0.1", "::1"]}),
+    )
+    assert ok is False
+    assert "link-local" in reason or "disallowed address" in reason
+
+
 def test_strict_mode_blocks_cgnat_shared_space():
     # RFC 6598 shared/CGNAT space (100.64.0.0/10) is not globally routable.
     # A public redirect into it must be rejected under full SSRF lockdown,
@@ -247,14 +282,3 @@ def test_ordinary_ipv6_behaviour_is_unchanged():
     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