diff --git a/src/fetch/src/mcp_server_fetch/server.py b/src/fetch/src/mcp_server_fetch/server.py index b42c7b1f6b..a9c9c3a83c 100644 --- a/src/fetch/src/mcp_server_fetch/server.py +++ b/src/fetch/src/mcp_server_fetch/server.py @@ -1,3 +1,5 @@ +import ipaddress +import socket from typing import Annotated, Tuple from urllib.parse import urlparse, urlunparse @@ -108,6 +110,23 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: )) +def _check_url_not_private(url: str) -> None: + """Raise McpError if the URL resolves to a private/internal IP address.""" + parsed = urlparse(url) + hostname = parsed.hostname + if not hostname: + raise McpError(ErrorData(code=INVALID_PARAMS, message="Invalid URL: no hostname")) + try: + ip = ipaddress.ip_address(socket.gethostbyname(hostname)) + except socket.gaierror as e: + raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"Failed to resolve hostname: {hostname}: {e}")) + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Access to private/internal network addresses is not allowed: {hostname}", + )) + + async def fetch_url( url: str, user_agent: str, force_raw: bool = False, proxy_url: str | None = None ) -> Tuple[str, str]: @@ -116,6 +135,8 @@ async def fetch_url( """ from httpx import AsyncClient, HTTPError + _check_url_not_private(url) + async with AsyncClient(proxy=proxy_url) as client: try: response = await client.get(