From 81fa7b6484af25702991dedf2c1beb46eeccf5cd Mon Sep 17 00:00:00 2001 From: Diwak4r Date: Tue, 4 Aug 2026 14:20:03 +0545 Subject: [PATCH] fix(openapi_plugin): block Azure WireServer and IPv6-embedded private addresses in server URL validation The server URL validator could be bypassed to reach cloud metadata endpoints: the Azure WireServer IP 168.63.129.16 is publicly routable so it passed the private-address checks, and IPv6 addresses embedding an IPv4 (NAT64, 6to4, Teredo) were classified purely as IPv6, letting link-local and other private IPv4 addresses through. - Denylist the Azure WireServer metadata endpoint in _try_classify_ipv4 and enforce it even when allow_private_network_access is enabled. - Decode IPv4 addresses embedded in IPv6 (6to4 via sixtofour, Teredo via teredo, NAT64 64:ff9b::/96 and 64:ff9b:1::/48) before classification so embedded private addresses are blocked. - Add tests covering the WireServer endpoint, all three IPv6 embedding forms, and the private-network-access override. --- .../openapi_plugin/server_url_validator.py | 64 ++++++++++++++++++- .../test_server_url_validator.py | 36 +++++++++++ 2 files changed, 98 insertions(+), 2 deletions(-) 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..123770a5ae89 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,15 @@ 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")) + class ServerUrlValidationOptions(KernelBaseModel): """Options for validating OpenAPI operation request URLs.""" @@ -59,6 +68,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 +82,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 +95,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 +225,35 @@ 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. + """ + 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 blocked and 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." + ) + + 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..e0f9e64a4c96 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,34 @@ 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)