diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..d006bb179c 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -7,6 +7,7 @@ from collections.abc import AsyncGenerator, Awaitable, Callable from contextlib import asynccontextmanager from dataclasses import dataclass +from typing import Any import anyio import httpx2 @@ -33,7 +34,7 @@ from mcp.client._transport import TransportStreams from mcp.shared._compat import resync_tracer from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams -from mcp.shared._httpx_utils import create_mcp_http_client +from mcp.shared._httpx_utils import RedirectPolicy, create_mcp_http_client from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER from mcp.shared.jsonrpc_dispatcher import cancelled_request_id_from_params from mcp.shared.message import ClientMessageMetadata, SessionMessage @@ -642,6 +643,7 @@ async def streamable_http_client( *, http_client: httpx2.AsyncClient | None = None, terminate_on_close: bool = True, + redirect_policy: RedirectPolicy | None = None, ) -> AsyncGenerator[TransportStreams, None]: """Client transport for StreamableHTTP. @@ -651,6 +653,10 @@ async def streamable_http_client( client with recommended MCP timeouts will be created. To configure headers, authentication, or other HTTP settings, create an httpx2.AsyncClient and pass it here. terminate_on_close: If True, send a DELETE request to terminate the session when the context exits. + redirect_policy: How to handle server 3xx redirects when the built-in + client is used (see ``RedirectPolicy``). Ignored when ``http_client`` + is provided — a caller-supplied client manages its own redirects and + is not protected from being bounced onto internal/loopback hosts. Yields: Tuple containing: @@ -666,7 +672,12 @@ async def streamable_http_client( if client is None: # Create default client with recommended MCP timeouts - client = create_mcp_http_client() + kwargs: dict[str, Any] = {} + if redirect_policy is not None: + kwargs["redirect_policy"] = redirect_policy + client = create_mcp_http_client(**kwargs) + else: + logger.debug("Using user-provided HTTP client; MCP redirect/SSRF protection is not applied") transport = StreamableHTTPTransport(url) diff --git a/src/mcp/shared/_httpx_utils.py b/src/mcp/shared/_httpx_utils.py index 6bb638886a..1761b16ffb 100644 --- a/src/mcp/shared/_httpx_utils.py +++ b/src/mcp/shared/_httpx_utils.py @@ -1,15 +1,112 @@ """Utilities for creating standardized httpx2 AsyncClient instances.""" +import ipaddress +import logging +from enum import Enum from typing import Any, Protocol import httpx2 -__all__ = ["create_mcp_http_client", "MCP_DEFAULT_TIMEOUT", "MCP_DEFAULT_SSE_READ_TIMEOUT"] +logger = logging.getLogger(__name__) + +__all__ = [ + "create_mcp_http_client", + "MCP_DEFAULT_TIMEOUT", + "MCP_DEFAULT_SSE_READ_TIMEOUT", + "RedirectPolicy", +] # Default MCP timeout configuration MCP_DEFAULT_TIMEOUT = 30.0 # General operations (seconds) MCP_DEFAULT_SSE_READ_TIMEOUT = 300.0 # SSE streams - 5 minutes (seconds) +# Well-known names that resolve to loopback (RFC 6761 reserves *.localhost). +_LOOPBACK_HOSTNAMES = ("localhost", ".localhost") + + +class RedirectPolicy(Enum): + """Controls how the MCP HTTP client handles server 3xx redirects. + + Streamable HTTP is JSON-RPC over HTTP; an attacker-influenced server can + return a ``307``/``308`` that bounces the client's JSON-RPC traffic onto an + internal or loopback endpoint (a local agent, a metadata service, a + registry), and the client will accept that endpoint's reply as the MCP + server's own. This is the client-side mirror of the server-side DNS + rebinding protection in ``mcp.server.transport_security``. + + Attributes: + NONE: Never follow redirects (``follow_redirects=False``). + SAME_HOST: Only follow redirects that stay on the same scheme and host. + SAFE: Follow any redirect whose target is not a loopback, link-local, + private, multicast or otherwise non-global address. This is the + default: legitimate public redirects (e.g. a migrated endpoint) + still work, while bounce-into-internal attacks are blocked. + ALL: Follow any redirect (the historical behavior). Primarily useful as + an explicit opt-out. + """ + + NONE = "none" + SAME_HOST = "same_host" + SAFE = "safe" + ALL = "all" + + +def _is_internal_or_non_global(host: str) -> bool: + """Return True when a literal host is loopback/link-local/private/etc. + + Hostnames (other than the reserved ``localhost``/``*.localhost`` names) are + treated as external, since a deterministic check would require resolving + them via DNS from the event hook. + """ + if not host: + return True + + lower = host.lower().rstrip(".") + if lower == "localhost" or lower.endswith(_LOOPBACK_HOSTNAMES): + return True + + try: + ip = ipaddress.ip_address(host) + except ValueError: + return False + return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified + + +def _make_redirect_guard(policy: RedirectPolicy): + """Build an httpx ``request`` event hook enforcing ``policy``. + + The hook is invoked for the initial request and for every redirect hop. + The first request's origin authorizes whatever host the caller explicitly + chose (a user may legitimately target their own loopback server); only + subsequent hops are validated. + """ + if policy is RedirectPolicy.NONE or policy is RedirectPolicy.ALL: + return None + + origin: tuple | None = None + + async def guard(request) -> None: + nonlocal origin + target = (request.url.scheme, request.url.host, request.url.port) + if origin is None: + origin = target + return + if target == origin: + return + if policy is RedirectPolicy.SAME_HOST: + if target[:2] != origin[:2]: + raise httpx2.ConnectError( + f"Blocked redirect to a different host '{request.url}' (redirect policy: {policy.value})" + ) + elif policy is RedirectPolicy.SAFE and _is_internal_or_non_global(target[1]): + raise httpx2.ConnectError( + f"Blocked redirect to internal/private host '{request.url}' " + f"(redirect policy: {policy.value}); refusing to send JSON-RPC " + f"traffic to a non-global address" + ) + + return guard + class McpHttpClientFactory(Protocol): # pragma: no branch def __call__( # pragma: no branch @@ -17,6 +114,7 @@ def __call__( # pragma: no branch headers: dict[str, str] | None = None, timeout: httpx2.Timeout | None = None, auth: httpx2.Auth | None = None, + redirect_policy: RedirectPolicy = RedirectPolicy.SAFE, ) -> httpx2.AsyncClient: ... @@ -24,16 +122,24 @@ def create_mcp_http_client( headers: dict[str, str] | None = None, timeout: httpx2.Timeout | None = None, auth: httpx2.Auth | None = None, + redirect_policy: RedirectPolicy = RedirectPolicy.SAFE, ) -> httpx2.AsyncClient: """Create a standardized httpx2 AsyncClient with MCP defaults. - Always enables follow_redirects and applies an SSE-friendly default timeout. + Builds a client that follows redirects by default, applies an SSE-friendly + default timeout, and protects against server-driven SSRF: redirect targets + are validated so the client never bounces JSON-RPC traffic onto internal or + loopback hosts (see ``RedirectPolicy``). Args: headers: Optional headers to include with all requests. timeout: Request timeout as httpx2.Timeout object. Defaults to 30s for connect/write/pool and 300s for read (for long-lived SSE streams). auth: Optional authentication handler. + redirect_policy: How to handle server 3xx redirects. Defaults to + ``RedirectPolicy.SAFE``, which blocks redirects into loopback, + link-local, private and other non-global addresses while still + following legitimate public redirects. Returns: Configured httpx2.AsyncClient instance with MCP defaults. @@ -76,7 +182,15 @@ def create_mcp_http_client( ``` """ # Set MCP defaults - kwargs: dict[str, Any] = {"follow_redirects": True} + kwargs: dict[str, Any] = {} + + if redirect_policy is RedirectPolicy.NONE: + kwargs["follow_redirects"] = False + else: + kwargs["follow_redirects"] = True + guard = _make_redirect_guard(redirect_policy) + if guard is not None: + kwargs["event_hooks"] = {"request": [guard]} # Handle timeout if timeout is None: diff --git a/tests/shared/test_httpx_utils.py b/tests/shared/test_httpx_utils.py index a94d7c9299..d754cb5139 100644 --- a/tests/shared/test_httpx_utils.py +++ b/tests/shared/test_httpx_utils.py @@ -1,8 +1,17 @@ -"""Tests for httpx2 utility functions.""" +"""Tests for httpx2 client factory and its SSRF redirect protection.""" + +import asyncio +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import httpx2 +import pytest -from mcp.shared._httpx_utils import create_mcp_http_client +from mcp.shared._httpx_utils import ( + RedirectPolicy, + _is_internal_or_non_global, + create_mcp_http_client, +) def test_default_settings(): @@ -22,3 +31,142 @@ def test_custom_parameters(): assert client.headers["Authorization"] == "Bearer token" assert client.timeout.connect == 60.0 + + +def test_redirect_policy_none_disables_follow(): + """NONE must set follow_redirects=False and install no guard.""" + client = create_mcp_http_client(redirect_policy=RedirectPolicy.NONE) + assert client.follow_redirects is False + assert not client._event_hooks["request"] + + +@pytest.mark.parametrize( + "host,expected", + [ + ("127.0.0.1", True), + ("localhost", True), + ("sub.localhost", True), + ("10.0.0.5", True), + ("172.16.1.1", True), + ("172.31.255.255", True), + ("192.168.1.10", True), + ("169.254.169.254", True), # cloud metadata endpoint + ("::1", True), + ("fc00::1", True), + ("fe80::1", True), + ("0.0.0.0", True), + # Public / non-literal hosts must be treated as external + ("93.184.216.34", False), + ("example.com", False), + ("1.2.3.4", False), + ], +) +def test_is_internal_or_non_global(host, expected): + assert _is_internal_or_non_global(host) is expected + + +# --------------------------------------------------------------------------- +# Redirect guard integration: a live local server that bounces HTTP to a target. +# --------------------------------------------------------------------------- + + +class _RedirectServerHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + redirect_status = 307 + target = None # set per-server + + def log_message(self, *args): # keep test output clean + pass + + def do_GET(self): + if not self.target or self.path != "/": + body = b"ok" + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + self.send_response(self.redirect_status) + self.send_header("Location", self.target) + self.send_header("Content-Length", "0") + self.end_headers() + + +class _TargetHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def do_GET(self): + body = b"internal-reply" + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +class _Server: + def __init__(self, handler): + self._httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler) + self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True) + self._thread.start() + + @property + def port(self): + return self._httpd.server_address[1] + + def close(self): + self._httpd.shutdown() + self._httpd.server_close() + + +@pytest.fixture(scope="module") +def servers(): + target = _Server(_TargetHandler) + redirector = _Server(_RedirectServerHandler) + yield redirector, target + redirector.close() + target.close() + + +def _get(client: httpx2.AsyncClient, url: str) -> httpx2.Response: + return asyncio.run(client.get(url)) + + +def test_safe_policy_blocks_redirect_to_internal(servers): + """Default SAFE policy must refuse to follow a redirect into loopback.""" + redirector, target = servers + _RedirectServerHandler.target = f"http://127.0.0.1:{target.port}/injected" + client = create_mcp_http_client() # default = SAFE + with pytest.raises(httpx2.ConnectError, match="internal/private host"): + _get(client, f"http://127.0.0.1:{redirector.port}/") + + +def test_all_policy_follows_redirect_to_internal(servers): + """Explicit ALL keeps legacy behavior: redirect into loopback is followed.""" + redirector, target = servers + _RedirectServerHandler.target = f"http://127.0.0.1:{target.port}/injected" + client = create_mcp_http_client(redirect_policy=RedirectPolicy.ALL) + resp = _get(client, f"http://127.0.0.1:{redirector.port}/") + assert resp.status_code == 200 + assert resp.text == "internal-reply" + + +def test_same_host_policy_still_follows_same_host_redirect(servers): + """SAME_HOST must not block the common same-host bounce.""" + redirector, _ = servers + _RedirectServerHandler.target = f"http://127.0.0.1:{redirector.port}/noop" + client = create_mcp_http_client(redirect_policy=RedirectPolicy.SAME_HOST) + resp = _get(client, f"http://127.0.0.1:{redirector.port}/") + assert resp.status_code == 200 + assert resp.text == "ok" + + +def test_none_policy_does_not_follow_redirect(servers): + """NONE must return the 3xx without following, so no rebinding occurs.""" + redirector, _ = servers + _RedirectServerHandler.target = "http://127.0.0.1:9999/nope" + client = create_mcp_http_client(redirect_policy=RedirectPolicy.NONE) + resp = _get(client, f"http://127.0.0.1:{redirector.port}/") + assert resp.status_code == 307