mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-24 06:24:18 +02:00
489 lines
20 KiB
Python
489 lines
20 KiB
Python
"""Tests for Azure (Microsoft Entra) OAuth provider."""
|
|
|
|
import os
|
|
from unittest.mock import patch
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import pytest
|
|
from mcp.server.auth.provider import AuthorizationParams
|
|
from mcp.shared.auth import OAuthClientInformationFull
|
|
from pydantic import AnyUrl
|
|
|
|
from fastmcp.server.auth.providers.azure import AzureProvider
|
|
|
|
|
|
class TestAzureProvider:
|
|
"""Test Azure OAuth provider functionality."""
|
|
|
|
def test_init_with_explicit_params(self):
|
|
"""Test AzureProvider initialization with explicit parameters."""
|
|
provider = AzureProvider(
|
|
client_id="12345678-1234-1234-1234-123456789012",
|
|
client_secret="azure_secret_123",
|
|
tenant_id="87654321-4321-4321-4321-210987654321",
|
|
base_url="https://myserver.com",
|
|
required_scopes=["read", "write"],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
assert provider._upstream_client_id == "12345678-1234-1234-1234-123456789012"
|
|
assert provider._upstream_client_secret.get_secret_value() == "azure_secret_123"
|
|
assert str(provider.base_url) == "https://myserver.com/"
|
|
# Check tenant is in the endpoints
|
|
parsed_auth = urlparse(provider._upstream_authorization_endpoint)
|
|
assert "87654321-4321-4321-4321-210987654321" in parsed_auth.path
|
|
parsed_token = urlparse(provider._upstream_token_endpoint)
|
|
assert "87654321-4321-4321-4321-210987654321" in parsed_token.path
|
|
|
|
@pytest.mark.parametrize(
|
|
"scopes_env",
|
|
[
|
|
"read,write",
|
|
'["read", "write"]',
|
|
],
|
|
)
|
|
def test_init_with_env_vars(self, scopes_env):
|
|
"""Test AzureProvider initialization from environment variables."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"FASTMCP_SERVER_AUTH_AZURE_CLIENT_ID": "env-client-id",
|
|
"FASTMCP_SERVER_AUTH_AZURE_CLIENT_SECRET": "env-secret",
|
|
"FASTMCP_SERVER_AUTH_AZURE_TENANT_ID": "env-tenant-id",
|
|
"FASTMCP_SERVER_AUTH_AZURE_BASE_URL": "https://envserver.com",
|
|
"FASTMCP_SERVER_AUTH_AZURE_REQUIRED_SCOPES": scopes_env,
|
|
"FASTMCP_SERVER_AUTH_AZURE_JWT_SIGNING_KEY": "test-secret",
|
|
},
|
|
):
|
|
provider = AzureProvider()
|
|
|
|
assert provider._upstream_client_id == "env-client-id"
|
|
assert provider._upstream_client_secret.get_secret_value() == "env-secret"
|
|
assert str(provider.base_url) == "https://envserver.com/"
|
|
# Scopes are stored unprefixed for token validation
|
|
# (Azure returns unprefixed scopes in JWT tokens)
|
|
assert provider._token_validator.required_scopes == [
|
|
"read",
|
|
"write",
|
|
]
|
|
# Check tenant is in the endpoints
|
|
parsed_auth = urlparse(provider._upstream_authorization_endpoint)
|
|
assert "env-tenant-id" in parsed_auth.path
|
|
parsed_token = urlparse(provider._upstream_token_endpoint)
|
|
assert "env-tenant-id" in parsed_token.path
|
|
|
|
def test_init_missing_client_id_raises_error(self):
|
|
"""Test that missing client_id raises ValueError."""
|
|
# Clear environment variables to ensure we're testing the parameter validation
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with pytest.raises(ValueError, match="client_id is required"):
|
|
AzureProvider(
|
|
client_secret="test_secret",
|
|
tenant_id="test-tenant",
|
|
required_scopes=["read"],
|
|
)
|
|
|
|
def test_init_missing_client_secret_raises_error(self):
|
|
"""Test that missing client_secret raises ValueError."""
|
|
# Clear environment variables to ensure we're testing the parameter validation
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with pytest.raises(ValueError, match="client_secret is required"):
|
|
AzureProvider(
|
|
client_id="test_client",
|
|
tenant_id="test-tenant",
|
|
required_scopes=["read"],
|
|
)
|
|
|
|
def test_init_missing_tenant_id_raises_error(self):
|
|
"""Test that missing tenant_id raises ValueError."""
|
|
# Clear environment variables to ensure we're testing the parameter validation
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with pytest.raises(ValueError, match="tenant_id is required"):
|
|
AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
required_scopes=["read"],
|
|
)
|
|
|
|
def test_init_missing_required_scopes_raises_error(self):
|
|
"""Test that missing required_scopes raises ValueError."""
|
|
# Clear environment variables to ensure we're testing the parameter validation
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with pytest.raises(
|
|
ValueError, match="required_scopes must include at least one scope"
|
|
):
|
|
AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="test-tenant",
|
|
)
|
|
|
|
def test_init_empty_required_scopes_raises_error(self):
|
|
"""Test that empty required_scopes raises ValueError."""
|
|
# Clear environment variables to ensure we're testing the parameter validation
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with pytest.raises(
|
|
ValueError, match="required_scopes must include at least one scope"
|
|
):
|
|
AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="test-tenant",
|
|
required_scopes=[],
|
|
)
|
|
|
|
def test_init_defaults(self):
|
|
"""Test that default values are applied correctly."""
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="test-tenant",
|
|
required_scopes=["read"],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
# Check defaults
|
|
assert provider.base_url is None
|
|
assert provider._redirect_path == "/auth/callback"
|
|
# Azure provider defaults are set but we can't easily verify them without accessing internals
|
|
|
|
def test_oauth_endpoints_configured_correctly(self):
|
|
"""Test that OAuth endpoints are configured correctly."""
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="my-tenant-id",
|
|
base_url="https://myserver.com",
|
|
required_scopes=["read"],
|
|
jwt_signing_key="test_secret",
|
|
)
|
|
|
|
# Check that endpoints use the correct Azure OAuth2 v2.0 endpoints with tenant
|
|
assert (
|
|
provider._upstream_authorization_endpoint
|
|
== "https://login.microsoftonline.com/my-tenant-id/oauth2/v2.0/authorize"
|
|
)
|
|
assert (
|
|
provider._upstream_token_endpoint
|
|
== "https://login.microsoftonline.com/my-tenant-id/oauth2/v2.0/token"
|
|
)
|
|
assert (
|
|
provider._upstream_revocation_endpoint is None
|
|
) # Azure doesn't support revocation
|
|
|
|
def test_special_tenant_values(self):
|
|
"""Test that special tenant values are accepted."""
|
|
# Test with "organizations"
|
|
provider1 = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="organizations",
|
|
required_scopes=["read"],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
parsed = urlparse(provider1._upstream_authorization_endpoint)
|
|
assert "/organizations/" in parsed.path
|
|
|
|
# Test with "consumers"
|
|
provider2 = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="consumers",
|
|
required_scopes=["read"],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
parsed = urlparse(provider2._upstream_authorization_endpoint)
|
|
assert "/consumers/" in parsed.path
|
|
|
|
def test_azure_specific_scopes(self):
|
|
"""Test handling of custom API scope formats."""
|
|
# Test that the provider accepts custom API scopes without error
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="test-tenant",
|
|
required_scopes=[
|
|
"read",
|
|
"write",
|
|
"admin",
|
|
],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
# Provider should initialize successfully with these scopes
|
|
assert provider is not None
|
|
# Scopes are stored unprefixed for token validation
|
|
# (Azure returns unprefixed scopes in JWT tokens)
|
|
assert provider._token_validator.required_scopes == [
|
|
"read",
|
|
"write",
|
|
"admin",
|
|
]
|
|
|
|
def test_init_does_not_require_api_client_id_anymore(self):
|
|
"""API client ID is no longer required; audience is client_id."""
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="test-tenant",
|
|
required_scopes=["read"],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
assert provider is not None
|
|
|
|
def test_init_with_custom_audience_uses_jwt_verifier(self):
|
|
"""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",
|
|
tenant_id="my-tenant",
|
|
identifier_uri="api://my-api",
|
|
required_scopes=[".default"],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
assert provider._token_validator is not None
|
|
assert isinstance(provider._token_validator, JWTVerifier)
|
|
verifier = provider._token_validator
|
|
assert verifier.jwks_uri is not None
|
|
assert verifier.jwks_uri.startswith(
|
|
"https://login.microsoftonline.com/my-tenant/discovery/v2.0/keys"
|
|
)
|
|
assert verifier.issuer == "https://login.microsoftonline.com/my-tenant/v2.0"
|
|
assert verifier.audience == "test_client"
|
|
# Scopes are stored unprefixed for token validation
|
|
# (Azure returns unprefixed scopes like ".default" in JWT tokens)
|
|
assert verifier.required_scopes == [".default"]
|
|
|
|
async def test_authorize_filters_resource_and_stores_unprefixed_scopes(self):
|
|
"""authorize() should drop resource parameter and store unprefixed scopes for MCP clients."""
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="common",
|
|
identifier_uri="api://my-api",
|
|
required_scopes=["read", "write"],
|
|
base_url="https://srv.example",
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
await provider.register_client(
|
|
OAuthClientInformationFull(
|
|
client_id="dummy",
|
|
client_secret="secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
)
|
|
)
|
|
|
|
client = OAuthClientInformationFull(
|
|
client_id="dummy",
|
|
client_secret="secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
)
|
|
|
|
params = AuthorizationParams(
|
|
redirect_uri=AnyUrl("http://localhost:12345/callback"),
|
|
redirect_uri_provided_explicitly=True,
|
|
scopes=[
|
|
"read",
|
|
"profile",
|
|
], # Client sends unprefixed scopes (from PRM which advertises unprefixed)
|
|
state="abc",
|
|
code_challenge="xyz",
|
|
resource="https://should.be.ignored",
|
|
)
|
|
|
|
url = await provider.authorize(client, params)
|
|
|
|
# Extract transaction ID from consent redirect
|
|
parsed = urlparse(url)
|
|
qs = parse_qs(parsed.query)
|
|
assert "txn_id" in qs, "Should redirect to consent page with transaction ID"
|
|
txn_id = qs["txn_id"][0]
|
|
|
|
# Verify transaction stores UNPREFIXED scopes for MCP clients
|
|
transaction = await provider._transaction_store.get(key=txn_id)
|
|
assert transaction is not None
|
|
assert "read" in transaction.scopes
|
|
assert "profile" in transaction.scopes
|
|
# Azure provider filters resource parameter (not stored in transaction)
|
|
assert transaction.resource is None
|
|
|
|
# Verify the upstream Azure URL will have PREFIXED scopes
|
|
upstream_url = provider._build_upstream_authorize_url(
|
|
txn_id, transaction.model_dump()
|
|
)
|
|
assert (
|
|
"api%3A%2F%2Fmy-api%2Fread" in upstream_url
|
|
or "api://my-api/read" in upstream_url
|
|
)
|
|
assert (
|
|
"api%3A%2F%2Fmy-api%2Fprofile" in upstream_url
|
|
or "api://my-api/profile" in upstream_url
|
|
)
|
|
|
|
async def test_authorize_appends_additional_scopes(self):
|
|
"""authorize() should append additional_authorize_scopes to the authorization request."""
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="common",
|
|
identifier_uri="api://my-api",
|
|
required_scopes=["read"],
|
|
base_url="https://srv.example",
|
|
additional_authorize_scopes=["Mail.Read", "User.Read"],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
await provider.register_client(
|
|
OAuthClientInformationFull(
|
|
client_id="dummy",
|
|
client_secret="secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
)
|
|
)
|
|
|
|
client = OAuthClientInformationFull(
|
|
client_id="dummy",
|
|
client_secret="secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
)
|
|
|
|
params = AuthorizationParams(
|
|
redirect_uri=AnyUrl("http://localhost:12345/callback"),
|
|
redirect_uri_provided_explicitly=True,
|
|
scopes=["read"], # Client sends unprefixed scopes
|
|
state="abc",
|
|
code_challenge="xyz",
|
|
)
|
|
|
|
url = await provider.authorize(client, params)
|
|
|
|
# Extract transaction ID from consent redirect
|
|
parsed = urlparse(url)
|
|
qs = parse_qs(parsed.query)
|
|
assert "txn_id" in qs, "Should redirect to consent page with transaction ID"
|
|
txn_id = qs["txn_id"][0]
|
|
|
|
# Verify transaction stores ONLY MCP scopes (unprefixed)
|
|
# additional_authorize_scopes are NOT stored in transaction
|
|
transaction = await provider._transaction_store.get(key=txn_id)
|
|
assert transaction is not None
|
|
assert "read" in transaction.scopes
|
|
assert "Mail.Read" not in transaction.scopes # Not in transaction
|
|
assert "User.Read" not in transaction.scopes # Not in transaction
|
|
|
|
# Verify upstream URL includes both MCP scopes (prefixed) AND additional Graph scopes
|
|
upstream_url = provider._build_upstream_authorize_url(
|
|
txn_id, transaction.model_dump()
|
|
)
|
|
assert (
|
|
"api%3A%2F%2Fmy-api%2Fread" in upstream_url
|
|
or "api://my-api/read" in upstream_url
|
|
)
|
|
assert "Mail.Read" in upstream_url
|
|
assert "User.Read" in upstream_url
|
|
|
|
def test_base_authority_defaults_to_public_cloud(self):
|
|
"""Test that base_authority defaults to login.microsoftonline.com."""
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="test-tenant",
|
|
required_scopes=["read"],
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
assert (
|
|
provider._upstream_authorization_endpoint
|
|
== "https://login.microsoftonline.com/test-tenant/oauth2/v2.0/authorize"
|
|
)
|
|
assert (
|
|
provider._upstream_token_endpoint
|
|
== "https://login.microsoftonline.com/test-tenant/oauth2/v2.0/token"
|
|
)
|
|
assert (
|
|
provider._token_validator.issuer # type: ignore[attr-defined]
|
|
== "https://login.microsoftonline.com/test-tenant/v2.0"
|
|
)
|
|
assert (
|
|
provider._token_validator.jwks_uri # type: ignore[attr-defined]
|
|
== "https://login.microsoftonline.com/test-tenant/discovery/v2.0/keys"
|
|
)
|
|
|
|
def test_base_authority_azure_government(self):
|
|
"""Test Azure Government endpoints with login.microsoftonline.us."""
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="gov-tenant-id",
|
|
required_scopes=["read"],
|
|
base_authority="login.microsoftonline.us",
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
assert (
|
|
provider._upstream_authorization_endpoint
|
|
== "https://login.microsoftonline.us/gov-tenant-id/oauth2/v2.0/authorize"
|
|
)
|
|
assert (
|
|
provider._upstream_token_endpoint
|
|
== "https://login.microsoftonline.us/gov-tenant-id/oauth2/v2.0/token"
|
|
)
|
|
assert (
|
|
provider._token_validator.issuer # type: ignore[attr-defined]
|
|
== "https://login.microsoftonline.us/gov-tenant-id/v2.0"
|
|
)
|
|
assert (
|
|
provider._token_validator.jwks_uri # type: ignore[attr-defined]
|
|
== "https://login.microsoftonline.us/gov-tenant-id/discovery/v2.0/keys"
|
|
)
|
|
|
|
def test_base_authority_from_environment_variable(self):
|
|
"""Test that base_authority can be set via environment variable."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"FASTMCP_SERVER_AUTH_AZURE_CLIENT_ID": "env-client-id",
|
|
"FASTMCP_SERVER_AUTH_AZURE_CLIENT_SECRET": "env-secret",
|
|
"FASTMCP_SERVER_AUTH_AZURE_TENANT_ID": "env-tenant-id",
|
|
"FASTMCP_SERVER_AUTH_AZURE_REQUIRED_SCOPES": "read",
|
|
"FASTMCP_SERVER_AUTH_AZURE_BASE_AUTHORITY": "login.microsoftonline.us",
|
|
"FASTMCP_SERVER_AUTH_AZURE_JWT_SIGNING_KEY": "test-secret",
|
|
},
|
|
):
|
|
provider = AzureProvider()
|
|
|
|
assert (
|
|
provider._upstream_authorization_endpoint
|
|
== "https://login.microsoftonline.us/env-tenant-id/oauth2/v2.0/authorize"
|
|
)
|
|
assert (
|
|
provider._upstream_token_endpoint
|
|
== "https://login.microsoftonline.us/env-tenant-id/oauth2/v2.0/token"
|
|
)
|
|
assert (
|
|
provider._token_validator.issuer # type: ignore[attr-defined]
|
|
== "https://login.microsoftonline.us/env-tenant-id/v2.0"
|
|
)
|
|
assert (
|
|
provider._token_validator.jwks_uri # type: ignore[attr-defined]
|
|
== "https://login.microsoftonline.us/env-tenant-id/discovery/v2.0/keys"
|
|
)
|
|
|
|
def test_base_authority_with_special_tenant_values(self):
|
|
"""Test that base_authority works with special tenant values like 'organizations'."""
|
|
provider = AzureProvider(
|
|
client_id="test_client",
|
|
client_secret="test_secret",
|
|
tenant_id="organizations",
|
|
required_scopes=["read"],
|
|
base_authority="login.microsoftonline.us",
|
|
jwt_signing_key="test-secret",
|
|
)
|
|
|
|
parsed = urlparse(provider._upstream_authorization_endpoint)
|
|
assert parsed.netloc == "login.microsoftonline.us"
|
|
assert "/organizations/" in parsed.path
|