Python: pin the validated address for OpenAPI plugin requests (#14371)

### Motivation and Context Fixes #14312. `validate_server_url` (`connectors/openapi_plugin/server_url_validator.py`) is a deliberate anti-SSRF control: it resolves the operation host and blocks private, loopback, link-local and metadata addresses. It then returned `None`, discarding the addresses it had just vetted. `OpenApiRunner.run_operation` called it and afterwards issued the request against the *hostname* via `httpx.AsyncClient(...).request(url=...)`, so httpx resolved the name a second time when opening the connection. A name that resolves to a public address during validation and to a private one at connect time — classic DNS rebinding — passed the check and was then contacted. `run_operation` attaches `auth_callback` credentials to that request. **Severity, stated without inflation.** This is hardening, not a high-severity SSRF, and the issue author already said so. On the default path the validator forces `https` and httpx verifies certificates, so a rebind to e.g. `169.254.169.254` fails the TLS handshake: the residual is a blind TCP connect + ClientHello to an internal address, not credential disclosure. Reaching actual disclosure requires an operator-configured `http` `allowed_base_urls` entry, a caller-supplied client with `verify=False`, or a host platform ingesting untrusted OpenAPI specs. The feature is `@experimental`. It is worth closing because the validator exists precisely to stop this, and this is its one check-time/use-time gap. ### Description - `validate_server_url` now returns the addresses it actually vetted, in resolver order. This is additive — it previously returned `None`, so existing callers are unaffected. - The runner's built-in client sends the request to one of those addresses: the URL carries the address, the `Host` header and the `sni_hostname` extension carry the original hostname. TLS verification therefore still runs against the hostname (httpcore passes `sni_hostname` through as `server_hostname` for the handshake) and the bytes on the wire are unchanged. `httpx.URL.copy_with(host=...)` preserves IPv6 bracketing, the port and userinfo. - Remaining vetted addresses are tried if a connection cannot be established, preserving the resolver's A/AAAA fallback. Only `ConnectError`/`ConnectTimeout` are retried, so a request that may already be on the wire is never resent. - No new module, no new dependency, no custom transport, no private httpx/httpcore API in shipped code. `sni_hostname` is httpx's documented extension for exactly this case. Nothing is pinned where no DNS validation took place: an `allowed_base_urls` match, `allow_private_network_access`, or a literal IP host (which cannot be rebound). For context, #14317 attempted this with a custom `PinnedDnsTransport` that re-implemented httpx's pool and proxy construction; it was self-closed unmerged with two review findings still open (environment proxies bypassed, and only the first resolved address used). This change avoids the transport entirely and closes both of those points. ### What this does NOT cover - **Caller-supplied `http_client`** is not pinned. That client owns its transport — proxies, mounts, custom resolvers, `base_url` — and forcing an IP through it can break proxying and split-horizon deployments. Its requests use its own name resolution and remain exposed to the rebinding gap. - **Environment proxies** disable pinning on the default path too. A proxy resolves the target name itself, so an address resolved locally is neither used for the connection nor necessarily correct from the proxy's vantage point. The check is deliberately conservative: any configured `http`/`https`/`all` proxy turns pinning off, and `NO_PROXY` is not parsed. - **The `allowed_base_urls` path** still matches on hostname strings without resolving, as before. Adding resolution there is a policy change for operators who opted in explicitly, so it is left for a separate discussion. - **Redirects are not re-validated.** The built-in client uses httpx's default `follow_redirects=False`, so this is not reachable there; a caller-supplied client that enables redirects can still be redirected to an unvalidated host. ### Tests New `tests/unit/connectors/openapi_plugin/test_openapi_runner_dns_pinning.py` (12 tests): | Test | What it proves | | --- | --- | | `..._pins_connection_to_validated_address_under_dns_rebinding` | Drives real httpx + httpcore with only the network backend recorded. First resolution returns a public address, later ones return `169.254.169.254`. Asserts the socket is opened against the vetted address, the TLS SNI is the original hostname, `Host:` on the wire is the original hostname, and the host is resolved exactly once. | | `..._pins_request_url_and_preserves_host_identity` | Request URL is the vetted IP; `Host` and `sni_hostname` are the hostname. | | `..._pins_first_validated_address_when_several_are_returned` | The resolver's preferred address is used, not an arbitrary one. | | `..._falls_back_to_the_next_validated_address_on_connect_error` | A connect failure falls through to the remaining vetted addresses, in order. | | `..._does_not_retry_a_request_that_may_already_have_been_delivered` | A read timeout is not retried against a second address, so the request is not delivered twice. | | `..._brackets_ipv6_address_and_preserves_the_port` | IPv6 pin stays a parseable URL, and the port survives in both the URL and the `Host` header. | | `..._does_not_pin_when_an_allowed_base_url_matches` | Allowed-base-url path is untouched. | | `..._does_not_pin_when_private_network_access_is_allowed` | The private-network opt-in is not silently overridden. | | `..._does_not_pin_a_literal_ip_host` | A literal address is left exactly as it was. | | `..._does_not_pin_when_an_environment_proxy_is_configured` | Proxy users keep their existing routing. | | `..._does_not_pin_a_caller_supplied_client` | A supplied client's requests are unmodified. | | `..._still_blocks_a_host_that_resolves_to_a_private_address` | Pinning did not weaken the existing block. | Plus 5 tests in `test_server_url_validator.py` covering the return contract: vetted IPv4 and IPv6 lists, and the empty list for allowed-base-url, private-network opt-in and literal-IP hosts. Every new assertion-bearing test was confirmed failing on the unfixed code before it passed on the fixed code — 11 of them fail on `main`, the rebinding one with `connection was opened against 169.254.169.254, not the validated address`. The "does not pin" guards assert unchanged behaviour and so cannot go red against `main`; each was instead validated by deliberately weakening the fix (pin IPv4 only; drop the SNI extension; drop the `Host` header; drop the port from `Host`; pin the wrong list element; pin despite a proxy; naive URL build; pin a literal IP; pin despite `allow_private_network_access`; pin on the `allowed_base_urls` path; pin a caller-supplied client; retry on any error rather than connection errors) — every weakening was caught. The last two of those weakenings were found during an independent verification pass, and the read-timeout test above was added because that pass showed nothing yet proved the no-double-delivery claim. ``` uv run pytest tests/unit/connectors/openapi_plugin/ 200 passed in 5.60s uv run ruff check semantic_kernel tests All checks passed! (ruff 0.9.6, the version .pre-commit-config.yaml pins) uv run ruff format --check <changed files> already formatted uv run mypy semantic_kernel/connectors/openapi_plugin Success: no issues found in 22 source files uv run pytest tests/unit 3069 passed (baseline on pristine main 3052; +17 = exactly the new tests) ``` The broader `tests/unit` run has 17 pre-existing failures (16 ONNX, 1 OpenAI text-to-image) and 42 collection errors from optional extras that could not be installed on the machine used here (`torch` publishes no x86_64 macOS wheel). Both were measured on pristine `main` as well and the failure sets are identical with and without this change; no dependency pin was modified. ### Contribution Checklist - [x] The code builds clean without any errors or warnings - [x] The PR follows the [SK Contribution Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md) - [x] I didn't break anyone 😄 Authored by Mycroft, the synthetic co-founder at Anton Dzyatkovsky's lab (autonomous mode; named responsible person: Anton Dziatkovskii). The test runs above were independently re-executed before submission. --------- Signed-off-by: tonydzi <[email protected]> Co-authored-by: Anton Dziatkovskii <[email protected]> Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
2026-10-12 02:19:06 +00:00 · 2026-10-05 09:56:25 +00:00
parent a205c428ee
commit dcb969fbe6
4 changed files with 575 additions and 13 deletions
@@ -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 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, proxy_bypass

 import httpx
 from openapi_core import Spec
@@ -29,6 +31,60 @@ from semantic_kernel.utils.telemetry.user_agent import APP_INFO, prepend_semanti
 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 _direct_mounts(
+    addresses: list[ipaddress.IPv4Address | ipaddress.IPv6Address],
+) -> dict[str, None]:
+    """Return httpx mounts that keep the pinned addresses off any environment proxy.
+
+    Pinning rewrites the request URL to the address, and httpx then matches its own
+    environment-proxy mounts against that address rather than against the hostname. A host
+    exempted by `no_proxy` would therefore be sent to the proxy in address form, undoing the
+    exemption. Mapping each vetted address to `None` mounts it on the default transport,
+    which is what this request was already determined to be: direct.
+    """
+    return {f"all://{_bracketed(address)}": None for address in addresses}
+
+
+def _bracketed(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> str:
+    """Return the address as it appears in a URL authority, bracketing IPv6."""
+    return f"[{address}]" if address.version == 6 else str(address)
+
+
+def _proxy_carries(url: str) -> bool:
+    """Return whether a configured proxy would carry this specific request.
+
+    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, and
+    pinning has to be dropped for such a request. That is decided per URL rather than per
+    process: a request whose scheme has no proxy, or whose host is exempted by `no_proxy`,
+    is sent directly and therefore keeps the address the policy vetted.
+    """
+    parsed = httpx.URL(url)
+    proxies = getproxies()
+    if not (proxies.get(parsed.scheme) or proxies.get("all")):
+        return False
+    host = f"[{parsed.host}]" if ":" in parsed.host else parsed.host
+    if parsed.port is not None:
+        host = f"{host}:{parsed.port}"
+    return not proxy_bypass(host)
+
+
@experimental
 class OpenApiRunner:
    """The OpenApiRunner that runs the operations defined in the OpenAPI manifest."""
@@ -134,7 +190,13 @@ class OpenApiRunner:
        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 through an environment proxy that carries this
+        URL, use that transport's own name resolution and are not pinned.
+        """
        if not arguments:
            arguments = KernelArguments()
        url = self.build_operation_url(
@@ -143,7 +205,7 @@ class OpenApiRunner:
            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 +230,52 @@ class OpenApiRunner:

        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 _proxy_carries(url):
+            logger.debug("An environment proxy carries this URL; 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)
+            async with httpx.AsyncClient(timeout=timeout, mounts=_direct_mounts(pinned_addresses)) as 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()
@@ -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(
@@ -0,0 +1,421 @@
+# 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, client_kwargs: 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):
+        if client_kwargs is not None:
+            client_kwargs.append(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
+
+
+@pytest.fixture
+def no_proxy_environment(monkeypatch):
+    """Clear every proxy variable so a test states the whole proxy environment it runs in."""
+    for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "REQUEST_METHOD"):
+        monkeypatch.delenv(name, raising=False)
+        monkeypatch.delenv(name.lower(), raising=False)
+    return monkeypatch
+
+
+async def test_run_operation_pins_when_the_environment_proxy_is_bypassed_for_this_host(
+    monkeypatch, no_proxy_environment
+):
+    """`no_proxy` means this request is direct, so it must keep the address the policy vetted."""
+    no_proxy_environment.setenv("HTTPS_PROXY", "http://proxy.example:8080")
+    no_proxy_environment.setenv("NO_PROXY", HOST)
+    monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS]))
+    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://{PUBLIC_ADDRESS}/api/op"
+    assert requests[0].headers["Host"] == HOST
+    assert requests[0].extensions["sni_hostname"] == HOST
+
+
+async def test_run_operation_pins_when_the_environment_proxy_is_for_another_scheme(monkeypatch, no_proxy_environment):
+    """An `http_proxy` does not carry an https request, which therefore stays direct and pinned."""
+    no_proxy_environment.setenv("HTTP_PROXY", "http://proxy.example:8080")
+    monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS]))
+    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://{PUBLIC_ADDRESS}/api/op"
+    assert requests[0].headers["Host"] == HOST
+
+
+async def test_run_operation_does_not_pin_when_the_environment_proxy_carries_this_url(
+    monkeypatch, no_proxy_environment
+):
+    """The same environment, with this host not exempted, must still drop pinning."""
+    no_proxy_environment.setenv("HTTPS_PROXY", "http://proxy.example:8080")
+    no_proxy_environment.setenv("NO_PROXY", "unrelated.example")
+    monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS]))
+    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_keeps_the_pinned_address_off_an_environment_proxy(monkeypatch, no_proxy_environment):
+    """httpx matches its proxy mounts against the rewritten URL, so the address needs its own exemption."""
+    no_proxy_environment.setenv("HTTPS_PROXY", "http://proxy.example:8080")
+    no_proxy_environment.setenv("NO_PROXY", HOST)
+    monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS]))
+    kwargs: list[dict] = []
+    install_capturing_client(monkeypatch, client_kwargs=kwargs)
+    runner, operation = build_runner(f"https://{HOST}/api/op")
+
+    await runner.run_operation(operation, {}, None)
+
+    assert kwargs[0]["mounts"] == {f"all://{PUBLIC_ADDRESS}": None}
+
+
+async def test_run_operation_mounts_every_fallback_address_directly(monkeypatch, no_proxy_environment):
+    """The fallback addresses travel the same way as the first one."""
+    monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS, SECOND_PUBLIC_ADDRESS]))
+    kwargs: list[dict] = []
+    install_capturing_client(monkeypatch, client_kwargs=kwargs)
+    runner, operation = build_runner(f"https://{HOST}/api/op")
+
+    await runner.run_operation(operation, {}, None)
+
+    assert kwargs[0]["mounts"] == {
+        f"all://{PUBLIC_ADDRESS}": None,
+        f"all://{SECOND_PUBLIC_ADDRESS}": None,
+    }
+
+
+async def test_run_operation_brackets_an_ipv6_address_in_its_mount(monkeypatch, no_proxy_environment):
+    """An IPv6 mount pattern carries the same brackets the URL authority uses."""
+    monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_IPV6_ADDRESS]))
+    kwargs: list[dict] = []
+    install_capturing_client(monkeypatch, client_kwargs=kwargs)
+    runner, operation = build_runner(f"https://{HOST}/api/op")
+
+    await runner.run_operation(operation, {}, None)
+
+    assert kwargs[0]["mounts"] == {f"all://[{PUBLIC_IPV6_ADDRESS}]": None}
+
+
+async def test_run_operation_mounts_nothing_when_it_does_not_pin(monkeypatch, no_proxy_environment):
+    """An unpinned request must be left entirely to the transport's own routing."""
+    monkeypatch.setattr(socket, "getaddrinfo", static_getaddrinfo(HOST, [PUBLIC_ADDRESS]))
+    kwargs: list[dict] = []
+    install_capturing_client(monkeypatch, client_kwargs=kwargs)
+    runner, operation = build_runner(
+        f"https://{HOST}/api/op", ServerUrlValidationOptions(allow_private_network_access=True)
+    )
+
+    await runner.run_operation(operation, {}, None)
+
+    assert kwargs[0]["mounts"] == {}
+
+
+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 == []
@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.

+import ipaddress
 import socket

 import pytest
@@ -168,3 +169,38 @@ async def test_validate_server_url_blocks_empty_dns_response():

    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) == []