Allow upstream client_id to be used directly without DCR (#3957)

Co-authored-by: Claude <noreply@anthropic.com>

Author: jlowin

src/fastmcp/server/auth/oauth_proxy/models.py

@@ -242,11 +242,19 @@ def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
             if pattern_matches:
                 return redirect_uri
 
-            # Patterns configured but didn't match
-            if self.allowed_redirect_uri_patterns:
+            # 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."
                 )
 
-        # No redirect_uri provided or no patterns configured — use base validation
-        return super().validate_redirect_uri(redirect_uri)
+        # 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
+        # the resolved URI against patterns so [] and other restrictions are enforced.
+        resolved = super().validate_redirect_uri(redirect_uri)
+        if self.allowed_redirect_uri_patterns is not None:
+            if not validate_redirect_uri(resolved, self.allowed_redirect_uri_patterns):
+                raise InvalidRedirectUriError(
+                    f"Redirect URI '{resolved}' does not match allowed patterns."
+                )
+        return resolved

src/fastmcp/server/auth/oauth_proxy/proxy.py

@@ -698,6 +698,24 @@ async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
                 await self._client_store.put(key=client_id, value=cimd_client)
                 return cimd_client
 
+        # Some MCP clients (e.g. claude.ai) skip Dynamic Client Registration and
+        # send the upstream OAuth App's client_id directly in the /authorize request.
+        # Synthesize a client on-the-fly so these clients aren't rejected with 400.
+        if client_id == self._upstream_client_id:
+            logger.debug(
+                "Client %s matched upstream client_id — synthesizing client without DCR",
+                client_id,
+            )
+            return ProxyDCRClient(
+                client_id=client_id,
+                client_secret=None,
+                redirect_uris=[AnyUrl("http://localhost")],
+                grant_types=["authorization_code", "refresh_token"],
+                scope=self._default_scope_str,
+                token_endpoint_auth_method="none",
+                allowed_redirect_uri_patterns=self._allowed_client_redirect_uris,
+            )
+
         return None
 
     @override

tests/server/auth/oauth_proxy/test_client_registration.py

@@ -1,8 +1,11 @@
 """Tests for OAuth proxy client registration (DCR)."""
 
+import pytest
 from mcp.shared.auth import OAuthClientInformationFull
 from pydantic import AnyUrl
 
+from fastmcp.server.auth.oauth_proxy.models import InvalidRedirectUriError
+
 
 class TestOAuthProxyClientRegistration:
     """Tests for OAuth proxy client registration (DCR)."""
@@ -67,3 +70,81 @@ async def test_enforcing_allowed_redirect_uris(self, oauth_proxy):
         assert retrieved.allowed_redirect_uri_patterns == [
             "http://localhost:12345/updated_callback"
         ]
+
+
+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):
+        """Any redirect URI is 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"
+
+    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)

tests/server/auth/test_oauth_proxy_redirect_validation.py

@@ -106,20 +106,18 @@ def test_default_not_applied_when_custom_patterns_supplied(self):
         with pytest.raises(InvalidRedirectUriError):
             client.validate_redirect_uri(AnyUrl("https://example.com"))
 
-    def test_empty_list_allows_none(self):
-        """Test that empty pattern list allows no URIs."""
+    def test_empty_list_blocks_all(self):
+        """Empty allowed_redirect_uri_patterns blocks all redirect URIs, including pre-registered ones."""
         client = ProxyDCRClient(
             client_id="test",
             client_secret="secret",
             redirect_uris=[AnyUrl("http://localhost:3000")],
             allowed_redirect_uri_patterns=[],
         )
 
-        # Nothing should be allowed (except the pre-registered redirect_uris via fallback)
-        # Pre-registered URI should work via fallback to base validation
-        assert client.validate_redirect_uri(AnyUrl("http://localhost:3000"))
-
-        # Non-registered URIs should be rejected
+        # All URIs must be rejected — [] means "block all", not "fall back to redirect_uris"
+        with pytest.raises(InvalidRedirectUriError):
+            client.validate_redirect_uri(AnyUrl("http://localhost:3000"))
         with pytest.raises(InvalidRedirectUriError):
             client.validate_redirect_uri(AnyUrl("http://example.com"))
         with pytest.raises(InvalidRedirectUriError):
@@ -139,6 +137,31 @@ def test_none_redirect_uri(self):
         result = client.validate_redirect_uri(None)
         assert result == AnyUrl("http://localhost:3000")
 
+    def test_none_redirect_uri_with_matching_patterns(self):
+        """DCR client with single URI and patterns: None resolves and validates against patterns."""
+        client = ProxyDCRClient(
+            client_id="test",
+            client_secret="secret",
+            redirect_uris=[AnyUrl("http://localhost:3000")],
+            allowed_redirect_uri_patterns=["http://localhost:*"],
+        )
+
+        # Resolves to registered URI which matches the pattern — should succeed
+        result = client.validate_redirect_uri(None)
+        assert result == AnyUrl("http://localhost:3000")
+
+    def test_none_redirect_uri_with_nonmatching_patterns(self):
+        """DCR client with single URI and patterns: None raises if resolved URI doesn't match."""
+        client = ProxyDCRClient(
+            client_id="test",
+            client_secret="secret",
+            redirect_uris=[AnyUrl("http://localhost:3000")],
+            allowed_redirect_uri_patterns=["https://myapp.example.com/*"],
+        )
+
+        with pytest.raises(InvalidRedirectUriError):
+            client.validate_redirect_uri(None)
+
     def test_cimd_none_redirect_uri_single_exact(self):
         """CIMD clients may omit redirect_uri only when a single exact URI exists."""
         cimd_doc = CIMDDocument(