Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions src/fetch/src/mcp_server_fetch/server.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import ipaddress
import socket
from typing import Annotated, Tuple
from urllib.parse import urlparse, urlunparse

Expand Down Expand Up @@ -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]:
Expand All @@ -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(
Expand Down
Loading