fastmcp/tests/server/auth/oauth_proxy/test_client_registration.py
Jeremiah Lowin b07f9ce9ae
Honor OAuth application_type in DCR (SEP-837) (#4621)
* Honor OAuth application_type in DCR (SEP-837)

* Simplify web redirect check per ruff SIM103

* Enforce application_type over HTTP, at auth time, and tighten native scheme rules

Recover the DCR application_type the SDK RegistrationHandler drops (P1), enforce the stored type on the authorization redirect path (P2), restrict native to loopback http + custom schemes (P2), and document the web/native rules (P2).

* Fix loopback range detection and use a positive scheme allowlist

Classify loopback hosts with ipaddress.is_loopback so all of 127.0.0.0/8 counts (a web client could bypass the non-loopback rule with 127.0.0.2). Replace the NON_REDIRECT_NETWORK_SCHEMES denylist with STANDARD_URI_SCHEMES: native now accepts only https, loopback http, and unregistered private-use schemes per RFC 8252, so smb/smtp/nfs and other unlisted standard schemes no longer pass.

* Vendor the IANA scheme registry and consolidate the loopback classifier

Replace the hand-picked STANDARD_URI_SCHEMES with a vendored snapshot of the IANA URI scheme registry (423 schemes), so registered transports nobody enumerated (coap, coaps, stun, turn, mqtt) fail closed instead of passing as private-use. Delete the stale duplicate _is_loopback_host in oauth_proxy/models.py and reuse the ipaddress-based classifier from redirect_validation, restoring loopback port flexibility across all of 127.0.0.0/8.

* Treat the reserved localhost namespace and absolute host forms as loopback

RFC 6761 6.3 reserves the whole localhost namespace for the local machine, so app.localhost and localhost. are loopback just as much as localhost. Previously a web client could register https://app.localhost/callback and bypass the non-loopback rule, while native clients were wrongly refused legitimate http://app.localhost:3000 dev callbacks. The suffix test is anchored on a leading dot so localhost.evil.com and notlocalhost stay non-loopback.

* Narrow scope: drop native scheme classification, keep the web rule

Registry membership cannot separate app-dispatch schemes from network transports (vscode is registered because it is an app scheme), so classifying a native client's scheme rejected callbacks that real MCP clients need. Remove the vendored registry and the private-use test; native now accepts any scheme outside the unsafe set, with cleartext http still limited to loopback. Also reject web registrations that omit redirect_uris rather than storing an unusable localhost placeholder.
2026-07-26 14:24:50 -04:00

520 lines
21 KiB
Python

"""Tests for OAuth proxy client registration (DCR)."""
import httpx2
import pytest
from mcp.server.auth.provider import RegistrationError
from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyUrl
from starlette.applications import Starlette
from fastmcp.server.auth.oauth_proxy.models import InvalidRedirectUriError
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_register_client_allows_external_https_by_default(self, oauth_proxy):
"""Default DCR registration accepts ordinary external HTTPS callbacks."""
client_info = OAuthClientInformationFull(
client_id="https-client",
client_secret="original-secret",
redirect_uris=[AnyUrl("https://client.example.com/callback")],
)
await oauth_proxy.register_client(client_info)
stored = await oauth_proxy.get_client("https-client")
assert stored is not None
assert stored.redirect_uris == [AnyUrl("https://client.example.com/callback")]
async def test_register_client_rejects_unsafe_redirect_scheme_by_default(
self, oauth_proxy
):
"""Default DCR registration rejects active browser redirect schemes."""
client_info = OAuthClientInformationFull(
client_id="javascript-client",
client_secret="original-secret",
redirect_uris=[AnyUrl("javascript:alert(document.cookie)//")],
)
with pytest.raises(RegistrationError, match="invalid_redirect_uri"):
await oauth_proxy.register_client(client_info)
async def test_register_client_without_redirect_uris_defers_allowlist_validation(
self, oauth_proxy
):
"""DCR clients may omit redirect_uris until the authorization request."""
oauth_proxy._allowed_client_redirect_uris = ["https://client.example/*"]
client_info = OAuthClientInformationFull(
client_id="deferred-client",
client_secret="original-secret",
redirect_uris=None,
)
await oauth_proxy.register_client(client_info)
stored = await oauth_proxy.get_client("deferred-client")
assert stored is not None
assert stored.redirect_uris is not None
assert str(stored.redirect_uris[0]).rstrip("/") == "http://localhost"
redirect_uri = stored.validate_redirect_uri(
AnyUrl("https://client.example/callback")
)
assert str(redirect_uri) == "https://client.example/callback"
with pytest.raises(InvalidRedirectUriError):
stored.validate_redirect_uri(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_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."""
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 = httpx2.ASGITransport(app=app)
async with httpx2.AsyncClient(
transport=transport,
base_url="https://myserver.com",
) as client:
response = await client.post(
"/register",
json={
"redirect_uris": ["https://client.example.com/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"
@pytest.mark.parametrize(
"requested_auth_method",
[None, "client_secret_post", "client_secret_basic"],
)
async def test_dcr_response_is_public_client(
self, oauth_proxy, requested_auth_method
):
"""The DCR response must describe the public client the proxy actually
stores — never a confidential method / secret the proxy does not enforce
and does not advertise in server metadata.
"""
registration = {"redirect_uris": ["https://client.example.com/callback"]}
if requested_auth_method is not None:
registration["token_endpoint_auth_method"] = requested_auth_method
app = Starlette(routes=oauth_proxy.get_routes())
transport = httpx2.ASGITransport(app=app)
async with httpx2.AsyncClient(
transport=transport,
base_url="https://myserver.com",
) as client:
response = await client.post("/register", json=registration)
assert response.status_code == 201
client_info = response.json()
assert client_info["token_endpoint_auth_method"] == "none"
assert client_info.get("client_secret") is None
class TestApplicationTypeRegistration:
"""SEP-837: DCR registration honors the client's application_type."""
async def test_default_application_type_is_native(self, oauth_proxy):
"""Omitting application_type defaults to native (the SDK default), so a
loopback redirect URI registers successfully."""
client_info = OAuthClientInformationFull(
client_id="default-client",
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
await oauth_proxy.register_client(client_info)
stored = await oauth_proxy.get_client("default-client")
assert stored is not None
assert stored.application_type == "native"
async def test_native_loopback_range_registers_then_authorizes_new_port(
self, oauth_proxy
):
"""A native client on 127.0.0.2 keeps loopback port flexibility.
Registration accepts the whole 127.0.0.0/8 range, and the stored client
must then authorize a different ephemeral port on that same address.
"""
client_info = OAuthClientInformationFull(
client_id="loopback-range-client",
redirect_uris=[AnyUrl("http://127.0.0.2:3000/callback")],
application_type="native",
)
await oauth_proxy.register_client(client_info)
stored = await oauth_proxy.get_client("loopback-range-client")
assert stored is not None
uri = stored.validate_redirect_uri(AnyUrl("http://127.0.0.2:54321/callback"))
assert str(uri) == "http://127.0.0.2:54321/callback"
# A different host is still rejected — flexibility is loopback-only.
with pytest.raises(InvalidRedirectUriError):
stored.validate_redirect_uri(
AnyUrl("http://evil.example.com:54321/callback")
)
async def test_native_client_accepts_loopback(self, oauth_proxy):
client_info = OAuthClientInformationFull(
client_id="native-client",
redirect_uris=[AnyUrl("http://127.0.0.1:55555/callback")],
application_type="native",
)
await oauth_proxy.register_client(client_info)
stored = await oauth_proxy.get_client("native-client")
assert stored is not None
assert stored.application_type == "native"
async def test_web_client_accepts_https(self, oauth_proxy):
client_info = OAuthClientInformationFull(
client_id="web-client",
redirect_uris=[AnyUrl("https://client.example.com/callback")],
application_type="web",
)
await oauth_proxy.register_client(client_info)
stored = await oauth_proxy.get_client("web-client")
assert stored is not None
assert stored.application_type == "web"
async def test_web_client_rejects_loopback(self, oauth_proxy):
client_info = OAuthClientInformationFull(
client_id="web-loopback-client",
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
application_type="web",
)
with pytest.raises(RegistrationError, match="application_type 'web'"):
await oauth_proxy.register_client(client_info)
async def test_web_client_rejects_custom_scheme(self, oauth_proxy):
client_info = OAuthClientInformationFull(
client_id="web-custom-client",
redirect_uris=[AnyUrl("com.example.app:/oauth/callback")],
application_type="web",
)
with pytest.raises(RegistrationError, match="application_type 'web'"):
await oauth_proxy.register_client(client_info)
async def test_web_client_without_redirect_uris_is_rejected(self, oauth_proxy):
"""A web client with no redirect_uris could never authorize.
Omitted redirect_uris fall back to the `http://localhost` placeholder,
which a web client can never use (loopback http fails its own rule), so
registering one would only create a client guaranteed to fail later.
"""
client_info = OAuthClientInformationFull(
client_id="web-no-uris",
redirect_uris=None,
application_type="web",
)
with pytest.raises(RegistrationError, match="required for application_type"):
await oauth_proxy.register_client(client_info)
async def test_native_client_without_redirect_uris_still_allowed(self, oauth_proxy):
"""Native clients may still defer redirect_uris to authorization time."""
client_info = OAuthClientInformationFull(
client_id="native-no-uris",
redirect_uris=None,
application_type="native",
)
await oauth_proxy.register_client(client_info)
stored = await oauth_proxy.get_client("native-no-uris")
assert stored is not None
@pytest.mark.parametrize("application_type", ["web", "native"])
async def test_unsafe_scheme_rejected_regardless_of_type(
self, oauth_proxy, application_type
):
client_info = OAuthClientInformationFull(
client_id="unsafe-client",
redirect_uris=[AnyUrl("javascript:alert(document.cookie)//")],
application_type=application_type,
)
with pytest.raises(RegistrationError, match="invalid_redirect_uri"):
await oauth_proxy.register_client(client_info)
class TestApplicationTypeRegistrationOverHTTP:
"""SEP-837: application_type is honored on the real POST /register route.
The SDK's RegistrationHandler parses application_type but drops it before
calling register_client, so these tests exercise the actual ASGI route to
prove FastMCP recovers the value end to end (a direct register_client call
would not catch the SDK dropping the field)."""
async def _register(self, oauth_proxy, payload: dict):
app = Starlette(routes=oauth_proxy.get_routes())
transport = httpx2.ASGITransport(app=app)
async with httpx2.AsyncClient(
transport=transport,
base_url="https://myserver.com",
) as client:
return await client.post("/register", json=payload)
async def test_web_client_with_loopback_rejected_over_http(self, oauth_proxy):
response = await self._register(
oauth_proxy,
{
"redirect_uris": ["http://localhost:12345/callback"],
"application_type": "web",
},
)
assert response.status_code == 400
body = response.json()
assert body["error"] == "invalid_redirect_uri"
assert "application_type 'web'" in body["error_description"]
async def test_web_client_with_https_accepted_over_http(self, oauth_proxy):
response = await self._register(
oauth_proxy,
{
"redirect_uris": ["https://client.example.com/callback"],
"application_type": "web",
},
)
assert response.status_code == 201
body = response.json()
assert body["application_type"] == "web"
stored = await oauth_proxy.get_client(body["client_id"])
assert stored is not None
assert stored.application_type == "web"
async def test_native_client_with_loopback_accepted_over_http(self, oauth_proxy):
response = await self._register(
oauth_proxy,
{
"redirect_uris": ["http://localhost:12345/callback"],
"application_type": "native",
},
)
assert response.status_code == 201
body = response.json()
assert body["application_type"] == "native"
async def test_default_application_type_is_native_over_http(self, oauth_proxy):
"""Omitting application_type over HTTP defaults to native, so a loopback
redirect is accepted (preserving pre-SEP-837 behavior)."""
response = await self._register(
oauth_proxy,
{"redirect_uris": ["http://localhost:12345/callback"]},
)
assert response.status_code == 201
body = response.json()
assert body["application_type"] == "native"
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):
"""Ordinary redirect URIs are accepted when allowed_client_redirect_uris is None."""
assert oauth_proxy._allowed_client_redirect_uris is None
client = await oauth_proxy.get_client("test-client-id")
assert client is not None
uri = client.validate_redirect_uri(AnyUrl("https://claude.ai/oauth/callback"))
assert str(uri) == "https://claude.ai/oauth/callback"
uri = client.validate_redirect_uri(
AnyUrl("cursor://anysphere.cursor-mcp/oauth/callback")
)
assert str(uri) == "cursor://anysphere.cursor-mcp/oauth/callback"
with pytest.raises(InvalidRedirectUriError):
client.validate_redirect_uri(AnyUrl("javascript:alert(document.cookie)//"))
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)