Python: block cloud metadata endpoints and IPv6-embedded IPv4 in server_url_validator

python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py

@@ -16,6 +16,39 @@
 
 DEFAULT_ALLOWED_SCHEME = "https"
 
+# Cloud metadata / credential endpoints. Most fall inside ranges `_try_classify_ipv4` already
+# rejects, but `168.63.129.16` (Azure WireServer) is publicly routable, so no range check reaches
+# it. They are listed explicitly and checked even when `allow_private_network_access` is set:
+# reaching a host on your own network and reaching the instance's credential endpoint are
+# different requests, and only the first is what that option is for.
+CLOUD_METADATA_ADDRESSES: frozenset[ipaddress.IPv4Address | ipaddress.IPv6Address] = frozenset(
+    ipaddress.ip_address(address)
+    for address in (
+        "169.254.169.254",  # AWS IMDS, GCP, Azure, OCI, DigitalOcean, Hetzner, OpenStack
+        "169.254.170.2",  # AWS ECS task IAM role credentials
+        "169.254.170.23",  # AWS EKS Pod Identity Agent
+        "168.63.129.16",  # Azure WireServer / platform channel (publicly routable)
+        "100.100.100.200",  # Alibaba Cloud
+        "192.0.0.192",  # Oracle Cloud (Classic)
+        "169.254.42.42",  # Scaleway
+        "fd00:ec2::254",  # AWS IMDS over IPv6
+        "fd00:ec2::23",  # AWS EKS Pod Identity Agent over IPv6
+    )
+)
+
+# NAT64 (RFC 6052) and 6to4 (RFC 3056) carry an IPv4 address inside the IPv6 one. Only the /96
+# embedding is decoded: it is the only length the well-known prefix allows, and guessing the
+# shorter lengths inside the RFC 8215 local-use prefix reads bytes that are not the embedded
+# address, which would reject legitimate NAT64 targets.
+_NAT64_IPV4_OFFSETS = (12, 13, 14, 15)
+_SIXTOFOUR_OFFSETS = (2, 3, 4, 5)
+_NAT64_NETWORKS: tuple[ipaddress.IPv6Network, ...] = (
+    ipaddress.IPv6Network("64:ff9b::/96"),
+    ipaddress.IPv6Network("64:ff9b:1::/48"),
+)
+_SIXTOFOUR_NETWORK = ipaddress.IPv6Network("2002::/16")
+_TEREDO_NETWORK = ipaddress.IPv6Network("2001::/32")
+
 
 class ServerUrlValidationOptions(KernelBaseModel):
     """Options for validating OpenAPI operation request URLs."""
@@ -58,10 +91,17 @@ async def validate_server_url(
             "To allow this URL, add it to server_url_validation_allowed_base_urls."
         )
 
-    if options.allow_private_network_access:
-        return
+    await _ensure_public_host(
+        parsed_url, dns_resolver, allow_private_network_access=options.allow_private_network_access
+    )
+
 
-    await _ensure_public_host(parsed_url, dns_resolver)
+def is_cloud_metadata_address(address: str | ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    """Return whether an address, or an IPv4 embedded in it, is a cloud metadata endpoint."""
+    ip_address = ipaddress.ip_address(address)
+    if ip_address in CLOUD_METADATA_ADDRESSES:
+        return True
+    return any(embedded in CLOUD_METADATA_ADDRESSES for embedded in _embedded_ipv4s(ip_address))
 
 
 def try_categorize_non_public_address(
@@ -76,7 +116,42 @@ def try_categorize_non_public_address(
     if isinstance(ip_address, ipaddress.IPv4Address):
         return _try_classify_ipv4(ip_address)
 
-    return _try_classify_ipv6(ip_address)
+    blocked, category = _try_classify_ipv6(ip_address)
+    if blocked:
+        return blocked, category
+
+    # 6to4, NAT64 and Teredo carry an IPv4 target inside an otherwise public-looking IPv6
+    # address. Decode those and classify the IPv4 they name.
+    for embedded in _embedded_ipv4s(ip_address):
+        blocked, category = _try_classify_ipv4(embedded)
+        if blocked:
+            return blocked, f"{category} (embedded in IPv6)"
+
+    return False, ""
+
+
+def _embedded_ipv4s(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> list[ipaddress.IPv4Address]:
+    """Decode the IPv4 addresses carried inside IPv4-mapped, NAT64, 6to4 and Teredo IPv6 forms."""
+    if not isinstance(address, ipaddress.IPv6Address):
+        return []
+
+    packed = address.packed
+    candidates: list[ipaddress.IPv4Address] = []
+
+    if address.ipv4_mapped is not None:
+        candidates.append(address.ipv4_mapped)
+
+    if any(address in network for network in _NAT64_NETWORKS):
+        candidates.append(ipaddress.IPv4Address(bytes(packed[offset] for offset in _NAT64_IPV4_OFFSETS)))
+
+    if address in _SIXTOFOUR_NETWORK:
+        candidates.append(ipaddress.IPv4Address(bytes(packed[offset] for offset in _SIXTOFOUR_OFFSETS)))
+
+    if address in _TEREDO_NETWORK:
+        # RFC 4380: the client IPv4 sits in the low 32 bits, obfuscated by XOR with all-ones.
+        candidates.append(ipaddress.IPv4Address(bytes(byte ^ 0xFF for byte in packed[12:16])))
+
+    return candidates
 
 
 def _parse_absolute_url(url: str, option_name: str = "url") -> ParseResult:
@@ -127,7 +202,9 @@ def _matches_path_prefix(url_path: str, base_path: str) -> bool:
     return url_path.lower().startswith(base_path_with_slash.lower())
 
 
-async def _ensure_public_host(parsed_url: ParseResult, dns_resolver: DnsResolver | None) -> None:
+async def _ensure_public_host(
+    parsed_url: ParseResult, dns_resolver: DnsResolver | None, allow_private_network_access: bool = False
+) -> None:
     host = parsed_url.hostname
     if host is None:
         raise FunctionExecutionException(f"The request URI '{parsed_url.geturl()}' does not contain a valid host.")
@@ -137,7 +214,7 @@ async def _ensure_public_host(parsed_url: ParseResult, dns_resolver: DnsResolver
     except ValueError:
         addresses = await _resolve_host(host, dns_resolver)
     else:
-        _ensure_public_address(parsed_url.geturl(), ip_address)
+        _ensure_public_address(parsed_url.geturl(), ip_address, allow_private_network_access)
         return
 
     if not addresses:
@@ -147,7 +224,7 @@ async def _ensure_public_host(parsed_url: ParseResult, dns_resolver: DnsResolver
         )
 
     for address in addresses:
-        _ensure_public_address(parsed_url.geturl(), address)
+        _ensure_public_address(parsed_url.geturl(), address, allow_private_network_access)
 
 
 async def _resolve_host(
@@ -180,7 +257,18 @@ async def _resolve_host(
     return addresses
 
 
-def _ensure_public_address(url: str, address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None:
+def _ensure_public_address(
+    url: str, address: ipaddress.IPv4Address | ipaddress.IPv6Address, allow_private_network_access: bool = False
+) -> None:
+    if is_cloud_metadata_address(address):
+        raise FunctionExecutionException(
+            f"The request URI '{url}' is not allowed: host resolves to a cloud metadata endpoint ({address}), "
+            "which is blocked to prevent Server-Side Request Forgery (SSRF). To allow this URL, add it to "
+            "server_url_validation_allowed_base_urls."
+        )
+    if allow_private_network_access:
+        return
+
     blocked, category = try_categorize_non_public_address(address)
     if blocked:
         raise FunctionExecutionException(

python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py

@@ -1,8 +1,10 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import ipaddress
 import socket
 
 import pytest
+from pytest import raises
 
 from semantic_kernel.connectors.openapi_plugin.server_url_validator import (
     ServerUrlValidationOptions,
@@ -17,7 +19,7 @@
     [
         ("127.0.0.1", "loopback"),
         ("127.255.255.254", "loopback"),
-        ("169.254.169.254", "link-local"),
+        ("169.254.10.10", "link-local"),
         ("169.254.0.1", "link-local"),
         ("10.0.0.1", "private (RFC1918)"),
         ("172.16.0.1", "private (RFC1918)"),
@@ -76,7 +78,7 @@ def test_try_categorize_non_public_address_allows_public_addresses(address: str)
 
 async def test_validate_server_url_rejects_literal_link_local_ipv4():
     with pytest.raises(FunctionExecutionException, match="link-local"):
-        await validate_server_url("https://169.254.169.254/latest/meta-data/")
+        await validate_server_url("https://169.254.10.10/latest/meta-data/")
 
 
 async def test_validate_server_url_rejects_literal_loopback_ipv6():
@@ -120,7 +122,7 @@ async def test_validate_server_url_allows_private_network_access_after_scheme_ga
 async def test_validate_server_url_blocks_hostname_resolving_to_link_local():
     async def fake_resolver(host: str):
         assert host == "evil.example.com"
-        return ["169.254.169.254"]
+        return ["169.254.10.10"]
 
     with pytest.raises(FunctionExecutionException, match="link-local"):
         await validate_server_url("https://evil.example.com/latest/meta-data/", dns_resolver=fake_resolver)
@@ -168,3 +170,76 @@ async def fake_resolver(host: str):
 
     with pytest.raises(FunctionExecutionException, match="returned no addresses"):
         await validate_server_url("https://empty-dns.example.com/", dns_resolver=fake_resolver)
+
+
+CLOUD_METADATA_ENDPOINTS = [
+    "169.254.169.254",  # AWS IMDS, GCP, Azure, OCI, DigitalOcean
+    "169.254.170.2",  # AWS ECS task IAM role credentials
+    "169.254.170.23",  # AWS EKS Pod Identity Agent
+    "168.63.129.16",  # Azure WireServer (publicly routable)
+    "100.100.100.200",  # Alibaba Cloud
+    "192.0.0.192",  # Oracle Cloud (Classic)
+    "169.254.42.42",  # Scaleway
+]
+
+
+@pytest.mark.parametrize("address", CLOUD_METADATA_ENDPOINTS)
+async def test_cloud_metadata_endpoints_blocked(address):
+    with raises(FunctionExecutionException):
+        await validate_server_url(f"https://{address}/latest/meta-data/")
+
+
+@pytest.mark.parametrize("address", CLOUD_METADATA_ENDPOINTS)
+async def test_cloud_metadata_endpoints_blocked_with_private_access(address):
+    """`allow_private_network_access` covers your own network, not the credential endpoint."""
+    options = ServerUrlValidationOptions(allow_private_network_access=True)
+    with raises(FunctionExecutionException, match="cloud metadata endpoint"):
+        await validate_server_url(f"https://{address}/latest/meta-data/", options)
+
+
+async def test_private_network_access_still_permits_rfc1918():
+    options = ServerUrlValidationOptions(allow_private_network_access=True)
+    await validate_server_url("https://10.0.0.5/resource", options)
+
+
+async def test_private_network_access_still_permits_loopback():
+    options = ServerUrlValidationOptions(allow_private_network_access=True)
+    await validate_server_url("https://127.0.0.1/resource", options)
+
+
+@pytest.mark.parametrize(
+    "address",
+    [
+        "64:ff9b::169.254.169.254",  # NAT64 well-known prefix (RFC 6052)
+        "64:ff9b:1::169.254.169.254",  # NAT64 local-use prefix (RFC 8215)
+        "2002:a9fe:a9fe::",  # 6to4 (RFC 3056)
+        "::ffff:169.254.169.254",  # IPv4-mapped
+    ],
+)
+async def test_ipv6_forms_carrying_a_blocked_ipv4_are_rejected(address):
+    with raises(FunctionExecutionException):
+        await validate_server_url(f"https://[{address}]/latest/meta-data/")
+
+
+async def test_teredo_carrying_a_blocked_ipv4_is_rejected():
+    """Teredo obfuscates the client IPv4 by XOR-ing the low 32 bits with all ones."""
+    target = ipaddress.IPv4Address("169.254.169.254")
+    low = bytes(byte ^ 0xFF for byte in target.packed)
+    teredo = ipaddress.IPv6Address(b"\x20\x01\x00\x00" + b"\x00" * 8 + low)
+
+    with raises(FunctionExecutionException):
+        await validate_server_url(f"https://[{teredo}]/latest/meta-data/")
+
+
+@pytest.mark.parametrize(
+    "address",
+    [
+        "64:ff9b::1.1.1.1",  # NAT64 pointing at a public IPv4
+        "64:ff9b:1::1.1.1.1",
+        "2002:0101:0101::",  # 6to4 pointing at 1.1.1.1
+        "2606:4700:4700::1111",  # plain public IPv6
+    ],
+)
+async def test_ipv6_forms_carrying_a_public_ipv4_are_allowed(address):
+    """Decoding must not reject legitimate NAT64/6to4 targets."""
+    await validate_server_url(f"https://[{address}]/resource")