From 211feaba7aa1d0178a6678e56be252985499dc28 Mon Sep 17 00:00:00 2001 From: Tomo1912 Date: Mon, 5 Jan 2026 10:07:20 +0100 Subject: [PATCH 1/2] feat(fetch): add SSRF protection and comprehensive security test suite This PR adds Server-Side Request Forgery (SSRF) protection and a comprehensive security test suite to the fetch MCP server. - URL scheme validation (only http/https allowed) - Private IP range blocking (10.x, 172.16-31.x, 192.168.x, 127.x, etc.) - IPv6 private address blocking (::1, fe80::, fc00::, etc.) - Dangerous hostname blocking (localhost, metadata services, etc.) - DNS resolution validation to prevent DNS rebinding - Configurable via MCP_FETCH_ALLOW_PRIVATE_IPS env var - Whitelist support via MCP_FETCH_ALLOWED_PRIVATE_HOSTS - Configurable SSL verification via MCP_FETCH_SSL_VERIFY env var - Comprehensive SSL error handling with helpful messages - SSRF protection tests - Private IP blocking tests - Input validation tests - URL scheme validation tests - Integration tests - Edge case tests ```bash export MCP_FETCH_SSL_VERIFY=false export MCP_FETCH_ALLOW_PRIVATE_IPS=true export MCP_FETCH_ALLOWED_PRIVATE_HOSTS=internal.company.com,api.local ``` fix: address security review feedback - Disable follow_redirects to prevent SSRF bypass via open redirects - Add explicit IP obfuscation detection (decimal/octal/hex formats) - Fix SSL parsing to be fail-secure (only 'false' disables verification) - Clean up test headers (remove enterprise roleplay language) - Add comprehensive tests for IP obfuscation parsing fix: add octal integer IP parsing and fix test naming - Add octal integer format parsing (017700000001 = 127.0.0.1) - Rename SSL test to reflect fail-secure behavior (stays_enabled, not defaults_to_false) - Add tests for octal integer IP obfuscation --- src/fetch/README.md | 39 ++ src/fetch/src/mcp_server_fetch/server.py | 391 +++++++++++++- src/fetch/tests/__init__.py | 2 + src/fetch/tests/test_security.py | 654 +++++++++++++++++++++++ src/fetch/tests/test_server.py | 11 +- 5 files changed, 1084 insertions(+), 13 deletions(-) create mode 100644 src/fetch/tests/test_security.py diff --git a/src/fetch/README.md b/src/fetch/README.md index 7e2869983f..7be8a34c7f 100644 --- a/src/fetch/README.md +++ b/src/fetch/README.md @@ -174,6 +174,45 @@ This can be customized by adding the argument `--user-agent=YourUserAgent` to th The server can be configured to use a proxy by using the `--proxy-url` argument. +### Customization - Private Network Access + +By default, the server blocks requests to private IP ranges (10.x.x.x, 192.168.x.x, 127.x.x.x, etc.) to prevent SSRF attacks. If you need to access internal services, you can configure this behavior: + +**Allow all private IPs (use with caution):** + +```json +{ + "mcpServers": { + "fetch": { + "command": "uvx", + "args": ["mcp-server-fetch"], + "env": { + "MCP_FETCH_ALLOW_PRIVATE_IPS": "true" + } + } + } +} +``` + +**Whitelist specific internal hosts:** + +```json +{ + "mcpServers": { + "fetch": { + "command": "uvx", + "args": ["mcp-server-fetch"], + "env": { + "MCP_FETCH_ALLOWED_PRIVATE_HOSTS": "internal.company.com,api.local" + } + } + } +} +``` + +> [!WARNING] +> Allowing private network access can expose internal services. Only enable this in trusted environments. + ## Windows Configuration If you're experiencing timeout issues on Windows, you may need to set the `PYTHONIOENCODING` environment variable to ensure proper character encoding: diff --git a/src/fetch/src/mcp_server_fetch/server.py b/src/fetch/src/mcp_server_fetch/server.py index b42c7b1f6b..cd7ccb348d 100644 --- a/src/fetch/src/mcp_server_fetch/server.py +++ b/src/fetch/src/mcp_server_fetch/server.py @@ -1,3 +1,7 @@ +import ipaddress +import os +import socket +import ssl from typing import Annotated, Tuple from urllib.parse import urlparse, urlunparse @@ -20,9 +24,313 @@ from protego import Protego from pydantic import BaseModel, Field, AnyUrl +# ============================================================================= +# SECURITY CONFIGURATION +# ============================================================================= + +# SSL Certificate Verification Configuration +# Set MCP_FETCH_SSL_VERIFY=false to disable SSL verification for internal/self-signed certificates +# NOTE: Only explicit "false" disables verification; any other value (including typos) keeps it enabled. +SSL_VERIFY = os.getenv("MCP_FETCH_SSL_VERIFY", "true").lower() != "false" + +# SSRF Protection Configuration +# Set MCP_FETCH_ALLOW_PRIVATE_IPS=true to allow fetching from private/internal networks +ALLOW_PRIVATE_IPS = os.getenv("MCP_FETCH_ALLOW_PRIVATE_IPS", "false").lower() == "true" + +# Comma-separated list of allowed private hosts (only used when ALLOW_PRIVATE_IPS=false) +# Example: "internal.company.com,api.internal.local" +ALLOWED_PRIVATE_HOSTS = [ + h.strip().lower() + for h in os.getenv("MCP_FETCH_ALLOWED_PRIVATE_HOSTS", "").split(",") + if h.strip() +] + DEFAULT_USER_AGENT_AUTONOMOUS = "ModelContextProtocol/1.0 (Autonomous; +https://github.com/modelcontextprotocol/servers)" DEFAULT_USER_AGENT_MANUAL = "ModelContextProtocol/1.0 (User-Specified; +https://github.com/modelcontextprotocol/servers)" +# ============================================================================= +# SSRF PROTECTION MODULE +# ============================================================================= + +# Blocked hostnames that resolve to internal services +BLOCKED_HOSTNAMES = frozenset([ + "localhost", + "localhost.localdomain", + "ip6-localhost", + "ip6-loopback", + "metadata.google.internal", # GCP metadata + "metadata.internal", # Generic cloud metadata + "kubernetes.default", # Kubernetes + "kubernetes.default.svc", + "kubernetes.default.svc.cluster.local", +]) + +# Cloud metadata IP addresses +CLOUD_METADATA_IPS = frozenset([ + "169.254.169.254", # AWS, Azure, GCP, DigitalOcean, Oracle Cloud + "169.254.170.2", # AWS ECS task metadata + "fd00:ec2::254", # AWS IPv6 metadata +]) + + +def _parse_obfuscated_ip(hostname: str) -> str | None: + """ + Detect and decode obfuscated IP address formats. + + Attackers may use alternative IP representations to bypass SSRF filters: + - Decimal: 2130706433 (= 127.0.0.1) + - Octal integer: 017700000001 (= 127.0.0.1) + - Octal dotted: 0177.0.0.1 (= 127.0.0.1) + - Hex: 0x7f000001 (= 127.0.0.1) + - Mixed: 0x7f.0.0.1 (= 127.0.0.1) + + Returns the normalized IP string if detected, None otherwise. + """ + hostname = hostname.strip() + + # Try octal integer format (e.g., 017700000001 = 127.0.0.1) + # Must check before decimal since octal strings are also digits + try: + if hostname.startswith("0") and len(hostname) > 1 and hostname.isdigit(): + ip_int = int(hostname, 8) + if 0 <= ip_int <= 0xFFFFFFFF: # Valid 32-bit range + return str(ipaddress.IPv4Address(ip_int)) + except (ValueError, ipaddress.AddressValueError): + pass + + # Try decimal integer format (e.g., 2130706433 = 127.0.0.1) + try: + if hostname.isdigit(): + ip_int = int(hostname) + if 0 <= ip_int <= 0xFFFFFFFF: # Valid 32-bit range + # Convert to dotted decimal + return str(ipaddress.IPv4Address(ip_int)) + except (ValueError, ipaddress.AddressValueError): + pass + + # Try hex format (e.g., 0x7f000001 = 127.0.0.1) + try: + if hostname.lower().startswith("0x") and "." not in hostname: + ip_int = int(hostname, 16) + if 0 <= ip_int <= 0xFFFFFFFF: + return str(ipaddress.IPv4Address(ip_int)) + except (ValueError, ipaddress.AddressValueError): + pass + + # Try octal/hex dotted format (e.g., 0177.0.0.1 or 0x7f.0.0.1) + # Only return if there's actual obfuscation (hex prefix or leading zeros) + if "." in hostname: + parts = hostname.split(".") + if len(parts) == 4: + try: + octets = [] + has_obfuscation = False + for part in parts: + part = part.strip() + if part.lower().startswith("0x"): + octets.append(int(part, 16)) + has_obfuscation = True + elif part.startswith("0") and len(part) > 1 and part.isdigit(): + # Octal format (leading zero with more digits) + octets.append(int(part, 8)) + has_obfuscation = True + else: + octets.append(int(part)) + + # Only return if we detected obfuscation AND result is valid + if has_obfuscation and all(0 <= o <= 255 for o in octets): + return f"{octets[0]}.{octets[1]}.{octets[2]}.{octets[3]}" + except ValueError: + pass + + return None + + +def _is_ip_private_or_reserved(ip_str: str) -> bool: + """ + Check if an IP address is private, reserved, loopback, or link-local. + + This function handles: + - IPv4 and IPv6 addresses + - Loopback (127.0.0.0/8, ::1) + - Private networks (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16) + - Link-local (169.254.0.0/16, fe80::/10) + - Reserved ranges + - Multicast addresses + - Unspecified addresses (0.0.0.0, ::) + """ + try: + ip = ipaddress.ip_address(ip_str) + + # Check all dangerous categories + if ip.is_private: + return True + if ip.is_loopback: + return True + if ip.is_link_local: + return True + if ip.is_reserved: + return True + if ip.is_multicast: + return True + if ip.is_unspecified: + return True + + # Additional check for IPv4-mapped IPv6 addresses (::ffff:127.0.0.1) + if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped: + return _is_ip_private_or_reserved(str(ip.ipv4_mapped)) + + # Check cloud metadata IPs explicitly + if ip_str in CLOUD_METADATA_IPS: + return True + + return False + except ValueError: + # If we can't parse it, block it to be safe + return True + + +def _normalize_hostname(hostname: str) -> str: + """Normalize hostname for comparison.""" + # Remove trailing dots (FQDN notation) + hostname = hostname.rstrip(".") + # Lowercase for case-insensitive comparison + return hostname.lower() + + +def _is_hostname_blocked(hostname: str) -> bool: + """Check if hostname is in the blocked list.""" + normalized = _normalize_hostname(hostname) + + # Direct match + if normalized in BLOCKED_HOSTNAMES: + return True + + # Check for subdomain matches of blocked hostnames + for blocked in BLOCKED_HOSTNAMES: + if normalized.endswith("." + blocked): + return True + + return False + + +def _is_hostname_whitelisted(hostname: str) -> bool: + """Check if hostname is explicitly whitelisted for private access.""" + if not ALLOWED_PRIVATE_HOSTS: + return False + + normalized = _normalize_hostname(hostname) + return normalized in ALLOWED_PRIVATE_HOSTS + + +def validate_url_for_ssrf(url: str) -> None: + """ + Validate a URL to prevent SSRF attacks. + + This function performs comprehensive SSRF protection: + 1. Validates URL scheme (only http/https allowed) + 2. Blocks known dangerous hostnames + 3. Resolves hostname to IP and validates against private ranges + 4. Handles IP address obfuscation (octal, hex, decimal encoding) + + Raises: + 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). + """ + try: + parsed = urlparse(url) + except Exception as e: + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Invalid URL format: {str(e)}", + )) + + # Validate scheme + if parsed.scheme not in ("http", "https"): + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"URL scheme '{parsed.scheme}' is not allowed. Only http and https are permitted.", + )) + + # Extract hostname + hostname = parsed.hostname + if not hostname: + raise McpError(ErrorData( + code=INVALID_PARAMS, + message="URL must contain a valid hostname.", + )) + + # Check if hostname is whitelisted (bypass other checks) + if _is_hostname_whitelisted(hostname): + return + + # Check blocked hostnames + if _is_hostname_blocked(hostname): + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Access to '{hostname}' is blocked for security reasons. " + f"This hostname is associated with internal services.", + )) + + # Check for obfuscated IP addresses (decimal, octal, hex encoding) + # Python's ipaddress module does NOT parse these from strings, so we handle them explicitly + obfuscated_ip = _parse_obfuscated_ip(hostname) + if obfuscated_ip: + if _is_ip_private_or_reserved(obfuscated_ip): + if not ALLOW_PRIVATE_IPS: + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Access to obfuscated private IP address '{hostname}' " + f"(decoded: {obfuscated_ip}) is blocked. " + f"Set MCP_FETCH_ALLOW_PRIVATE_IPS=true to allow internal network access.", + )) + return + + # Try to parse hostname as standard IP address + try: + ip = ipaddress.ip_address(hostname) + if _is_ip_private_or_reserved(str(ip)): + if not ALLOW_PRIVATE_IPS: + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Access to private/internal IP address '{hostname}' is blocked. " + f"Set MCP_FETCH_ALLOW_PRIVATE_IPS=true to allow internal network access.", + )) + return + except ValueError: + # Not an IP address, continue with DNS resolution + pass + + # Resolve hostname to IP addresses + try: + # Get all IP addresses for the hostname + addr_info = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) + resolved_ips = set() + for family, _, _, _, sockaddr in addr_info: + ip_str = sockaddr[0] + resolved_ips.add(ip_str) + except socket.gaierror as e: + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Failed to resolve hostname '{hostname}': {str(e)}", + )) + + # Validate all resolved IPs + if not ALLOW_PRIVATE_IPS: + for ip_str in resolved_ips: + if _is_ip_private_or_reserved(ip_str): + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Hostname '{hostname}' resolves to private/internal IP address '{ip_str}'. " + f"Access to internal networks is blocked for security. " + f"Set MCP_FETCH_ALLOW_PRIVATE_IPS=true or add the host to " + f"MCP_FETCH_ALLOWED_PRIVATE_HOSTS to allow access.", + )) + def extract_content_from_html(html: str) -> str: """Extract and convert HTML content to Markdown format. @@ -67,22 +375,54 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: """ Check if the URL can be fetched by the user agent according to the robots.txt file. Raises a McpError if not. + + Security Features: + - SSRF protection via URL validation + - SSL certificate verification (configurable via SSL_VERIFY) + - Comprehensive SSL error handling """ - from httpx import AsyncClient, HTTPError + import httpx robot_txt_url = get_robots_txt_url(url) - async with AsyncClient(proxy=proxy_url) as client: + # 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: try: response = await client.get( robot_txt_url, - follow_redirects=True, + follow_redirects=False, headers={"User-Agent": user_agent}, + timeout=30, ) - except HTTPError: + except ssl.SSLError as e: + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"SSL Certificate verification failed for {robot_txt_url}. " + f"If this is an internal server with a self-signed certificate, " + f"set MCP_FETCH_SSL_VERIFY=false in your environment. " + f"Error details: {str(e)}", + )) + except httpx.ConnectError as e: + # httpx wraps SSL errors in ConnectError in some cases + error_str = str(e).lower() + if "ssl" in error_str or "certificate" in error_str or "verify" in error_str: + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"SSL Certificate verification failed for {robot_txt_url}. " + f"If this is an internal server with a self-signed certificate, " + f"set MCP_FETCH_SSL_VERIFY=false in your environment. " + f"Error details: {str(e)}", + )) + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"Failed to connect to {robot_txt_url}: {str(e)}", + )) + except httpx.HTTPError as e: raise McpError(ErrorData( code=INTERNAL_ERROR, - message=f"Failed to fetch robots.txt {robot_txt_url} due to a connection issue", + message=f"Failed to fetch robots.txt {robot_txt_url} due to a connection issue: {str(e)}", )) if response.status_code in (401, 403): raise McpError(ErrorData( @@ -113,18 +453,51 @@ async def fetch_url( ) -> Tuple[str, str]: """ Fetch the URL and return the content in a form ready for the LLM, as well as a prefix string with status information. + + Security Features: + - SSRF protection via comprehensive URL validation + - SSL certificate verification (configurable via SSL_VERIFY) + - Timeout protection (30 seconds) to prevent resource exhaustion + - User-Agent header for transparency + - Comprehensive SSL error handling (catches wrapped exceptions) """ - from httpx import AsyncClient, HTTPError + import httpx + + # SSRF Protection: Validate URL before fetching + validate_url_for_ssrf(url) - async with AsyncClient(proxy=proxy_url) as client: + async with httpx.AsyncClient(proxy=proxy_url, verify=SSL_VERIFY) as client: try: response = await client.get( url, - follow_redirects=True, + follow_redirects=False, headers={"User-Agent": user_agent}, timeout=30, ) - except HTTPError as e: + except ssl.SSLError as e: + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"SSL Certificate verification failed for {url}. " + f"If this is an internal server with a self-signed certificate, " + f"set MCP_FETCH_SSL_VERIFY=false in your environment. " + f"Error details: {str(e)}", + )) + except httpx.ConnectError as e: + # httpx wraps SSL errors in ConnectError in some cases + error_str = str(e).lower() + if "ssl" in error_str or "certificate" in error_str or "verify" in error_str: + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"SSL Certificate verification failed for {url}. " + f"If this is an internal server with a self-signed certificate, " + f"set MCP_FETCH_SSL_VERIFY=false in your environment. " + f"Error details: {str(e)}", + )) + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"Failed to connect to {url}: {str(e)}", + )) + except httpx.HTTPError as e: raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"Failed to fetch {url}: {e!r}")) if response.status_code >= 400: raise McpError(ErrorData( diff --git a/src/fetch/tests/__init__.py b/src/fetch/tests/__init__.py index e69de29bb2..2d8640af3c 100644 --- a/src/fetch/tests/__init__.py +++ b/src/fetch/tests/__init__.py @@ -0,0 +1,2 @@ +# MCP Fetch Server - Security Test Suite + diff --git a/src/fetch/tests/test_security.py b/src/fetch/tests/test_security.py new file mode 100644 index 0000000000..8dbddca4f2 --- /dev/null +++ b/src/fetch/tests/test_security.py @@ -0,0 +1,654 @@ +""" +Security regression tests for fetch server. + +This test suite validates the security controls implemented in server.py: +1. SSL Certificate Verification Toggle +2. SSRF (Server-Side Request Forgery) Protection +3. Payload Size Limits +4. URL Scheme Validation +""" + +import os +import sys +import pytest +from unittest.mock import patch, AsyncMock, MagicMock + +# Add the source directory to the path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) + +from mcp_server_fetch.server import ( + validate_url_for_ssrf, + _is_ip_private_or_reserved, + _is_hostname_blocked, + _is_hostname_whitelisted, + _normalize_hostname, + _parse_obfuscated_ip, + fetch_url, + extract_content_from_html, + BLOCKED_HOSTNAMES, + CLOUD_METADATA_IPS, +) +from mcp.shared.exceptions import McpError + + +# ============================================================================= +# TEST FIXTURES +# ============================================================================= + +@pytest.fixture +def reset_env(): + """Reset environment variables after each test.""" + original_ssl = os.environ.get('MCP_FETCH_SSL_VERIFY') + original_private = os.environ.get('MCP_FETCH_ALLOW_PRIVATE_IPS') + original_hosts = os.environ.get('MCP_FETCH_ALLOWED_PRIVATE_HOSTS') + + yield + + # Restore original values + if original_ssl is not None: + os.environ['MCP_FETCH_SSL_VERIFY'] = original_ssl + elif 'MCP_FETCH_SSL_VERIFY' in os.environ: + del os.environ['MCP_FETCH_SSL_VERIFY'] + + if original_private is not None: + os.environ['MCP_FETCH_ALLOW_PRIVATE_IPS'] = original_private + elif 'MCP_FETCH_ALLOW_PRIVATE_IPS' in os.environ: + del os.environ['MCP_FETCH_ALLOW_PRIVATE_IPS'] + + if original_hosts is not None: + os.environ['MCP_FETCH_ALLOWED_PRIVATE_HOSTS'] = original_hosts + elif 'MCP_FETCH_ALLOWED_PRIVATE_HOSTS' in os.environ: + del os.environ['MCP_FETCH_ALLOWED_PRIVATE_HOSTS'] + + +# ============================================================================= +# 1. SSL TOGGLE TESTS +# ============================================================================= + +class TestSSLToggle: + """Test suite for SSL certificate verification toggle (Issue #508).""" + + @pytest.mark.asyncio + async def test_ssl_verify_enabled_by_default(self): + """SSL verification should be enabled when env var is not set.""" + # Remove env var if set + if 'MCP_FETCH_SSL_VERIFY' in os.environ: + del os.environ['MCP_FETCH_SSL_VERIFY'] + + # Re-import to get fresh value + import importlib + import mcp_server_fetch.server as server_module + importlib.reload(server_module) + + assert server_module.SSL_VERIFY is True + + @pytest.mark.asyncio + async def test_ssl_verify_disabled_when_false(self, reset_env): + """SSL verification should be disabled when env var is 'false'.""" + os.environ['MCP_FETCH_SSL_VERIFY'] = 'false' + + import importlib + import mcp_server_fetch.server as server_module + importlib.reload(server_module) + + assert server_module.SSL_VERIFY is False + + @pytest.mark.asyncio + async def test_ssl_verify_case_insensitive(self, reset_env): + """SSL toggle should handle case variations.""" + test_cases = [ + ('TRUE', True), + ('True', True), + ('true', True), + ('FALSE', False), + ('False', False), + ('false', False), + ] + + import importlib + import mcp_server_fetch.server as server_module + + for value, expected in test_cases: + os.environ['MCP_FETCH_SSL_VERIFY'] = value + importlib.reload(server_module) + assert server_module.SSL_VERIFY is expected, f"Failed for value: {value}" + + @pytest.mark.asyncio + async def test_ssl_verify_invalid_value_stays_enabled(self, reset_env): + """Invalid/unknown values should keep SSL verification ENABLED (fail-secure).""" + os.environ['MCP_FETCH_SSL_VERIFY'] = 'invalid' + + import importlib + import mcp_server_fetch.server as server_module + importlib.reload(server_module) + + # Fail-secure: only explicit "false" disables SSL verification + assert server_module.SSL_VERIFY is True + + @pytest.mark.asyncio + async def test_ssl_disabled_allows_self_signed(self, reset_env): + """When SSL is disabled, self-signed certificates should work.""" + os.environ['MCP_FETCH_SSL_VERIFY'] = 'false' + + import importlib + 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_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = "Test" + mock_response.headers = {'content-type': 'text/html'} + + mock_instance = AsyncMock() + mock_instance.get = AsyncMock(return_value=mock_response) + mock_instance.__aenter__ = AsyncMock(return_value=mock_instance) + mock_instance.__aexit__ = AsyncMock(return_value=None) + mock_client.return_value = mock_instance + + # Mock SSRF validation to allow test URL (DNS won't resolve for fake domain) + with patch.object(server_module, 'validate_url_for_ssrf'): + # This should not raise an SSL error + await server_module.fetch_url( + "https://self-signed.example.com", + "TestAgent/1.0" + ) + + # Verify AsyncClient was called with verify=False + mock_client.assert_called_once() + call_kwargs = mock_client.call_args[1] + assert call_kwargs.get('verify') is False + + +# ============================================================================= +# 2. IP OBFUSCATION PARSING TESTS +# ============================================================================= + +class TestIPObfuscationParsing: + """Test suite for IP obfuscation detection and parsing.""" + + @pytest.mark.parametrize("obfuscated,expected", [ + # Decimal encoding (127.0.0.1 = 2130706433) + ("2130706433", "127.0.0.1"), + # Decimal encoding (169.254.169.254 = 2852039166) + ("2852039166", "169.254.169.254"), + # Hex encoding + ("0x7f000001", "127.0.0.1"), + ("0x7F000001", "127.0.0.1"), # uppercase + # Octal integer format (without dots) + ("017700000001", "127.0.0.1"), + ("025177524776", "169.254.169.254"), # metadata IP in octal + # Octal dotted format + ("0177.0.0.1", "127.0.0.1"), + ("0177.0.0.01", "127.0.0.1"), + # Hex dotted format + ("0x7f.0.0.1", "127.0.0.1"), + ("0x7f.0x0.0x0.0x1", "127.0.0.1"), + ]) + def test_parses_obfuscated_ips(self, obfuscated, expected): + """Obfuscated IP formats should be correctly decoded.""" + result = _parse_obfuscated_ip(obfuscated) + assert result == expected, f"Failed to parse {obfuscated}" + + @pytest.mark.parametrize("normal_input", [ + "127.0.0.1", # Normal IP - not obfuscated + "example.com", # Hostname + "google.com", + "192.168.1.1", # Normal private IP + "", # Empty + "not-an-ip", + ]) + def test_returns_none_for_normal_inputs(self, normal_input): + """Normal hostnames and IPs should return None (not obfuscated).""" + result = _parse_obfuscated_ip(normal_input) + assert result is None, f"Should not parse {normal_input} as obfuscated" + + def test_blocks_obfuscated_loopback_via_validation(self): + """Obfuscated loopback should be blocked by validate_url_for_ssrf.""" + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf("http://2130706433/") + assert "obfuscated" in str(exc_info.value).lower() or "blocked" in str(exc_info.value).lower() + + def test_blocks_obfuscated_metadata_via_validation(self): + """Obfuscated metadata IP should be blocked by validate_url_for_ssrf.""" + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf("http://2852039166/") # 169.254.169.254 + assert "obfuscated" in str(exc_info.value).lower() or "blocked" in str(exc_info.value).lower() + + +# ============================================================================= +# 3. SSRF PROTECTION TESTS +# ============================================================================= + +class TestSSRFProtection: + """Test suite for SSRF (Server-Side Request Forgery) protection.""" + + # ------------------------------------------------------------------------- + # Private IP Range Tests + # ------------------------------------------------------------------------- + + @pytest.mark.parametrize("private_ip", [ + # Loopback (127.0.0.0/8) + "127.0.0.1", + "127.0.0.2", + "127.255.255.255", + + # Class A Private (10.0.0.0/8) + "10.0.0.1", + "10.255.255.255", + + # Class B Private (172.16.0.0/12) + "172.16.0.1", + "172.31.255.255", + + # Class C Private (192.168.0.0/16) + "192.168.0.1", + "192.168.1.1", + "192.168.255.255", + + # Link-local (169.254.0.0/16) + "169.254.0.1", + "169.254.169.254", # AWS/GCP/Azure metadata + "169.254.170.2", # AWS ECS metadata + ]) + def test_blocks_private_ip_addresses(self, private_ip): + """Private and reserved IP addresses must be blocked.""" + assert _is_ip_private_or_reserved(private_ip) is True + + @pytest.mark.parametrize("public_ip", [ + "8.8.8.8", # Google DNS + "1.1.1.1", # Cloudflare DNS + "142.250.80.46", # google.com + "151.101.1.140", # reddit.com + ]) + def test_allows_public_ip_addresses(self, public_ip): + """Public IP addresses should be allowed.""" + assert _is_ip_private_or_reserved(public_ip) is False + + # ------------------------------------------------------------------------- + # Cloud Metadata Endpoint Tests + # ------------------------------------------------------------------------- + + @pytest.mark.parametrize("metadata_url", [ + "http://169.254.169.254/", + "http://169.254.169.254/latest/meta-data/", + "http://169.254.169.254/latest/meta-data/iam/security-credentials/", + "http://169.254.170.2/v2/credentials/", + ]) + def test_blocks_cloud_metadata_ips(self, metadata_url): + """Cloud metadata IP addresses must be blocked.""" + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf(metadata_url) + + error_msg = str(exc_info.value).lower() + assert "private" in error_msg or "blocked" in error_msg + + @pytest.mark.parametrize("metadata_hostname", [ + "http://metadata.google.internal/", + "http://metadata.google.internal/computeMetadata/v1/", + "http://kubernetes.default.svc/", + ]) + def test_blocks_cloud_metadata_hostnames(self, metadata_hostname): + """Cloud metadata hostnames must be blocked.""" + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf(metadata_hostname) + + assert "blocked" in str(exc_info.value).lower() + + # ------------------------------------------------------------------------- + # Localhost Tests + # ------------------------------------------------------------------------- + + @pytest.mark.parametrize("localhost_url", [ + "http://localhost/", + "http://localhost:8080/", + "http://localhost:6379/", # Redis + "http://localhost:9200/", # Elasticsearch + "http://127.0.0.1/", + "http://127.0.0.1:3306/", # MySQL + "http://[::1]/", # IPv6 loopback + ]) + def test_blocks_localhost_access(self, localhost_url): + """Localhost and loopback addresses must be blocked.""" + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf(localhost_url) + + error_msg = str(exc_info.value).lower() + assert "blocked" in error_msg or "private" in error_msg + + # ------------------------------------------------------------------------- + # IP Obfuscation Tests + # ------------------------------------------------------------------------- + + @pytest.mark.parametrize("obfuscated_url,description", [ + ("http://2130706433/", "Decimal encoding of 127.0.0.1"), + ("http://017700000001/", "Octal encoding of 127.0.0.1"), + ("http://0x7f000001/", "Hex encoding of 127.0.0.1"), + ("http://[::ffff:127.0.0.1]/", "IPv4-mapped IPv6 loopback"), + ("http://[::ffff:169.254.169.254]/", "IPv4-mapped IPv6 metadata"), + ]) + def test_blocks_ip_obfuscation(self, obfuscated_url, description): + """IP obfuscation techniques must be detected and blocked.""" + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf(obfuscated_url) + + # Should be blocked regardless of encoding + error_msg = str(exc_info.value).lower() + assert any(w in error_msg for w in ["blocked", "private", "internal"]), \ + f"Failed to block: {description}" + + # ------------------------------------------------------------------------- + # Hostname Blocklist Tests + # ------------------------------------------------------------------------- + + def test_blocked_hostnames_list(self): + """Verify all critical hostnames are in the blocklist.""" + required_blocked = [ + "localhost", + "metadata.google.internal", + "kubernetes.default", + ] + + for hostname in required_blocked: + assert hostname in BLOCKED_HOSTNAMES, \ + f"Critical hostname '{hostname}' missing from blocklist" + + def test_hostname_normalization(self): + """Hostname normalization should handle edge cases.""" + assert _normalize_hostname("LOCALHOST") == "localhost" + assert _normalize_hostname("Localhost.") == "localhost" + assert _normalize_hostname("EXAMPLE.COM.") == "example.com" + + def test_subdomain_blocking(self): + """Subdomains of blocked hosts should also be blocked.""" + assert _is_hostname_blocked("evil.localhost") is True + assert _is_hostname_blocked("sub.metadata.google.internal") is True + + +# ============================================================================= +# 3. PAYLOAD LIMIT TESTS +# ============================================================================= + +class TestPayloadLimits: + """Test suite for payload size limits and resource exhaustion prevention.""" + + @pytest.mark.asyncio + async def test_large_response_truncated(self): + """Responses larger than max_length should be truncated.""" + # Create a large HTML response (10MB) + large_content = "" + ("X" * 10_000_000) + "" + + with patch('httpx.AsyncClient') as mock_client: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = large_content + mock_response.headers = {'content-type': 'text/html'} + + mock_instance = AsyncMock() + mock_instance.get = AsyncMock(return_value=mock_response) + mock_instance.__aenter__ = AsyncMock(return_value=mock_instance) + mock_instance.__aexit__ = AsyncMock(return_value=None) + mock_client.return_value = mock_instance + + # Mock SSRF validation to allow the test URL + with patch('mcp_server_fetch.server.validate_url_for_ssrf'): + content, prefix = await fetch_url( + "https://example.com/large-file", + "TestAgent/1.0" + ) + + # Content should be returned (will be truncated by caller) + # The fetch_url function returns the full content; + # truncation happens in call_tool based on max_length parameter + assert len(content) > 0 + + @pytest.mark.asyncio + async def test_billion_laughs_protection(self): + """XML bomb / billion laughs attacks should be mitigated by size limits.""" + # Simulated expanded XML bomb content + billion_laughs_expanded = "LOL" * 5_000_000 # 15MB of "LOL" + + with patch('httpx.AsyncClient') as mock_client: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = billion_laughs_expanded + mock_response.headers = {'content-type': 'text/xml'} + + mock_instance = AsyncMock() + mock_instance.get = AsyncMock(return_value=mock_response) + mock_instance.__aenter__ = AsyncMock(return_value=mock_instance) + mock_instance.__aexit__ = AsyncMock(return_value=None) + mock_client.return_value = mock_instance + + with patch('mcp_server_fetch.server.validate_url_for_ssrf'): + content, prefix = await fetch_url( + "https://example.com/xml", + "TestAgent/1.0" + ) + + # Response is returned but will be truncated by max_length in call_tool + assert content is not None + + def test_max_length_parameter_validation(self): + """max_length parameter should have upper bounds.""" + from mcp_server_fetch.server import Fetch + from pydantic import AnyUrl, ValidationError + + test_url = AnyUrl("https://example.com") + + # Valid max_length + fetch = Fetch(url=test_url, max_length=5000, start_index=0, raw=False) + assert fetch.max_length == 5000 + + # max_length at upper bound (lt=1000000 means < 1,000,000) + fetch = Fetch(url=test_url, max_length=999999, start_index=0, raw=False) + assert fetch.max_length == 999999 + + # max_length exceeding upper bound should fail + with pytest.raises(ValidationError): + Fetch(url=test_url, max_length=1_000_001, start_index=0, raw=False) + + # max_length must be positive (gt=0) + with pytest.raises(ValidationError): + Fetch(url=test_url, max_length=0, start_index=0, raw=False) + + with pytest.raises(ValidationError): + Fetch(url=test_url, max_length=-1, start_index=0, raw=False) + + +# ============================================================================= +# 4. URL SCHEME VALIDATION TESTS +# ============================================================================= + +class TestURLSchemeValidation: + """Test suite for URL scheme validation.""" + + @pytest.mark.parametrize("valid_url", [ + "http://example.com/", + "https://example.com/", + "http://example.com:8080/path", + "https://api.example.com/v1/resource", + "HTTP://EXAMPLE.COM/", # Case insensitive + "HTTPS://EXAMPLE.COM/", + ]) + def test_allows_http_and_https(self, valid_url): + """HTTP and HTTPS schemes should be allowed.""" + # Should not raise for valid schemes (may raise for other reasons like DNS) + # We mock DNS resolution to isolate scheme testing + with patch('socket.getaddrinfo') as mock_dns: + mock_dns.return_value = [ + (2, 1, 6, '', ('93.184.216.34', 0)) # example.com public IP + ] + try: + validate_url_for_ssrf(valid_url) + except McpError as e: + # Should not be a scheme error + assert "scheme" not in str(e).lower() + + @pytest.mark.parametrize("invalid_scheme_url,scheme", [ + ("file:///etc/passwd", "file"), + ("ftp://ftp.example.com/file.txt", "ftp"), + ("gopher://localhost/", "gopher"), + ("data:text/html,", "data"), + ("javascript:alert(1)", "javascript"), + ("ldap://localhost/", "ldap"), + ("dict://localhost/", "dict"), + ("sftp://example.com/file", "sftp"), + ]) + def test_blocks_dangerous_schemes(self, invalid_scheme_url, scheme): + """Non-HTTP(S) schemes must be blocked.""" + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf(invalid_scheme_url) + + error_msg = str(exc_info.value).lower() + assert "scheme" in error_msg or "not allowed" in error_msg + + def test_blocks_empty_scheme(self): + """URLs without schemes should be rejected.""" + with pytest.raises(McpError): + validate_url_for_ssrf("//example.com/path") + + def test_blocks_malformed_urls(self): + """Malformed URLs should be rejected.""" + malformed_urls = [ + "", + "not-a-url", + "://missing-scheme.com", + "http://", # No hostname + ] + + for url in malformed_urls: + with pytest.raises(McpError): + validate_url_for_ssrf(url) + + +# ============================================================================= +# 5. INTEGRATION TESTS +# ============================================================================= + +class TestSecurityIntegration: + """Integration tests combining multiple security controls.""" + + @pytest.mark.asyncio + async def test_full_security_chain(self): + """Test the complete security validation chain.""" + # This test verifies that all security checks are applied in order + + # 1. Scheme validation happens first + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf("file:///etc/passwd") + assert "scheme" in str(exc_info.value).lower() + + # 2. Hostname blocklist check + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf("http://localhost/") + assert "blocked" in str(exc_info.value).lower() + + # 3. IP validation + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf("http://127.0.0.1/") + assert "private" in str(exc_info.value).lower() + + # 4. DNS resolution and IP check + with patch('socket.getaddrinfo') as mock_dns: + # Simulate DNS resolving to private IP + mock_dns.return_value = [ + (2, 1, 6, '', ('10.0.0.1', 0)) + ] + with pytest.raises(McpError) as exc_info: + validate_url_for_ssrf("http://internal.evil.com/") + assert "private" in str(exc_info.value).lower() + + @pytest.mark.asyncio + async def test_timeout_protection(self): + """Verify timeout is set on requests.""" + with patch('httpx.AsyncClient') as mock_client: + mock_instance = AsyncMock() + mock_instance.__aenter__ = AsyncMock(return_value=mock_instance) + mock_instance.__aexit__ = AsyncMock(return_value=None) + + # Simulate a timeout + import httpx + mock_instance.get = AsyncMock( + side_effect=httpx.TimeoutException("Connection timed out") + ) + mock_client.return_value = mock_instance + + with patch('mcp_server_fetch.server.validate_url_for_ssrf'): + with pytest.raises(McpError) as exc_info: + await fetch_url("https://slow.example.com/", "TestAgent/1.0") + + assert "failed" in str(exc_info.value).lower() + + def test_security_constants_immutable(self): + """Security constants should be immutable (frozenset).""" + assert isinstance(BLOCKED_HOSTNAMES, frozenset) + assert isinstance(CLOUD_METADATA_IPS, frozenset) + + # frozenset has no .add() method - it's truly immutable + assert not hasattr(BLOCKED_HOSTNAMES, 'add') or not callable(getattr(BLOCKED_HOSTNAMES, 'add', None)) + + # Verify we cannot create a modified version that affects the original + original_len = len(BLOCKED_HOSTNAMES) + _ = BLOCKED_HOSTNAMES | {"new-host"} # Creates new frozenset, doesn't modify original + assert len(BLOCKED_HOSTNAMES) == original_len + + +# ============================================================================= +# 6. EDGE CASE TESTS +# ============================================================================= + +class TestEdgeCases: + """Test edge cases and boundary conditions.""" + + def test_ipv6_addresses(self): + """IPv6 addresses should be properly validated.""" + # IPv6 loopback + assert _is_ip_private_or_reserved("::1") is True + + # IPv6 link-local + assert _is_ip_private_or_reserved("fe80::1") is True + + # IPv6 private (unique local) + assert _is_ip_private_or_reserved("fc00::1") is True + assert _is_ip_private_or_reserved("fd00::1") is True + + def test_unspecified_addresses(self): + """Unspecified addresses (0.0.0.0, ::) should be blocked.""" + assert _is_ip_private_or_reserved("0.0.0.0") is True + assert _is_ip_private_or_reserved("::") is True + + def test_multicast_addresses(self): + """Multicast addresses should be blocked.""" + assert _is_ip_private_or_reserved("224.0.0.1") is True + assert _is_ip_private_or_reserved("239.255.255.255") is True + assert _is_ip_private_or_reserved("ff02::1") is True + + def test_url_with_credentials(self): + """URLs with embedded credentials should be handled.""" + # These should still be validated for SSRF + with pytest.raises(McpError): + validate_url_for_ssrf("http://user:pass@localhost/") + + with pytest.raises(McpError): + validate_url_for_ssrf("http://admin:admin@169.254.169.254/") + + def test_url_with_port_bypass_attempt(self): + """Port variations should not bypass security.""" + with pytest.raises(McpError): + validate_url_for_ssrf("http://localhost:80/") + + with pytest.raises(McpError): + validate_url_for_ssrf("http://localhost:443/") + + with pytest.raises(McpError): + validate_url_for_ssrf("http://127.0.0.1:65535/") + + +# ============================================================================= +# RUN CONFIGURATION +# ============================================================================= + +if __name__ == "__main__": + pytest.main([__file__, "-v", "--tb=short"]) \ No newline at end of file diff --git a/src/fetch/tests/test_server.py b/src/fetch/tests/test_server.py index 96c1cb38c7..97f64d25b8 100644 --- a/src/fetch/tests/test_server.py +++ b/src/fetch/tests/test_server.py @@ -252,7 +252,8 @@ async def test_fetch_json_returns_raw(self): mock_response.text = json_content mock_response.headers = {"content-type": "application/json"} - with patch("httpx.AsyncClient") as mock_client_class: + with patch("httpx.AsyncClient") as mock_client_class, \ + patch("mcp_server_fetch.server.validate_url_for_ssrf"): mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client) @@ -310,7 +311,9 @@ async def test_fetch_with_proxy(self): mock_response.text = '{"data": "test"}' mock_response.headers = {"content-type": "application/json"} - with patch("httpx.AsyncClient") as mock_client_class: + with patch("httpx.AsyncClient") as mock_client_class, \ + patch("mcp_server_fetch.server.validate_url_for_ssrf"), \ + patch("mcp_server_fetch.server.SSL_VERIFY", True): mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client) @@ -322,5 +325,5 @@ async def test_fetch_with_proxy(self): proxy_url="http://proxy.example.com:8080" ) - # Verify AsyncClient was called with proxy - mock_client_class.assert_called_once_with(proxy="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) From ad1c2a1af2b53f2ffbaa47a6b37d0aad3e4f074d Mon Sep 17 00:00:00 2001 From: Tomo1912 Date: Thu, 12 Feb 2026 18:25:22 +0100 Subject: [PATCH 2/2] 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. --- src/fetch/src/mcp_server_fetch/server.py | 106 ++++++++++++++-- src/fetch/tests/test_security.py | 147 ++++++++++++++++++++++- src/fetch/tests/test_server.py | 9 +- 3 files changed, 245 insertions(+), 17 deletions(-) diff --git a/src/fetch/src/mcp_server_fetch/server.py b/src/fetch/src/mcp_server_fetch/server.py index cd7ccb348d..c9e0390acc 100644 --- a/src/fetch/src/mcp_server_fetch/server.py +++ b/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, diff --git a/src/fetch/tests/test_security.py b/src/fetch/tests/test_security.py index 8dbddca4f2..62d928e9be 100644 --- a/src/fetch/tests/test_security.py +++ b/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 = "Test" @@ -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 # ============================================================================= diff --git a/src/fetch/tests/test_server.py b/src/fetch/tests/test_server.py index 97f64d25b8..2719d24eb4 100644 --- a/src/fetch/tests/test_server.py +++ b/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" + )