From 9c178c3af928316d7ec6e61a42f60da2971d6556 Mon Sep 17 00:00:00 2001 From: Diwak4r Date: Tue, 4 Aug 2026 14:20:03 +0545 Subject: [PATCH 1/2] 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) From 673aefe749d03999b2db4d785ff7cd56a2e4b8c3 Mon Sep 17 00:00:00 2001 From: Diwak4r Date: Fri, 11 Sep 2026 09:06:22 +0545 Subject: [PATCH 2/2] fix(openapi_plugin): block cloud metadata addresses in private-network mode The call site documents that the host agent's cloud metadata endpoint "is always blocked", but _reject_cloud_metadata_host only rejected the WireServer category, so every other metadata endpoint passed when allow_private_network_access=True: 168.63.129.16 Azure metadata (WireServer) rejected 169.254.169.254 link-local allowed 169.254.170.2 link-local allowed 100.100.100.200 carrier-grade NAT allowed fd00:ec2::254 private (IPv6 ULA) allowed 169.254.169.254 is the IMDS address shared by AWS, GCP and Azure, so the gap covered the endpoint most SSRF write-ups mean. The category is not a usable signal here: "link-local", "carrier-grade NAT" and "private (IPv6 ULA)" also cover ordinary private addresses that the option exists to permit, and widening the gate to those categories would remove the feature. Match the named metadata addresses instead, resolving IPv4-mapped and NAT64-embedded IPv6 forms to the address they carry so ::ffff:169.254.169.254 and 64:ff9b::169.254.169.254 are caught as well. Ordinary private addresses are unaffected: 10.0.0.5, 172.16.0.1, 192.168.1.50 and 100.64.0.1 (same category as the Alibaba Cloud address) still pass. Adds 10 tests. Verified: 72 passed with this change; 6 failed, 66 passed without it, the failures being exactly the six metadata cases above. AI-assisted change, reviewed against the test output above. --- .../openapi_plugin/server_url_validator.py | 48 ++++++++++++++++++- .../test_server_url_validator.py | 37 ++++++++++++++ 2 files changed, 84 insertions(+), 1 deletion(-) 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 123770a5ae89..ddc860053267 100644 --- a/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py +++ b/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py @@ -25,6 +25,21 @@ # 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.""" @@ -230,6 +245,12 @@ def _reject_cloud_metadata_host(parsed_url: ParseResult) -> None: 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: @@ -241,13 +262,38 @@ def _reject_cloud_metadata_host(parsed_url: ParseResult) -> None: return blocked, category = try_categorize_non_public_address(ip_address) - if blocked and category == _AZURE_WIRE_SERVER_CATEGORY: + 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 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 e0f9e64a4c96..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 @@ -204,3 +204,40 @@ async def fake_resolver(host: str): 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)