mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
231 lines
9.7 KiB
Python
231 lines
9.7 KiB
Python
"""Tests for OAuth proxy client registration (DCR)."""
|
|
|
|
import httpx
|
|
import pytest
|
|
from mcp.shared.auth import OAuthClientInformationFull
|
|
from pydantic import AnyUrl
|
|
from starlette.applications import Starlette
|
|
|
|
from fastmcp.server.auth.oauth_proxy.models import InvalidRedirectUriError
|
|
from fastmcp.server.auth.redirect_validation import DEFAULT_LOCALHOST_PATTERNS
|
|
|
|
|
|
class TestOAuthProxyClientRegistration:
|
|
"""Tests for OAuth proxy client registration (DCR)."""
|
|
|
|
async def test_register_client(self, oauth_proxy):
|
|
"""Test client registration creates ProxyDCRClient."""
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="original-client",
|
|
client_secret="original-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
)
|
|
|
|
await oauth_proxy.register_client(client_info)
|
|
|
|
# Client should be retrievable with original credentials
|
|
stored = await oauth_proxy.get_client("original-client")
|
|
assert stored is not None
|
|
assert stored.client_id == "original-client"
|
|
# Proxy uses token_endpoint_auth_method="none", so client_secret is not stored
|
|
assert stored.client_secret is None
|
|
|
|
async def test_get_registered_client(self, oauth_proxy):
|
|
"""Test retrieving a registered client."""
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="test-client",
|
|
client_secret="test-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:8080/callback")],
|
|
)
|
|
await oauth_proxy.register_client(client_info)
|
|
|
|
retrieved = await oauth_proxy.get_client("test-client")
|
|
assert retrieved is not None
|
|
assert retrieved.client_id == "test-client"
|
|
|
|
async def test_get_unregistered_client_returns_none(self, oauth_proxy):
|
|
"""Test that unregistered clients return None."""
|
|
client = await oauth_proxy.get_client("unknown-client")
|
|
assert client is None
|
|
|
|
async def test_enforcing_allowed_redirect_uris(self, oauth_proxy):
|
|
"""Test enforcing allowed redirect uris configuration."""
|
|
|
|
oauth_proxy._allowed_client_redirect_uris = ["http://localhost:12345/callback"]
|
|
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="original-client",
|
|
client_secret="original-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
)
|
|
|
|
await oauth_proxy.register_client(client_info)
|
|
retrieved = await oauth_proxy.get_client("original-client")
|
|
assert retrieved.allowed_redirect_uri_patterns == [
|
|
"http://localhost:12345/callback"
|
|
]
|
|
|
|
oauth_proxy._allowed_client_redirect_uris = [
|
|
"http://localhost:12345/updated_callback"
|
|
]
|
|
|
|
retrieved = await oauth_proxy.get_client("original-client")
|
|
assert retrieved.allowed_redirect_uri_patterns == [
|
|
"http://localhost:12345/updated_callback"
|
|
]
|
|
|
|
async def test_update_default_scopes_applies_to_dcr_registration(self, oauth_proxy):
|
|
"""DCR clients without scope should receive the updated default scopes."""
|
|
oauth_proxy.update_default_scopes(["read", "write", "calendar"])
|
|
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
transport = httpx.ASGITransport(app=app)
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="https://myserver.com",
|
|
) as client:
|
|
response = await client.post(
|
|
"/register",
|
|
json={
|
|
"redirect_uris": ["http://localhost:43210/callback"],
|
|
"client_name": "Test Client",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
client_info = response.json()
|
|
assert client_info["scope"] == "read write calendar"
|
|
|
|
registered_client = await oauth_proxy.get_client(client_info["client_id"])
|
|
assert registered_client is not None
|
|
assert registered_client.scope == "read write calendar"
|
|
|
|
async def test_register_client_rejects_external_redirect_by_default(
|
|
self, oauth_proxy
|
|
):
|
|
"""DCR defaults to loopback redirects rather than arbitrary HTTPS targets."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
transport = httpx.ASGITransport(app=app)
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="https://myserver.com",
|
|
) as client:
|
|
response = await client.post(
|
|
"/register",
|
|
json={
|
|
"redirect_uris": ["https://attacker.example/callback"],
|
|
"client_name": "Test Client",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
body = response.json()
|
|
assert body["error"] == "invalid_redirect_uri"
|
|
|
|
|
|
class TestUpstreamClientIdFallback:
|
|
"""Tests for clients that skip DCR and use the upstream client_id directly."""
|
|
|
|
async def test_upstream_client_id_returns_synthetic_client(self, oauth_proxy):
|
|
"""Clients that skip DCR and use upstream client_id directly are accepted."""
|
|
# oauth_proxy fixture uses "test-client-id" as upstream_client_id
|
|
client = await oauth_proxy.get_client("test-client-id")
|
|
assert client is not None
|
|
assert client.client_id == "test-client-id"
|
|
assert client.client_secret is None
|
|
assert client.token_endpoint_auth_method == "none"
|
|
|
|
async def test_upstream_client_id_inherits_allowed_redirect_uris(self, oauth_proxy):
|
|
"""Synthetic upstream client respects the proxy's redirect URI restrictions."""
|
|
oauth_proxy._allowed_client_redirect_uris = ["http://localhost:*"]
|
|
client = await oauth_proxy.get_client("test-client-id")
|
|
assert client is not None
|
|
assert client.allowed_redirect_uri_patterns == ["http://localhost:*"]
|
|
|
|
async def test_unknown_client_id_still_returns_none(self, oauth_proxy):
|
|
"""Non-upstream, unregistered IDs still return None."""
|
|
client = await oauth_proxy.get_client("some-random-client-id")
|
|
assert client is None
|
|
|
|
async def test_redirect_uri_allowed_when_no_pattern_restriction(self, oauth_proxy):
|
|
"""Default redirect URI validation accepts loopback redirects."""
|
|
assert oauth_proxy._allowed_client_redirect_uris == DEFAULT_LOCALHOST_PATTERNS
|
|
client = await oauth_proxy.get_client("test-client-id")
|
|
assert client is not None
|
|
uri = client.validate_redirect_uri(AnyUrl("http://localhost:12345/callback"))
|
|
assert str(uri) == "http://localhost:12345/callback"
|
|
|
|
with pytest.raises(InvalidRedirectUriError):
|
|
client.validate_redirect_uri(AnyUrl("https://attacker.example/callback"))
|
|
|
|
async def test_authorize_does_not_redirect_to_external_uri_by_default(
|
|
self, oauth_proxy
|
|
):
|
|
"""Synthetic upstream clients cannot turn /authorize into an open redirect."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
transport = httpx.ASGITransport(app=app)
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="https://myserver.com",
|
|
follow_redirects=False,
|
|
) as client:
|
|
response = await client.get(
|
|
"/authorize",
|
|
params={
|
|
"response_type": "code",
|
|
"client_id": "test-client-id",
|
|
"redirect_uri": "https://attacker.example/callback",
|
|
"state": "abc123",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "location" not in response.headers
|
|
|
|
async def test_redirect_uri_validated_against_patterns(self, oauth_proxy):
|
|
"""Redirect URI validation honours allowed_client_redirect_uris when set."""
|
|
oauth_proxy._allowed_client_redirect_uris = ["http://localhost:*"]
|
|
client = await oauth_proxy.get_client("test-client-id")
|
|
assert client is not None
|
|
|
|
# Allowed URI passes
|
|
uri = client.validate_redirect_uri(AnyUrl("http://localhost:12345/callback"))
|
|
assert str(uri) == "http://localhost:12345/callback"
|
|
|
|
# Disallowed URI raises
|
|
with pytest.raises(InvalidRedirectUriError):
|
|
client.validate_redirect_uri(AnyUrl("https://evil.example.com/callback"))
|
|
|
|
async def test_redirect_uri_blocked_when_empty_allowlist(self, oauth_proxy):
|
|
"""Empty allowed_client_redirect_uris blocks all redirect URIs, including localhost."""
|
|
oauth_proxy._allowed_client_redirect_uris = []
|
|
client = await oauth_proxy.get_client("test-client-id")
|
|
assert client is not None
|
|
|
|
with pytest.raises(InvalidRedirectUriError):
|
|
client.validate_redirect_uri(AnyUrl("http://localhost/callback"))
|
|
|
|
with pytest.raises(InvalidRedirectUriError):
|
|
client.validate_redirect_uri(AnyUrl("https://claude.ai/oauth/callback"))
|
|
|
|
async def test_none_redirect_uri_validated_against_patterns(self, oauth_proxy):
|
|
"""redirect_uri=None resolves to the placeholder then validates against patterns."""
|
|
# Placeholder is http://localhost — a pattern that can't match it forces rejection.
|
|
oauth_proxy._allowed_client_redirect_uris = ["https://myapp.example.com/*"]
|
|
client = await oauth_proxy.get_client("test-client-id")
|
|
assert client is not None
|
|
|
|
with pytest.raises(InvalidRedirectUriError):
|
|
client.validate_redirect_uri(None)
|
|
|
|
async def test_none_redirect_uri_rejected_when_empty_allowlist(self, oauth_proxy):
|
|
"""redirect_uri=None is rejected when allowlist is empty ([] blocks the resolved URI too)."""
|
|
oauth_proxy._allowed_client_redirect_uris = []
|
|
client = await oauth_proxy.get_client("test-client-id")
|
|
assert client is not None
|
|
|
|
with pytest.raises(InvalidRedirectUriError):
|
|
client.validate_redirect_uri(None)
|