Jeremiah Lowin Claude commited on
Commit
5c6cd30
·
1 Parent(s): 08c7eb3

Support 'scp' claim for OAuth scopes in BearerAuthProvider

Browse files

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 <noreply@anthropic.com>

src/fastmcp/server/auth/providers/bearer.py CHANGED
@@ -399,12 +399,23 @@ class BearerAuthProvider(OAuthProvider):
399
  return None
400
 
401
  def _extract_scopes(self, claims: dict[str, Any]) -> list[str]:
402
- """Extract scopes from JWT claims."""
403
- scope_claim = claims.get("scope", "")
404
- if isinstance(scope_claim, str):
405
- return scope_claim.split()
406
- elif isinstance(scope_claim, list):
407
- return scope_claim
 
 
 
 
 
 
 
 
 
 
 
408
  return []
409
 
410
  async def verify_token(self, token: str) -> AccessToken | None:
 
399
  return None
400
 
401
  def _extract_scopes(self, claims: dict[str, Any]) -> list[str]:
402
+ """Extract scopes from JWT claims. Supports both 'scope' and 'scp' claims."""
403
+ # Check for 'scope' claim first (standard OAuth2 claim)
404
+ scope_claim = claims.get("scope")
405
+ if scope_claim is not None:
406
+ if isinstance(scope_claim, str):
407
+ return scope_claim.split()
408
+ elif isinstance(scope_claim, list):
409
+ return scope_claim
410
+
411
+ # Check for 'scp' claim (used by some Identity Providers)
412
+ scp_claim = claims.get("scp")
413
+ if scp_claim is not None:
414
+ if isinstance(scp_claim, str):
415
+ return scp_claim.split()
416
+ elif isinstance(scp_claim, list):
417
+ return scp_claim
418
+
419
  return []
420
 
421
  async def verify_token(self, token: str) -> AccessToken | None:
tests/auth/providers/test_bearer.py CHANGED
@@ -533,6 +533,59 @@ class TestBearerToken:
533
  assert access_token is not None
534
  assert access_token.scopes == []
535
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
536
  async def test_malformed_token_rejection(self, bearer_provider: BearerAuthProvider):
537
  """Test rejection of malformed tokens."""
538
  malformed_tokens = [
 
533
  assert access_token is not None
534
  assert access_token.scopes == []
535
 
536
+ async def test_scp_claim_extraction_string(
537
+ self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
538
+ ):
539
+ """Test scope extraction from 'scp' claim with space-separated string."""
540
+ token = rsa_key_pair.create_token(
541
+ subject="test-user",
542
+ issuer="https://test.example.com",
543
+ audience="https://api.example.com",
544
+ additional_claims={"scp": "read write admin"}, # 'scp' claim as string
545
+ )
546
+
547
+ access_token = await bearer_provider.load_access_token(token)
548
+
549
+ assert access_token is not None
550
+ assert set(access_token.scopes) == {"read", "write", "admin"}
551
+
552
+ async def test_scp_claim_extraction_list(
553
+ self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
554
+ ):
555
+ """Test scope extraction from 'scp' claim with list format."""
556
+ token = rsa_key_pair.create_token(
557
+ subject="test-user",
558
+ issuer="https://test.example.com",
559
+ audience="https://api.example.com",
560
+ additional_claims={
561
+ "scp": ["read", "write", "admin"]
562
+ }, # 'scp' claim as list
563
+ )
564
+
565
+ access_token = await bearer_provider.load_access_token(token)
566
+
567
+ assert access_token is not None
568
+ assert set(access_token.scopes) == {"read", "write", "admin"}
569
+
570
+ async def test_scope_precedence_over_scp(
571
+ self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
572
+ ):
573
+ """Test that 'scope' claim takes precedence over 'scp' claim when both are present."""
574
+ token = rsa_key_pair.create_token(
575
+ subject="test-user",
576
+ issuer="https://test.example.com",
577
+ audience="https://api.example.com",
578
+ additional_claims={
579
+ "scope": "read write", # Standard OAuth2 claim
580
+ "scp": "admin delete", # Should be ignored when 'scope' is present
581
+ },
582
+ )
583
+
584
+ access_token = await bearer_provider.load_access_token(token)
585
+
586
+ assert access_token is not None
587
+ assert set(access_token.scopes) == {"read", "write"} # Only 'scope' claim used
588
+
589
  async def test_malformed_token_rejection(self, bearer_provider: BearerAuthProvider):
590
  """Test rejection of malformed tokens."""
591
  malformed_tokens = [