diff --git a/python/semantic_kernel/connectors/openapi_plugin/openapi_runner.py b/python/semantic_kernel/connectors/openapi_plugin/openapi_runner.py index ee88d4111ffb..b21bdcfa57e4 100644 --- a/python/semantic_kernel/connectors/openapi_plugin/openapi_runner.py +++ b/python/semantic_kernel/connectors/openapi_plugin/openapi_runner.py @@ -1,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. +import ipaddress import json import logging from collections import OrderedDict @@ -7,6 +8,7 @@ from inspect import isawaitable from typing import Any from urllib.parse import urlparse, urlunparse +from urllib.request import getproxies import httpx from openapi_core import Spec @@ -29,6 +31,33 @@ logger: logging.Logger = logging.getLogger(__name__) +def _pin_url_to_address(url: str, address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> tuple[str, str, str]: + """Rewrite a URL so the connection targets `address` while keeping the original host identity. + + Returns the address-form URL, the `Host` header value and the TLS SNI hostname, all + derived from the original URL so that the request on the wire is unchanged apart from + the address it is delivered to. IPv6 bracketing, the port and any userinfo are + preserved by `httpx.URL.copy_with`. + """ + original = httpx.URL(url) + return ( + str(original.copy_with(host=str(address))), + original.netloc.decode("ascii"), + original.raw_host.decode("ascii"), + ) + + +def _has_environment_proxy() -> bool: + """Return whether a proxy is configured in the environment for outbound HTTP requests. + + Deliberately conservative: any configured proxy disables address pinning, because a + proxy resolves the target name itself, so an address resolved locally is neither the + one used for the connection nor necessarily reachable or correct from the proxy. + """ + proxies = getproxies() + return any(proxies.get(scheme) for scheme in ("http", "https", "all")) + + @experimental class OpenApiRunner: """The OpenApiRunner that runs the operations defined in the OpenAPI manifest.""" @@ -134,7 +163,13 @@ async def run_operation( arguments: KernelArguments | None = None, options: RestApiRunOptions | None = None, ) -> str: - """Runs the operation defined in the OpenAPI manifest.""" + """Runs the operation defined in the OpenAPI manifest. + + When the URL is validated by DNS resolution, the request issued by the built-in + client is pinned to one of the validated addresses. Requests made through a + caller-supplied `http_client`, or while an environment proxy is configured, use + that transport's own name resolution and are not pinned. + """ if not arguments: arguments = KernelArguments() url = self.build_operation_url( @@ -143,7 +178,7 @@ async def run_operation( server_url_override=options.server_url_override if options else None, api_host_url=options.api_host_url if options else None, ) - await validate_server_url(url, self.server_url_validation_options) + validated_addresses = await validate_server_url(url, self.server_url_validation_options) headers = operation.build_headers(arguments=arguments) payload, _ = self.build_operation_payload(operation=operation, arguments=arguments) @@ -168,22 +203,52 @@ async def run_operation( timeout = options.timeout if options and hasattr(options, "timeout") and options.timeout is not None else None + # Pin the connection to an address the validator actually vetted so that a name which + # resolves to a public address during validation cannot resolve to a private one at + # connect time (DNS rebinding). The list is empty when there is nothing to pin. + pinned_addresses = validated_addresses + if pinned_addresses and _has_environment_proxy(): + logger.debug("An environment proxy is configured; the OpenAPI request address is not pinned.") + pinned_addresses = [] + async def fetch(): - async def make_request(client: httpx.AsyncClient): + async def make_request( + client: httpx.AsyncClient, + pin_to: ipaddress.IPv4Address | ipaddress.IPv6Address | None = None, + ): merged_headers = client.headers.copy() merged_headers.update(headers) + request_url = url + extensions: dict[str, Any] = {} + if pin_to is not None: + request_url, merged_headers["Host"], extensions["sni_hostname"] = _pin_url_to_address(url, pin_to) response = await client.request( method=operation.method, - url=url, + url=request_url, headers=merged_headers, json=json.loads(payload) if payload else None, + extensions=extensions, ) response.raise_for_status() return response.text if hasattr(self, "http_client") and self.http_client is not None: + # A caller-supplied client owns its transport configuration (proxies, mounts, + # custom resolvers), so its connections are left untouched. return await make_request(self.http_client) async with httpx.AsyncClient(timeout=timeout) as client: - return await make_request(client) + if not pinned_addresses: + return await make_request(client) + # Every vetted address is an acceptable target, so keep the resolver's + # fallback behaviour by trying the next one when a connection cannot be + # established. Only connection failures are retried, so no request is + # ever delivered more than once. + *fallback_addresses, final_address = pinned_addresses + for address in fallback_addresses: + try: + return await make_request(client, pin_to=address) + except (httpx.ConnectError, httpx.ConnectTimeout): + logger.debug("Could not connect to validated address %s, trying the next one.", address) + return await make_request(client, pin_to=final_address) return await fetch() 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..5daed30d5a74 100644 --- a/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py +++ b/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py @@ -33,8 +33,17 @@ async def validate_server_url( url: str, options: ServerUrlValidationOptions | None = None, dns_resolver: DnsResolver | None = None, -) -> None: - """Validate a fully resolved OpenAPI operation URL against the supplied policy.""" +) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + """Validate a fully resolved OpenAPI operation URL against the supplied policy. + + Returns the DNS-resolved addresses that were vetted by this call, in resolver order, + so that the caller can pin the connection to an address the policy actually approved + (closing the DNS check-time/use-time gap known as DNS rebinding). + + The list is empty whenever there is nothing to pin, and callers must then connect + normally: when an allowed base URL matched, when ``allow_private_network_access`` + is set, or when the host is already a literal IP address (which cannot be rebound). + """ options = options or ServerUrlValidationOptions() try: parsed_url = _parse_absolute_url(url) @@ -44,7 +53,7 @@ async def validate_server_url( ) from exc if _matches_allowed_base_url(parsed_url, options.allowed_base_urls): - return + return [] if options.allowed_base_urls: raise FunctionExecutionException( @@ -59,9 +68,9 @@ async def validate_server_url( ) if options.allow_private_network_access: - return + return [] - await _ensure_public_host(parsed_url, dns_resolver) + return await _ensure_public_host(parsed_url, dns_resolver) def try_categorize_non_public_address( @@ -127,7 +136,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 +) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: host = parsed_url.hostname if host is None: raise FunctionExecutionException(f"The request URI '{parsed_url.geturl()}' does not contain a valid host.") @@ -138,7 +149,8 @@ async def _ensure_public_host(parsed_url: ParseResult, dns_resolver: DnsResolver addresses = await _resolve_host(host, dns_resolver) else: _ensure_public_address(parsed_url.geturl(), ip_address) - return + # A literal IP address cannot be rebound between validation and connection. + return [] if not addresses: raise FunctionExecutionException( @@ -148,6 +160,7 @@ async def _ensure_public_host(parsed_url: ParseResult, dns_resolver: DnsResolver for address in addresses: _ensure_public_address(parsed_url.geturl(), address) + return addresses async def _resolve_host( diff --git a/python/tests/unit/connectors/openapi_plugin/test_openapi_runner_dns_pinning.py b/python/tests/unit/connectors/openapi_plugin/test_openapi_runner_dns_pinning.py new file mode 100644 index 000000000000..5cdb7d38ef20 --- /dev/null +++ b/python/tests/unit/connectors/openapi_plugin/test_openapi_runner_dns_pinning.py @@ -0,0 +1,307 @@ +# Copyright (c) Microsoft. All rights reserved. + +import ipaddress +import socket +from collections import OrderedDict +from unittest.mock import MagicMock + +import httpcore +import httpx +import pytest + +from semantic_kernel.connectors.openapi_plugin.openapi_runner import OpenApiRunner +from semantic_kernel.connectors.openapi_plugin.server_url_validator import ( + ServerUrlValidationOptions, + try_categorize_non_public_address, +) + +HOST = "rebind.example" +PUBLIC_ADDRESS = "93.184.216.34" +SECOND_PUBLIC_ADDRESS = "198.41.0.4" +REBOUND_ADDRESS = "169.254.169.254" +PUBLIC_IPV6_ADDRESS = "2606:2800:220:1:248:1893:25c8:1946" + +RAW_RESPONSE = b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 13\r\n\r\nresponse text" + + +def build_runner(url: str, options: ServerUrlValidationOptions | None = None) -> tuple[OpenApiRunner, MagicMock]: + """Build a runner whose operation resolves to `url` and carries no payload.""" + runner = OpenApiRunner({}, server_url_validation_options=options) + operation = MagicMock() + operation.method = "GET" + operation.build_headers.return_value = {} + operation.responses = OrderedDict() + runner.build_operation_url = MagicMock(return_value=url) + runner.build_operation_payload = MagicMock(return_value=(None, None)) + return runner, operation + + +def static_getaddrinfo(host_name: str, addresses: list[str]): + """Return a `socket.getaddrinfo` replacement that answers `host_name` with `addresses`.""" + real_getaddrinfo = socket.getaddrinfo + + def fake_getaddrinfo(host, port=None, *args, **kwargs): + if host == host_name: + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, port or 0)) for address in addresses] + return real_getaddrinfo(host, port, *args, **kwargs) + + return fake_getaddrinfo + + +def rebinding_getaddrinfo(host_name: str, first_address: str, later_address: str, lookups: list[str]): + """Return a `socket.getaddrinfo` replacement that answers `host_name` differently after the first lookup.""" + real_getaddrinfo = socket.getaddrinfo + + def fake_getaddrinfo(host, port=None, *args, **kwargs): + if host == host_name: + lookups.append(host) + address = first_address if len(lookups) == 1 else later_address + family = socket.AF_INET6 if ":" in address else socket.AF_INET + return [(family, socket.SOCK_STREAM, 6, "", (address, port or 0))] + return real_getaddrinfo(host, port, *args, **kwargs) + + return fake_getaddrinfo + + +class RecordingStream(httpcore.AsyncMockStream): + """A mock network stream that records the TLS SNI hostname and the bytes written to the wire.""" + + def __init__(self, buffer: list[bytes], record: dict) -> None: + super().__init__(buffer) + self._record = record + + async def write(self, buffer: bytes, timeout: float | None = None) -> None: + self._record["written"] += buffer + + async def start_tls(self, ssl_context, server_hostname=None, timeout=None): + self._record["sni_hostname"] = server_hostname + return self + + +class RecordingBackend(httpcore.AsyncNetworkBackend): + """A network backend that records the address the connection is actually opened against. + + A real backend resolves a hostname at connect time, which is exactly the second, unvalidated + lookup this test grid is about, so hostnames are resolved here the same way. + """ + + def __init__(self, record: dict) -> None: + self._record = record + + async def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None): + try: + ipaddress.ip_address(host) + except ValueError: + connect_target = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)[0][4][0] + else: + connect_target = host + self._record["connect_target"] = connect_target + return RecordingStream([RAW_RESPONSE], self._record) + + +def install_recording_client(monkeypatch) -> dict: + """Make the runner's built-in client speak to a recording backend through real httpx machinery.""" + record: dict = {"written": b"", "connect_target": None, "sni_hostname": None} + real_client_type = httpx.AsyncClient + + def client_factory(**kwargs): + transport = httpx.AsyncHTTPTransport() + # httpx exposes no public seam for the network backend, so the test reaches into the + # transport's pool. Everything above it (URL handling, headers, extensions) is real. + transport._pool._network_backend = RecordingBackend(record) + return real_client_type(transport=transport, **kwargs) + + monkeypatch.setattr(httpx, "AsyncClient", client_factory) + return record + + +def install_capturing_client(monkeypatch, responses: list | None = None) -> list[httpx.Request]: + """Make the runner's built-in client capture the request it sends instead of connecting.""" + requests: list[httpx.Request] = [] + real_client_type = httpx.AsyncClient + outcomes = list(responses or []) + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + outcome = outcomes.pop(0) if outcomes else None + if isinstance(outcome, Exception): + raise outcome + return httpx.Response(200, text="response text") + + def client_factory(**kwargs): + return real_client_type(transport=httpx.MockTransport(handler), **kwargs) + + monkeypatch.setattr(httpx, "AsyncClient", client_factory) + return requests + + +async def test_run_operation_pins_connection_to_validated_address_under_dns_rebinding(monkeypatch): + """A host that resolves public at validation time and private at connect time must not be reached.""" + lookups: list[str] = [] + monkeypatch.setattr(socket, "getaddrinfo", rebinding_getaddrinfo(HOST, PUBLIC_ADDRESS, REBOUND_ADDRESS, lookups)) + record = install_recording_client(monkeypatch) + runner, operation = build_runner(f"https://{HOST}/api/op") + + assert await runner.run_operation(operation, {}, None) == "response text" + + assert record["connect_target"] == PUBLIC_ADDRESS, ( + f"connection was opened against {record['connect_target']}, not the validated address" + ) + assert try_categorize_non_public_address(record["connect_target"]) == (False, "") + assert record["sni_hostname"] == HOST + assert b"Host: rebind.example\r\n" in record["written"] + assert len(lookups) == 1, "the host was resolved a second time at connect time" + + +async def test_run_operation_pins_request_url_and_preserves_host_identity(monkeypatch): + """The request is addressed to the validated IP while keeping the original Host and SNI.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS])) + requests = install_capturing_client(monkeypatch) + runner, operation = build_runner(f"https://{HOST}/api/op?a=1") + + await runner.run_operation(operation, {}, None) + + assert str(requests[0].url) == f"https://{PUBLIC_ADDRESS}/api/op?a=1" + assert requests[0].headers["Host"] == HOST + assert requests[0].extensions["sni_hostname"] == HOST + + +async def test_run_operation_pins_first_validated_address_when_several_are_returned(monkeypatch): + """Pinning uses the resolver's preferred address, not an arbitrary one.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS, SECOND_PUBLIC_ADDRESS])) + requests = install_capturing_client(monkeypatch) + runner, operation = build_runner(f"https://{HOST}/api/op") + + await runner.run_operation(operation, {}, None) + + assert len(requests) == 1 + assert requests[0].url.host == PUBLIC_ADDRESS + + +async def test_run_operation_falls_back_to_the_next_validated_address_on_connect_error(monkeypatch): + """A connection failure falls through to the remaining validated addresses, as the resolver would.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS, SECOND_PUBLIC_ADDRESS])) + requests = install_capturing_client(monkeypatch, responses=[httpx.ConnectError("no route")]) + runner, operation = build_runner(f"https://{HOST}/api/op") + + assert await runner.run_operation(operation, {}, None) == "response text" + + assert [request.url.host for request in requests] == [PUBLIC_ADDRESS, SECOND_PUBLIC_ADDRESS] + + +async def test_run_operation_does_not_retry_a_request_that_may_already_have_been_delivered(monkeypatch): + """Only connection failures fall through. A later failure means the request may already be on the wire.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS, SECOND_PUBLIC_ADDRESS])) + requests = install_capturing_client(monkeypatch, responses=[httpx.ReadTimeout("timed out")]) + runner, operation = build_runner(f"https://{HOST}/api/op") + + with pytest.raises(httpx.ReadTimeout): + await runner.run_operation(operation, {}, None) + + assert [request.url.host for request in requests] == [PUBLIC_ADDRESS], ( + "a request that may already have been delivered was resent to a second address" + ) + + +async def test_run_operation_brackets_ipv6_address_and_preserves_the_port(monkeypatch): + """An IPv6 pin keeps the URL parseable and does not move the request to another port.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_IPV6_ADDRESS])) + requests = install_capturing_client(monkeypatch) + runner, operation = build_runner(f"https://{HOST}:8443/api/op") + + await runner.run_operation(operation, {}, None) + + assert str(requests[0].url) == f"https://[{PUBLIC_IPV6_ADDRESS}]:8443/api/op" + assert requests[0].url.port == 8443 + assert requests[0].headers["Host"] == f"{HOST}:8443" + assert requests[0].extensions["sni_hostname"] == HOST + + +async def test_run_operation_does_not_pin_when_an_allowed_base_url_matches(monkeypatch): + """The allowed-base-url path never resolves the host, so there is no vetted address to pin.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo("api.example.com", [PUBLIC_ADDRESS])) + requests = install_capturing_client(monkeypatch) + runner, operation = build_runner( + "https://api.example.com/api/op", + ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com"]), + ) + + await runner.run_operation(operation, {}, None) + + assert str(requests[0].url) == "https://api.example.com/api/op" + assert "sni_hostname" not in requests[0].extensions + + +async def test_run_operation_does_not_pin_when_private_network_access_is_allowed(monkeypatch): + """Opting into private network access skips resolution, so nothing may be pinned.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo("internal.example", ["10.0.0.5"])) + requests = install_capturing_client(monkeypatch) + runner, operation = build_runner( + "https://internal.example/api/op", + ServerUrlValidationOptions(allow_private_network_access=True), + ) + + await runner.run_operation(operation, {}, None) + + assert str(requests[0].url) == "https://internal.example/api/op" + assert "sni_hostname" not in requests[0].extensions + + +async def test_run_operation_does_not_pin_a_literal_ip_host(monkeypatch): + """A literal address cannot be rebound, so the request is left exactly as it was.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS])) + requests = install_capturing_client(monkeypatch) + runner, operation = build_runner(f"https://{PUBLIC_ADDRESS}/api/op") + + await runner.run_operation(operation, {}, None) + + assert str(requests[0].url) == f"https://{PUBLIC_ADDRESS}/api/op" + assert requests[0].headers["Host"] == PUBLIC_ADDRESS + assert "sni_hostname" not in requests[0].extensions + + +async def test_run_operation_does_not_pin_when_an_environment_proxy_is_configured(monkeypatch): + """A proxy resolves the target name itself, so a locally resolved address must not be forced on it.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS])) + monkeypatch.setattr( + "semantic_kernel.connectors.openapi_plugin.openapi_runner.getproxies", + lambda: {"https": "http://proxy.example:8080"}, + ) + requests = install_capturing_client(monkeypatch) + runner, operation = build_runner(f"https://{HOST}/api/op") + + await runner.run_operation(operation, {}, None) + + assert str(requests[0].url) == f"https://{HOST}/api/op" + assert "sni_hostname" not in requests[0].extensions + + +async def test_run_operation_does_not_pin_a_caller_supplied_client(monkeypatch): + """A caller-supplied client owns its transport, so its requests are left untouched.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS])) + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, text="response text") + + runner, operation = build_runner(f"https://{HOST}/api/op") + runner.http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + await runner.run_operation(operation, {}, None) + await runner.http_client.aclose() + + assert str(requests[0].url) == f"https://{HOST}/api/op" + assert "sni_hostname" not in requests[0].extensions + + +async def test_run_operation_still_blocks_a_host_that_resolves_to_a_private_address(monkeypatch): + """Pinning must not weaken the existing block on non-public resolutions.""" + monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [REBOUND_ADDRESS])) + requests = install_capturing_client(monkeypatch) + runner, operation = build_runner(f"https://{HOST}/api/op") + + with pytest.raises(Exception, match="link-local"): + await runner.run_operation(operation, {}, None) + + assert requests == [] 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..c90356c7cc26 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,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. +import ipaddress import socket import pytest @@ -168,3 +169,38 @@ 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_returns_validated_addresses_for_pinning(): + async def fake_resolver(host: str): + assert host == "api.example.com" + return ["93.184.216.34", "198.41.0.4"] + + assert await validate_server_url("https://api.example.com/", dns_resolver=fake_resolver) == [ + ipaddress.ip_address("93.184.216.34"), + ipaddress.ip_address("198.41.0.4"), + ] + + +async def test_validate_server_url_returns_validated_ipv6_address_for_pinning(): + async def fake_resolver(host: str): + assert host == "api.example.com" + return ["2606:2800:220:1:248:1893:25c8:1946"] + + assert await validate_server_url("https://api.example.com/", dns_resolver=fake_resolver) == [ + ipaddress.ip_address("2606:2800:220:1:248:1893:25c8:1946") + ] + + +async def test_validate_server_url_returns_no_addresses_for_literal_ip_host(): + assert await validate_server_url("https://93.184.216.34/api") == [] + + +async def test_validate_server_url_returns_no_addresses_for_allowed_base_url(): + options = ServerUrlValidationOptions(allowed_base_urls=["http://api.example.com"]) + assert await validate_server_url("http://api.example.com/api", options) == [] + + +async def test_validate_server_url_returns_no_addresses_when_private_access_is_allowed(): + options = ServerUrlValidationOptions(allow_private_network_access=True) + assert await validate_server_url("https://internal.example/api", options) == []