diff --git a/modules/security_utils.py b/modules/security_utils.py index 4c4fa27..4c12276 100644 --- a/modules/security_utils.py +++ b/modules/security_utils.py @@ -124,6 +124,13 @@ class SafeUrlPolicy: return scheme, hostname, port or (443 if scheme == "https" else 80) def _check_address(self, address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None: + # Canonicalize IPv4-mapped IPv6 (e.g. ``::ffff:169.254.169.254``) to the + # IPv4 target the socket layer will actually dial. The mapped form is a + # distinct address object that is absent from _METADATA_ADDRESSES and + # reports is_reserved=False/is_global=False, so without this it would + # bypass the metadata and non-unicast checks under allow_private=True. + if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped is not None: + address = address.ipv4_mapped if address in _METADATA_ADDRESSES: raise UnsafeUrlError(f"Cloud metadata address is not allowed: {address}") # These cannot be meaningful unicast HTTP server destinations. They @@ -240,8 +247,8 @@ class SafeAiohttpResolver(AbstractResolver): return [ { "hostname": host, - "host": record[4][0], - "port": record[4][1], + "host": str(record[4][0]), + "port": int(record[4][1]), "family": record[0], "proto": record[2], "flags": socket.AI_NUMERICHOST, diff --git a/tests/test_security_utils.py b/tests/test_security_utils.py index e35b897..ac52412 100644 --- a/tests/test_security_utils.py +++ b/tests/test_security_utils.py @@ -178,8 +178,22 @@ class TestSafeUrlPolicy: ): assert validate_external_url("https://mixed.example/") is False - def test_allow_private_does_not_allow_metadata(self): - with patch("socket.getaddrinfo", return_value=_addrinfo("169.254.169.254")): + @pytest.mark.parametrize( + "resolved", + [ + "169.254.169.254", # AWS/Azure/GCP IMDS + "169.254.170.2", # AWS ECS task credentials + "100.100.100.200", # Alibaba Cloud + # IPv4-mapped IPv6 spellings must not bypass the metadata block: + # the mapped form is a distinct address object absent from the + # metadata set and reports is_reserved=False. + "::ffff:169.254.169.254", + "::ffff:169.254.170.2", + "::ffff:100.100.100.200", + ], + ) + def test_allow_private_does_not_allow_metadata(self, resolved): + with patch("socket.getaddrinfo", return_value=_addrinfo(resolved)): assert validate_external_url( "http://metadata.example/", allow_private=True,