Implement TokenVerifier protocol for mcp-python-sdk compatibility

Fixes breaking changes from mcp-python-sdk PR #982 which updated
BearerAuthBackend to use TokenVerifier protocol instead of OAuth providers.

Changes:
- Add verify_token() method to BearerAuthProvider implementing TokenVerifier protocol
- Add verify_token() method to InMemoryOAuthProvider implementing TokenVerifier protocol
- Update BearerAuthBackend usage in setup_auth_middleware_and_routes() to pass TokenVerifier
- Add comprehensive unit tests for TokenVerifier implementations
- Add integration tests for BearerAuthBackend with TokenVerifier
- Add tests for HTTP auth setup functions

All existing functionality preserved with full backwards compatibility.

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

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Jeremiah Lowin 2025-06-26 15:08:31 -04:00
commit b98b080c98
6 changed files with 575 additions and 1 deletions

View file

@ -0,0 +1,179 @@
"""Tests for TokenVerifier protocol implementation in auth providers."""
import pytest
from mcp.server.auth.provider import AccessToken
from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider
class TestBearerAuthProviderTokenVerifier:
"""Test that BearerAuthProvider implements TokenVerifier protocol correctly."""
@pytest.fixture
def rsa_key_pair(self) -> RSAKeyPair:
"""Generate RSA key pair for testing."""
return RSAKeyPair.generate()
@pytest.fixture
def bearer_provider(self, rsa_key_pair: RSAKeyPair) -> BearerAuthProvider:
"""Create BearerAuthProvider for testing."""
return BearerAuthProvider(
public_key=rsa_key_pair.public_key,
issuer="https://test.example.com",
audience="https://api.example.com",
)
@pytest.fixture
def valid_token(self, rsa_key_pair: RSAKeyPair) -> str:
"""Create a valid test token."""
return rsa_key_pair.create_token(
subject="test-user",
issuer="https://test.example.com",
audience="https://api.example.com",
scopes=["read", "write"],
)
@pytest.fixture
def expired_token(self, rsa_key_pair: RSAKeyPair) -> str:
"""Create an expired test token."""
return rsa_key_pair.create_token(
subject="test-user",
issuer="https://test.example.com",
audience="https://api.example.com",
expires_in_seconds=-3600, # Expired 1 hour ago
)
async def test_verify_token_with_valid_token(
self, bearer_provider: BearerAuthProvider, valid_token: str
):
"""Test that verify_token returns AccessToken for valid token."""
result = await bearer_provider.verify_token(valid_token)
assert result is not None
assert isinstance(result, AccessToken)
assert result.token == valid_token
assert result.client_id == "test-user"
assert "read" in result.scopes
assert "write" in result.scopes
async def test_verify_token_with_expired_token(
self, bearer_provider: BearerAuthProvider, expired_token: str
):
"""Test that verify_token returns None for expired token."""
result = await bearer_provider.verify_token(expired_token)
assert result is None
async def test_verify_token_with_invalid_token(
self, bearer_provider: BearerAuthProvider
):
"""Test that verify_token returns None for invalid token."""
result = await bearer_provider.verify_token("invalid.token.here")
assert result is None
async def test_verify_token_with_malformed_token(
self, bearer_provider: BearerAuthProvider
):
"""Test that verify_token returns None for malformed token."""
result = await bearer_provider.verify_token("not-a-jwt")
assert result is None
async def test_verify_token_delegation_to_load_access_token(
self, bearer_provider: BearerAuthProvider, valid_token: str
):
"""Test that verify_token delegates to load_access_token."""
# Both methods should return the same result
verify_result = await bearer_provider.verify_token(valid_token)
load_result = await bearer_provider.load_access_token(valid_token)
assert verify_result == load_result
if verify_result is not None and load_result is not None:
assert verify_result.token == load_result.token
assert verify_result.client_id == load_result.client_id
assert verify_result.scopes == load_result.scopes
class TestInMemoryOAuthProviderTokenVerifier:
"""Test that InMemoryOAuthProvider implements TokenVerifier protocol correctly."""
@pytest.fixture
def in_memory_provider(self) -> InMemoryOAuthProvider:
"""Create InMemoryOAuthProvider for testing."""
return InMemoryOAuthProvider(
issuer_url="https://test.example.com",
required_scopes=["user"],
)
async def test_verify_token_with_nonexistent_token(
self, in_memory_provider: InMemoryOAuthProvider
):
"""Test that verify_token returns None for nonexistent token."""
result = await in_memory_provider.verify_token("nonexistent-token")
assert result is None
async def test_verify_token_delegation_to_load_access_token(
self, in_memory_provider: InMemoryOAuthProvider
):
"""Test that verify_token delegates to load_access_token."""
# Create a test token in the provider's storage
test_token = "test-access-token"
test_access_token = AccessToken(
token=test_token,
client_id="test-client",
scopes=["user"],
expires_at=None, # No expiry
)
in_memory_provider.access_tokens[test_token] = test_access_token
# Both methods should return the same result
verify_result = await in_memory_provider.verify_token(test_token)
load_result = await in_memory_provider.load_access_token(test_token)
assert verify_result == load_result
assert verify_result is not None
assert verify_result.token == test_token
assert verify_result.client_id == "test-client"
assert verify_result.scopes == ["user"]
async def test_verify_token_with_expired_token(
self, in_memory_provider: InMemoryOAuthProvider
):
"""Test that verify_token returns None for expired token."""
import time
# Create an expired token
expired_token = "expired-token"
expired_access_token = AccessToken(
token=expired_token,
client_id="test-client",
scopes=["user"],
expires_at=int(time.time()) - 3600, # Expired 1 hour ago
)
in_memory_provider.access_tokens[expired_token] = expired_access_token
result = await in_memory_provider.verify_token(expired_token)
assert result is None
# Token should be cleaned up from storage
assert expired_token not in in_memory_provider.access_tokens
class TestTokenVerifierProtocolCompliance:
"""Test that our providers properly implement the TokenVerifier protocol."""
async def test_bearer_provider_implements_protocol(self):
"""Test that BearerAuthProvider can be used as TokenVerifier."""
key_pair = RSAKeyPair.generate()
provider = BearerAuthProvider(public_key=key_pair.public_key)
# Should have the required method for TokenVerifier protocol
assert hasattr(provider, "verify_token")
assert callable(provider.verify_token)
async def test_in_memory_provider_implements_protocol(self):
"""Test that InMemoryOAuthProvider can be used as TokenVerifier."""
provider = InMemoryOAuthProvider()
# Should have the required method for TokenVerifier protocol
assert hasattr(provider, "verify_token")
assert callable(provider.verify_token)