From 63f3940c6f6b6b2b30d511f149803fdf9cb7fc75 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Thu, 3 Jul 2025 20:18:54 -0400 Subject: [PATCH] Support 'scp' claim for OAuth scopes in BearerAuthProvider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Some Identity Providers use 'scp' claim instead of the standard 'scope' claim for OAuth scopes. This change updates the scope extraction logic to support both claims, with 'scope' taking precedence when both are present. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- src/fastmcp/server/auth/providers/bearer.py | 23 ++++++--- tests/auth/providers/test_bearer.py | 53 +++++++++++++++++++++ 2 files changed, 70 insertions(+), 6 deletions(-) diff --git a/src/fastmcp/server/auth/providers/bearer.py b/src/fastmcp/server/auth/providers/bearer.py index 9d5d40f7e..77cf2289c 100644 --- a/src/fastmcp/server/auth/providers/bearer.py +++ b/src/fastmcp/server/auth/providers/bearer.py @@ -399,12 +399,23 @@ class BearerAuthProvider(OAuthProvider): return None def _extract_scopes(self, claims: dict[str, Any]) -> list[str]: - """Extract scopes from JWT claims.""" - scope_claim = claims.get("scope", "") - if isinstance(scope_claim, str): - return scope_claim.split() - elif isinstance(scope_claim, list): - return scope_claim + """Extract scopes from JWT claims. Supports both 'scope' and 'scp' claims.""" + # Check for 'scope' claim first (standard OAuth2 claim) + scope_claim = claims.get("scope") + if scope_claim is not None: + if isinstance(scope_claim, str): + return scope_claim.split() + elif isinstance(scope_claim, list): + return scope_claim + + # Check for 'scp' claim (used by some Identity Providers) + scp_claim = claims.get("scp") + if scp_claim is not None: + if isinstance(scp_claim, str): + return scp_claim.split() + elif isinstance(scp_claim, list): + return scp_claim + return [] async def verify_token(self, token: str) -> AccessToken | None: diff --git a/tests/auth/providers/test_bearer.py b/tests/auth/providers/test_bearer.py index 07790ef48..381012dd9 100644 --- a/tests/auth/providers/test_bearer.py +++ b/tests/auth/providers/test_bearer.py @@ -533,6 +533,59 @@ class TestBearerToken: assert access_token is not None assert access_token.scopes == [] + async def test_scp_claim_extraction_string( + self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider + ): + """Test scope extraction from 'scp' claim with space-separated string.""" + token = rsa_key_pair.create_token( + subject="test-user", + issuer="https://test.example.com", + audience="https://api.example.com", + additional_claims={"scp": "read write admin"}, # 'scp' claim as string + ) + + access_token = await bearer_provider.load_access_token(token) + + assert access_token is not None + assert set(access_token.scopes) == {"read", "write", "admin"} + + async def test_scp_claim_extraction_list( + self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider + ): + """Test scope extraction from 'scp' claim with list format.""" + token = rsa_key_pair.create_token( + subject="test-user", + issuer="https://test.example.com", + audience="https://api.example.com", + additional_claims={ + "scp": ["read", "write", "admin"] + }, # 'scp' claim as list + ) + + access_token = await bearer_provider.load_access_token(token) + + assert access_token is not None + assert set(access_token.scopes) == {"read", "write", "admin"} + + async def test_scope_precedence_over_scp( + self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider + ): + """Test that 'scope' claim takes precedence over 'scp' claim when both are present.""" + token = rsa_key_pair.create_token( + subject="test-user", + issuer="https://test.example.com", + audience="https://api.example.com", + additional_claims={ + "scope": "read write", # Standard OAuth2 claim + "scp": "admin delete", # Should be ignored when 'scope' is present + }, + ) + + access_token = await bearer_provider.load_access_token(token) + + assert access_token is not None + assert set(access_token.scopes) == {"read", "write"} # Only 'scope' claim used + async def test_malformed_token_rejection(self, bearer_provider: BearerAuthProvider): """Test rejection of malformed tokens.""" malformed_tokens = [