Add configurable redirect URI validation for OAuth providers (#1582)

This commit is contained in:
Jeremiah Lowin 2025-08-22 15:52:56 -04:00 committed by GitHub
commit 839a022b07
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 475 additions and 17 deletions

View file

@ -161,6 +161,15 @@ The `OAuthProxy` class provides the complete proxy implementation:
<ParamField body="resource_server_url" type="AnyHttpUrl | str | None">
Resource server URL (defaults to base_url)
</ParamField>
<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): Only localhost redirect URIs allowed (`http://localhost:*`, `http://127.0.0.1:*`)
- Empty list `[]`: All redirect URIs allowed (not recommended for production)
- Custom list: Only matching patterns allowed
These patterns apply to MCP client loopback redirects, NOT the upstream OAuth app redirect URI.
</ParamField>
</Card>
```python
@ -219,6 +228,32 @@ The proxy automatically:
- Validates tokens using your provider's public keys or API
- Maintains PKCE security throughout the flow
## Client Redirect URI Security
<Warning>
By default, OAuth Proxy only accepts localhost redirect URIs from MCP clients for security. You can customize this with the `allowed_client_redirect_uris` parameter:
```python
# Default: localhost only (secure)
auth = OAuthProxy(...)
# Custom patterns with wildcards
auth = OAuthProxy(
...,
allowed_client_redirect_uris=[
"http://localhost:*",
"https://app.example.com/auth/*"
]
)
# Allow all (NOT recommended for production)
auth = OAuthProxy(
...,
allowed_client_redirect_uris=[]
)
```
</Warning>
## Client Compatibility
<Tip>

View file

@ -111,7 +111,9 @@ token_verifier = JWTVerifier(
auth = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.yourcompany.com")],
resource_server_url="https://api.yourcompany.com"
resource_server_url="https://api.yourcompany.com",
# Optional: customize allowed client redirect URIs (defaults to localhost only)
allowed_client_redirect_uris=["http://localhost:*", "http://127.0.0.1:*"]
)
mcp = FastMCP(name="Company API", auth=auth)
@ -192,6 +194,18 @@ WorkOS's support for Dynamic Client Registration makes it particularly well-suit
→ **Complete WorkOS tutorial**: [AuthKit Integration Guide](/integrations/authkit)
## Client Redirect URI Security
<Note>
`RemoteAuthProvider` also supports the `allowed_client_redirect_uris` parameter for controlling which redirect URIs are accepted from MCP clients during DCR:
- `None` (default): Only localhost patterns allowed
- Custom list: Specify allowed patterns with wildcard support
- Empty list `[]`: Allow all (not recommended)
This provides defense-in-depth even though DCR providers typically validate redirect URIs themselves.
</Note>
## Implementation Considerations
Remote OAuth integration requires careful attention to several technical details that affect reliability and security.

View file

@ -6,6 +6,7 @@ from fastmcp.settings import Settings
from fastmcp.utilities.logging import configure_logging as _configure_logging
settings = Settings()
# if False:
_configure_logging(
level=settings.log_level,
enable_rich_tracebacks=settings.enable_rich_tracebacks,

View file

@ -43,6 +43,7 @@ from starlette.responses import JSONResponse, RedirectResponse
from starlette.routing import Route
from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier
from fastmcp.server.auth.redirect_validation import validate_redirect_uri
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
@ -52,7 +53,7 @@ logger = get_logger(__name__)
class ProxyDCRClient(OAuthClientInformationFull):
"""Client for DCR proxy that accepts any localhost redirect URI.
"""Client for DCR proxy with configurable redirect URI validation.
This special client class is critical for the OAuth proxy to work correctly
with Dynamic Client Registration (DCR). Here's why it exists:
@ -61,36 +62,48 @@ class ProxyDCRClient(OAuthClientInformationFull):
--------
When MCP clients use OAuth, they dynamically register with random localhost
ports (e.g., http://localhost:55454/callback). The OAuth proxy needs to:
1. Accept these dynamic redirect URIs from clients
1. Accept these dynamic redirect URIs from clients based on configured patterns
2. Use its own fixed redirect URI with the upstream provider (Google, GitHub, etc.)
3. Forward the authorization code back to the client's dynamic URI
Solution:
---------
This class overrides redirect_uri validation to accept ANY localhost URI,
This class validates redirect URIs against configurable patterns,
while the proxy internally uses its own fixed redirect URI with the upstream
provider. This allows the flow to work even when clients reconnect with
different ports or when tokens are cached.
Without this class, clients would get "Redirect URI not registered" errors
when trying to authenticate with cached tokens, because the stored client
would have fixed redirect URIs that don't match the new dynamic port.
Without proper validation, clients could get "Redirect URI not registered" errors
when trying to authenticate with cached tokens, or security vulnerabilities could
arise from accepting arbitrary redirect URIs.
"""
def __init__(
self, *args, allowed_redirect_uri_patterns: list[str] | None = None, **kwargs
):
"""Initialize with allowed redirect URI patterns.
Args:
allowed_redirect_uri_patterns: List of allowed redirect URI patterns with wildcard support.
If None, defaults to localhost-only patterns.
If empty list, allows all redirect URIs.
"""
super().__init__(*args, **kwargs)
self._allowed_redirect_uri_patterns = allowed_redirect_uri_patterns
def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
"""Accept any localhost redirect URI for DCR clients.
"""Validate redirect URI against allowed patterns.
Since we're acting as a proxy and clients register dynamically,
we need to accept their localhost redirect URIs even though they're
not pre-registered with us. This is essential for cached token
scenarios where the client may reconnect with a different port.
we validate their redirect URIs against configurable patterns.
This is essential for cached token scenarios where the client may
reconnect with a different port.
"""
if redirect_uri is not None:
# Accept any localhost redirect URI for DCR clients
uri_str = str(redirect_uri)
if uri_str.startswith(("http://localhost", "http://127.0.0.1")):
# Validate against allowed patterns
if validate_redirect_uri(redirect_uri, self._allowed_redirect_uri_patterns):
return redirect_uri
# Fall back to normal validation for non-localhost URIs
# Fall back to normal validation if not in allowed patterns
return super().validate_redirect_uri(redirect_uri)
# If no redirect_uri provided, use default behavior
return super().validate_redirect_uri(redirect_uri)
@ -229,6 +242,8 @@ class OAuthProxy(OAuthProvider):
issuer_url: AnyHttpUrl | str | None = None,
service_documentation_url: AnyHttpUrl | str | None = None,
resource_server_url: AnyHttpUrl | str | None = None,
# Client redirect URI validation
allowed_client_redirect_uris: list[str] | None = None,
):
"""Initialize the OAuth proxy provider.
@ -244,6 +259,11 @@ class OAuthProxy(OAuthProvider):
issuer_url: Issuer URL for OAuth metadata (defaults to base_url)
service_documentation_url: Optional service documentation URL
resource_server_url: Resource server URL (defaults to base_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), only localhost redirect URIs are allowed.
If empty list, all redirect URIs are allowed (not recommended for production).
These are for MCP clients performing loopback redirects, NOT for the upstream OAuth app.
"""
# Convert string URLs to AnyHttpUrl for parent class
base_url_parsed = (
@ -302,6 +322,7 @@ class OAuthProxy(OAuthProvider):
self._redirect_path = (
redirect_path if redirect_path.startswith("/") else f"/{redirect_path}"
)
self._allowed_client_redirect_uris = allowed_client_redirect_uris
# Local state for DCR and token bookkeeping
self._clients: dict[str, OAuthClientInformationFull] = {}
@ -353,9 +374,10 @@ class OAuthProxy(OAuthProvider):
client_secret=None,
redirect_uris=[
AnyUrl("http://localhost")
], # Placeholder - we accept any localhost URI
], # Placeholder, validation uses allowed_patterns
grant_types=["authorization_code", "refresh_token"],
token_endpoint_auth_method="none",
allowed_redirect_uri_patterns=self._allowed_client_redirect_uris,
)
logger.debug("Created ProxyDCRClient for unregistered client %s", client_id)
@ -386,7 +408,7 @@ class OAuthProxy(OAuthProvider):
upstream_id = self._upstream_client_id
upstream_secret = self._upstream_client_secret.get_secret_value()
# Create a ProxyDCRClient that accepts any localhost redirect URI
# Create a ProxyDCRClient with configured redirect URI validation
proxy_client = ProxyDCRClient(
client_id=upstream_id,
client_secret=upstream_secret,
@ -394,6 +416,7 @@ class OAuthProxy(OAuthProvider):
grant_types=client_info.grant_types
or ["authorization_code", "refresh_token"],
token_endpoint_auth_method="none",
allowed_redirect_uri_patterns=self._allowed_client_redirect_uris,
)
# Modify the client_info object in place (framework ignores return values)

View file

@ -0,0 +1,70 @@
"""Utilities for validating client redirect URIs in OAuth flows."""
import fnmatch
from pydantic import AnyUrl
def matches_allowed_pattern(uri: str, pattern: str) -> bool:
"""Check if a URI matches an allowed pattern with wildcard support.
Patterns support * wildcard matching:
- http://localhost:* matches any localhost port
- http://127.0.0.1:* matches any 127.0.0.1 port
- https://*.example.com/* matches any subdomain of example.com
- https://app.example.com/auth/* matches any path under /auth/
Args:
uri: The redirect URI to validate
pattern: The allowed pattern (may contain wildcards)
Returns:
True if the URI matches the pattern
"""
# Use fnmatch for wildcard matching
return fnmatch.fnmatch(uri, pattern)
def validate_redirect_uri(
redirect_uri: str | AnyUrl | None,
allowed_patterns: list[str] | None,
) -> bool:
"""Validate a redirect URI against allowed patterns.
Args:
redirect_uri: The redirect URI to validate
allowed_patterns: List of allowed patterns. If None, defaults to localhost.
If empty list, all URIs are allowed.
Returns:
True if the redirect URI is allowed
"""
if redirect_uri is None:
return True # None is allowed (will use client's default)
uri_str = str(redirect_uri)
# If no patterns specified, default to localhost only
if allowed_patterns is None:
allowed_patterns = [
"http://localhost:*",
"http://127.0.0.1:*",
]
# Empty list means allow all
if len(allowed_patterns) == 0:
return True
# Check if URI matches any allowed pattern
for pattern in allowed_patterns:
if matches_allowed_pattern(uri_str, pattern):
return True
return False
# Default patterns for localhost-only validation
DEFAULT_LOCALHOST_PATTERNS = [
"http://localhost:*",
"http://127.0.0.1:*",
]

View file

@ -0,0 +1,191 @@
"""Tests for OAuth proxy redirect URI validation."""
import pytest
from mcp.shared.auth import InvalidRedirectUriError
from pydantic import AnyUrl
from fastmcp.server.auth.auth import TokenVerifier
from fastmcp.server.auth.oauth_proxy import OAuthProxy, ProxyDCRClient
class MockTokenVerifier(TokenVerifier):
"""Mock token verifier for testing."""
def __init__(self):
self.required_scopes = []
async def verify_token(self, token: str) -> dict | None:
return {"sub": "test-user"}
class TestProxyDCRClient:
"""Test ProxyDCRClient redirect URI validation."""
def test_default_localhost_only(self):
"""Test that default configuration only allows localhost."""
client = ProxyDCRClient(
client_id="test",
client_secret="secret",
redirect_uris=[AnyUrl("http://localhost:3000")],
)
# Localhost should be allowed
assert client.validate_redirect_uri(AnyUrl("http://localhost:3000")) == AnyUrl(
"http://localhost:3000"
)
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"
)
# Non-localhost should fallback to base validation
# This will check against registered redirect_uris
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(AnyUrl("http://example.com"))
def test_custom_patterns(self):
"""Test custom redirect URI patterns."""
client = ProxyDCRClient(
client_id="test",
client_secret="secret",
redirect_uris=[AnyUrl("http://localhost:3000")],
allowed_redirect_uri_patterns=[
"http://localhost:*",
"https://app.example.com/*",
],
)
# Allowed by patterns
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
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(AnyUrl("http://127.0.0.1:3000"))
def test_empty_list_allows_all(self):
"""Test that empty pattern list allows all URIs."""
client = ProxyDCRClient(
client_id="test",
client_secret="secret",
redirect_uris=[AnyUrl("http://localhost:3000")],
allowed_redirect_uri_patterns=[],
)
# Everything should be allowed
assert client.validate_redirect_uri(AnyUrl("http://localhost:3000"))
assert client.validate_redirect_uri(AnyUrl("http://example.com"))
assert client.validate_redirect_uri(AnyUrl("https://anywhere.com:9999/path"))
def test_none_redirect_uri(self):
"""Test that None redirect URI uses default behavior."""
client = ProxyDCRClient(
client_id="test",
client_secret="secret",
redirect_uris=[AnyUrl("http://localhost:3000")],
)
# None should use the first registered URI
result = client.validate_redirect_uri(None)
assert result == AnyUrl("http://localhost:3000")
class TestOAuthProxyRedirectValidation:
"""Test OAuth proxy with redirect URI validation."""
def test_proxy_default_localhost_validation(self):
"""Test that OAuth proxy defaults to localhost-only validation."""
proxy = OAuthProxy(
upstream_authorization_endpoint="https://auth.example.com/authorize",
upstream_token_endpoint="https://auth.example.com/token",
upstream_client_id="test-client",
upstream_client_secret="test-secret",
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
)
# The proxy should store None for default localhost patterns
assert proxy._allowed_client_redirect_uris is None
def test_proxy_custom_patterns(self):
"""Test OAuth proxy with custom redirect patterns."""
custom_patterns = ["http://localhost:*", "https://*.myapp.com/*"]
proxy = OAuthProxy(
upstream_authorization_endpoint="https://auth.example.com/authorize",
upstream_token_endpoint="https://auth.example.com/token",
upstream_client_id="test-client",
upstream_client_secret="test-secret",
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
allowed_client_redirect_uris=custom_patterns,
)
assert proxy._allowed_client_redirect_uris == custom_patterns
def test_proxy_empty_list_validation(self):
"""Test OAuth proxy with empty list (allow all)."""
proxy = OAuthProxy(
upstream_authorization_endpoint="https://auth.example.com/authorize",
upstream_token_endpoint="https://auth.example.com/token",
upstream_client_id="test-client",
upstream_client_secret="test-secret",
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
allowed_client_redirect_uris=[],
)
assert proxy._allowed_client_redirect_uris == []
@pytest.mark.asyncio
async def test_proxy_register_client_uses_patterns(self):
"""Test that registered clients use the configured patterns."""
custom_patterns = ["https://app.example.com/*"]
proxy = OAuthProxy(
upstream_authorization_endpoint="https://auth.example.com/authorize",
upstream_token_endpoint="https://auth.example.com/token",
upstream_client_id="test-client",
upstream_client_secret="test-secret",
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
allowed_client_redirect_uris=custom_patterns,
)
# Register a client
from mcp.shared.auth import OAuthClientInformationFull
client_info = OAuthClientInformationFull(
client_id="new-client",
client_secret="new-secret",
redirect_uris=[AnyUrl("https://app.example.com/callback")],
)
await proxy.register_client(client_info)
# Get the registered client
registered = await proxy.get_client("test-client") # Uses upstream ID
assert isinstance(registered, ProxyDCRClient)
assert registered._allowed_redirect_uri_patterns == custom_patterns
@pytest.mark.asyncio
async def test_proxy_unregistered_client_uses_patterns(self):
"""Test that unregistered clients also use configured patterns."""
custom_patterns = ["http://localhost:*", "http://127.0.0.1:*"]
proxy = OAuthProxy(
upstream_authorization_endpoint="https://auth.example.com/authorize",
upstream_token_endpoint="https://auth.example.com/token",
upstream_client_id="test-client",
upstream_client_secret="test-secret",
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
allowed_client_redirect_uris=custom_patterns,
)
# Get an unregistered client
client = await proxy.get_client("unknown-client")
assert isinstance(client, ProxyDCRClient)
assert client._allowed_redirect_uri_patterns == custom_patterns

View file

@ -0,0 +1,124 @@
"""Tests for redirect URI validation in OAuth flows."""
from pydantic import AnyUrl
from fastmcp.server.auth.redirect_validation import (
DEFAULT_LOCALHOST_PATTERNS,
matches_allowed_pattern,
validate_redirect_uri,
)
class TestMatchesAllowedPattern:
"""Test wildcard pattern matching for redirect URIs."""
def test_exact_match(self):
"""Test exact URI matching without wildcards."""
assert matches_allowed_pattern(
"http://localhost:3000/callback", "http://localhost:3000/callback"
)
assert not matches_allowed_pattern(
"http://localhost:3000/callback", "http://localhost:3001/callback"
)
def test_port_wildcard(self):
"""Test wildcard matching for ports."""
pattern = "http://localhost:*/callback"
assert matches_allowed_pattern("http://localhost:3000/callback", pattern)
assert matches_allowed_pattern("http://localhost:54321/callback", pattern)
assert not matches_allowed_pattern("http://example.com:3000/callback", pattern)
def test_path_wildcard(self):
"""Test wildcard matching for paths."""
pattern = "http://localhost:3000/*"
assert matches_allowed_pattern("http://localhost:3000/callback", pattern)
assert matches_allowed_pattern("http://localhost:3000/auth/callback", pattern)
assert not matches_allowed_pattern("http://localhost:3001/callback", pattern)
def test_subdomain_wildcard(self):
"""Test wildcard matching for subdomains."""
pattern = "https://*.example.com/callback"
assert matches_allowed_pattern("https://app.example.com/callback", pattern)
assert matches_allowed_pattern("https://api.example.com/callback", pattern)
assert not matches_allowed_pattern("https://example.com/callback", pattern)
assert not matches_allowed_pattern("http://app.example.com/callback", pattern)
def test_multiple_wildcards(self):
"""Test patterns with multiple wildcards."""
pattern = "https://*.example.com:*/auth/*"
assert matches_allowed_pattern(
"https://app.example.com:8080/auth/callback", pattern
)
assert matches_allowed_pattern(
"https://api.example.com:3000/auth/redirect", pattern
)
assert not matches_allowed_pattern(
"http://app.example.com:8080/auth/callback", pattern
)
class TestValidateRedirectUri:
"""Test redirect URI validation with pattern lists."""
def test_none_redirect_uri_allowed(self):
"""Test that None redirect URI is always allowed."""
assert validate_redirect_uri(None, None)
assert validate_redirect_uri(None, [])
assert validate_redirect_uri(None, ["http://localhost:*"])
def test_default_localhost_patterns(self):
"""Test default localhost-only patterns when None is provided."""
# Localhost patterns should be allowed by default
assert validate_redirect_uri("http://localhost:3000", None)
assert validate_redirect_uri("http://127.0.0.1:8080", None)
# Non-localhost should be rejected by default
assert not validate_redirect_uri("http://example.com", None)
assert not validate_redirect_uri("https://app.example.com", None)
def test_empty_list_allows_all(self):
"""Test that empty list allows all redirect URIs."""
assert validate_redirect_uri("http://localhost:3000", [])
assert validate_redirect_uri("http://example.com", [])
assert validate_redirect_uri("https://anywhere.com:9999/path", [])
def test_custom_patterns(self):
"""Test validation with custom pattern list."""
patterns = [
"http://localhost:*",
"https://app.example.com/*",
"https://*.trusted.io/*",
]
# Allowed URIs
assert validate_redirect_uri("http://localhost:3000", patterns)
assert validate_redirect_uri("https://app.example.com/callback", patterns)
assert validate_redirect_uri("https://api.trusted.io/auth", patterns)
# Rejected URIs
assert not validate_redirect_uri("http://127.0.0.1:3000", patterns)
assert not validate_redirect_uri("https://other.example.com/callback", patterns)
assert not validate_redirect_uri("http://app.example.com/callback", patterns)
def test_anyurl_conversion(self):
"""Test that AnyUrl objects are properly converted to strings."""
patterns = ["http://localhost:*"]
uri = AnyUrl("http://localhost:3000/callback")
assert validate_redirect_uri(uri, patterns)
uri = AnyUrl("http://example.com/callback")
assert not validate_redirect_uri(uri, patterns)
class TestDefaultPatterns:
"""Test the default localhost patterns constant."""
def test_default_patterns_exist(self):
"""Test that default patterns are defined."""
assert DEFAULT_LOCALHOST_PATTERNS is not None
assert len(DEFAULT_LOCALHOST_PATTERNS) > 0
def test_default_patterns_include_localhost(self):
"""Test that default patterns include localhost variations."""
assert "http://localhost:*" in DEFAULT_LOCALHOST_PATTERNS
assert "http://127.0.0.1:*" in DEFAULT_LOCALHOST_PATTERNS