Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 98 additions & 6 deletions python/semantic_kernel/connectors/openapi_plugin/openapi_runner.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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."""
Expand Down Expand Up @@ -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(
Expand All @@ -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)

Expand All @@ -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:
Comment thread
eavanvalkenburg marked this conversation as resolved.
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()
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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(
Expand All @@ -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(
Expand Down Expand Up @@ -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.")
Expand All @@ -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(
Expand All @@ -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(
Expand Down
Loading
Loading