@@ -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, proxy_bypass
 
 import httpx
 from openapi_core import Spec
@@ -29,6 +31,60 @@
 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 @@ 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 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 @@ 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 +230,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 _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(