fix(fetch): add DNS rebinding TOCTOU protection via SSRFSafeTransport

Address review feedback on SSRF protection:

- Add SSRFSafeTransport custom async transport that resolves DNS,
  validates the resolved IP, and replaces the hostname with the
  validated IP before connecting. This eliminates the TOCTOU window
  between validate_url_for_ssrf() and the actual HTTP request.
- Integrate SSRFSafeTransport into fetch_url() and
  check_may_autonomously_fetch_url() replacing direct AsyncClient usage.
- Add 6 DNS rebinding tests including full attack scenario simulation.
- Update existing tests to match new transport-based architecture.

Author: Tomo1912

src/fetch/src/mcp_server_fetch/server.py

@@ -2,9 +2,11 @@
 import os
 import socket
 import ssl
+from types import TracebackType
 from typing import Annotated, Tuple
 from urllib.parse import urlparse, urlunparse
 
+import httpx
 import markdownify
 import readabilipy.simple_json
 from mcp.shared.exceptions import McpError
@@ -237,9 +239,9 @@ def validate_url_for_ssrf(url: str) -> None:
         McpError: If the URL is potentially dangerous
 
     Security Note:
-        This validation happens BEFORE the request is made, but DNS rebinding
-        attacks could still occur. For maximum security, use network-level
-        controls (firewall rules, egress filtering).
+        This validation provides early rejection of obviously dangerous URLs.
+        DNS rebinding protection is handled at the transport layer by
+        SSRFSafeTransport, which validates resolved IPs at connection time.
     """
     try:
         parsed = urlparse(url)
@@ -332,6 +334,94 @@ def validate_url_for_ssrf(url: str) -> None:
                 ))
 
 
+class SSRFSafeTransport(httpx.AsyncBaseTransport):
+    """
+    Custom async transport that prevents DNS rebinding attacks.
+
+    DNS rebinding TOCTOU (Time-of-Check-Time-of-Use) attack:
+    1. validate_url_for_ssrf() resolves DNS → gets public IP → passes check
+    2. Attacker's DNS server changes the record to a private IP (e.g., 169.254.169.254)
+    3. httpx resolves DNS again → gets private IP → connects to internal service
+
+    This transport eliminates the TOCTOU window by:
+    1. Resolving DNS ourselves
+    2. Validating the resolved IP
+    3. Replacing the hostname in the URL with the validated IP
+    4. Preserving the original Host header for correct HTTP routing
+    """
+
+    def __init__(self, proxy: str | None = None, verify: bool = True):
+        kwargs: dict = {"verify": verify}
+        if proxy:
+            kwargs["proxy"] = proxy
+        self._transport = httpx.AsyncHTTPTransport(**kwargs)
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        hostname = request.url.host
+        # Skip IP validation for already-resolved IPs
+        try:
+            ipaddress.ip_address(hostname)
+            # Already an IP - validation was done in validate_url_for_ssrf()
+            return await self._transport.handle_async_request(request)
+        except ValueError:
+            pass  # It's a hostname, resolve it
+
+        # Resolve DNS
+        try:
+            addr_info = socket.getaddrinfo(
+                hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM
+            )
+            if not addr_info:
+                raise McpError(ErrorData(
+                    code=INVALID_PARAMS,
+                    message=f"Failed to resolve hostname '{hostname}': no addresses found",
+                ))
+            resolved_ip = addr_info[0][4][0]
+        except socket.gaierror as e:
+            raise McpError(ErrorData(
+                code=INVALID_PARAMS,
+                message=f"Failed to resolve hostname '{hostname}': {str(e)}",
+            ))
+
+        # Validate resolved IP against SSRF rules
+        if not ALLOW_PRIVATE_IPS and _is_ip_private_or_reserved(resolved_ip):
+            raise McpError(ErrorData(
+                code=INVALID_PARAMS,
+                message=f"DNS rebinding protection: hostname '{hostname}' resolved to "
+                f"private/internal IP '{resolved_ip}' at connection time. "
+                f"Set MCP_FETCH_ALLOW_PRIVATE_IPS=true to allow internal network access.",
+            ))
+
+        # Replace hostname with validated IP to prevent DNS rebinding
+        # The Host header is already set to the original hostname by httpx
+        new_url = request.url.copy_with(host=resolved_ip)
+        # Create new request with the IP-based URL but same headers (including Host)
+        new_request = httpx.Request(
+            method=request.method,
+            url=new_url,
+            headers=request.headers,
+            stream=request.stream,
+            extensions=request.extensions,
+        )
+
+        return await self._transport.handle_async_request(new_request)
+
+    async def aclose(self):
+        await self._transport.aclose()
+
+    async def __aenter__(self):
+        await self._transport.__aenter__()
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None = None,
+        exc_val: BaseException | None = None,
+        exc_tb: TracebackType | None = None,
+    ) -> None:
+        await self._transport.__aexit__(exc_type, exc_val, exc_tb)
+
+
 def extract_content_from_html(html: str) -> str:
     """Extract and convert HTML content to Markdown format.
 
@@ -381,14 +471,13 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url:
     - SSL certificate verification (configurable via SSL_VERIFY)
     - Comprehensive SSL error handling
     """
-    import httpx
-
     robot_txt_url = get_robots_txt_url(url)
 
     # SSRF Protection: Validate robots.txt URL before fetching
     validate_url_for_ssrf(robot_txt_url)
 
-    async with httpx.AsyncClient(proxy=proxy_url, verify=SSL_VERIFY) as client:
+    transport = SSRFSafeTransport(proxy=proxy_url, verify=SSL_VERIFY)
+    async with httpx.AsyncClient(transport=transport) as client:
         try:
             response = await client.get(
                 robot_txt_url,
@@ -461,12 +550,11 @@ async def fetch_url(
     - User-Agent header for transparency
     - Comprehensive SSL error handling (catches wrapped exceptions)
     """
-    import httpx
-
     # SSRF Protection: Validate URL before fetching
     validate_url_for_ssrf(url)
 
-    async with httpx.AsyncClient(proxy=proxy_url, verify=SSL_VERIFY) as client:
+    transport = SSRFSafeTransport(proxy=proxy_url, verify=SSL_VERIFY)
+    async with httpx.AsyncClient(transport=transport) as client:
         try:
             response = await client.get(
                 url,

src/fetch/tests/test_security.py

@@ -25,6 +25,7 @@
     _parse_obfuscated_ip,
     fetch_url,
     extract_content_from_html,
+    SSRFSafeTransport,
     BLOCKED_HOSTNAMES,
     CLOUD_METADATA_IPS,
 )
@@ -134,8 +135,9 @@ async def test_ssl_disabled_allows_self_signed(self, reset_env):
         import mcp_server_fetch.server as server_module
         importlib.reload(server_module)
 
-        # Mock httpx.AsyncClient to verify verify=False is passed
-        with patch('httpx.AsyncClient') as mock_client:
+        # Mock httpx.AsyncClient and AsyncHTTPTransport to verify verify=False is passed
+        with patch('httpx.AsyncClient') as mock_client, \
+             patch('httpx.AsyncHTTPTransport') as mock_transport_class:
             mock_response = MagicMock()
             mock_response.status_code = 200
             mock_response.text = "<html><body>Test</body></html>"
@@ -155,9 +157,9 @@ async def test_ssl_disabled_allows_self_signed(self, reset_env):
                     "TestAgent/1.0"
                 )
 
-            # Verify AsyncClient was called with verify=False
-            mock_client.assert_called_once()
-            call_kwargs = mock_client.call_args[1]
+            # Verify AsyncHTTPTransport was created with verify=False
+            mock_transport_class.assert_called_once()
+            call_kwargs = mock_transport_class.call_args[1]
             assert call_kwargs.get('verify') is False
 
 
@@ -646,6 +648,141 @@ def test_url_with_port_bypass_attempt(self):
             validate_url_for_ssrf("http://127.0.0.1:65535/")
 
 
+# =============================================================================
+# 8. DNS REBINDING PROTECTION TESTS
+# =============================================================================
+
+class TestDNSRebindingProtection:
+    """Test suite for DNS rebinding TOCTOU protection via SSRFSafeTransport."""
+
+    @pytest.mark.asyncio
+    async def test_transport_blocks_private_ip_at_connection_time(self):
+        """SSRFSafeTransport must block requests when DNS resolves to private IP."""
+        import httpx
+
+        transport = SSRFSafeTransport(verify=False)
+
+        # Simulate DNS resolving to a private IP (127.0.0.1)
+        with patch("socket.getaddrinfo") as mock_dns:
+            mock_dns.return_value = [
+                (2, 1, 6, '', ('127.0.0.1', 0)),
+            ]
+            request = httpx.Request("GET", "http://evil-rebind.example.com/secret")
+
+            with pytest.raises(McpError, match="DNS rebinding protection"):
+                await transport.handle_async_request(request)
+
+    @pytest.mark.asyncio
+    async def test_transport_blocks_metadata_ip_at_connection_time(self):
+        """SSRFSafeTransport must block DNS rebinding to cloud metadata IP."""
+        import httpx
+
+        transport = SSRFSafeTransport(verify=False)
+
+        # Simulate DNS rebinding: attacker DNS returns metadata IP
+        with patch("socket.getaddrinfo") as mock_dns:
+            mock_dns.return_value = [
+                (2, 1, 6, '', ('169.254.169.254', 0)),
+            ]
+            request = httpx.Request("GET", "http://evil-rebind.example.com/metadata")
+
+            with pytest.raises(McpError, match="DNS rebinding protection"):
+                await transport.handle_async_request(request)
+
+    @pytest.mark.asyncio
+    async def test_transport_allows_public_ip(self):
+        """SSRFSafeTransport must allow requests when DNS resolves to public IP."""
+        import httpx
+
+        transport = SSRFSafeTransport(verify=False)
+
+        # Simulate DNS resolving to a public IP
+        with patch("socket.getaddrinfo") as mock_dns, \
+             patch.object(transport, '_transport') as mock_inner:
+            mock_dns.return_value = [
+                (2, 1, 6, '', ('93.184.216.34', 0)),
+            ]
+            mock_response = httpx.Response(200, text="OK")
+            mock_inner.handle_async_request = AsyncMock(return_value=mock_response)
+
+            request = httpx.Request("GET", "http://example.com/page")
+            response = await transport.handle_async_request(request)
+
+            assert response.status_code == 200
+            # Verify the inner transport was called with the IP-based URL
+            called_request = mock_inner.handle_async_request.call_args[0][0]
+            assert called_request.url.host == "93.184.216.34"
+            # Verify Host header preserved
+            assert called_request.headers["host"] == "example.com"
+
+    @pytest.mark.asyncio
+    async def test_transport_skips_validation_for_direct_ip(self):
+        """SSRFSafeTransport should skip DNS resolution for direct IP URLs."""
+        import httpx
+
+        transport = SSRFSafeTransport(verify=False)
+
+        # Direct IP URL - should go straight to inner transport (IP already validated by validate_url_for_ssrf)
+        with patch.object(transport, '_transport') as mock_inner, \
+             patch("socket.getaddrinfo") as mock_dns:
+            mock_response = httpx.Response(200, text="OK")
+            mock_inner.handle_async_request = AsyncMock(return_value=mock_response)
+
+            request = httpx.Request("GET", "http://93.184.216.34/page")
+            await transport.handle_async_request(request)
+
+            # DNS should NOT be called for direct IP
+            mock_dns.assert_not_called()
+            mock_inner.handle_async_request.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_transport_blocks_dns_failure(self):
+        """SSRFSafeTransport must raise error when DNS resolution fails."""
+        import httpx
+        import socket as socket_module
+
+        transport = SSRFSafeTransport(verify=False)
+
+        with patch("socket.getaddrinfo") as mock_dns:
+            mock_dns.side_effect = socket_module.gaierror("Name resolution failed")
+            request = httpx.Request("GET", "http://nonexistent.example.com/")
+
+            with pytest.raises(McpError, match="Failed to resolve"):
+                await transport.handle_async_request(request)
+
+    @pytest.mark.asyncio
+    async def test_dns_rebinding_scenario(self):
+        """
+        Full DNS rebinding attack scenario:
+        1. validate_url_for_ssrf() sees public IP (passes)
+        2. SSRFSafeTransport resolves DNS again and sees private IP (blocks)
+        """
+        import httpx
+
+        call_count = 0
+
+        def rebinding_dns(hostname, *args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                # First call (validate_url_for_ssrf): return public IP
+                return [(2, 1, 6, '', ('93.184.216.34', 0))]
+            else:
+                # Second call (SSRFSafeTransport): return private IP (rebinding!)
+                return [(2, 1, 6, '', ('169.254.169.254', 0))]
+
+        with patch("socket.getaddrinfo", side_effect=rebinding_dns):
+            # First validation passes (public IP)
+            validate_url_for_ssrf("http://evil-rebind.example.com/")
+
+            # But transport-level check catches the rebinding
+            transport = SSRFSafeTransport(verify=False)
+            request = httpx.Request("GET", "http://evil-rebind.example.com/metadata")
+
+            with pytest.raises(McpError, match="DNS rebinding protection"):
+                await transport.handle_async_request(request)
+
+
 # =============================================================================
 # RUN CONFIGURATION
 # =============================================================================

src/fetch/tests/test_server.py

@@ -305,13 +305,14 @@ async def test_fetch_500_raises_error(self):
 
     @pytest.mark.asyncio
     async def test_fetch_with_proxy(self):
-        """Test that proxy URL is passed to client."""
+        """Test that proxy URL is passed to SSRFSafeTransport."""
         mock_response = MagicMock()
         mock_response.status_code = 200
         mock_response.text = '{"data": "test"}'
         mock_response.headers = {"content-type": "application/json"}
 
         with patch("httpx.AsyncClient") as mock_client_class, \
+             patch("httpx.AsyncHTTPTransport") as mock_transport_class, \
              patch("mcp_server_fetch.server.validate_url_for_ssrf"), \
              patch("mcp_server_fetch.server.SSL_VERIFY", True):
             mock_client = AsyncMock()
@@ -325,5 +326,7 @@ async def test_fetch_with_proxy(self):
                 proxy_url="http://proxy.example.com:8080"
             )
 
-            # Verify AsyncClient was called with proxy and verify
-            mock_client_class.assert_called_once_with(proxy="http://proxy.example.com:8080", verify=True)
+            # Verify AsyncHTTPTransport was created with proxy and verify
+            mock_transport_class.assert_called_once_with(
+                verify=True, proxy="http://proxy.example.com:8080"
+            )