security(connectors): reject tunneled metadata addresses

src/govoplan_core/security/outbound_http.py

@@ -18,6 +18,15 @@ _READ_CHUNK_BYTES: Final = 64 * 1024
 _TRUE_VALUES: Final = frozenset({"1", "true", "yes", "on"})
 _FALSE_VALUES: Final = frozenset({"0", "false", "no", "off"})
 _IPV4_LIMITED_BROADCAST: Final = ipaddress.IPv4Address("255.255.255.255")
+_IPV4_COMPATIBLE_NETWORK: Final = ipaddress.IPv6Network("::/96")
+_NAT64_WELL_KNOWN_NETWORK: Final = ipaddress.IPv6Network("64:ff9b::/96")
+_KNOWN_METADATA_ADDRESSES: Final = frozenset(
+    {
+        ipaddress.ip_address("100.100.100.200"),
+        ipaddress.ip_address("169.254.169.254"),
+        ipaddress.ip_address("fd00:ec2::254"),
+    }
+)
 
 
 class OutboundHttpError(RuntimeError):
@@ -276,9 +285,15 @@ def _content_length(headers: Mapping[str, str]) -> int | None:
 
 def _is_public_address(value: str) -> bool:
     try:
-        return ipaddress.ip_address(value).is_global
+        address = ipaddress.ip_address(value)
     except ValueError:
         return False
+    if address.is_reserved or getattr(address, "is_site_local", False) or not address.is_global:
+        return False
+    return all(
+        embedded.is_global and not embedded.is_reserved
+        for embedded in _embedded_ipv4_addresses(address)
+    )
 
 
 def _is_forbidden_special_address(value: str) -> bool:
@@ -293,17 +308,38 @@ def _is_forbidden_special_address(value: str) -> bool:
         address = ipaddress.ip_address(value)
     except ValueError:
         return True
-    mapped = getattr(address, "ipv4_mapped", None)
+    candidates = (address, *_embedded_ipv4_addresses(address))
-    if mapped is not None:
+    return any(
-        address = mapped
+        candidate in _KNOWN_METADATA_ADDRESSES
-    return (
+        or candidate == _IPV4_LIMITED_BROADCAST
-        address == _IPV4_LIMITED_BROADCAST
+        or candidate.is_link_local
-        or address.is_link_local
+        or candidate.is_multicast
-        or address.is_multicast
+        or candidate.is_unspecified
-        or address.is_unspecified
+        for candidate in candidates
     )
 
 
+def _embedded_ipv4_addresses(
+    address: ipaddress.IPv4Address | ipaddress.IPv6Address,
+) -> tuple[ipaddress.IPv4Address, ...]:
+    """Return IPv4 destinations encoded by standard IPv6 transition forms."""
+
+    if isinstance(address, ipaddress.IPv4Address):
+        return ()
+    candidates: list[ipaddress.IPv4Address] = []
+    if address.ipv4_mapped is not None:
+        candidates.append(address.ipv4_mapped)
+    elif address in _IPV4_COMPATIBLE_NETWORK:
+        candidates.append(ipaddress.IPv4Address(int(address) & 0xFFFFFFFF))
+    if address in _NAT64_WELL_KNOWN_NETWORK:
+        candidates.append(ipaddress.IPv4Address(int(address) & 0xFFFFFFFF))
+    if address.sixtofour is not None:
+        candidates.append(address.sixtofour)
+    if address.teredo is not None:
+        candidates.extend(address.teredo)
+    return tuple(dict.fromkeys(candidates))
+
+
 def _resolved_address_records(
     hostname: str,
     *,

tests/test_http_fetch.py

@@ -86,6 +86,9 @@ class HttpFetchTests(unittest.TestCase):
             "255.255.255.255",
             "::",
             "::ffff:255.255.255.255",
+            "64:ff9b::169.254.169.254",
+            "64:ff9b::255.255.255.255",
+            "fd00:ec2::254",
             "fe80::1",
             "ff02::1",
         ):
@@ -95,6 +98,21 @@ class HttpFetchTests(unittest.TestCase):
             ), self.assertRaisesRegex(OutboundHttpBlocked, "forbidden special-purpose network"):
                 validate_outbound_http_url("https://connector.example.test/path", policy=policy)
 
+    def test_public_policy_rejects_reserved_and_ipv4_embedded_ipv6_addresses(self) -> None:
+        policy = outbound_http_policy({"APP_ENV": "production"})
+        for address in (
+            "fec0::1",
+            "::127.0.0.1",
+            "64:ff9b::8.8.8.8",
+            "64:ff9b::127.0.0.1",
+            "2002:7f00:1::",
+        ):
+            with self.subTest(address=address), patch(
+                "govoplan_core.security.outbound_http.socket.getaddrinfo",
+                return_value=[(10, 1, 6, "", (address, 443, 0, 0))],
+            ), self.assertRaises(OutboundHttpBlocked):
+                validate_outbound_http_url("https://connector.example.test/path", policy=policy)
+
     def test_outbound_policy_rejects_mixed_public_and_private_dns_answers(self) -> None:
         policy = outbound_http_policy({"APP_ENV": "production"})
         with patch(