diff --git a/src/govoplan_core/security/outbound_http.py b/src/govoplan_core/security/outbound_http.py index 976cf98..c3f557b 100644 --- a/src/govoplan_core/security/outbound_http.py +++ b/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) - if mapped is not None: - address = mapped - return ( - address == _IPV4_LIMITED_BROADCAST - or address.is_link_local - or address.is_multicast - or address.is_unspecified + candidates = (address, *_embedded_ipv4_addresses(address)) + return any( + candidate in _KNOWN_METADATA_ADDRESSES + or candidate == _IPV4_LIMITED_BROADCAST + or candidate.is_link_local + or candidate.is_multicast + or candidate.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, *, diff --git a/tests/test_http_fetch.py b/tests/test_http_fetch.py index 2d46a75..cb54812 100644 --- a/tests/test_http_fetch.py +++ b/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(