diff --git a/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py b/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py index a3ede15f0f3c..ddc860053267 100644 --- a/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py +++ b/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py @@ -16,6 +16,30 @@ DEFAULT_ALLOWED_SCHEME = "https" +# Azure instance metadata service (WireServer). Unlike the AWS/GCP equivalents, this +# address is publicly routable, so it would otherwise pass the private-address checks. +_AZURE_WIRE_SERVER = ipaddress.ip_address("168.63.129.16") +_AZURE_WIRE_SERVER_CATEGORY = "Azure metadata (WireServer)" + +# Well-known NAT64 prefixes (RFC 6052): the global 64:ff9b::/96 and the local-use +# 64:ff9b:1::/48. IPv4 addresses embedded in these ranges must be classified too. +_NAT64_PREFIXES = (ipaddress.ip_network("64:ff9b::/96"), ipaddress.ip_network("64:ff9b:1::/48")) + +# Cloud instance metadata services other than WireServer. These answer with instance +# credentials, so they stay blocked even when allow_private_network_access=True. Each one +# sits inside a range the private-address checks already flag, but the category it lands in +# is generic -- "link-local", "carrier-grade NAT", "private (IPv6 ULA)" -- and those +# categories also cover ordinary private addresses that the option is meant to permit, so +# the gate cannot identify these by category and has to name them. +# 169.254.169.254 IMDS, shared by AWS, Azure, GCP and OCI +# 169.254.170.2 ECS task metadata +# 100.100.100.200 Alibaba Cloud +# fd00:ec2::254 AWS IMDS over IPv6 +_CLOUD_METADATA_ADDRESSES = frozenset( + ipaddress.ip_address(address) + for address in ("169.254.169.254", "169.254.170.2", "100.100.100.200", "fd00:ec2::254") +) + class ServerUrlValidationOptions(KernelBaseModel): """Options for validating OpenAPI operation request URLs.""" @@ -59,6 +83,9 @@ async def validate_server_url( ) if options.allow_private_network_access: + # Allowing access to a private network is not the same as allowing access to + # the host agent's cloud metadata endpoint, which is always blocked. + _reject_cloud_metadata_host(parsed_url) return await _ensure_public_host(parsed_url, dns_resolver) @@ -70,8 +97,12 @@ def try_categorize_non_public_address( """Return whether an IP address is non-public and the category when blocked.""" ip_address = ipaddress.ip_address(address) - if isinstance(ip_address, ipaddress.IPv6Address) and ip_address.ipv4_mapped: - ip_address = ip_address.ipv4_mapped + if isinstance(ip_address, ipaddress.IPv6Address): + embedded_ipv4 = _extract_embedded_ipv4(ip_address) + if embedded_ipv4 is not None: + ip_address = embedded_ipv4 + elif ip_address.ipv4_mapped: + ip_address = ip_address.ipv4_mapped if isinstance(ip_address, ipaddress.IPv4Address): return _try_classify_ipv4(ip_address) @@ -79,6 +110,24 @@ def try_categorize_non_public_address( return _try_classify_ipv6(ip_address) +def _extract_embedded_ipv4(address: ipaddress.IPv6Address) -> ipaddress.IPv4Address | None: + """Decode an IPv4 address embedded in an IPv6 address, if any. + + Covers IPv4-mapped (``::ffff:a.b.c.d``), 6to4 (``2002::/16``, RFC 3056), Teredo + (``2001::/32``, RFC 4380) and NAT64 (RFC 6052) addresses. The embedded IPv4 is + classified separately so a private address cannot slip through an otherwise + public-looking IPv6 address. + """ + if address.sixtofour is not None: + return address.sixtofour + if address.teredo is not None: + _, teredo_client = address.teredo + return teredo_client + if any(address in prefix for prefix in _NAT64_PREFIXES): + return ipaddress.ip_address(address.packed[-4:]) + return None + + def _parse_absolute_url(url: str, option_name: str = "url") -> ParseResult: parsed_url = urlparse(url) try: @@ -191,9 +240,66 @@ def _ensure_public_address(url: str, address: ipaddress.IPv4Address | ipaddress. ) +def _reject_cloud_metadata_host(parsed_url: ParseResult) -> None: + """Block cloud metadata endpoints even when private network access is allowed. + + Only literal IP hosts are checked: private-network mode deliberately does not + resolve hostnames, so a hostname pointing at a metadata endpoint is out of scope. + + WireServer is recognised by its category. The remaining metadata endpoints share a + category with ordinary private addresses, so they are matched by address instead; + without that, ``169.254.169.254`` -- the IMDS address for AWS, GCP and Azure alike -- + would be permitted here while the call site claims metadata endpoints are always + blocked. + """ + host = parsed_url.hostname + if host is None: + return + + try: + ip_address = ipaddress.ip_address(host) + except ValueError: + return + + blocked, category = try_categorize_non_public_address(ip_address) + if not blocked: + return + + if category == _AZURE_WIRE_SERVER_CATEGORY: + raise FunctionExecutionException( + f"The request URI '{parsed_url.geturl()}' is not allowed: the host is the Azure " + f"metadata endpoint (WireServer, {ip_address}), which is blocked even when " + "allow_private_network_access=True to prevent SSRF against cloud metadata services." + ) + + if _is_cloud_metadata_address(ip_address): + raise FunctionExecutionException( + f"The request URI '{parsed_url.geturl()}' is not allowed: the host is a cloud " + f"instance metadata endpoint ({ip_address}), which is blocked even when " + "allow_private_network_access=True to prevent SSRF against cloud metadata services." + ) + + +def _is_cloud_metadata_address(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + """Return whether an address is a well-known cloud instance metadata endpoint. + + IPv6 forms that embed an IPv4 address are resolved to the embedded address first, so + ``::ffff:169.254.169.254`` and ``64:ff9b::169.254.169.254`` are recognised as well. + """ + if isinstance(address, ipaddress.IPv6Address): + embedded_ipv4 = _extract_embedded_ipv4(address) + if embedded_ipv4 is not None: + return embedded_ipv4 in _CLOUD_METADATA_ADDRESSES + if address.ipv4_mapped is not None: + return address.ipv4_mapped in _CLOUD_METADATA_ADDRESSES + return address in _CLOUD_METADATA_ADDRESSES + + def _try_classify_ipv4(address: ipaddress.IPv4Address) -> tuple[bool, str]: b0, b1, b2, _ = address.packed + if address == _AZURE_WIRE_SERVER: + return True, _AZURE_WIRE_SERVER_CATEGORY if b0 == 0: return True, "unspecified" if b0 == 10: diff --git a/python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py b/python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py index 177ff15cb2c6..ec8927b078f8 100644 --- a/python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py +++ b/python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py @@ -43,6 +43,11 @@ ("2001:db8::1", "reserved"), ("::ffff:127.0.0.1", "loopback"), ("::ffff:169.254.169.254", "link-local"), + ("168.63.129.16", "Azure metadata (WireServer)"), + ("64:ff9b::169.254.169.254", "link-local"), + ("64:ff9b::a83f:8110", "Azure metadata (WireServer)"), + ("2002:a9fe:a9fe::", "link-local"), + ("2001:0:4136:e378:8000:63bf:3fff:fdd2", "reserved"), ], ) def test_try_categorize_non_public_address(address: str, expected_category: str): @@ -168,3 +173,71 @@ 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) + + +async def test_validate_server_url_rejects_literal_azure_wire_server(): + with pytest.raises(FunctionExecutionException, match="Azure metadata"): + await validate_server_url("https://168.63.129.16/machine/") + + +async def test_validate_server_url_rejects_nat64_embedded_link_local(): + with pytest.raises(FunctionExecutionException, match="link-local"): + await validate_server_url("https://[64:ff9b::169.254.169.254]/latest/meta-data/") + + +async def test_validate_server_url_rejects_6to4_embedded_link_local(): + with pytest.raises(FunctionExecutionException, match="link-local"): + await validate_server_url("https://[2002:a9fe:a9fe::]/latest/meta-data/") + + +async def test_validate_server_url_blocks_wireserver_even_with_private_network_access(): + options = ServerUrlValidationOptions(allow_private_network_access=True) + + with pytest.raises(FunctionExecutionException, match="Azure metadata"): + await validate_server_url("https://168.63.129.16/machine/", options) + + +async def test_validate_server_url_blocks_hostname_resolving_to_azure_wire_server(): + async def fake_resolver(host: str): + assert host == "wireserver-host.example.com" + return ["168.63.129.16"] + + with pytest.raises(FunctionExecutionException, match="Azure metadata"): + await validate_server_url("https://wireserver-host.example.com/machine/", dns_resolver=fake_resolver) + + +@pytest.mark.parametrize( + "url", + [ + "https://169.254.169.254/latest/meta-data/", + "https://169.254.170.2/v2/credentials", + "https://100.100.100.200/latest/meta-data/", + "https://[fd00:ec2::254]/latest/meta-data/", + "https://[::ffff:169.254.169.254]/latest/meta-data/", + "https://[64:ff9b::169.254.169.254]/latest/meta-data/", + ], +) +async def test_validate_server_url_blocks_cloud_metadata_even_with_private_network_access(url: str): + """The call site promises metadata endpoints are always blocked; these are the ones that + share a category with ordinary private addresses and so are not caught by category.""" + options = ServerUrlValidationOptions(allow_private_network_access=True) + + with pytest.raises(FunctionExecutionException, match="cloud instance metadata endpoint"): + await validate_server_url(url, options) + + +@pytest.mark.parametrize( + "url", + [ + "https://10.0.0.5/", + "https://172.16.0.1/", + "https://192.168.1.50/v1", + # Same category as the Alibaba Cloud metadata address above, so this also pins the + # blocklist to named addresses rather than to the category they fall in. + "https://100.64.0.1/", + ], +) +async def test_validate_server_url_still_allows_ordinary_private_addresses(url: str): + options = ServerUrlValidationOptions(allow_private_network_access=True) + + await validate_server_url(url, options)