Python: fix(python): Pin HttpPlugin requests to validated DNS addresses

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()

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):