diff --git a/src/fetch/src/mcp_server_fetch/server.py b/src/fetch/src/mcp_server_fetch/server.py index b42c7b1f6b..053e7e6b7f 100644 --- a/src/fetch/src/mcp_server_fetch/server.py +++ b/src/fetch/src/mcp_server_fetch/server.py @@ -1,5 +1,7 @@ +import ipaddress +import socket from typing import Annotated, Tuple -from urllib.parse import urlparse, urlunparse +from urllib.parse import urljoin, urlparse, urlunparse import markdownify import readabilipy.simple_json @@ -23,6 +25,123 @@ 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)" +# Only plain web URLs are ever allowed; anything else (file://, data://, +# ftp://, ...) is rejected up front to avoid protocol-abuse SSRF. +_ALLOWED_SCHEMES = ("http", "https") +_MAX_REDIRECTS = 5 + +# Address ranges a public-facing fetch server must never reach. This includes +# loopback, RFC1918 / IPv6 unique-local, link-local (which covers the cloud +# metadata endpoint 169.254.169.254), and reserved / multicast ranges for both +# address families. See https://github.com/modelcontextprotocol/servers/issues/3741 +_BLOCKED_NETWORKS = ( + ipaddress.ip_network("0.0.0.0/8"), + ipaddress.ip_network("10.0.0.0/8"), + ipaddress.ip_network("100.64.0.0/10"), + ipaddress.ip_network("127.0.0.0/8"), + ipaddress.ip_network("169.254.0.0/16"), + ipaddress.ip_network("172.16.0.0/12"), + ipaddress.ip_network("192.168.0.0/16"), + ipaddress.ip_network("198.18.0.0/15"), + ipaddress.ip_network("224.0.0.0/4"), + ipaddress.ip_network("240.0.0.0/4"), + ipaddress.ip_network("::/128"), + ipaddress.ip_network("::1/128"), + ipaddress.ip_network("fc00::/7"), + ipaddress.ip_network("fe80::/10"), + ipaddress.ip_network("ff00::/8"), +) + + +def _is_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + # IPv4-mapped IPv6 addresses (e.g. ::ffff:127.0.0.1) must be checked as + # IPv4, otherwise the ::-prefixed form slips past the IPv4 rules. + if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: + ip = ip.ipv4_mapped + return any(ip in net for net in _BLOCKED_NETWORKS) + + +def _resolved_ips(host: str) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + """Every IP ``host`` resolves to, deduplicated. + + A literal IP host yields that single address; a hostname is resolved via + DNS and all A / AAAA records are returned so a host that maps to *any* + blocked network is rejected (fail-closed). + """ + try: + return [ipaddress.ip_address(host)] + except ValueError: + pass + try: + candidates = socket.getaddrinfo(host, None) + except socket.gaierror as e: + raise McpError( + ErrorData( + code=INVALID_PARAMS, + message=f"Unable to resolve host '{host}': {e}", + ) + ) + ips: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = [] + for _, _, _, _, sockaddr in candidates: + ip = ipaddress.ip_address(sockaddr[0]) + if ip not in ips: + ips.append(ip) + return ips + + +def _validate_url(url: str) -> None: + """Ensure ``url`` is an http(s) URL pointing at a public, non-private host. + + Blocks scheme abuse (file://, data://, ...) and SSRF targets: literal + internal IPs, and hostnames that resolve to any blocked network such as + loopback, RFC1918, unique-local, link-local and cloud metadata endpoints. + Raises McpError otherwise. + """ + parsed = urlparse(url) + if parsed.scheme not in _ALLOWED_SCHEMES: + raise McpError( + ErrorData( + code=INVALID_PARAMS, + message=f"Only http and https URLs are supported, got scheme '{parsed.scheme}'", + ) + ) + host = parsed.hostname + if not host: + raise McpError( + ErrorData(code=INVALID_PARAMS, message=f"Invalid URL: '{url}' has no host") + ) + for ip in _resolved_ips(host): + if _is_blocked_ip(ip): + raise McpError( + ErrorData( + code=INTERNAL_ERROR, + message=f"Blocked {url}: resolves to a non-public address ({ip}). " + "Fetching loopback, private, link-local or metadata endpoints is not allowed.", + ) + ) + + +async def _get_with_validation(client, url: str, headers: dict, timeout: int = 30): + """GET ``url`` while validating scheme + host on every redirect hop. + + Redirects are followed manually so each intermediate target is re-validated; + ``follow_redirects=True`` in httpx would let a redirect re-point the request + at an internal address after the initial URL passed validation. + """ + current = url + for _ in range(_MAX_REDIRECTS): + _validate_url(current) + response = await client.get( + current, follow_redirects=False, headers=headers, timeout=timeout + ) + if response.is_redirect and "location" in response.headers: + current = urljoin(current, response.headers["location"]) + continue + return response + raise McpError( + ErrorData(code=INTERNAL_ERROR, message=f"Too many redirects fetching {url}") + ) + def extract_content_from_html(html: str) -> str: """Extract and convert HTML content to Markdown format. @@ -63,7 +182,9 @@ 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: +async def check_may_autonomously_fetch_url( + url: str, user_agent: str, proxy_url: str | None = None +) -> None: """ Check if the URL can be fetched by the user agent according to the robots.txt file. Raises a McpError if not. @@ -74,21 +195,23 @@ 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( - robot_txt_url, - follow_redirects=True, - headers={"User-Agent": user_agent}, + response = await _get_with_validation( + client, robot_txt_url, {"User-Agent": user_agent} ) except HTTPError: - raise McpError(ErrorData( - code=INTERNAL_ERROR, - message=f"Failed to fetch robots.txt {robot_txt_url} due to a connection issue", - )) + raise McpError( + ErrorData( + code=INTERNAL_ERROR, + message=f"Failed to fetch robots.txt {robot_txt_url} due to a connection issue", + ) + ) if response.status_code in (401, 403): - raise McpError(ErrorData( - code=INTERNAL_ERROR, - message=f"When fetching robots.txt ({robot_txt_url}), received status {response.status_code} so assuming that autonomous fetching is not allowed, the user can try manually fetching by using the fetch prompt", - )) + raise McpError( + ErrorData( + code=INTERNAL_ERROR, + message=f"When fetching robots.txt ({robot_txt_url}), received status {response.status_code} so assuming that autonomous fetching is not allowed, the user can try manually fetching by using the fetch prompt", + ) + ) elif 400 <= response.status_code < 500: return robot_txt = response.text @@ -97,15 +220,17 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: ) robot_parser = Protego.parse(processed_robot_txt) if not robot_parser.can_fetch(str(url), user_agent): - raise McpError(ErrorData( - code=INTERNAL_ERROR, - message=f"The sites robots.txt ({robot_txt_url}), specifies that autonomous fetching of this page is not allowed, " - f"{user_agent}\n" - f"{url}" - f"\n{robot_txt}\n\n" - f"The assistant must let the user know that it failed to view the page. The assistant may provide further guidance based on the above information.\n" - f"The assistant can tell the user that they can try manually fetching the page by using the fetch prompt within their UI.", - )) + raise McpError( + ErrorData( + code=INTERNAL_ERROR, + message=f"The sites robots.txt ({robot_txt_url}), specifies that autonomous fetching of this page is not allowed, " + f"{user_agent}\n" + f"{url}" + f"\n{robot_txt}\n\n" + f"The assistant must let the user know that it failed to view the page. The assistant may provide further guidance based on the above information.\n" + f"The assistant can tell the user that they can try manually fetching the page by using the fetch prompt within their UI.", + ) + ) async def fetch_url( @@ -118,19 +243,20 @@ async def fetch_url( async with AsyncClient(proxy=proxy_url) as client: try: - response = await client.get( - url, - follow_redirects=True, - headers={"User-Agent": user_agent}, - timeout=30, + response = await _get_with_validation( + client, url, {"User-Agent": user_agent} ) except HTTPError as e: - raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"Failed to fetch {url}: {e!r}")) + raise McpError( + ErrorData(code=INTERNAL_ERROR, message=f"Failed to fetch {url}: {e!r}") + ) if response.status_code >= 400: - raise McpError(ErrorData( - code=INTERNAL_ERROR, - message=f"Failed to fetch {url} - status code {response.status_code}", - )) + raise McpError( + ErrorData( + code=INTERNAL_ERROR, + message=f"Failed to fetch {url} - status code {response.status_code}", + ) + ) page_raw = response.text @@ -232,7 +358,9 @@ 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 + ) content, prefix = await fetch_url( url, user_agent_autonomous, force_raw=args.raw, proxy_url=proxy_url @@ -241,13 +369,17 @@ async def call_tool(name, arguments: dict) -> list[TextContent]: if args.start_index >= original_length: content = "No more content available." else: - truncated_content = content[args.start_index : args.start_index + args.max_length] + truncated_content = content[ + args.start_index : args.start_index + args.max_length + ] if not truncated_content: content = "No more content available." else: content = truncated_content actual_content_length = len(truncated_content) - remaining_content = original_length - (args.start_index + actual_content_length) + remaining_content = original_length - ( + args.start_index + actual_content_length + ) # Only add the prompt to continue fetching if there is still remaining content if actual_content_length == args.max_length and remaining_content > 0: next_start = args.start_index + actual_content_length @@ -262,7 +394,9 @@ 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 + ) # 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..5b598619f0 100644 --- a/src/fetch/tests/test_server.py +++ b/src/fetch/tests/test_server.py @@ -1,5 +1,9 @@ """Tests for the fetch MCP server.""" +import ipaddress +import socket + +import httpx import pytest from unittest.mock import AsyncMock, patch, MagicMock from mcp.shared.exceptions import McpError @@ -9,10 +13,28 @@ get_robots_txt_url, check_may_autonomously_fetch_url, fetch_url, + _get_with_validation, + _is_blocked_ip, + _validate_url, DEFAULT_USER_AGENT_AUTONOMOUS, ) +@pytest.fixture(autouse=True) +def _resolve_hosts_to_public_ip(monkeypatch): + """Resolve every hostname to a single public IP so tests stay hermetic. + + The SSRF validation performs real hostname resolution; pinning the resolver + to a public address lets each URL pass ``_validate_url`` without touching + the network, while still exercising the real code path. + """ + + def fake_getaddrinfo(host, port): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 0))] + + monkeypatch.setattr("socket.getaddrinfo", fake_getaddrinfo) + + class TestGetRobotsTxtUrl: """Tests for get_robots_txt_url function.""" @@ -100,13 +122,14 @@ async def test_allows_when_robots_txt_404(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) # Should not raise await check_may_autonomously_fetch_url( - "https://example.com/page", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://example.com/page", DEFAULT_USER_AGENT_AUTONOMOUS ) @pytest.mark.asyncio @@ -118,13 +141,14 @@ async def test_blocks_when_robots_txt_401(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) with pytest.raises(McpError): await check_may_autonomously_fetch_url( - "https://example.com/page", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://example.com/page", DEFAULT_USER_AGENT_AUTONOMOUS ) @pytest.mark.asyncio @@ -136,13 +160,14 @@ async def test_blocks_when_robots_txt_403(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) with pytest.raises(McpError): await check_may_autonomously_fetch_url( - "https://example.com/page", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://example.com/page", DEFAULT_USER_AGENT_AUTONOMOUS ) @pytest.mark.asyncio @@ -155,13 +180,14 @@ async def test_allows_when_robots_txt_allows_all(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) # Should not raise await check_may_autonomously_fetch_url( - "https://example.com/page", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://example.com/page", DEFAULT_USER_AGENT_AUTONOMOUS ) @pytest.mark.asyncio @@ -174,13 +200,14 @@ async def test_blocks_when_robots_txt_disallows_all(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) with pytest.raises(McpError): await check_may_autonomously_fetch_url( - "https://example.com/page", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://example.com/page", DEFAULT_USER_AGENT_AUTONOMOUS ) @@ -207,12 +234,13 @@ async def test_fetch_html_page(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) content, prefix = await fetch_url( - "https://example.com/page", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://example.com/page", DEFAULT_USER_AGENT_AUTONOMOUS ) # HTML is processed, so we check it returns something @@ -231,13 +259,15 @@ async def test_fetch_html_page_raw(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) content, prefix = await fetch_url( "https://example.com/page", DEFAULT_USER_AGENT_AUTONOMOUS, - force_raw=True + force_raw=True, ) assert content == html_content @@ -255,12 +285,13 @@ async def test_fetch_json_returns_raw(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) content, prefix = await fetch_url( - "https://api.example.com/data", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://api.example.com/data", DEFAULT_USER_AGENT_AUTONOMOUS ) assert content == json_content @@ -275,13 +306,14 @@ async def test_fetch_404_raises_error(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) with pytest.raises(McpError): await fetch_url( - "https://example.com/notfound", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://example.com/notfound", DEFAULT_USER_AGENT_AUTONOMOUS ) @pytest.mark.asyncio @@ -293,13 +325,14 @@ async def test_fetch_500_raises_error(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) with pytest.raises(McpError): await fetch_url( - "https://example.com/error", - DEFAULT_USER_AGENT_AUTONOMOUS + "https://example.com/error", DEFAULT_USER_AGENT_AUTONOMOUS ) @pytest.mark.asyncio @@ -313,14 +346,146 @@ async def test_fetch_with_proxy(self): 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.__aenter__ = AsyncMock( + return_value=mock_client + ) mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None) await fetch_url( "https://example.com/data", DEFAULT_USER_AGENT_AUTONOMOUS, - proxy_url="http://proxy.example.com:8080" + 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") + mock_client_class.assert_called_once_with( + proxy="http://proxy.example.com:8080" + ) + + +class TestIsBlockedIp: + """Unit tests for the blocked-address matcher.""" + + @pytest.mark.parametrize( + "ip_str", + [ + "127.0.0.1", # loopback + "0.0.0.0", # this network + "10.0.0.1", # RFC1918 + "172.16.0.1", # RFC1918 + "192.168.1.5", # RFC1918 + "169.254.169.254", # link-local / cloud metadata + "100.64.0.1", # CGNAT + "198.18.0.1", # benchmarking + "224.0.0.1", # multicast + "::1", # IPv6 loopback + "fc00::1", # IPv6 unique-local + "fe80::1", # IPv6 link-local + "::ffff:127.0.0.1", # IPv4-mapped loopback + ], + ) + def test_blocked(self, ip_str): + assert _is_blocked_ip(ipaddress.ip_address(ip_str)) is True + + @pytest.mark.parametrize("ip_str", ["8.8.8.8", "1.1.1.1", "2606:4700:4700::1111"]) + def test_public_allowed(self, ip_str): + assert _is_blocked_ip(ipaddress.ip_address(ip_str)) is False + + +class TestValidateUrl: + """Tests for SSRF / scheme validation of target URLs.""" + + @pytest.mark.parametrize( + "url", + [ + "file:///etc/passwd", + "ftp://example.com/file", + "data:text/plain,hello", + "gopher://example.com/", + ], + ) + def test_non_http_scheme_rejected(self, url): + with pytest.raises(McpError): + _validate_url(url) + + @pytest.mark.parametrize( + "url", + [ + "https://127.0.0.1/", + "http://10.0.0.1/", + "http://169.254.169.254/latest/meta-data/", + "https://192.168.1.5/", + "http://[::1]/", + "http://[fc00::1]/", + "http://[fe80::1]/", + "http://[::ffff:127.0.0.1]/", + ], + ) + def test_literal_private_ip_rejected(self, url): + with pytest.raises(McpError): + _validate_url(url) + + @pytest.mark.parametrize( + "url", + ["http://8.8.8.8/", "https://example.com/", "http://[2606:4700:4700::1111]/"], + ) + def test_public_url_accepted(self, url): + # Hostname arms rely on the autouse fixture resolving to 8.8.8.8 (public). + _validate_url(url) # should not raise + + def test_hostname_resolving_to_private_is_rejected(self, monkeypatch): + def fake_getaddrinfo(host, port): + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", 0)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 0)), + ] + + monkeypatch.setattr("socket.getaddrinfo", fake_getaddrinfo) + # Fail-closed: a host that maps to *any* blocked address is rejected, + # even when a second A record points at a public IP. + with pytest.raises(McpError): + _validate_url("https://metadata.internal/latest/meta-data/") + + +class _StubAsyncClient: + """Async client that yields a pre-scripted sequence of httpx responses.""" + + def __init__(self, responses): + self.responses = list(responses) + + async def get(self, *args, **kwargs): + return self.responses.pop(0) + + +class TestRedirectValidation: + """Redirect handling must validate every hop, not just the first URL.""" + + @pytest.mark.asyncio + async def test_redirect_into_private_network_is_blocked(self): + redirect = httpx.Response( + 302, headers={"location": "http://169.254.169.254/latest/meta-data/"} + ) + client = _StubAsyncClient([redirect]) + # First hop (example.com -> public) is fine, but following the 302 into + # the metadata endpoint must be rejected before the request is issued. + with pytest.raises(McpError): + await _get_with_validation(client, "https://example.com/start", {}) + + @pytest.mark.asyncio + async def test_redirect_to_public_url_is_followed(self): + redirect = httpx.Response( + 302, headers={"location": "https://example.com/final"} + ) + ok = httpx.Response( + 200, request=httpx.Request("GET", "https://example.com/final") + ) + client = _StubAsyncClient([redirect, ok]) + response = await _get_with_validation(client, "https://example.com/start", {}) + assert response.status_code == 200 + + @pytest.mark.asyncio + async def test_redirect_to_non_http_scheme_is_blocked(self): + redirect = httpx.Response(302, headers={"location": "file:///etc/passwd"}) + client = _StubAsyncClient([redirect]) + with pytest.raises(McpError): + await _get_with_validation(client, "https://example.com/start", {})