From 41f8a3a18515ce0f450e1031d6ec95ed5d30799c Mon Sep 17 00:00:00 2001 From: olaservo Date: Wed, 8 Jul 2026 07:43:24 -0700 Subject: [PATCH 1/2] fix(fetch): block SSRF to internal/metadata IPs by default The fetch tool issued server-side requests to any URL with no validation of the destination host, and followed redirects without re-checking the target. A prompt-injection-steered URL (or a public URL that 302-redirects to an internal address) could reach loopback, private (RFC1918), link-local, and cloud-metadata endpoints (e.g. 169.254.169.254) and return their contents into the model context. Resolve the host and reject non-public IP addresses (loopback, private, link-local/metadata, multicast, reserved, unspecified), restrict schemes to http/https, and follow redirects manually so every hop is re-validated. The same guard is applied to the robots.txt pre-check. Blocking is on by default; operators who need to fetch internal hosts can opt out with --allow-internal-ips. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/fetch/README.md | 6 +- src/fetch/src/mcp_server_fetch/__init__.py | 8 +- src/fetch/src/mcp_server_fetch/server.py | 164 +++++++++++++++++++-- src/fetch/tests/test_server.py | 94 ++++++++++++ 4 files changed, 261 insertions(+), 11 deletions(-) diff --git a/src/fetch/README.md b/src/fetch/README.md index ed6d2262f4..3d348b58a6 100644 --- a/src/fetch/README.md +++ b/src/fetch/README.md @@ -7,7 +7,7 @@ A Model Context Protocol server that provides web content fetching capabilities. Source: https://github.com/modelcontextprotocol/servers/tree/main/src/fetch > [!CAUTION] -> This server can access local/internal IP addresses and may represent a security risk. Exercise caution when using this MCP server to ensure this does not expose any sensitive data. +> By default this server refuses to fetch loopback, private (RFC1918), link-local, and cloud-metadata IP addresses to mitigate server-side request forgery (SSRF). If you deliberately need to fetch internal hosts, you can re-enable this with `--allow-internal-ips` (see [Customization - Internal IPs](#customization---internal-ips)) — doing so may expose internal services and cloud instance metadata to the model, so only use it in trusted environments. The fetch tool will truncate the response, but by using the `start_index` argument, you can specify where to start the content extraction. This lets models read a webpage in chunks, until they find the information they need. @@ -172,6 +172,10 @@ 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 - Internal IPs + +By default the server blocks requests to loopback, private (RFC1918), link-local, cloud-metadata (169.254.169.254), and other non-public IP addresses to prevent SSRF, and it re-validates the destination on every redirect hop. If you need to fetch internal hosts (for example a service on `localhost`), add the argument `--allow-internal-ips` to the `args` list in the configuration. This disables the SSRF protection, so only enable it 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/__init__.py b/src/fetch/src/mcp_server_fetch/__init__.py index 09744ce319..a2a83a3fbe 100644 --- a/src/fetch/src/mcp_server_fetch/__init__.py +++ b/src/fetch/src/mcp_server_fetch/__init__.py @@ -16,9 +16,15 @@ def main(): help="Ignore robots.txt restrictions", ) parser.add_argument("--proxy-url", type=str, help="Proxy URL to use for requests") + parser.add_argument( + "--allow-internal-ips", + action="store_true", + help="Allow fetching loopback, private, link-local, and cloud-metadata " + "addresses. Disables SSRF protection; only use in trusted environments.", + ) args = parser.parse_args() - asyncio.run(serve(args.user_agent, args.ignore_robots_txt, args.proxy_url)) + asyncio.run(serve(args.user_agent, args.ignore_robots_txt, args.proxy_url, args.allow_internal_ips)) if __name__ == "__main__": diff --git a/src/fetch/src/mcp_server_fetch/server.py b/src/fetch/src/mcp_server_fetch/server.py index b42c7b1f6b..9eb5c698cd 100644 --- a/src/fetch/src/mcp_server_fetch/server.py +++ b/src/fetch/src/mcp_server_fetch/server.py @@ -1,3 +1,6 @@ +import asyncio +import ipaddress +import socket from typing import Annotated, Tuple from urllib.parse import urlparse, urlunparse @@ -63,7 +66,144 @@ def get_robots_txt_url(url: str) -> str: return robots_url -async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: str | None = None) -> None: +# Cap on the number of redirect hops we will follow (and re-validate). +MAX_REDIRECTS = 20 + +# HTTP status codes that indicate a redirect with a Location header. +REDIRECT_STATUS_CODES = (301, 302, 303, 307, 308) + + +def _is_blocked_ip( + ip: ipaddress.IPv4Address | ipaddress.IPv6Address, +) -> bool: + """Return True if an IP address is not safe to fetch (SSRF target). + + Blocks loopback, private (RFC1918), link-local (including the cloud + metadata address 169.254.169.254), unique-local, multicast, reserved, + and unspecified addresses. + """ + # Unwrap IPv4-mapped IPv6 addresses (e.g. ::ffff:127.0.0.1) so the + # underlying IPv4 address is classified rather than the wrapper. + if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: + ip = ip.ipv4_mapped + return ( + ip.is_loopback + or ip.is_private + or ip.is_link_local + or ip.is_multicast + or ip.is_reserved + or ip.is_unspecified + ) + + +async def _resolve_host_ips( + host: str, port: int | None, scheme: str +) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + """Resolve a host to the IP addresses it points at. + + If the host is already an IP literal it is returned directly; otherwise + DNS resolution is performed and every returned address is checked. + """ + try: + return [ipaddress.ip_address(host)] + except ValueError: + pass + + default_port = 443 if scheme == "https" else 80 + try: + infos = await asyncio.get_running_loop().getaddrinfo( + host, port or default_port, type=socket.SOCK_STREAM + ) + except socket.gaierror as e: + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"Failed to resolve host {host}: {e}", + )) + + ips: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = [] + for info in infos: + try: + ips.append(ipaddress.ip_address(info[4][0])) + except ValueError: + continue + return ips + + +async def _validate_url_is_safe(url: str) -> None: + """Guard a URL against SSRF before it is fetched. + + Rejects non-http(s) schemes and any URL whose host resolves to a + non-public IP address. Servers started with --allow-internal-ips skip + this check entirely. + """ + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Cannot fetch {url}: only http and https URLs are supported.", + )) + + host = parsed.hostname + if not host: + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Cannot fetch {url}: URL has no host.", + )) + + for ip in await _resolve_host_ips(host, parsed.port, parsed.scheme): + if _is_blocked_ip(ip): + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=( + f"Cannot fetch {url}: host {host} resolves to non-public IP " + f"address {ip}. Fetching loopback, private, link-local, and " + f"cloud-metadata addresses is blocked to prevent SSRF. Start " + f"the server with --allow-internal-ips to override this." + ), + )) + + +async def _get_following_redirects( + client, + url: str, + *, + headers: dict, + timeout: float, + allow_internal_ips: bool, +): + """GET a URL, following redirects manually so every hop is re-validated. + + httpx's automatic redirect handling never re-checks the destination, so + a public URL that 302-redirects to an internal address would bypass the + guard. Following redirects manually lets us validate each hop. + """ + from httpx import URL + + current_url = url + for _ in range(MAX_REDIRECTS + 1): + if not allow_internal_ips: + await _validate_url_is_safe(current_url) + response = await client.get( + current_url, + follow_redirects=False, + headers=headers, + timeout=timeout, + ) + if response.status_code in REDIRECT_STATUS_CODES: + location = response.headers.get("location") + if not location: + return response + current_url = str(URL(current_url).join(location)) + continue + return response + + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"Cannot fetch {url}: exceeded the maximum of {MAX_REDIRECTS} redirects.", + )) + + +async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: str | None = None, allow_internal_ips: bool = False) -> None: """ Check if the URL can be fetched by the user agent according to the robots.txt file. Raises a McpError if not. @@ -74,10 +214,12 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: async with AsyncClient(proxy=proxy_url) as client: try: - response = await client.get( + response = await _get_following_redirects( + client, robot_txt_url, - follow_redirects=True, headers={"User-Agent": user_agent}, + timeout=30, + allow_internal_ips=allow_internal_ips, ) except HTTPError: raise McpError(ErrorData( @@ -109,7 +251,7 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: async def fetch_url( - url: str, user_agent: str, force_raw: bool = False, proxy_url: str | None = None + url: str, user_agent: str, force_raw: bool = False, proxy_url: str | None = None, allow_internal_ips: bool = False ) -> 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. @@ -118,11 +260,12 @@ async def fetch_url( async with AsyncClient(proxy=proxy_url) as client: try: - response = await client.get( + response = await _get_following_redirects( + client, url, - follow_redirects=True, headers={"User-Agent": user_agent}, timeout=30, + allow_internal_ips=allow_internal_ips, ) except HTTPError as e: raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"Failed to fetch {url}: {e!r}")) @@ -182,6 +325,7 @@ async def serve( custom_user_agent: str | None = None, ignore_robots_txt: bool = False, proxy_url: str | None = None, + allow_internal_ips: bool = False, ) -> None: """Run the fetch MCP server. @@ -189,6 +333,8 @@ async def serve( custom_user_agent: Optional custom User-Agent string to use for requests ignore_robots_txt: Whether to ignore robots.txt restrictions proxy_url: Optional proxy URL to use for requests + allow_internal_ips: Allow fetching loopback/private/link-local/metadata + addresses (disables SSRF protection). Off by default. """ server = Server("mcp-fetch") user_agent_autonomous = custom_user_agent or DEFAULT_USER_AGENT_AUTONOMOUS @@ -232,10 +378,10 @@ async def call_tool(name, arguments: dict) -> list[TextContent]: raise McpError(ErrorData(code=INVALID_PARAMS, message="URL is required")) if not ignore_robots_txt: - await check_may_autonomously_fetch_url(url, user_agent_autonomous, proxy_url) + await check_may_autonomously_fetch_url(url, user_agent_autonomous, proxy_url, allow_internal_ips) content, prefix = await fetch_url( - url, user_agent_autonomous, force_raw=args.raw, proxy_url=proxy_url + url, user_agent_autonomous, force_raw=args.raw, proxy_url=proxy_url, allow_internal_ips=allow_internal_ips ) original_length = len(content) if args.start_index >= original_length: @@ -262,7 +408,7 @@ async def get_prompt(name: str, arguments: dict | None) -> GetPromptResult: url = arguments["url"] try: - content, prefix = await fetch_url(url, user_agent_manual, proxy_url=proxy_url) + content, prefix = await fetch_url(url, user_agent_manual, proxy_url=proxy_url, allow_internal_ips=allow_internal_ips) # TODO: after SDK bug is addressed, don't catch the exception except McpError as e: return GetPromptResult( diff --git a/src/fetch/tests/test_server.py b/src/fetch/tests/test_server.py index 96c1cb38c7..25269034f7 100644 --- a/src/fetch/tests/test_server.py +++ b/src/fetch/tests/test_server.py @@ -9,6 +9,7 @@ get_robots_txt_url, check_may_autonomously_fetch_url, fetch_url, + _validate_url_is_safe, DEFAULT_USER_AGENT_AUTONOMOUS, ) @@ -91,6 +92,13 @@ def test_empty_content_returns_error(self): class TestCheckMayAutonomouslyFetchUrl: """Tests for check_may_autonomously_fetch_url function.""" + @pytest.fixture(autouse=True) + def _skip_ssrf_guard(self): + """These tests exercise robots.txt handling, not the SSRF guard, and + mock the HTTP client, so stub host validation to avoid real DNS.""" + with patch("mcp_server_fetch.server._validate_url_is_safe", new=AsyncMock()): + yield + @pytest.mark.asyncio async def test_allows_when_robots_txt_404(self): """Test that fetching is allowed when robots.txt returns 404.""" @@ -187,6 +195,13 @@ async def test_blocks_when_robots_txt_disallows_all(self): class TestFetchUrl: """Tests for fetch_url function.""" + @pytest.fixture(autouse=True) + def _skip_ssrf_guard(self): + """These tests exercise fetch/content handling, not the SSRF guard, and + mock the HTTP client, so stub host validation to avoid real DNS.""" + with patch("mcp_server_fetch.server._validate_url_is_safe", new=AsyncMock()): + yield + @pytest.mark.asyncio async def test_fetch_html_page(self): """Test fetching an HTML page returns markdown content.""" @@ -324,3 +339,82 @@ async def test_fetch_with_proxy(self): # Verify AsyncClient was called with proxy mock_client_class.assert_called_once_with(proxy="http://proxy.example.com:8080") + + +class TestValidateUrlIsSafe: + """Tests for the SSRF guard (_validate_url_is_safe). + + These use IP literals so no DNS resolution (and no network) is required. + """ + + @pytest.mark.asyncio + async def test_blocks_loopback_ipv4(self): + with pytest.raises(McpError): + await _validate_url_is_safe("http://127.0.0.1/") + + @pytest.mark.asyncio + async def test_blocks_cloud_metadata_address(self): + with pytest.raises(McpError): + await _validate_url_is_safe("http://169.254.169.254/latest/meta-data/") + + @pytest.mark.asyncio + async def test_blocks_private_ranges(self): + for host in ("10.0.0.1", "192.168.1.1", "172.16.0.1"): + with pytest.raises(McpError): + await _validate_url_is_safe(f"http://{host}/") + + @pytest.mark.asyncio + async def test_blocks_unspecified_address(self): + with pytest.raises(McpError): + await _validate_url_is_safe("http://0.0.0.0/") + + @pytest.mark.asyncio + async def test_blocks_ipv6_loopback(self): + with pytest.raises(McpError): + await _validate_url_is_safe("http://[::1]/") + + @pytest.mark.asyncio + async def test_blocks_ipv4_mapped_ipv6_loopback(self): + with pytest.raises(McpError): + await _validate_url_is_safe("http://[::ffff:127.0.0.1]/") + + @pytest.mark.asyncio + async def test_blocks_non_http_scheme(self): + with pytest.raises(McpError): + await _validate_url_is_safe("file:///etc/passwd") + + @pytest.mark.asyncio + async def test_allows_public_ip_literal(self): + # Public IP literal: should not raise (no DNS needed). + await _validate_url_is_safe("https://1.1.1.1/") + + @pytest.mark.asyncio + async def test_blocked_url_rejected_by_fetch_url(self): + """The guard is enforced end-to-end through fetch_url by default.""" + with pytest.raises(McpError): + await fetch_url( + "http://169.254.169.254/latest/meta-data/", + DEFAULT_USER_AGENT_AUTONOMOUS, + ) + + @pytest.mark.asyncio + async def test_allow_internal_ips_bypasses_guard(self): + """With allow_internal_ips=True the guard is skipped (no McpError from + validation); the request proceeds to the mocked client.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = "internal" + mock_response.headers = {"content-type": "text/plain"} + + with patch("httpx.AsyncClient") as mock_client_class: + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) + + content, _ = await fetch_url( + "http://127.0.0.1/secret", + DEFAULT_USER_AGENT_AUTONOMOUS, + allow_internal_ips=True, + ) + assert content == "internal" From 0e8f64c18d5c093db944fc26059ba92346a33b25 Mon Sep 17 00:00:00 2001 From: olaservo Date: Fri, 10 Jul 2026 08:02:36 -0700 Subject: [PATCH 2/2] fix(fetch): block carrier-grade NAT and IPv4-compatible IPv6 SSRF targets Explicitly block 100.64.0.0/10 (not flagged by is_private before Python 3.13) and unwrap deprecated IPv4-compatible IPv6 addresses (::a.b.c.d, ::/96) so forms like [::127.0.0.1] are classified by their embedded IPv4 address. Brings the fetch guard to parity with the everything server's SSRF classifier. Adds tests for both. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/fetch/src/mcp_server_fetch/server.py | 23 ++++++++++++++++++++--- src/fetch/tests/test_server.py | 10 ++++++++++ 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/src/fetch/src/mcp_server_fetch/server.py b/src/fetch/src/mcp_server_fetch/server.py index 9eb5c698cd..21e225d224 100644 --- a/src/fetch/src/mcp_server_fetch/server.py +++ b/src/fetch/src/mcp_server_fetch/server.py @@ -72,20 +72,37 @@ def get_robots_txt_url(url: str) -> str: # HTTP status codes that indicate a redirect with a Location header. REDIRECT_STATUS_CODES = (301, 302, 303, 307, 308) +# Carrier-grade NAT range: not globally reachable, but not flagged by +# is_private on Python < 3.13, so it is checked explicitly. +_CGNAT_NETWORK = ipaddress.ip_network("100.64.0.0/10") + def _is_blocked_ip( ip: ipaddress.IPv4Address | ipaddress.IPv6Address, ) -> bool: """Return True if an IP address is not safe to fetch (SSRF target). - Blocks loopback, private (RFC1918), link-local (including the cloud - metadata address 169.254.169.254), unique-local, multicast, reserved, - and unspecified addresses. + Blocks loopback, private (RFC1918), carrier-grade NAT, link-local + (including the cloud metadata address 169.254.169.254), unique-local, + multicast, reserved, and unspecified addresses. """ # Unwrap IPv4-mapped IPv6 addresses (e.g. ::ffff:127.0.0.1) so the # underlying IPv4 address is classified rather than the wrapper. if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: ip = ip.ipv4_mapped + # Unwrap deprecated IPv4-compatible IPv6 addresses (::a.b.c.d, ::/96), + # other than :: and ::1 which are classified as unspecified/loopback + # below, so the embedded IPv4 address (e.g. ::127.0.0.1) is checked. + elif ( + isinstance(ip, ipaddress.IPv6Address) + and int(ip) >> 32 == 0 + and int(ip) not in (0, 1) + ): + ip = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF) + + if isinstance(ip, ipaddress.IPv4Address) and ip in _CGNAT_NETWORK: + return True + return ( ip.is_loopback or ip.is_private diff --git a/src/fetch/tests/test_server.py b/src/fetch/tests/test_server.py index 25269034f7..a267d7876d 100644 --- a/src/fetch/tests/test_server.py +++ b/src/fetch/tests/test_server.py @@ -368,6 +368,16 @@ async def test_blocks_unspecified_address(self): with pytest.raises(McpError): await _validate_url_is_safe("http://0.0.0.0/") + @pytest.mark.asyncio + async def test_blocks_carrier_grade_nat(self): + with pytest.raises(McpError): + await _validate_url_is_safe("http://100.64.0.1/") + + @pytest.mark.asyncio + async def test_blocks_ipv4_compatible_ipv6_loopback(self): + with pytest.raises(McpError): + await _validate_url_is_safe("http://[::127.0.0.1]/") + @pytest.mark.asyncio async def test_blocks_ipv6_loopback(self): with pytest.raises(McpError):