diff --git a/python/semantic_kernel/core_plugins/http_plugin.py b/python/semantic_kernel/core_plugins/http_plugin.py index 762351f308cb..5203079d20d3 100644 --- a/python/semantic_kernel/core_plugins/http_plugin.py +++ b/python/semantic_kernel/core_plugins/http_plugin.py @@ -1,7 +1,10 @@ # Copyright (c) Microsoft. All rights reserved. +import ipaddress import json import logging +import socket +from collections.abc import Sequence from typing import Annotated, Any, ClassVar from urllib.parse import urlparse @@ -15,6 +18,56 @@ logger = logging.getLogger(__name__) +_NetworkAddress = ipaddress.IPv4Address | ipaddress.IPv6Address + + +class _ValidatedAddressResolver(aiohttp.abc.AbstractResolver): + """Resolve one validated host only to addresses that were already vetted.""" + + def __init__(self, host: str, addresses: Sequence[_NetworkAddress]) -> None: + self._host = host + self._addresses = addresses + + async def resolve( + self, + host: str, + port: int = 0, + family: socket.AddressFamily = socket.AF_INET, + ) -> list[Any]: + """Return vetted addresses without performing another DNS lookup.""" + if host != self._host: + raise OSError(f"Refusing to resolve host '{host}'; only '{self._host}' was validated.") + + results: list[Any] = [] + for address in self._addresses: + address_family = socket.AF_INET6 if isinstance(address, ipaddress.IPv6Address) else socket.AF_INET + if family not in (socket.AF_UNSPEC, address_family): + continue + results.append({ + "hostname": host, + "host": str(address), + "port": port, + "family": address_family, + "proto": socket.IPPROTO_TCP, + "flags": socket.AI_NUMERICHOST, + }) + return results + + async def close(self) -> None: + """Release resolver resources.""" + + +def _create_client_session(request_url: URL, validated_addresses: Sequence[_NetworkAddress]) -> aiohttp.ClientSession: + if not validated_addresses: + return aiohttp.ClientSession() + + host = request_url.raw_host + if host is None: + raise FunctionExecutionException("The request URL does not contain a valid host.") + + resolver = _ValidatedAddressResolver(host, validated_addresses) + return aiohttp.ClientSession(connector=aiohttp.TCPConnector(resolver=resolver)) + class HttpPlugin(KernelBaseModel): """A plugin that provides HTTP functionality. @@ -45,11 +98,11 @@ class HttpPlugin(KernelBaseModel): to permit additional ports. Port validation is skipped when ``allow_all_domains`` is True. - Allowed hosts are normalized using aiohttp's IDNA rules before DNS validation, - and the same normalized URL is used for the request. Non-public addresses and - DNS failures are blocked unless ``allow_private_network_access`` is True. - ``allow_all_domains=True`` also bypasses this check. - - This pre-request check does not pin the connection's IP address and does not - protect against DNS changing between validation and connection. + and the same normalized URL is used for the request. Connections are pinned to + the validated DNS addresses so the host cannot resolve to a different address + when the connection is opened. Non-public addresses and DNS failures are blocked + unless ``allow_private_network_access`` is True. ``allow_all_domains=True`` also + bypasses this check. """ allowed_domains: set[str] | None = None @@ -141,7 +194,7 @@ def _is_uri_allowed(self, url: str) -> bool: # Default: deny all return False - async def _validate_url(self, url: str) -> URL: + async def _validate_url(self, url: str) -> tuple[URL, list[_NetworkAddress]]: """Validate and normalize the URL before sending a request. Always checks that the URL is non-empty, uses an allowed scheme, and has a @@ -153,7 +206,7 @@ async def _validate_url(self, url: str) -> URL: url: The URL to validate. Returns: - The normalized URL to use for the request. + The normalized URL and any DNS addresses that must be pinned for the request. Raises: FunctionExecutionException: If the URL is empty, uses a disallowed scheme, @@ -167,13 +220,14 @@ async def _validate_url(self, url: str) -> URL: raise FunctionExecutionException("Sending requests to the provided location is not allowed.") request_url = URL(url) + validated_addresses: list[_NetworkAddress] = [] if not self.allow_all_domains and not self.allow_private_network_access: - await ensure_public_host( + validated_addresses = await ensure_public_host( urlparse(str(request_url)), configuration_hint="To allow trusted internal endpoints, set allow_private_network_access=True.", ) - return request_url + return request_url, validated_addresses @kernel_function(description="Makes a GET request to a url", name="getAsync") async def get(self, url: Annotated[str, "The URL to send the request to."]) -> str: @@ -185,10 +239,10 @@ async def get(self, url: Annotated[str, "The URL to send the request to."]) -> s Returns: The response body as a string. """ - request_url = await self._validate_url(url) + request_url, validated_addresses = await self._validate_url(url) async with ( - aiohttp.ClientSession() as session, + _create_client_session(request_url, validated_addresses) as session, session.get(request_url, raise_for_status=True, allow_redirects=self._allow_redirects) as response, ): return await response.text() @@ -207,12 +261,12 @@ async def post( returns: The response body as a string. """ - request_url = await self._validate_url(url) + request_url, validated_addresses = await self._validate_url(url) headers = {"Content-Type": "application/json"} data = json.dumps(body) if body is not None else None async with ( - aiohttp.ClientSession() as session, + _create_client_session(request_url, validated_addresses) as session, session.post( request_url, headers=headers, data=data, raise_for_status=True, allow_redirects=self._allow_redirects ) as response, @@ -234,12 +288,12 @@ async def put( Returns: The response body as a string. """ - request_url = await self._validate_url(url) + request_url, validated_addresses = await self._validate_url(url) headers = {"Content-Type": "application/json"} data = json.dumps(body) if body is not None else None async with ( - aiohttp.ClientSession() as session, + _create_client_session(request_url, validated_addresses) as session, session.put( request_url, headers=headers, data=data, raise_for_status=True, allow_redirects=self._allow_redirects ) as response, @@ -256,10 +310,10 @@ async def delete(self, url: Annotated[str, "The URI to send the request to."]) - Returns: The response body as a string. """ - request_url = await self._validate_url(url) + request_url, validated_addresses = await self._validate_url(url) async with ( - aiohttp.ClientSession() as session, + _create_client_session(request_url, validated_addresses) as session, session.delete(request_url, raise_for_status=True, allow_redirects=self._allow_redirects) as response, ): return await response.text() diff --git a/python/tests/unit/core_plugins/test_http_plugin.py b/python/tests/unit/core_plugins/test_http_plugin.py index 6fd98324bf80..7f04bbbff757 100644 --- a/python/tests/unit/core_plugins/test_http_plugin.py +++ b/python/tests/unit/core_plugins/test_http_plugin.py @@ -1,5 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. +import asyncio +import ipaddress import logging import socket from unittest.mock import AsyncMock, patch @@ -576,6 +578,89 @@ async def test_public_dns_addresses_are_allowed(method, mock_dns): mock_dns.assert_awaited_once() +@pytest.mark.parametrize("method", ["get", "post", "put", "delete"]) +async def test_http_methods_pin_validated_dns_addresses(method, mock_dns): + mock_dns.return_value = dns_response("93.184.216.34", "2606:4700:4700::1111") + plugin = HttpPlugin(allowed_domains={"example.com"}) + + with ( + patch("aiohttp.TCPConnector") as connector_type, + patch("aiohttp.ClientSession") as session_type, + ): + session = session_type.return_value + session.__aenter__.return_value = session + request = getattr(session, method) + request.return_value.__aenter__.return_value.text.return_value = "public response" + + kwargs = {"body": {"key": "value"}} if method in {"post", "put"} else {} + assert await getattr(plugin, method)("http://example.com/", **kwargs) == "public response" + + session_type.assert_called_once_with(connector=connector_type.return_value) + resolver = connector_type.call_args.kwargs["resolver"] + assert await resolver.resolve("example.com", 80, socket.AF_UNSPEC) == [ + { + "hostname": "example.com", + "host": "93.184.216.34", + "port": 80, + "family": socket.AF_INET, + "proto": socket.IPPROTO_TCP, + "flags": socket.AI_NUMERICHOST, + }, + { + "hostname": "example.com", + "host": "2606:4700:4700::1111", + "port": 80, + "family": socket.AF_INET6, + "proto": socket.IPPROTO_TCP, + "flags": socket.AI_NUMERICHOST, + }, + ] + with pytest.raises(OSError, match="only 'example.com' was validated"): + await resolver.resolve("rebound.example", 80, socket.AF_UNSPEC) + mock_dns.assert_awaited_once_with("example.com", None, type=socket.SOCK_STREAM) + + +async def test_request_pins_connection_to_validated_address_under_dns_rebinding(mock_dns): + response_body = b"validated address response" + + async def handle_request(reader, writer): + await reader.readuntil(b"\r\n\r\n") + writer.write( + b"HTTP/1.1 200 OK\r\n" + + f"Content-Length: {len(response_body)}\r\n".encode() + + b"Connection: close\r\n\r\n" + + response_body + ) + await writer.drain() + writer.close() + await writer.wait_closed() + + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(("127.0.0.1", 0)) + listener.listen() + listener.setblocking(False) + port = listener.getsockname()[1] + server = await asyncio.start_server(handle_request, sock=listener) + + mock_dns.return_value = dns_response("127.0.0.2") + plugin = HttpPlugin(allowed_domains={"rebind.example"}, allowed_ports={port}) + + try: + with patch( + "semantic_kernel.core_plugins.http_plugin.ensure_public_host", + new_callable=AsyncMock, + return_value=[ipaddress.ip_address("127.0.0.1")], + ) as validate_host: + assert await plugin.get(f"http://rebind.example:{port}/") == response_body.decode() + finally: + server.close() + await server.wait_closed() + + validate_host.assert_awaited_once() + mock_dns.assert_not_awaited() + + @pytest.mark.parametrize("method", ["get", "post", "put", "delete"]) async def test_idna_hostname_resolving_to_loopback_is_blocked(method, mock_dns): def resolve(host, *args, **kwargs):