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 = [