fix(client): add SSRF redirect protection to httpx client factory

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)
 

src/mcp/shared/_httpx_utils.py

@@ -1,39 +1,145 @@
 """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
         self,
         headers: dict[str, str] | None = None,
         timeout: httpx2.Timeout | None = None,
         auth: httpx2.Auth | None = None,
+        redirect_policy: RedirectPolicy = RedirectPolicy.SAFE,
     ) -> httpx2.AsyncClient: ...
 
 
 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:

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