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

Fix #4292: SSRF guard breaks OAuth/JWKS fetches behind a corporate HTTP proxy (#4412)

Browse source 
* Add FASTMCP_SSRF_TRUST_PROXY to allow SSRF fetches through a corporate proxy

🤖 Generated with Claude Code

* Make SSRF fetch client trust_env explicit for proxy routing

🤖 Generated with Claude Code

* Warn when SSRF proxy trust is enabled without a configured proxy

🤖 Generated with Claude Code

* Warn when NO_PROXY would send an SSRF-trust-proxy fetch direct

🤖 Generated with Claude Code

* Refuse SSRF-trust-proxy fetches when no proxy would route the target

🤖 Generated with Claude Code

* Fix TestProxyMode mocks to patch httpx2.AsyncClient

main's httpx -> httpx2 migration (#4503) landed after these tests were
written; ssrf.py's fetch path already uses httpx2.AsyncClient, but
TestProxyMode still patched the old httpx module, so the mock silently
stopped intercepting and requests escaped to the real network.

* Fix port-qualified NO_PROXY bypass in SSRF proxy-trust guard

proxy_bypass(hostname) discarded the port, so a NO_PROXY entry like
127.0.0.1:8443 went undetected while httpx2 honored it and sent the
request direct with the blocklist already disabled. Pass host:port
instead, except for IPv6 literals, where httpx2 ignores port when
matching NO_PROXY and neither bracketed nor unbracketed host:port
reliably matches through proxy_bypass()'s own parser.

* Replace NO_PROXY prediction with explicit proxy control in SSRF trust-proxy mode

Predicting httpx2's proxy routing (via proxy_bypass(), then via httpx2's own
get_environment_proxies()/URLPattern internals) kept diverging from its real
NO_PROXY handling — three rounds, three different divergences, always in the
unsafe direction. Read HTTPS_PROXY/ALL_PROXY directly and pass it to httpx2
explicitly with trust_env=False, so the request provably goes through that
proxy instead of being predicted to. NO_PROXY is no longer evaluated in this
mode: a NO_PROXY'd host is now routed through the proxy rather than refused,
since that's strictly safer than the alternative (direct with the blocklist
already off).

---------

Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
This commit is contained in:
Alexander Savchuk 2026-07-19 13:42:52 +12:00 • committed by GitHub
commit 2899ffb6f3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 547 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

20
docs/more/settings.mdx
View file

@ -96,6 +96,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, with the client's own environment-based proxy routing turned off — so the request either goes through that exact proxy or fails outright, with no routing decision left for the client to make on its own. One consequence: `NO_PROXY` is **not honored** in this mode. 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 |

176
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 httpx2 explicitly with ``trust_env`` disabled, so the request is provably
routed through the proxy rather than predicted to be — NO_PROXY is not evaluated in
this mode. 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 httpx2
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 httpx2 explicitly (with
``trust_env`` disabled) so there is no routing decision left for httpx2 to make
differently than this function assumed — 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,54 @@ 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 httpx2's routing decision,
# first by approximating NO_PROXY handling with urllib.request.proxy_bypass(),
# then by replicating httpx2's own get_environment_proxies()/URLPattern
# 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
# httpx2 explicitly below, with trust_env disabled. With an explicit
# `proxy=` and `trust_env=False`, httpx2 has no routing decision left to make
# differently than assumed here: the request provably goes through that
# proxy or the connection fails. NO_PROXY is therefore not evaluated in this
# mode at all — 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 +362,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 +433,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 +473,19 @@ async def ssrf_safe_fetch_response(
),
follow_redirects=False,
verify=True,
# Default (pinned) mode has no proxy_url and keeps trust_env's
# normal default (True). Proxy-trust mode sets an explicit
# proxy_url and turns trust_env off, so httpx2 has no environment
# -based routing decision left to make — see validate_url() above
# for why that matters.
proxy=target.proxy_url,
trust_env=target.proxy_url is None,
) 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 +537,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

@ -344,6 +344,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",

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

@ -3,11 +3,13 @@
This module tests the ssrf.py module which provides SSRF-protected HTTP fetching.
"""
import socket
from unittest.mock import AsyncMock, MagicMock, patch
import httpx2
import pytest
import fastmcp
from fastmcp.server.auth.ssrf import (
SSRFError,
SSRFFetchError,
@ -15,6 +17,41 @@ from fastmcp.server.auth.ssrf import (
ssrf_safe_fetch,
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 httpx2.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 +543,316 @@ 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 httpx2 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 httpx2 explicitly with trust_env=False, so the request is provably
routed through that proxy rather than predicted to be. NO_PROXY is therefore not
evaluated in this mode. 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("httpx2.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 httpx2 explicitly with trust_env disabled, and a
single request goes to the original hostname URL — not an IP literal.
This is the property the whole redesign rests on: with an explicit proxy=
and trust_env=False, httpx2 has no environment-based routing decision left
to make differently than assumed.
"""
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") as mock_resolve,
patch("httpx2.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 False
# 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("httpx2.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 False
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("httpx2.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("httpx2.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 False
async def test_fetch_preserves_request_headers_but_drops_host(self, monkeypatch):
"""Caller headers pass through, but a caller-supplied Host is dropped."""
from fastmcp.server.auth.ssrf import ssrf_safe_fetch_response
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("httpx2.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("httpx2.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("httpx2.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("httpx2.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 False
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("httpx2.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("httpx2.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