Skip to content

fix(fetch): harden URL fetching against SSRF #4773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Ethanz11-creat wants to merge 1 commit into
base: main
Choose a base branch
from
+368 −69
Open

fix(fetch): harden URL fetching against SSRF #4773

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
208 changes: 171 additions & 37 deletions src/fetch/src/mcp_server_fetch/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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.
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand All @@ -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"<useragent>{user_agent}</useragent>\n"
f"<url>{url}</url>"
f"<robots>\n{robot_txt}\n</robots>\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"<useragent>{user_agent}</useragent>\n"
f"<url>{url}</url>"
f"<robots>\n{robot_txt}\n</robots>\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(
Expand All @@ -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

Expand Down Expand Up @@ -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
Expand All @@ -241,13 +369,17 @@ async def call_tool(name, arguments: dict) -> list[TextContent]:
if args.start_index >= original_length:
content = "<error>No more content available.</error>"
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 = "<error>No more content available.</error>"
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
Expand All @@ -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(
Expand Down
Loading