fix: pin validated DNS addresses for OpenAPI requests - #14317
fix: pin validated DNS addresses for OpenAPI requests#14317mikemikimike wants to merge 1 commit into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR closes a DNS check-time vs use-time gap in the Python OpenAPI plugin by returning the validated DNS result from validate_server_url and using a custom HTTP transport to connect to the validated IP (while preserving the original hostname for Host/SNI handling).
Changes:
- Change
validate_server_urlto return the validated host→IP mapping needed for connection pinning. - Add
PinnedDnsTransport(and backend) to connect validated hostnames to the validated IP address. - Update
OpenApiRunnerto use the pinned transport for default requests and add a regression assertion for the new return contract.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
| python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py | Updates tests to assert the new {host: ip} return contract. |
| python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py | Returns validated host→IP mapping (or {} when pinning is not used) instead of None. |
| python/semantic_kernel/connectors/openapi_plugin/pinned_http_transport.py | Introduces a custom httpx transport that pins the TCP connection to a validated address. |
| python/semantic_kernel/connectors/openapi_plugin/openapi_runner.py | Uses the pinned transport for default OpenAPI HTTP requests when validation returns a pinned host. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| ssl_context = httpx.create_ssl_context( | ||
| verify=kwargs.pop("verify", True), cert=kwargs.pop("cert", None), trust_env=kwargs.pop("trust_env", True) | ||
| ) | ||
| limits = kwargs.pop("limits", httpx.Limits()) | ||
| self._pool = httpcore.AsyncConnectionPool( | ||
| ssl_context=ssl_context, | ||
| max_connections=limits.max_connections, | ||
| max_keepalive_connections=limits.max_keepalive_connections, | ||
| keepalive_expiry=limits.keepalive_expiry, | ||
| http1=kwargs.pop("http1", True), | ||
| http2=kwargs.pop("http2", False), | ||
| network_backend=_PinnedNetworkBackend(pinned_hosts), | ||
| ) |
There was a problem hiding this comment.
Acknowledged. The pinned transport currently applies only to the default direct-client path and does not preserve environment proxy selection. I will handle proxy-aware transport selection in a follow-up revision; the explicit http_client path remains unchanged.
| return httpx.Response( | ||
| status_code=response.status, | ||
| headers=response.headers, | ||
| stream=httpx._transports.default.AsyncResponseStream(response.stream), | ||
| extensions=response.extensions, |
There was a problem hiding this comment.
Fixed in c5c8db7 by replacing the private httpx response-stream adapter with a public httpx.AsyncByteStream implementation.
| def __init__(self, pinned_hosts: Mapping[str, str], **kwargs): | ||
| """Initialize a transport using fixed addresses for validated hosts.""" | ||
| ssl_context = httpx.create_ssl_context( | ||
| verify=kwargs.pop("verify", True), cert=kwargs.pop("cert", None), trust_env=kwargs.pop("trust_env", True) | ||
| ) | ||
| limits = kwargs.pop("limits", httpx.Limits()) | ||
| self._pool = httpcore.AsyncConnectionPool( | ||
| ssl_context=ssl_context, | ||
| max_connections=limits.max_connections, | ||
| max_keepalive_connections=limits.max_keepalive_connections, | ||
| keepalive_expiry=limits.keepalive_expiry, | ||
| http1=kwargs.pop("http1", True), | ||
| http2=kwargs.pop("http2", False), | ||
| network_backend=_PinnedNetworkBackend(pinned_hosts), | ||
| ) |
There was a problem hiding this comment.
Fixed in c5c8db7: unsupported transport options now fail fast with TypeError instead of being silently ignored.
| async def validate_server_url( | ||
| url: str, | ||
| options: ServerUrlValidationOptions | None = None, | ||
| dns_resolver: DnsResolver | None = None, | ||
| ) -> None: | ||
| ) -> dict[str, str]: | ||
| """Validate a fully resolved OpenAPI operation URL against the supplied policy.""" | ||
| options = options or ServerUrlValidationOptions() |
There was a problem hiding this comment.
Fixed in c5c8db7: the validator docstring now documents the host-to-address return contract and the empty-mapping bypass cases.
|
|
||
| for address in addresses: | ||
| _ensure_public_address(parsed_url.geturl(), address) | ||
| return {host: str(addresses[0])} |
There was a problem hiding this comment.
Confirmed. The current mapping pins the first validated address, so multi-address DNS results can lose the connector's normal fallback behavior. I will address this by retaining the validated address set and trying the remaining validated public addresses on connection failure.
e26bf5b to
c5c8db7
Compare
Fixes #14312
Background
OpenApiRunnervalidates a hostname with DNS before making an HTTP request, but the defaulthttpxtransport resolves the hostname again when opening the connection. DNS rebinding can therefore cause the connection to target a different, private address than the one that passed SSRF validation.Changes
validate_server_url.PinnedDnsTransport, which connects validated hostnames to the resolved address while preserving the original request hostname for HTTP Host handling and TLS SNI.Requests using an explicitly supplied
http_clientremain under that client's transport configuration. Explicitly allowed base URLs and private-network opt-in retain their existing behavior and do not use pinning.Tests
uv run --project python --python 3.12 pytest python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py python/tests/unit/connectors/openapi_plugin/test_openapi_runner.py -q— 73 passed.uv run --project python ruff check python/semantic_kernel/connectors/openapi_plugin python/tests/unit/connectors/openapi_plugin— passed.git diff --check— passed.The full test suite was not run because it is outside the scope of this change.