From 463652a878bc7dc57801e484fd5ef8524498580b Mon Sep 17 00:00:00 2001 From: Nestor Qin Date: Thu, 20 Nov 2025 15:08:50 -0800 Subject: [PATCH] Add unit tests --- src/fastmcp/server/auth/providers/azure.py | 8 +- tests/server/auth/providers/test_azure.py | 173 +++++++++++++++++++++ 2 files changed, 178 insertions(+), 3 deletions(-) diff --git a/src/fastmcp/server/auth/providers/azure.py b/src/fastmcp/server/auth/providers/azure.py index e206ed6ec..42b252eb2 100644 --- a/src/fastmcp/server/auth/providers/azure.py +++ b/src/fastmcp/server/auth/providers/azure.py @@ -6,11 +6,9 @@ using the OAuth Proxy pattern for non-DCR OAuth flows. from __future__ import annotations -from typing import Any +from typing import TYPE_CHECKING, Any from key_value.aio.protocols import AsyncKeyValue -from mcp.server.auth.provider import AuthorizationParams -from mcp.shared.auth import OAuthClientInformationFull from pydantic import SecretStr, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -21,6 +19,10 @@ from fastmcp.utilities.auth import parse_scopes from fastmcp.utilities.logging import get_logger from fastmcp.utilities.types import NotSet, NotSetT +if TYPE_CHECKING: + from mcp.server.auth.provider import AuthorizationParams + from mcp.shared.auth import OAuthClientInformationFull + logger = get_logger(__name__) diff --git a/tests/server/auth/providers/test_azure.py b/tests/server/auth/providers/test_azure.py index 168384eb0..894cd7531 100644 --- a/tests/server/auth/providers/test_azure.py +++ b/tests/server/auth/providers/test_azure.py @@ -487,3 +487,176 @@ class TestAzureProvider: parsed = urlparse(provider._upstream_authorization_endpoint) assert parsed.netloc == "login.microsoftonline.us" assert "/organizations/" in parsed.path + + def test_prepare_scopes_for_upstream_refresh_basic_prefixing(self): + """Test that unprefixed scopes are correctly prefixed for Azure token refresh.""" + provider = AzureProvider( + client_id="test_client", + client_secret="test_secret", + tenant_id="test-tenant", + identifier_uri="api://my-api", + required_scopes=["read", "write"], + jwt_signing_key="test-secret", + ) + + # Unprefixed scopes from storage should be prefixed + result = provider._prepare_scopes_for_upstream_refresh(["read", "write"]) + + assert "api://my-api/read" in result + assert "api://my-api/write" in result + assert len(result) == 2 + + def test_prepare_scopes_for_upstream_refresh_already_prefixed(self): + """Test that already-prefixed scopes remain unchanged.""" + provider = AzureProvider( + client_id="test_client", + client_secret="test_secret", + tenant_id="test-tenant", + identifier_uri="api://my-api", + required_scopes=["read"], + jwt_signing_key="test-secret", + ) + + # Already prefixed scopes should pass through unchanged + result = provider._prepare_scopes_for_upstream_refresh( + ["api://my-api/read", "api://other-api/admin"] + ) + + assert "api://my-api/read" in result + assert "api://other-api/admin" in result + assert len(result) == 2 + + def test_prepare_scopes_for_upstream_refresh_with_additional_scopes(self): + """Test that additional_authorize_scopes are added during token refresh.""" + provider = AzureProvider( + client_id="test_client", + client_secret="test_secret", + tenant_id="test-tenant", + identifier_uri="api://my-api", + required_scopes=["read"], + additional_authorize_scopes=[ + "User.Read", + "openid", + "profile", + "offline_access", + ], + jwt_signing_key="test-secret", + ) + + # Base scopes should be prefixed, additional scopes appended + result = provider._prepare_scopes_for_upstream_refresh(["read", "write"]) + + assert "api://my-api/read" in result + assert "api://my-api/write" in result + assert "User.Read" in result + assert "openid" in result + assert "profile" in result + assert "offline_access" in result + assert len(result) == 6 + + def test_prepare_scopes_for_upstream_refresh_filters_duplicate_additional_scopes( + self, + ): + """Test that accidentally stored additional_authorize_scopes are filtered out.""" + provider = AzureProvider( + client_id="test_client", + client_secret="test_secret", + tenant_id="test-tenant", + identifier_uri="api://my-api", + required_scopes=["read"], + additional_authorize_scopes=["User.Read", "openid"], + jwt_signing_key="test-secret", + ) + + # If additional scopes were accidentally stored, they should be filtered + # to prevent accumulation + result = provider._prepare_scopes_for_upstream_refresh( + ["read", "User.Read", "openid"] + ) + + # Should have: api://my-api/read (prefixed) + User.Read + openid (added once) + assert "api://my-api/read" in result + assert result.count("User.Read") == 1 + assert result.count("openid") == 1 + assert len(result) == 3 + + def test_prepare_scopes_for_upstream_refresh_mixed_scopes(self): + """Test mixed scenario with both prefixed and unprefixed scopes.""" + provider = AzureProvider( + client_id="test_client", + client_secret="test_secret", + tenant_id="test-tenant", + identifier_uri="api://my-api", + required_scopes=["read"], + additional_authorize_scopes=["User.Read"], + jwt_signing_key="test-secret", + ) + + # Mix of prefixed and unprefixed scopes + result = provider._prepare_scopes_for_upstream_refresh( + ["read", "api://other-api/admin", "write"] + ) + + assert "api://my-api/read" in result + assert "api://other-api/admin" in result # Already prefixed, unchanged + assert "api://my-api/write" in result + assert "User.Read" in result + assert len(result) == 4 + + def test_prepare_scopes_for_upstream_refresh_scope_with_slash(self): + """Test that scopes containing '/' are not prefixed.""" + provider = AzureProvider( + client_id="test_client", + client_secret="test_secret", + tenant_id="test-tenant", + identifier_uri="api://my-api", + required_scopes=["read"], + jwt_signing_key="test-secret", + ) + + # Scopes with "/" should not be prefixed (already fully qualified) + result = provider._prepare_scopes_for_upstream_refresh( + ["read", "https://graph.microsoft.com/.default"] + ) + + assert "api://my-api/read" in result + assert ( + "https://graph.microsoft.com/.default" in result + ) # Not prefixed (contains ://) + + def test_prepare_scopes_for_upstream_refresh_empty_scopes(self): + """Test behavior with empty scopes list.""" + provider = AzureProvider( + client_id="test_client", + client_secret="test_secret", + tenant_id="test-tenant", + identifier_uri="api://my-api", + required_scopes=["read"], + additional_authorize_scopes=["User.Read", "openid"], + jwt_signing_key="test-secret", + ) + + # Empty scopes should still add additional_authorize_scopes + result = provider._prepare_scopes_for_upstream_refresh([]) + + assert "User.Read" in result + assert "openid" in result + assert len(result) == 2 + + def test_prepare_scopes_for_upstream_refresh_no_additional_scopes(self): + """Test behavior when no additional_authorize_scopes are configured.""" + provider = AzureProvider( + client_id="test_client", + client_secret="test_secret", + tenant_id="test-tenant", + identifier_uri="api://my-api", + required_scopes=["read"], + jwt_signing_key="test-secret", + ) + + # Should only prefix base scopes, no additional scopes added + result = provider._prepare_scopes_for_upstream_refresh(["read", "write"]) + + assert "api://my-api/read" in result + assert "api://my-api/write" in result + assert len(result) == 2