ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

Backport #4412 to 3.x: support trusted SSRF proxies (#4755)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-08-05 10:29:38 -04:00 • committed by GitHub
commit fcdf422e54
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 548 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

20
docs/more/settings.mdx
View file

@ -88,6 +88,26 @@ When setting Docket values in a `.env` file, use a **double** underscore: `FASTM
| `FASTMCP_DOCKET_RECONNECTION_DELAY` | `timedelta` | `5s` | Delay between reconnection attempts when the worker loses its backend connection. |
| `FASTMCP_DOCKET_MINIMUM_CHECK_INTERVAL` | `timedelta` | `50ms` | How frequently the worker polls for new tasks. Lower values reduce latency at the cost of more CPU usage. |
## Security
These control FastMCP's SSRF protection for the outbound fetches it makes during authentication (OAuth client metadata and JWKS).
| Environment Variable | Type | Default | Description |
|---|---|---|---|
| `FASTMCP_SSRF_TRUST_PROXY` | `bool` | `false` | Trust an outbound HTTP proxy for SSRF-protected fetches. When `false`, FastMCP resolves the target hostname itself and refuses to connect if it maps to a private, loopback, link-local, or reserved IP. When `true`, FastMCP routes auth metadata and JWKS fetches through the configured `HTTPS_PROXY`/`ALL_PROXY` and does not honor `NO_PROXY`; if no proxy is configured the fetch is refused. |
By default, FastMCP protects its OAuth and JWKS fetches against [SSRF](https://owasp.org/www-community/attacks/Server_Side_Request_Forgery) by resolving the target hostname, rejecting any address that maps to a private, loopback, link-local, or reserved IP, and then pinning the connection to that validated IP.
This breaks when a corporate `CONNECT` proxy is the only egress path: the container often cannot resolve external DNS at all (only the proxy can), and even when it can, pinning to the IP makes TLS verification fail because public certificates list hostnames, not IP addresses.
Set `FASTMCP_SSRF_TRUST_PROXY=true` when a trusted proxy is your mandated egress. FastMCP then skips DNS resolution and the IP blocklist entirely and makes a single request to the hostname URL, explicitly routed through the proxy named by the standard `HTTPS_PROXY` / `ALL_PROXY` environment variables (checked in that order). The HTTPS-only and hostname checks still apply.
<Warning>
This is a deliberate trust shift: the IP blocklist cannot be enforced through a proxy (the proxy does its own DNS, so an address FastMCP resolved is not the one the proxy dials). Only enable it when the proxy itself is trusted to mediate egress.
FastMCP reads the proxy URL from the environment and passes it to the HTTP client explicitly, so environment proxy selection and `NO_PROXY` do not participate — the request either goes through that exact proxy or fails outright. TLS trust environment variables remain enabled, so `SSL_CERT_FILE` and `SSL_CERT_DIR` continue to work for deployments that install a corporate CA. A host that `NO_PROXY` would otherwise exclude is still routed through the configured proxy rather than fetched direct with the IP blocklist disabled — the safer of the two options, since the blocklist cannot apply to a direct fetch here anyway. If you set `FASTMCP_SSRF_TRUST_PROXY=true` but neither `HTTPS_PROXY` nor `ALL_PROXY` is present in the server process's environment (an `HTTP_PROXY` alone never routes these HTTPS-only fetches), the request would otherwise go out **direct with the IP blocklist disabled** — no SSRF protection at all. Rather than send it, FastMCP refuses the fetch and raises `SSRFError` with an actionable message. The contract is crisp: proxy-trust mode delegates SSRF protection to the proxy, and with no proxy configured the fetch cannot proceed. Enable this setting only together with an active proxy that routes your auth endpoints.
</Warning>
## Advanced
| Environment Variable | Type | Default | Description |

175
fastmcp_slim/fastmcp/server/auth/ssrf.py
View file

@ -4,12 +4,22 @@ This module provides SSRF-protected HTTP fetching with:
- DNS resolution and IP validation before requests
- DNS pinning to prevent rebinding TOCTOU attacks
- Support for both CIMD and JWKS fetches
When ``FASTMCP_SSRF_TRUST_PROXY`` is set, DNS resolution and the IP blocklist are
skipped and a single request is made to the hostname URL through the configured
HTTPS_PROXY/ALL_PROXY, delegating DNS and egress to that trusted proxy (the scheme
and hostname checks still apply). The proxy URL is read from the environment and
passed to httpx explicitly, so environment proxy selection and NO_PROXY are not
evaluated. ``trust_env`` remains enabled so SSL_CERT_FILE and SSL_CERT_DIR continue
to provide corporate CA trust. If no proxy is configured, the fetch is refused
rather than sent direct with the blocklist disabled.
"""
from __future__ import annotations
import asyncio
import ipaddress
import os
import socket
import time
from collections.abc import Mapping
@ -18,6 +28,7 @@ from urllib.parse import urlparse
import httpx
import fastmcp
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
@ -62,6 +73,26 @@ def format_ip_for_url(ip_str: str) -> str:
return ip_str
def _configured_proxy_url() -> str | None:
"""Return the proxy URL to route proxy-trust fetches through, if any is set.
Reads ``HTTPS_PROXY``/``https_proxy`` first, falling back to ``ALL_PROXY``/
``all_proxy``. This is a simple presence check: no host matching, no ``NO_PROXY``
evaluation. The caller passes the returned URL to httpx explicitly, which takes
precedence over environment proxy selection and NO_PROXY while preserving
environment-provided CA trust — see the module docstring and :func:`validate_url`
for why that matters.
Returns:
The configured proxy URL, or None if none of the supported variables are set.
"""
for name in ("HTTPS_PROXY", "https_proxy", "ALL_PROXY", "all_proxy"):
value = os.environ.get(name)
if value:
return value
return None
class SSRFError(Exception):
"""Raised when an SSRF protection check fails."""
@ -179,6 +210,7 @@ class ValidatedURL:
port: int
path: str
resolved_ips: list[str]
proxy_url: str | None = None
@dataclass
@ -190,6 +222,55 @@ class SSRFFetchResponse:
headers: dict[str, str]
@dataclass
class _FetchTarget:
"""A single connection attempt for an SSRF-safe fetch.
In pinned (default) mode there is one target per resolved IP: the request goes to
an IP-literal URL with Host and SNI pinned to the validated hostname. In proxy
mode (FASTMCP_SSRF_TRUST_PROXY) there is a single target: the original hostname
URL with no pinning and an explicit ``proxy_url``, so the request is dialed
through the trusted proxy and the proxy (not httpx's environment-proxy routing)
owns DNS and TLS.
"""
url: str
host_header: str | None
sni_hostname: str | None
proxy_url: str | None = None
def _build_fetch_targets(validated: ValidatedURL) -> list[_FetchTarget]:
"""Build the ordered connection attempts for a validated URL.
An empty ``resolved_ips`` means proxy mode (see :func:`validate_url`): a single
unpinned request to the original hostname URL, explicitly routed through
``validated.proxy_url``. Otherwise, one pinned IP-literal request per resolved
IP, tried in order with fallback on connection error.
"""
if not validated.resolved_ips:
# Proxy mode: dial the original hostname URL verbatim and let the proxy parse
# and resolve it. validated.hostname is informational here — it does not
# constrain what gets dialed — so do not pin Host or SNI from it.
return [
_FetchTarget(
url=validated.original_url,
host_header=None,
sni_hostname=None,
proxy_url=validated.proxy_url,
)
]
return [
_FetchTarget(
url=f"https://{format_ip_for_url(ip)}:{validated.port}{validated.path}",
host_header=validated.hostname,
sni_hostname=validated.hostname,
)
for ip in validated.resolved_ips
]
async def validate_url(url: str, require_path: bool = False) -> ValidatedURL:
"""Validate URL for SSRF and resolve to IPs.
@ -201,7 +282,8 @@ async def validate_url(url: str, require_path: bool = False) -> ValidatedURL:
ValidatedURL with resolved IPs
Raises:
SSRFError: If URL is invalid or resolves to blocked IPs
SSRFError: If the URL is invalid, resolves to blocked IPs, or proxy-trust
mode is enabled but no configured proxy will route the request.
"""
try:
parsed = urlparse(url)
@ -219,8 +301,55 @@ async def validate_url(url: str, require_path: bool = False) -> ValidatedURL:
hostname = parsed.hostname or parsed.netloc
port = parsed.port or 443
path = parsed.path + ("?" + parsed.query if parsed.query else "")
# Resolve and validate IPs
# Proxy mode (FASTMCP_SSRF_TRUST_PROXY): a trusted outbound proxy owns DNS and
# egress, so resolving the hostname here is pointless — the IP we'd pin is not
# the one the proxy dials, making the blocklist unenforceable theater. Skip
# resolution and the blocklist entirely and signal proxy mode downstream with an
# empty resolved_ips list. The scheme (HTTPS) and host checks above still run.
if fastmcp.settings.ssrf_trust_proxy:
# Skipping the blocklist is only safe if the request is *actually* routed
# through a trusted proxy, so this does not try to predict whether it will
# be — it controls it. Earlier revisions predicted the HTTP client's routing
# decision, first by approximating NO_PROXY handling with urllib's proxy
# bypass helper,
# then by replicating the client's own environment-proxy matching
# internally. Both were still predictions of a library with
# open-ended NO_PROXY semantics, and each was found wrong for a different
# NO_PROXY form (port-qualified, IPv6, scheme-qualified entries each broke a
# different revision) — always in the dangerous direction of assuming
# "proxied" for a request that actually went out direct.
#
# Instead, read the proxy URL directly from the environment and hand it to
# httpx explicitly below. An explicit `proxy=` fixes httpx's proxy map without
# consulting NO_PROXY, even with `trust_env=True`; keeping trust_env enabled
# preserves SSL_CERT_FILE and SSL_CERT_DIR for corporate CA trust. The request
# therefore goes through that proxy or the connection fails. A NO_PROXY'd
# host is routed through the proxy rather than fetched direct with the
# blocklist already disabled, which is strictly safer than the alternative
# (see the module docstring). If no proxy is configured, there is nothing to
# route through, so refuse rather than fetch unprotected.
proxy_url = _configured_proxy_url()
if proxy_url is None:
raise SSRFError(
f"FASTMCP_SSRF_TRUST_PROXY is enabled but no HTTPS_PROXY/ALL_PROXY is "
f"configured, so the request to {hostname} would go direct with SSRF "
f"protection disabled. Set HTTPS_PROXY (or ALL_PROXY) to the trusted "
f"proxy, or unset FASTMCP_SSRF_TRUST_PROXY to restore DNS/IP "
f"validation."
)
return ValidatedURL(
original_url=url,
hostname=hostname,
port=port,
path=path,
resolved_ips=[],
proxy_url=proxy_url,
)
# Resolve and validate IPs (resolve_hostname raises rather than returning [], so a
# successful return here always yields a non-empty list — see ssrf_safe_fetch_response).
resolved_ips = await resolve_hostname(hostname, port)
blocked = [ip for ip in resolved_ips if not is_ip_allowed(ip)]
@ -234,7 +363,7 @@ async def validate_url(url: str, require_path: bool = False) -> ValidatedURL:
original_url=url,
hostname=hostname,
port=port,
path=parsed.path + ("?" + parsed.query if parsed.query else ""),
path=path,
resolved_ips=resolved_ips,
)
@ -305,31 +434,34 @@ async def ssrf_safe_fetch_response(
last_error: Exception | None = None
expected_statuses = allowed_status_codes or {200}
for pinned_ip in validated.resolved_ips:
# One target per pinned IP in default mode; a single unpinned target in proxy mode.
targets = _build_fetch_targets(validated)
for target in targets:
elapsed = time.monotonic() - start_time
if elapsed > overall_timeout:
raise SSRFFetchError(f"Overall timeout exceeded: {url}")
remaining = max(1.0, overall_timeout - elapsed)
pinned_url = (
f"https://{format_ip_for_url(pinned_ip)}:{validated.port}{validated.path}"
)
logger.debug("SSRF-safe fetch: %s -> %s", url, target.url)
logger.debug(
"SSRF-safe fetch: %s -> %s (pinned to %s)",
url,
pinned_url,
pinned_ip,
)
headers = {"Host": validated.hostname}
# In pinned mode Host is forced to the validated hostname; in proxy mode httpx
# derives it from the hostname URL. Either way, never let a caller override it.
headers: dict[str, str] = {}
if target.host_header is not None:
headers["Host"] = target.host_header
if request_headers:
for key, value in request_headers.items():
# Host must remain pinned to the validated hostname.
if key.lower() == "host":
continue
headers[key] = value
# Pin SNI to the hostname when connecting to an IP literal; in proxy mode httpx
# derives SNI from the URL, so no override is sent.
extensions: dict[str, str] = {}
if target.sni_hostname is not None:
extensions["sni_hostname"] = target.sni_hostname
try:
# Use httpx with streaming to enforce size limit during download
async with (
@ -342,12 +474,17 @@ async def ssrf_safe_fetch_response(
),
follow_redirects=False,
verify=True,
# An explicit proxy_url controls routing without consulting
# environment proxy selection or NO_PROXY. Keep trust_env enabled
# in both modes so SSL_CERT_FILE and SSL_CERT_DIR remain effective.
proxy=target.proxy_url,
trust_env=True,
) as client,
client.stream(
"GET",
pinned_url,
target.url,
headers=headers,
extensions={"sni_hostname": validated.hostname},
extensions=extensions,
) as response,
):
if time.monotonic() - start_time > overall_timeout:
@ -399,4 +536,4 @@ async def ssrf_safe_fetch_response(
raise SSRFFetchError(f"Timeout fetching {url}") from last_error
raise SSRFFetchError(f"Error fetching {url}: {last_error}") from last_error
raise SSRFFetchError(f"Error fetching {url}: no resolved IPs succeeded")
raise SSRFFetchError(f"Error fetching {url}: no fetch targets succeeded")

20
fastmcp_slim/fastmcp/settings.py
View file

@ -310,6 +310,26 @@ class Settings(BaseSettings):
),
] = False
ssrf_trust_proxy: Annotated[
bool,
Field(
description=inspect.cleandoc(
"""
Trust an outbound HTTP proxy for SSRF-protected fetches (OAuth client
metadata and JWKS). When False (default), FastMCP resolves the target
hostname itself and refuses to connect if it maps to a private,
loopback, link-local, or otherwise reserved IP. When True, FastMCP
routes auth metadata and JWKS fetches through the configured
HTTPS_PROXY/ALL_PROXY and does not honor NO_PROXY; if no proxy is
configured the fetch is refused (raising SSRFError) rather than sent
direct with the blocklist disabled. Only enable this when a trusted
corporate proxy is the mandated egress path: it shifts SSRF trust to
that proxy. Scheme (HTTPS-only) and hostname checks still apply.
"""
),
),
] = False
server_dependencies: list[str] = Field(
default_factory=list,
description="List of dependencies to install in the server environment",

352
tests/server/auth/test_ssrf_protection.py
View file

@ -3,18 +3,56 @@
This module tests the ssrf.py module which provides SSRF-protected HTTP fetching.
"""
import socket
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
import fastmcp
from fastmcp.server.auth.ssrf import (
SSRFError,
SSRFFetchError,
is_ip_allowed,
ssrf_safe_fetch,
ssrf_safe_fetch_response,
validate_url,
)
from fastmcp.utilities.tests import temporary_settings
def _mock_httpx_client(
*,
status_code: int = 200,
headers: dict[str, str] | None = None,
body_chunks: list[bytes] | None = None,
) -> AsyncMock:
"""Build a mock httpx.AsyncClient whose stream() yields a canned response.
The returned client's ``.stream.call_args`` exposes the request that was made.
"""
if headers is None:
headers = {"content-length": "2"}
if body_chunks is None:
body_chunks = [b"ok"]
mock_stream = MagicMock()
mock_stream.status_code = status_code
mock_stream.headers = headers
mock_stream.__aenter__ = AsyncMock(return_value=mock_stream)
mock_stream.__aexit__ = AsyncMock(return_value=None)
async def aiter_bytes():
for chunk in body_chunks:
yield chunk
mock_stream.aiter_bytes = aiter_bytes
mock_client = AsyncMock()
mock_client.stream = MagicMock(return_value=mock_stream)
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__ = AsyncMock(return_value=None)
return mock_client
class TestIsIPAllowed:
@ -506,3 +544,317 @@ class TestStreamingResponseSizeLimit:
with pytest.raises(SSRFFetchError, match="too large"):
await ssrf_safe_fetch("https://example.com/api", max_size=5120)
class TestProxyMode:
"""Tests for FASTMCP_SSRF_TRUST_PROXY (proxy trust) mode.
In proxy mode FastMCP skips its own DNS resolution and IP blocklist. Rather than
predicting whether the HTTP client would route a request through a proxy -- a strategy
that broke three times chasing different NO_PROXY forms (port-qualified, IPv6,
scheme-qualified) -- it reads the proxy URL directly from the environment and
hands it to httpx explicitly, so the request is provably routed through that
proxy rather than predicted to be. NO_PROXY is therefore not evaluated, while
trust_env remains enabled for environment-provided CA trust. The scheme (HTTPS)
and host checks still apply.
"""
@pytest.fixture(autouse=True)
def _clear_proxy_env(self, monkeypatch):
"""Start every test from a clean slate for both spellings of every proxy
variable, so a proxy inherited from the host/CI environment (or left behind
by another test) can't leak in and make behavior non-deterministic."""
for name in (
"HTTP_PROXY",
"http_proxy",
"HTTPS_PROXY",
"https_proxy",
"ALL_PROXY",
"all_proxy",
"NO_PROXY",
"no_proxy",
):
monkeypatch.delenv(name, raising=False)
def test_flag_defaults_to_false(self):
"""The trust-proxy flag must be off by default (no silent weakening)."""
assert fastmcp.settings.ssrf_trust_proxy is False
async def test_validate_url_skips_resolution_and_blocklist(self, monkeypatch):
"""Proxy mode returns resolved_ips=[] without resolving or blocklisting, and
carries the configured proxy URL for the fetch to use."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname") as mock_resolve,
patch("fastmcp.server.auth.ssrf.is_ip_allowed") as mock_blocklist,
):
result = await validate_url("https://example.com/path")
assert result.resolved_ips == []
assert result.original_url == "https://example.com/path"
assert result.hostname == "example.com"
assert result.proxy_url == "http://proxy.internal:3128"
mock_resolve.assert_not_called()
mock_blocklist.assert_not_called()
async def test_validate_url_still_rejects_http(self, monkeypatch):
"""Proxy mode keeps the HTTPS-only scheme check."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
with temporary_settings(ssrf_trust_proxy=True):
with pytest.raises(SSRFError, match="must use HTTPS"):
await validate_url("http://example.com/path")
async def test_validate_url_still_rejects_missing_host(self, monkeypatch):
"""Proxy mode keeps the host check."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
with temporary_settings(ssrf_trust_proxy=True):
with pytest.raises(SSRFError, match="must have a host"):
await validate_url("https:///path")
async def test_validate_url_still_enforces_require_path(self, monkeypatch):
"""Proxy mode keeps the require_path check (CIMD)."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
with temporary_settings(ssrf_trust_proxy=True):
with pytest.raises(SSRFError, match="non-root path"):
await validate_url("https://example.com/", require_path=True)
async def test_raises_when_no_proxy_is_configured(self):
"""No proxy in the environment → refuse rather than fetch unprotected."""
with temporary_settings(ssrf_trust_proxy=True):
with pytest.raises(SSRFError, match="no HTTPS_PROXY/ALL_PROXY"):
await validate_url("https://example.com/path")
async def test_fetch_refuses_end_to_end_when_no_proxy_configured(self):
"""The refusal surfaces through ssrf_safe_fetch: no client is ever built.
The whole point of the hard failure is that the *fetch* cannot proceed, so
this drives it through the public entrypoint and asserts no httpx client is
ever constructed — the request never leaves the process with the blocklist
disabled.
"""
with (
temporary_settings(ssrf_trust_proxy=True),
patch("httpx.AsyncClient") as mock_client_class,
):
with pytest.raises(SSRFError, match="no HTTPS_PROXY/ALL_PROXY"):
await ssrf_safe_fetch("https://example.com/api")
mock_client_class.assert_not_called()
async def test_https_proxy_used_explicitly(self, monkeypatch):
"""HTTPS_PROXY is passed to httpx explicitly while CA environment handling
remains enabled, and a single request goes to the original hostname URL — not
an IP literal.
This is the property the whole redesign rests on: an explicit proxy= fixes
httpx's proxy map without consulting NO_PROXY, while trust_env=True preserves
SSL_CERT_FILE and SSL_CERT_DIR handling.
"""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
monkeypatch.setenv("SSL_CERT_FILE", "/corporate-ca.pem")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname") as mock_resolve,
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
content = await ssrf_safe_fetch("https://example.com/api")
assert content == b"ok"
mock_resolve.assert_not_called()
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] == "http://proxy.internal:3128"
assert client_kwargs["trust_env"] is True
# A single request to the original hostname URL — not an IP literal.
assert mock_client.stream.call_count == 1
url_called = mock_client.stream.call_args[0][1]
assert url_called == "https://example.com/api"
# No Host override and no SNI override — the client derives both from the URL.
call_kwargs = mock_client.stream.call_args[1]
assert "Host" not in call_kwargs["headers"]
assert call_kwargs["extensions"] == {}
# Redirects stay disabled and TLS verification stays on.
assert client_kwargs["follow_redirects"] is False
assert client_kwargs["verify"] is True
async def test_all_proxy_used_as_fallback(self, monkeypatch):
"""ALL_PROXY routes the fetch when HTTPS_PROXY is not set."""
monkeypatch.setenv("ALL_PROXY", "http://all-proxy.internal:3128")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
content = await ssrf_safe_fetch("https://example.com/api")
assert content == b"ok"
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] == "http://all-proxy.internal:3128"
assert client_kwargs["trust_env"] is True
async def test_https_proxy_preferred_over_all_proxy(self, monkeypatch):
"""When both are set, HTTPS_PROXY takes priority."""
monkeypatch.setenv("HTTPS_PROXY", "http://https-proxy.internal:3128")
monkeypatch.setenv("ALL_PROXY", "http://all-proxy.internal:3128")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
await ssrf_safe_fetch("https://example.com/api")
proxy_used = mock_client_class.call_args[1]["proxy"]
assert proxy_used == "http://https-proxy.internal:3128"
async def test_no_proxy_is_not_honored(self, monkeypatch):
"""Documents the behavior change: a NO_PROXY entry that would previously have
matched the target host no longer excludes it. The fetch still proceeds
through the configured proxy rather than being refused, because routing a
NO_PROXY'd host through the proxy is strictly safer than the alternative —
fetching it direct with the IP blocklist already disabled.
"""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
monkeypatch.setenv("NO_PROXY", "example.com")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
content = await ssrf_safe_fetch("https://example.com/api")
assert content == b"ok"
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] == "http://proxy.internal:3128"
assert client_kwargs["trust_env"] is True
async def test_fetch_preserves_request_headers_but_drops_host(self, monkeypatch):
"""Caller headers pass through, but a caller-supplied Host is dropped."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client),
):
await ssrf_safe_fetch_response(
"https://example.com/api",
request_headers={"If-None-Match": "etag", "Host": "evil.example"},
)
sent_headers = mock_client.stream.call_args[1]["headers"]
assert sent_headers["If-None-Match"] == "etag"
assert "Host" not in sent_headers
async def test_fetch_size_limit_preserved(self, monkeypatch):
"""Proxy mode still enforces the response size limit during streaming."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
big_chunks = [b"x" * 1024 for _ in range(10)]
mock_client = _mock_httpx_client(headers={}, body_chunks=big_chunks)
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client),
):
with pytest.raises(SSRFFetchError, match="too large"):
await ssrf_safe_fetch("https://example.com/api", max_size=5120)
async def test_fetch_status_check_preserved(self, monkeypatch):
"""Proxy mode still rejects non-allowed status codes."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
mock_client = _mock_httpx_client(status_code=404, body_chunks=[b"no"])
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client),
):
with pytest.raises(SSRFFetchError, match="HTTP 404"):
await ssrf_safe_fetch("https://example.com/api")
async def test_gaierror_repro_succeeds_through_proxy(self, monkeypatch):
"""Reproduces issue #4292: on a host with no external DNS at all (every
getaddrinfo() call raises gaierror), the OAuth/JWKS fetch still succeeds in
proxy-trust mode, because DNS resolution is never attempted — only HTTPS_PROXY
is read and the proxy resolves the target. This is the reporter's exact
failure mode, and the strongest proof the redesign closes the issue: unlike
other tests in this class, resolve_hostname itself is *not* mocked, so if
proxy mode ever regressed into calling it, this test would fail with SSRFError
instead of succeeding.
"""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
def _no_dns(*args, **kwargs):
raise socket.gaierror("Name or service not known")
monkeypatch.setattr(socket, "getaddrinfo", _no_dns)
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
content = await ssrf_safe_fetch("https://example.com/api")
assert content == b"ok"
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] == "http://proxy.internal:3128"
assert client_kwargs["trust_env"] is True
assert mock_client.stream.call_args[0][1] == "https://example.com/api"
async def test_default_mode_still_resolves_and_pins(self):
"""Regression: with the flag off, resolution + blocklist + IP pinning still
apply, and no explicit proxy is passed to the client."""
resolved_ip = "93.184.216.34"
mock_client = _mock_httpx_client()
with (
patch(
"fastmcp.server.auth.ssrf.resolve_hostname",
return_value=[resolved_ip],
) as mock_resolve,
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
assert fastmcp.settings.ssrf_trust_proxy is False
await ssrf_safe_fetch("https://example.com/api")
mock_resolve.assert_called_once()
# Connection is pinned to the resolved IP literal, with Host + SNI = hostname.
call_args = mock_client.stream.call_args
url_called = call_args[0][1]
assert resolved_ip in url_called
assert call_args[1]["headers"]["Host"] == "example.com"
assert call_args[1]["extensions"] == {"sni_hostname": "example.com"}
# No explicit proxy is passed, and trust_env keeps its normal default.
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] is None
assert client_kwargs["trust_env"] is True
async def test_default_mode_ignores_proxy_env_vars(self, monkeypatch):
"""Regression: proxy env vars — including a hostile NO_PROXY that previously
caused non-deterministic failures — must not affect the default (non-trust)
path at all, since it never reads them."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
monkeypatch.setenv("NO_PROXY", "localhost,127.0.0.1,::1")
resolved_ip = "93.184.216.34"
mock_client = _mock_httpx_client()
with (
patch(
"fastmcp.server.auth.ssrf.resolve_hostname",
return_value=[resolved_ip],
) as mock_resolve,
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
assert fastmcp.settings.ssrf_trust_proxy is False
await ssrf_safe_fetch("https://example.com/api")
mock_resolve.assert_called_once()
assert mock_client_class.call_args[1]["proxy"] is None
assert mock_client_class.call_args[1]["trust_env"] is True