Spaces:
Running
Running
File size: 7,116 Bytes
ac19f92 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | """Tests for authentication setup in HTTP apps."""
import pytest
from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend
from mcp.server.auth.provider import AccessToken
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider
from fastmcp.server.http import setup_auth_middleware_and_routes
class TestSetupAuthMiddlewareAndRoutes:
"""Test setup_auth_middleware_and_routes with TokenVerifier providers."""
@pytest.fixture
def bearer_provider(self) -> BearerAuthProvider:
"""Create BearerAuthProvider for testing."""
key_pair = RSAKeyPair.generate()
return BearerAuthProvider(
public_key=key_pair.public_key,
issuer="https://test.example.com",
audience="https://api.example.com",
required_scopes=["read", "write"],
)
@pytest.fixture
def in_memory_provider(self) -> InMemoryOAuthProvider:
"""Create InMemoryOAuthProvider for testing."""
return InMemoryOAuthProvider(
issuer_url="https://test.example.com",
required_scopes=["user"],
)
def test_setup_with_bearer_provider(self, bearer_provider: BearerAuthProvider):
"""Test that setup works with BearerAuthProvider as TokenVerifier."""
middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
bearer_provider
)
# Should return middleware list
assert isinstance(middleware, list)
assert len(middleware) == 2 # AuthenticationMiddleware + AuthContextMiddleware
# First middleware should be AuthenticationMiddleware with BearerAuthBackend
auth_middleware = middleware[0]
assert isinstance(auth_middleware, Middleware)
assert auth_middleware.cls == AuthenticationMiddleware
assert "backend" in auth_middleware.kwargs
backend = auth_middleware.kwargs["backend"]
assert isinstance(backend, BearerAuthBackend)
assert backend.token_verifier is bearer_provider # type: ignore[attr-defined]
# Should return auth routes
assert isinstance(auth_routes, list)
assert len(auth_routes) > 0 # Should have OAuth routes
# Should return required scopes
assert required_scopes == ["read", "write"]
def test_setup_with_in_memory_provider(
self, in_memory_provider: InMemoryOAuthProvider
):
"""Test that setup works with InMemoryOAuthProvider as TokenVerifier."""
middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
in_memory_provider
)
# Should return middleware list
assert isinstance(middleware, list)
assert len(middleware) == 2
# Backend should use the provider as token verifier
auth_middleware = middleware[0]
backend = auth_middleware.kwargs["backend"]
assert isinstance(backend, BearerAuthBackend)
assert backend.token_verifier is in_memory_provider # type: ignore[attr-defined]
# Should return required scopes
assert required_scopes == ["user"]
def test_setup_preserves_provider_functionality(
self, bearer_provider: BearerAuthProvider
):
"""Test that setup doesn't break the provider's functionality."""
# Setup should not modify the provider
original_issuer = bearer_provider.issuer
original_scopes = bearer_provider.required_scopes
middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
bearer_provider
)
# Provider should be unchanged
assert bearer_provider.issuer == original_issuer
assert bearer_provider.required_scopes == original_scopes
# Provider should still work as TokenVerifier
assert hasattr(bearer_provider, "verify_token")
assert callable(bearer_provider.verify_token)
class MockOAuthProvider:
"""Mock OAuth provider that implements TokenVerifier."""
def __init__(self, required_scopes=None, issuer_url="http://localhost:8000"):
from pydantic import AnyHttpUrl
from fastmcp.server.auth.auth import (
ClientRegistrationOptions,
RevocationOptions,
)
self.required_scopes = required_scopes or []
self.issuer_url = AnyHttpUrl(issuer_url)
self.service_documentation_url = None
self.client_registration_options = ClientRegistrationOptions(enabled=False)
self.revocation_options = RevocationOptions(enabled=False)
async def verify_token(self, token: str) -> AccessToken | None:
"""Mock verify_token implementation."""
if token == "valid-token":
return AccessToken(
token=token,
client_id="mock-client",
scopes=self.required_scopes,
expires_at=None,
)
return None
class TestSetupWithMockProvider:
"""Test setup function with mock provider."""
def test_setup_with_mock_token_verifier(self):
"""Test that setup works with any TokenVerifier implementation."""
mock_provider = MockOAuthProvider(required_scopes=["mock-scope"])
middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
mock_provider # type: ignore[arg-type]
)
# Should work with any TokenVerifier
assert len(middleware) == 2
auth_middleware = middleware[0]
backend = auth_middleware.kwargs["backend"]
assert isinstance(backend, BearerAuthBackend)
assert backend.token_verifier is mock_provider # type: ignore[attr-defined]
assert required_scopes == ["mock-scope"]
async def test_setup_middleware_can_authenticate(self):
"""Test that the setup middleware can actually authenticate requests."""
mock_provider = MockOAuthProvider()
middleware, _, _ = setup_auth_middleware_and_routes(mock_provider) # type: ignore[arg-type]
# Extract the BearerAuthBackend
auth_middleware = middleware[0]
backend = auth_middleware.kwargs["backend"]
# Test authentication with valid token
from starlette.requests import HTTPConnection
scope = {
"type": "http",
"headers": [(b"authorization", b"Bearer valid-token")],
}
conn = HTTPConnection(scope)
result = await backend.authenticate(conn) # type: ignore[attr-defined]
assert result is not None
credentials, user = result
assert user.username == "mock-client"
# Test authentication with invalid token
scope = {
"type": "http",
"headers": [(b"authorization", b"Bearer invalid-token")],
}
conn = HTTPConnection(scope)
result = await backend.authenticate(conn) # type: ignore[attr-defined]
assert result is None
|