Jeremiah Lowin Claude commited on
Commit
ca53851
·
1 Parent(s): 5d876b5

Fix BearerAuthProvider audience type annotations to support List[str]

Browse files

The audience parameter was typed as `str | None` but the implementation
already supported `List[str]`. This fix aligns the type annotations with
the actual functionality and adds comprehensive validation logic for all
audience type combinations.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

src/fastmcp/server/auth/providers/bearer.py CHANGED
@@ -89,7 +89,7 @@ class RSAKeyPair:
89
  self,
90
  subject: str = "fastmcp-user",
91
  issuer: str = "https://fastmcp.example.com",
92
- audience: str | None = None,
93
  scopes: list[str] | None = None,
94
  expires_in_seconds: int = 3600,
95
  additional_claims: dict[str, Any] | None = None,
@@ -102,7 +102,7 @@ class RSAKeyPair:
102
  private_key_pem: RSA private key in PEM format
103
  subject: Subject claim (usually user ID)
104
  issuer: Issuer claim
105
- audience: Audience claim (optional)
106
  scopes: List of scopes to include
107
  expires_in_seconds: Token expiration time in seconds
108
  additional_claims: Any additional claims to include
@@ -161,7 +161,7 @@ class BearerAuthProvider(OAuthProvider):
161
  public_key: str | None = None,
162
  jwks_uri: str | None = None,
163
  issuer: str | None = None,
164
- audience: str | None = None,
165
  required_scopes: list[str] | None = None,
166
  ):
167
  """
@@ -171,7 +171,7 @@ class BearerAuthProvider(OAuthProvider):
171
  public_key: RSA public key in PEM format (for static key)
172
  jwks_uri: URI to fetch keys from (for key rotation)
173
  issuer: Expected issuer claim (optional)
174
- audience: Expected audience claim (optional)
175
  required_scopes: List of required scopes for access (optional)
176
  """
177
  if not (public_key or jwks_uri):
@@ -312,11 +312,25 @@ class BearerAuthProvider(OAuthProvider):
312
  # Validate audience if configured
313
  if self.audience:
314
  aud = claims.get("aud")
315
- if isinstance(aud, list):
316
- if self.audience not in aud:
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
317
  return None
318
- elif aud != self.audience:
319
- return None
320
 
321
  # Extract claims - prefer client_id over sub for OAuth application identification
322
  client_id = claims.get("client_id") or claims.get("sub") or "unknown"
 
89
  self,
90
  subject: str = "fastmcp-user",
91
  issuer: str = "https://fastmcp.example.com",
92
+ audience: str | list[str] | None = None,
93
  scopes: list[str] | None = None,
94
  expires_in_seconds: int = 3600,
95
  additional_claims: dict[str, Any] | None = None,
 
102
  private_key_pem: RSA private key in PEM format
103
  subject: Subject claim (usually user ID)
104
  issuer: Issuer claim
105
+ audience: Audience claim - can be a string or list of strings (optional)
106
  scopes: List of scopes to include
107
  expires_in_seconds: Token expiration time in seconds
108
  additional_claims: Any additional claims to include
 
161
  public_key: str | None = None,
162
  jwks_uri: str | None = None,
163
  issuer: str | None = None,
164
+ audience: str | list[str] | None = None,
165
  required_scopes: list[str] | None = None,
166
  ):
167
  """
 
171
  public_key: RSA public key in PEM format (for static key)
172
  jwks_uri: URI to fetch keys from (for key rotation)
173
  issuer: Expected issuer claim (optional)
174
+ audience: Expected audience claim - can be a string or list of strings (optional)
175
  required_scopes: List of required scopes for access (optional)
176
  """
177
  if not (public_key or jwks_uri):
 
312
  # Validate audience if configured
313
  if self.audience:
314
  aud = claims.get("aud")
315
+
316
+ # Handle different combinations of audience types
317
+ if isinstance(self.audience, list):
318
+ # self.audience is a list - check if any expected audience is present
319
+ if isinstance(aud, list):
320
+ # Both are lists - check for intersection
321
+ if not any(expected in aud for expected in self.audience):
322
+ return None
323
+ else:
324
+ # aud is a string - check if it's in our expected list
325
+ if aud not in self.audience:
326
+ return None
327
+ else:
328
+ # self.audience is a string - use original logic
329
+ if isinstance(aud, list):
330
+ if self.audience not in aud:
331
+ return None
332
+ elif aud != self.audience:
333
  return None
 
 
334
 
335
  # Extract claims - prefer client_id over sub for OAuth application identification
336
  client_id = claims.get("client_id") or claims.get("sub") or "unknown"
tests/auth/providers/test_bearer.py CHANGED
@@ -446,6 +446,43 @@ class TestBearerToken:
446
  access_token = await provider.load_access_token(token)
447
  assert access_token is not None
448
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
449
  async def test_scope_extraction_string(
450
  self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
451
  ):
 
446
  access_token = await provider.load_access_token(token)
447
  assert access_token is not None
448
 
449
+ async def test_provider_with_multiple_expected_audiences(self, rsa_key_pair: RSAKeyPair):
450
+ """Test provider configured with multiple expected audiences."""
451
+ provider = BearerAuthProvider(
452
+ public_key=rsa_key_pair.public_key,
453
+ issuer="https://test.example.com",
454
+ audience=["https://api.example.com", "https://other-api.example.com"],
455
+ )
456
+
457
+ # Token with single audience that matches one of the expected
458
+ token1 = rsa_key_pair.create_token(
459
+ subject="test-user",
460
+ issuer="https://test.example.com",
461
+ audience="https://api.example.com",
462
+ )
463
+ access_token1 = await provider.load_access_token(token1)
464
+ assert access_token1 is not None
465
+
466
+ # Token with multiple audiences, one of which matches
467
+ token2 = rsa_key_pair.create_token(
468
+ subject="test-user",
469
+ issuer="https://test.example.com",
470
+ additional_claims={
471
+ "aud": ["https://api.example.com", "https://third-party.example.com"]
472
+ },
473
+ )
474
+ access_token2 = await provider.load_access_token(token2)
475
+ assert access_token2 is not None
476
+
477
+ # Token with audience that doesn't match any expected
478
+ token3 = rsa_key_pair.create_token(
479
+ subject="test-user",
480
+ issuer="https://test.example.com",
481
+ audience="https://wrong-api.example.com",
482
+ )
483
+ access_token3 = await provider.load_access_token(token3)
484
+ assert access_token3 is None
485
+
486
  async def test_scope_extraction_string(
487
  self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
488
  ):