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")