Fix DCR redirect URI validation (#4408)

This commit is contained in:
Jeremiah Lowin 2026-07-05 14:05:13 -07:00 committed by GitHub
commit cccb529f50
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 182 additions and 39 deletions

View file

@ -192,11 +192,14 @@ mcp = FastMCP(name="My Server", auth=auth)
<ParamField body="allowed_client_redirect_uris" type="list[str] | None">
List of allowed redirect URI patterns for MCP clients. Patterns support
wildcards (e.g., `"http://localhost:*"`, `"https://*.example.com/*"`). -
`None` (default): All redirect URIs allowed (for MCP/DCR compatibility) -
Empty list `[]`: No redirect URIs allowed - Custom list: Only matching
patterns allowed These patterns apply to MCP client loopback redirects, NOT
the upstream OAuth app redirect URI.
wildcards (e.g., `"http://localhost:*"`, `"https://*.example.com/*"`).
- `None` (default): DCR clients use registered redirect URIs, with loopback
ports allowed to vary for MCP compatibility
- Empty list `[]`: No redirect URIs allowed
- Custom list: Only matching patterns allowed
These patterns apply to MCP client loopback redirects. Configure the upstream
OAuth app redirect URI separately with `redirect_path`.
</ParamField>
<ParamField body="valid_scopes" type="list[str] | None">
@ -556,7 +559,7 @@ auth = OAuthProxy(
### Redirect URI Validation
While the OAuth proxy accepts all redirect URIs by default (for DCR compatibility), you can restrict which clients can connect by specifying allowed patterns:
By default, the OAuth proxy validates DCR clients against their registered redirect URIs while allowing loopback ports to vary for MCP compatibility. You can restrict which clients can connect at the server level by specifying allowed patterns:
```python
# Allow only localhost clients (common for development)

View file

@ -124,11 +124,11 @@ mcp = FastMCP(name="My Server", auth=auth)
<ParamField body="allowed_client_redirect_uris" type="list[str] | None">
List of allowed redirect URI patterns for MCP clients. Patterns support wildcards (e.g., `"http://localhost:*"`, `"https://*.example.com/*"`).
- `None` (default): All redirect URIs allowed (for MCP/DCR compatibility)
- `None` (default): DCR clients use registered redirect URIs, with loopback ports allowed to vary for MCP compatibility
- Empty list `[]`: No redirect URIs allowed
- Custom list: Only matching patterns allowed
These patterns apply to MCP client loopback redirects, NOT the upstream OAuth app redirect URI.
These patterns apply to MCP client loopback redirects. Configure the upstream OAuth app redirect URI separately with `redirect_path`.
</ParamField>

View file

@ -7,9 +7,10 @@ from __future__ import annotations
import hashlib
from typing import Any, Final
from urllib.parse import urlparse
from mcp.shared.auth import InvalidRedirectUriError, OAuthClientInformationFull
from pydantic import AnyUrl, BaseModel, Field
from pydantic import AnyUrl, BaseModel, Field, ValidationError
from fastmcp.server.auth.cimd import CIMDDocument
from fastmcp.server.auth.redirect_validation import (
@ -134,6 +135,57 @@ def _hash_token(token: str) -> str:
return hashlib.sha256(token.encode()).hexdigest()
def _redirect_uri_path(uri_path: str) -> str:
return uri_path or "/"
def _is_loopback_host(host: str | None) -> bool:
return host is not None and host.lower() in {"localhost", "127.0.0.1", "::1"}
def _matches_registered_loopback_redirect_uri(
redirect_uri: AnyUrl,
registered_uri: AnyUrl,
) -> bool:
requested = urlparse(str(redirect_uri))
registered = urlparse(str(registered_uri))
if requested.username or requested.password:
return False
if registered.username or registered.password:
return False
requested_host = requested.hostname.lower() if requested.hostname else None
registered_host = registered.hostname.lower() if registered.hostname else None
if not _is_loopback_host(registered_host):
return False
if requested_host != registered_host:
return False
return (
requested.scheme.lower() == registered.scheme.lower()
and _redirect_uri_path(requested.path) == _redirect_uri_path(registered.path)
and requested.params == registered.params
and requested.query == registered.query
and requested.fragment == registered.fragment
)
def _matches_registered_redirect_uri(
redirect_uri: AnyUrl,
registered_uris: list[AnyUrl] | None,
) -> bool:
if not registered_uris:
return False
return any(
redirect_uri == registered_uri
or _matches_registered_loopback_redirect_uri(redirect_uri, registered_uri)
for registered_uri in registered_uris
)
class ProxyDCRClient(OAuthClientInformationFull):
"""Client for DCR proxy with configurable redirect URI validation.
@ -164,6 +216,7 @@ class ProxyDCRClient(OAuthClientInformationFull):
client_name: str | None = Field(default=None)
cimd_document: CIMDDocument | None = Field(default=None)
cimd_fetched_at: float | None = Field(default=None)
allow_unregistered_redirect_uris: bool = Field(default=False, exclude=True)
def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
"""Validate redirect URI against proxy patterns and optionally CIMD redirect_uris.
@ -171,8 +224,9 @@ class ProxyDCRClient(OAuthClientInformationFull):
For CIMD clients: validates against BOTH the CIMD document's redirect_uris
AND the proxy's allowed patterns (if configured). Both must pass.
For DCR clients: validates against proxy patterns first, falling back to
base validation (registered redirect_uris) if patterns don't match.
For DCR clients: validates against proxy patterns when configured. Without
proxy patterns, validates against registered redirect_uris while allowing
loopback ports to vary for MCP client compatibility.
"""
if redirect_uri is None and self.cimd_document is not None:
cimd_redirect_uris = self.cimd_document.redirect_uris
@ -184,7 +238,7 @@ class ProxyDCRClient(OAuthClientInformationFull):
)
try:
resolved = AnyUrl(candidate)
except Exception as e:
except ValidationError as e:
raise InvalidRedirectUriError(
f"Invalid CIMD redirect_uri: {e}"
) from e
@ -235,19 +289,24 @@ class ProxyDCRClient(OAuthClientInformationFull):
return redirect_uri
pattern_matches = validate_redirect_uri(
if self.allowed_redirect_uri_patterns is None:
if self.allow_unregistered_redirect_uris:
return redirect_uri
if _matches_registered_redirect_uri(redirect_uri, self.redirect_uris):
return redirect_uri
raise InvalidRedirectUriError(
f"Redirect URI '{redirect_uri}' not registered for client"
)
if validate_redirect_uri(
redirect_uri=redirect_uri,
allowed_patterns=self.allowed_redirect_uri_patterns,
)
if pattern_matches:
):
return redirect_uri
# Patterns configured but didn't match (None means "allow all"; [] means "block all")
if self.allowed_redirect_uri_patterns is not None:
raise InvalidRedirectUriError(
f"Redirect URI '{redirect_uri}' does not match allowed patterns."
)
raise InvalidRedirectUriError(
f"Redirect URI '{redirect_uri}' does not match allowed patterns."
)
# redirect_uri is None with no CIMD document: let base class resolve the URI
# (handles the single-registered-URI shortcut for DCR clients), then validate

View file

@ -300,7 +300,8 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
service_documentation_url: Optional service documentation URL
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
Patterns support wildcards (e.g., "http://localhost:*", "https://*.example.com/*").
If None (default), all redirect URIs are allowed (for DCR compatibility).
If None (default), DCR clients use registered redirect URIs, with loopback
ports allowed to vary for MCP compatibility.
If empty list, no redirect URIs are allowed.
These are for MCP clients performing loopback redirects, NOT for the upstream OAuth app.
valid_scopes: List of all the possible valid scopes for a client.
@ -820,6 +821,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
scope=self._default_scope_str,
token_endpoint_auth_method="none",
allowed_redirect_uri_patterns=self._allowed_client_redirect_uris,
allow_unregistered_redirect_uris=True,
)
return None

View file

@ -277,7 +277,8 @@ class OIDCProxy(OAuthProxy):
redirect_path: Redirect path configured in upstream OAuth app (defaults to "/auth/callback")
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
Patterns support wildcards (e.g., "http://localhost:*", "https://*.example.com/*").
If None (default), all redirect URIs are allowed (for DCR compatibility).
If None (default), DCR clients use registered redirect URIs, with loopback
ports allowed to vary for MCP compatibility.
If empty list, no redirect URIs are allowed.
These are for MCP clients performing loopback redirects, NOT for the upstream OAuth app.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).

View file

@ -47,6 +47,49 @@ class TestOAuthProxyClientRegistration:
client = await oauth_proxy.get_client("unknown-client")
assert client is None
async def test_dcr_client_rejects_unregistered_redirect_uri(self, oauth_proxy):
"""DCR clients honor their registered redirect_uris by default."""
client_info = OAuthClientInformationFull(
client_id="original-client",
client_secret="original-secret",
redirect_uris=[AnyUrl("http://localhost:6274/oauth/callback")],
)
await oauth_proxy.register_client(client_info)
retrieved = await oauth_proxy.get_client("original-client")
assert retrieved is not None
with pytest.raises(InvalidRedirectUriError):
retrieved.validate_redirect_uri(AnyUrl("http://evil.com/anything"))
with pytest.raises(InvalidRedirectUriError):
retrieved.validate_redirect_uri(AnyUrl("http://localhost:6274/other"))
uri = retrieved.validate_redirect_uri(
AnyUrl("http://localhost:51353/oauth/callback")
)
assert str(uri) == "http://localhost:51353/oauth/callback"
async def test_dcr_client_accepts_registered_external_redirect_uri(
self, oauth_proxy
):
"""Open DCR still accepts arbitrary redirect URIs that clients register."""
client_info = OAuthClientInformationFull(
client_id="external-client",
client_secret="external-secret",
redirect_uris=[AnyUrl("https://client.example.com/oauth/callback")],
)
await oauth_proxy.register_client(client_info)
retrieved = await oauth_proxy.get_client("external-client")
assert retrieved is not None
uri = retrieved.validate_redirect_uri(
AnyUrl("https://client.example.com/oauth/callback")
)
assert str(uri) == "https://client.example.com/oauth/callback"
async def test_enforcing_allowed_redirect_uris(self, oauth_proxy):
"""Test enforcing allowed redirect uris configuration."""

View file

@ -29,27 +29,62 @@ class MockTokenVerifier(TokenVerifier):
class TestProxyDCRClient:
"""Test ProxyDCRClient redirect URI validation."""
def test_default_allows_all(self):
"""Test that default configuration allows all URIs for DCR compatibility."""
def test_default_uses_registered_redirect_uris_with_loopback_port_flexibility(self):
"""Default DCR clients allow registered loopback callbacks to vary ports."""
client = ProxyDCRClient(
client_id="test",
client_secret="secret",
redirect_uris=[AnyUrl("http://localhost:3000")],
redirect_uris=[AnyUrl("http://localhost:3000/callback")],
)
# All URIs should be allowed by default for DCR compatibility
assert client.validate_redirect_uri(AnyUrl("http://localhost:3000")) == AnyUrl(
"http://localhost:3000"
assert client.validate_redirect_uri(
AnyUrl("http://localhost:3000/callback")
) == AnyUrl("http://localhost:3000/callback")
assert client.validate_redirect_uri(
AnyUrl("http://localhost:8080/callback")
) == AnyUrl("http://localhost:8080/callback")
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(AnyUrl("http://localhost:8080/other"))
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(AnyUrl("http://127.0.0.1:3000/callback"))
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(AnyUrl("http://example.com/callback"))
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(
AnyUrl("https://claude.ai/api/mcp/auth_callback")
)
def test_default_uses_exact_registered_external_redirect_uri(self):
"""Default DCR clients require exact matches for non-loopback callbacks."""
client = ProxyDCRClient(
client_id="test",
client_secret="secret",
redirect_uris=[AnyUrl("https://client.example.com/oauth/callback")],
)
assert client.validate_redirect_uri(
AnyUrl("https://client.example.com/oauth/callback")
) == AnyUrl("https://client.example.com/oauth/callback")
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(
AnyUrl("https://client.example.com:8443/oauth/callback")
)
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(AnyUrl("https://evil.example.com/callback"))
def test_synthetic_client_can_allow_unregistered_redirect_uris(self):
"""Synthetic clients can opt in to broad redirect URI compatibility."""
client = ProxyDCRClient(
client_id="test",
client_secret="secret",
redirect_uris=[AnyUrl("http://localhost")],
allow_unregistered_redirect_uris=True,
)
assert client.validate_redirect_uri(AnyUrl("http://localhost:8080")) == AnyUrl(
"http://localhost:8080"
)
assert client.validate_redirect_uri(AnyUrl("http://127.0.0.1:3000")) == AnyUrl(
"http://127.0.0.1:3000"
)
assert client.validate_redirect_uri(AnyUrl("http://example.com")) == AnyUrl(
"http://example.com"
)
assert client.validate_redirect_uri(
AnyUrl("https://claude.ai/api/mcp/auth_callback")
) == AnyUrl("https://claude.ai/api/mcp/auth_callback")
@ -70,7 +105,7 @@ class TestProxyDCRClient:
assert client.validate_redirect_uri(AnyUrl("http://localhost:3000"))
assert client.validate_redirect_uri(AnyUrl("https://app.example.com/callback"))
# Not allowed by patterns - will fallback to base validation
# Not allowed by patterns
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(AnyUrl("http://127.0.0.1:3000"))
with pytest.raises(InvalidRedirectUriError):
@ -256,8 +291,8 @@ class TestProxyDCRClient:
class TestOAuthProxyRedirectValidation:
"""Test OAuth proxy with redirect URI validation."""
def test_proxy_default_allows_all(self):
"""Test that OAuth proxy defaults to allowing all URIs for DCR compatibility."""
def test_proxy_default_has_no_server_redirect_pattern_restriction(self):
"""OAuth proxy defaults to no server-level redirect URI pattern restriction."""
proxy = OAuthProxy(
upstream_authorization_endpoint="https://auth.example.com/authorize",
upstream_token_endpoint="https://auth.example.com/token",
@ -269,7 +304,7 @@ class TestOAuthProxyRedirectValidation:
client_storage=MemoryStore(),
)
# The proxy should store None for default (allow all)
# The proxy stores None when no server-level pattern restriction is configured.
assert proxy._allowed_client_redirect_uris is None
def test_proxy_custom_patterns(self):