ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-24 06:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: accept both client_id and identifier_uri as Azure audience (#3797)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-04-08 20:25:42 -04:00 • committed by GitHub
commit 6592aaa70e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 157 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/fastmcp/server/auth/providers/azure.py
View file

@ -220,7 +220,7 @@ class AzureProvider(OAuthProxy):
token_verifier = JWTVerifier(
jwks_uri=jwks_uri,
issuer=issuer,
audience=self.identifier_uri,
audience=[client_id, self.identifier_uri],
algorithm="RS256",
required_scopes=validation_scopes, # Only validate non-OIDC scopes
http_client=http_client,
@ -627,7 +627,7 @@ class AzureJWTVerifier(JWTVerifier):
super().__init__(
jwks_uri=f"https://{base_authority}/{tenant_id}/discovery/v2.0/keys",
issuer=issuer,
audience=self._identifier_uri,
audience=[client_id, self._identifier_uri],
algorithm="RS256",
required_scopes=required_scopes,
)

95
tests/server/auth/providers/test_azure.py
View file

@ -9,7 +9,7 @@ from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyUrl
from fastmcp.server.auth.providers.azure import AzureProvider
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
@pytest.fixture
@ -190,8 +190,6 @@ class TestAzureProvider:
self, memory_storage: MemoryStore
):
"""When audience is provided, JWTVerifier is configured with JWKS and issuer."""
from fastmcp.server.auth.providers.jwt import JWTVerifier
provider = AzureProvider(
client_id="test_client",
client_secret="test_secret",
@ -211,11 +209,100 @@ class TestAzureProvider:
"https://login.microsoftonline.com/my-tenant/discovery/v2.0/keys"
)
assert verifier.issuer == "https://login.microsoftonline.com/my-tenant/v2.0"
assert verifier.audience == "api://my-api"
assert verifier.audience == ["test_client", "api://my-api"]
# Scopes are stored unprefixed for token validation
# (Azure returns unprefixed scopes like ".default" in JWT tokens)
assert verifier.required_scopes == [".default"]
async def test_token_accepted_with_client_id_audience(
self, memory_storage: MemoryStore
):
"""Azure AD v2 tokens use the bare client_id as aud — must be accepted."""
key_pair = RSAKeyPair.generate()
provider = AzureProvider(
client_id="test_client",
client_secret="test_secret",
tenant_id="my-tenant",
base_url="https://myserver.com",
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert isinstance(provider._token_validator, JWTVerifier)
verifier = provider._token_validator
verifier.public_key = key_pair.public_key
verifier.jwks_uri = None
token = key_pair.create_token(
subject="test-user",
issuer="https://login.microsoftonline.com/my-tenant/v2.0",
audience="test_client",
additional_claims={"scp": "read"},
)
result = await verifier.load_access_token(token)
assert result is not None
async def test_token_accepted_with_identifier_uri_audience(
self, memory_storage: MemoryStore
):
"""Azure AD v1 tokens use the identifier_uri as aud — must be accepted."""
key_pair = RSAKeyPair.generate()
provider = AzureProvider(
client_id="test_client",
client_secret="test_secret",
tenant_id="my-tenant",
base_url="https://myserver.com",
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert isinstance(provider._token_validator, JWTVerifier)
verifier = provider._token_validator
verifier.public_key = key_pair.public_key
verifier.jwks_uri = None
token = key_pair.create_token(
subject="test-user",
issuer="https://login.microsoftonline.com/my-tenant/v2.0",
audience="api://my-api",
additional_claims={"scp": "read"},
)
result = await verifier.load_access_token(token)
assert result is not None
async def test_token_rejected_with_wrong_audience(
self, memory_storage: MemoryStore
):
"""Tokens for a different application must be rejected."""
key_pair = RSAKeyPair.generate()
provider = AzureProvider(
client_id="test_client",
client_secret="test_secret",
tenant_id="my-tenant",
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert isinstance(provider._token_validator, JWTVerifier)
verifier = provider._token_validator
verifier.public_key = key_pair.public_key
verifier.jwks_uri = None
token = key_pair.create_token(
subject="test-user",
issuer="https://login.microsoftonline.com/my-tenant/v2.0",
audience="wrong-app-id",
additional_claims={"scp": "read"},
)
result = await verifier.load_access_token(token)
assert result is None
async def test_authorize_filters_resource_and_stores_unprefixed_scopes(
self, memory_storage: MemoryStore
):

65
tests/server/auth/providers/test_azure_scopes.py
View file

@ -413,7 +413,7 @@ class TestAzureJWTVerifier:
== "https://login.microsoftonline.com/my-tenant-id/discovery/v2.0/keys"
)
assert verifier.issuer == "https://login.microsoftonline.com/my-tenant-id/v2.0"
assert verifier.audience == "api://my-client-id"
assert verifier.audience == ["my-client-id", "api://my-client-id"]
assert verifier.algorithm == "RS256"
assert verifier.required_scopes == ["access_as_user"]
@ -438,6 +438,69 @@ class TestAzureJWTVerifier:
assert result is not None
assert "access_as_user" in result.scopes
async def test_validates_token_with_client_id_audience(self):
"""Azure AD v2 tokens use the bare client_id GUID as audience."""
key_pair = RSAKeyPair.generate()
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["access_as_user"],
)
verifier.public_key = key_pair.public_key
verifier.jwks_uri = None
token = key_pair.create_token(
subject="test-user",
issuer="https://login.microsoftonline.com/my-tenant-id/v2.0",
audience="my-client-id",
additional_claims={"scp": "access_as_user"},
)
result = await verifier.load_access_token(token)
assert result is not None
assert "access_as_user" in result.scopes
async def test_validates_token_with_custom_identifier_uri_audience(self):
"""Custom identifier_uri (e.g. Bicep deployments) accepted as audience."""
key_pair = RSAKeyPair.generate()
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["read"],
identifier_uri="api://my-app-name",
)
verifier.public_key = key_pair.public_key
verifier.jwks_uri = None
token = key_pair.create_token(
subject="test-user",
issuer="https://login.microsoftonline.com/my-tenant-id/v2.0",
audience="api://my-app-name",
additional_claims={"scp": "read"},
)
result = await verifier.load_access_token(token)
assert result is not None
assert "read" in result.scopes
async def test_rejects_token_with_wrong_audience(self):
"""Tokens for a different application must be rejected."""
key_pair = RSAKeyPair.generate()
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["read"],
)
verifier.public_key = key_pair.public_key
verifier.jwks_uri = None
token = key_pair.create_token(
subject="test-user",
issuer="https://login.microsoftonline.com/my-tenant-id/v2.0",
audience="some-other-app-id",
additional_claims={"scp": "read"},
)
result = await verifier.load_access_token(token)
assert result is None
def test_scopes_supported_returns_prefixed_form(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",