Spaces:
Running
Running
File size: 6,284 Bytes
ac19f92 341332a 3959209 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 ac19f92 b6176a5 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 | """Tests for BearerAuthBackend integration with TokenVerifier."""
import pytest
from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend
from starlette.requests import HTTPConnection
from fastmcp.server.auth import AccessToken
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
class TestBearerAuthBackendTokenVerifierIntegration:
"""Test BearerAuthBackend works with TokenVerifier protocol."""
@pytest.fixture
def rsa_key_pair(self) -> RSAKeyPair:
"""Generate RSA key pair for testing."""
return RSAKeyPair.generate()
@pytest.fixture
def jwt_verifier(self, rsa_key_pair: RSAKeyPair) -> JWTVerifier:
"""Create JWTVerifier for testing."""
return JWTVerifier(
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"],
)
def test_bearer_auth_backend_constructor_accepts_token_verifier(
self, jwt_verifier: JWTVerifier
):
"""Test that BearerAuthBackend constructor accepts TokenVerifier."""
# This should not raise an error
backend = BearerAuthBackend(jwt_verifier)
assert backend.token_verifier is jwt_verifier # type: ignore[attr-defined]
async def test_bearer_auth_backend_authenticate_with_valid_token(
self, jwt_verifier: JWTVerifier, valid_token: str
):
"""Test BearerAuthBackend authentication with valid token."""
backend = BearerAuthBackend(jwt_verifier)
# Create mock HTTPConnection with Authorization header
scope = {
"type": "http",
"headers": [(b"authorization", f"Bearer {valid_token}".encode())],
}
conn = HTTPConnection(scope)
result = await backend.authenticate(conn)
assert result is not None
credentials, user = result
assert credentials.scopes == ["read", "write"]
assert user.username == "test-user"
assert hasattr(user, "access_token")
assert user.access_token.token == valid_token
async def test_bearer_auth_backend_authenticate_with_invalid_token(
self, jwt_verifier: JWTVerifier
):
"""Test BearerAuthBackend authentication with invalid token."""
backend = BearerAuthBackend(jwt_verifier)
# Create mock HTTPConnection with invalid Authorization header
scope = {
"type": "http",
"headers": [(b"authorization", b"Bearer invalid-token")],
}
conn = HTTPConnection(scope)
result = await backend.authenticate(conn)
assert result is None
async def test_bearer_auth_backend_authenticate_with_no_header(
self, jwt_verifier: JWTVerifier
):
"""Test BearerAuthBackend authentication with no Authorization header."""
backend = BearerAuthBackend(jwt_verifier)
# Create mock HTTPConnection without Authorization header
scope = {
"type": "http",
"headers": [],
}
conn = HTTPConnection(scope)
result = await backend.authenticate(conn)
assert result is None
async def test_bearer_auth_backend_authenticate_with_non_bearer_token(
self, jwt_verifier: JWTVerifier
):
"""Test BearerAuthBackend authentication with non-Bearer token."""
backend = BearerAuthBackend(jwt_verifier)
# Create mock HTTPConnection with Basic auth header
scope = {
"type": "http",
"headers": [(b"authorization", b"Basic dXNlcjpwYXNz")],
}
conn = HTTPConnection(scope)
result = await backend.authenticate(conn)
assert result is None
class MockTokenVerifier:
"""Mock TokenVerifier for testing backend integration."""
def __init__(self, return_value: AccessToken | None = None):
self.return_value = return_value
self.verify_token_calls = []
async def verify_token(self, token: str) -> AccessToken | None:
"""Mock verify_token method."""
self.verify_token_calls.append(token)
return self.return_value
class TestBearerAuthBackendWithMockVerifier:
"""Test BearerAuthBackend with mock TokenVerifier."""
async def test_backend_calls_verify_token_method(self):
"""Test that BearerAuthBackend calls verify_token on the verifier."""
mock_access_token = AccessToken(
token="test-token",
client_id="test-client",
scopes=["read"],
expires_at=None,
)
mock_verifier = MockTokenVerifier(return_value=mock_access_token)
backend = BearerAuthBackend(mock_verifier) # type: ignore[arg-type]
scope = {
"type": "http",
"headers": [(b"authorization", b"Bearer test-token")],
}
conn = HTTPConnection(scope)
result = await backend.authenticate(conn)
# Should have called verify_token with the token
assert mock_verifier.verify_token_calls == ["test-token"]
# Should return authentication result
assert result is not None
credentials, user = result
assert credentials.scopes == ["read"]
assert user.username == "test-client"
async def test_backend_handles_verify_token_none_result(self):
"""Test that BearerAuthBackend handles None result from verify_token."""
mock_verifier = MockTokenVerifier(return_value=None)
backend = BearerAuthBackend(mock_verifier) # type: ignore[arg-type]
scope = {
"type": "http",
"headers": [(b"authorization", b"Bearer invalid-token")],
}
conn = HTTPConnection(scope)
result = await backend.authenticate(conn)
# Should have called verify_token
assert mock_verifier.verify_token_calls == ["invalid-token"]
# Should return None for authentication failure
assert result is None
|