Python: pin the validated address for OpenAPI plugin requests

python/semantic_kernel/connectors/openapi_plugin/openapi_runner.py

@@ -1,12 +1,14 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import ipaddress
 import json
 import logging
 from collections import OrderedDict
 from collections.abc import Awaitable, Callable, Mapping
 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()

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(