From 7239c47546e03f66541e629233ba4039bafb542c Mon Sep 17 00:00:00 2001 From: ErenAta16 Date: Wed, 29 Jul 2026 22:25:42 +0300 Subject: [PATCH] Python: block cloud metadata endpoints and IPv6-embedded IPv4 in server_url_validator server_url_validator is the default SSRF guard for OpenAPI plugin calls, and the URL it sees is shaped by the model through server variables and path parameters. Two categories get through it: 168.63.129.16 -- Azure WireServer, reachable from every Azure VM. It is publicly routable, so none of the _try_classify_ipv4 branches match it. NAT64, 6to4 and Teredo addresses. _try_classify_ipv6 unwraps ::ffff: via ipv4_mapped but does not decode the other encodings, so 64:ff9b::169.254.169.254 and 2002:a9fe:a9fe:: both pass. Add an explicit metadata denylist, checked before the range logic and also 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. RFC1918 and loopback still pass with the option on. Decode the embedded IPv4 for NAT64 (RFC 6052 / RFC 8215), 6to4 (RFC 3056) and Teredo (RFC 4380) and classify that. Only the /96 embedding is decoded: guessing the shorter lengths inside the local-use prefix reads bytes that are not the embedded address, which rejected legitimate NAT64 targets like 64:ff9b:1::1.1.1.1 -- there is a test pinning that those stay allowed. Two existing tests moved off 169.254.169.254 onto a plain link-local address. It is still blocked, now with the more specific metadata message, and the new tests assert on it directly. Closes #14240 --- .../openapi_plugin/server_url_validator.py | 104 ++++++++++++++++-- .../test_server_url_validator.py | 81 +++++++++++++- 2 files changed, 174 insertions(+), 11 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..df0c16f4f81d 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,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( 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..e5bcc5bfb3f3 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 @@ -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")