mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-15 18:19:10 +02:00
3321644ad3* Use anyio as testing backend * Remove asyncio markers * Update streamable http tests * Replace all subprocess tests * Replace anyio task groups with asyncio context managers in tests - Convert run_server_async from anyio task group pattern to asyncio.create_task with async context manager - Remove task_group fixture from conftest - Update all test fixtures to use async with run_server_async pattern - Remove TaskGroup imports from all test files - Tests now work with pytest-asyncio instead of pytest-anyio * Update test_github_provider_integration.py
1078 lines
39 KiB
Python
1078 lines
39 KiB
Python
from collections.abc import AsyncGenerator
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
from pytest_httpx import HTTPXMock
|
|
|
|
from fastmcp import Client, FastMCP
|
|
from fastmcp.client.auth.bearer import BearerAuth
|
|
from fastmcp.server.auth.providers.jwt import JWKData, JWKSData, JWTVerifier, RSAKeyPair
|
|
from fastmcp.utilities.tests import run_server_async
|
|
|
|
|
|
class SymmetricKeyHelper:
|
|
"""Helper class for generating symmetric key JWT tokens for testing."""
|
|
|
|
def __init__(self, secret: str):
|
|
"""Initialize with a secret key."""
|
|
self.secret = secret
|
|
|
|
def create_token(
|
|
self,
|
|
subject: str = "fastmcp-user",
|
|
issuer: str = "https://fastmcp.example.com",
|
|
audience: str | list[str] | None = None,
|
|
scopes: list[str] | None = None,
|
|
expires_in_seconds: int = 3600,
|
|
additional_claims: dict[str, Any] | None = None,
|
|
algorithm: str = "HS256",
|
|
) -> str:
|
|
"""
|
|
Generate a test JWT token using symmetric key for testing purposes.
|
|
|
|
Args:
|
|
subject: Subject claim (usually user ID)
|
|
issuer: Issuer claim
|
|
audience: Audience claim - can be a string or list of strings (optional)
|
|
scopes: List of scopes to include
|
|
expires_in_seconds: Token expiration time in seconds
|
|
additional_claims: Any additional claims to include
|
|
algorithm: JWT signing algorithm (HS256, HS384, or HS512)
|
|
"""
|
|
import time
|
|
|
|
from authlib.jose import JsonWebToken
|
|
|
|
# Create header
|
|
header = {"alg": algorithm}
|
|
|
|
# Create payload
|
|
payload = {
|
|
"sub": subject,
|
|
"iss": issuer,
|
|
"iat": int(time.time()),
|
|
"exp": int(time.time()) + expires_in_seconds,
|
|
}
|
|
|
|
if audience:
|
|
payload["aud"] = audience
|
|
|
|
if scopes:
|
|
payload["scope"] = " ".join(scopes)
|
|
|
|
if additional_claims:
|
|
payload.update(additional_claims)
|
|
|
|
# Create JWT
|
|
jwt_lib = JsonWebToken([algorithm])
|
|
token_bytes = jwt_lib.encode(header, payload, self.secret)
|
|
|
|
return token_bytes.decode("utf-8")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def rsa_key_pair() -> RSAKeyPair:
|
|
return RSAKeyPair.generate()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def symmetric_key_helper() -> SymmetricKeyHelper:
|
|
"""Generate a symmetric key helper for testing."""
|
|
return SymmetricKeyHelper("test-secret-key-for-hmac-signing")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def bearer_token(rsa_key_pair: RSAKeyPair) -> str:
|
|
return rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def bearer_provider(rsa_key_pair: RSAKeyPair) -> JWTVerifier:
|
|
return JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def symmetric_provider(symmetric_key_helper: SymmetricKeyHelper) -> JWTVerifier:
|
|
"""Create JWTVerifier configured for symmetric key verification."""
|
|
return JWTVerifier(
|
|
public_key=symmetric_key_helper.secret,
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
algorithm="HS256",
|
|
)
|
|
|
|
|
|
def create_mcp_server(
|
|
public_key: str,
|
|
auth_kwargs: dict[str, Any] | None = None,
|
|
) -> FastMCP:
|
|
mcp = FastMCP(
|
|
auth=JWTVerifier(
|
|
public_key=public_key,
|
|
**auth_kwargs or {},
|
|
)
|
|
)
|
|
|
|
@mcp.tool
|
|
def add(a: int, b: int) -> int:
|
|
return a + b
|
|
|
|
return mcp
|
|
|
|
|
|
@pytest.fixture
|
|
async def mcp_server_url(rsa_key_pair: RSAKeyPair) -> AsyncGenerator[str, None]:
|
|
server = create_mcp_server(
|
|
public_key=rsa_key_pair.public_key,
|
|
auth_kwargs=dict(
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
),
|
|
)
|
|
async with run_server_async(server, transport="http") as url:
|
|
yield url
|
|
|
|
|
|
class TestRSAKeyPair:
|
|
def test_generate_key_pair(self):
|
|
"""Test RSA key pair generation."""
|
|
key_pair = RSAKeyPair.generate()
|
|
|
|
assert key_pair.private_key is not None
|
|
assert key_pair.public_key is not None
|
|
|
|
# Check that keys are in PEM format
|
|
private_pem = key_pair.private_key.get_secret_value()
|
|
public_pem = key_pair.public_key
|
|
|
|
assert "-----BEGIN PRIVATE KEY-----" in private_pem
|
|
assert "-----END PRIVATE KEY-----" in private_pem
|
|
assert "-----BEGIN PUBLIC KEY-----" in public_pem
|
|
assert "-----END PUBLIC KEY-----" in public_pem
|
|
|
|
def test_create_basic_token(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test basic token creation."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
)
|
|
|
|
assert isinstance(token, str)
|
|
assert len(token.split(".")) == 3 # JWT has 3 parts
|
|
|
|
def test_create_token_with_scopes(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test token creation with scopes."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
scopes=["read", "write"],
|
|
)
|
|
|
|
assert isinstance(token, str)
|
|
# We'll validate the scopes in the BearerToken tests
|
|
|
|
|
|
class TestSymmetricKeyJWT:
|
|
"""Tests for JWT verification using symmetric keys (HMAC algorithms)."""
|
|
|
|
def test_initialization_with_symmetric_key(
|
|
self, symmetric_key_helper: SymmetricKeyHelper
|
|
):
|
|
"""Test JWTVerifier initialization with symmetric key."""
|
|
provider = JWTVerifier(
|
|
public_key=symmetric_key_helper.secret,
|
|
issuer="https://test.example.com",
|
|
algorithm="HS256",
|
|
)
|
|
|
|
assert provider.issuer == "https://test.example.com"
|
|
assert provider.public_key == symmetric_key_helper.secret
|
|
assert provider.algorithm == "HS256"
|
|
assert provider.jwks_uri is None
|
|
|
|
def test_initialization_with_different_symmetric_algorithms(
|
|
self, symmetric_key_helper: SymmetricKeyHelper
|
|
):
|
|
"""Test JWTVerifier initialization with different HMAC algorithms."""
|
|
algorithms = ["HS256", "HS384", "HS512"]
|
|
|
|
for algorithm in algorithms:
|
|
provider = JWTVerifier(
|
|
public_key=symmetric_key_helper.secret,
|
|
issuer="https://test.example.com",
|
|
algorithm=algorithm,
|
|
)
|
|
assert provider.algorithm == algorithm
|
|
|
|
async def test_valid_symmetric_token_validation(
|
|
self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
|
|
):
|
|
"""Test validation of a valid token signed with symmetric key."""
|
|
token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
scopes=["read", "write"],
|
|
algorithm="HS256",
|
|
)
|
|
|
|
access_token = await symmetric_provider.load_access_token(token)
|
|
|
|
assert access_token is not None
|
|
assert access_token.client_id == "test-user"
|
|
assert "read" in access_token.scopes
|
|
assert "write" in access_token.scopes
|
|
assert access_token.expires_at is not None
|
|
|
|
async def test_symmetric_token_with_different_algorithms(
|
|
self, symmetric_key_helper: SymmetricKeyHelper
|
|
):
|
|
"""Test that different HMAC algorithms work correctly."""
|
|
algorithms = ["HS256", "HS384", "HS512"]
|
|
|
|
for algorithm in algorithms:
|
|
provider = JWTVerifier(
|
|
public_key=symmetric_key_helper.secret,
|
|
issuer="https://test.example.com",
|
|
algorithm=algorithm,
|
|
)
|
|
|
|
token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
algorithm=algorithm,
|
|
)
|
|
|
|
access_token = await provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert access_token.client_id == "test-user"
|
|
|
|
async def test_symmetric_token_issuer_validation(
|
|
self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
|
|
):
|
|
"""Test issuer validation with symmetric key tokens."""
|
|
# Valid issuer
|
|
valid_token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
access_token = await symmetric_provider.load_access_token(valid_token)
|
|
assert access_token is not None
|
|
|
|
# Invalid issuer
|
|
invalid_token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://evil.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
access_token = await symmetric_provider.load_access_token(invalid_token)
|
|
assert access_token is None
|
|
|
|
async def test_symmetric_token_audience_validation(
|
|
self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
|
|
):
|
|
"""Test audience validation with symmetric key tokens."""
|
|
# Valid audience
|
|
valid_token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
access_token = await symmetric_provider.load_access_token(valid_token)
|
|
assert access_token is not None
|
|
|
|
# Invalid audience
|
|
invalid_token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://wrong-api.example.com",
|
|
)
|
|
access_token = await symmetric_provider.load_access_token(invalid_token)
|
|
assert access_token is None
|
|
|
|
async def test_symmetric_token_scope_extraction(
|
|
self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
|
|
):
|
|
"""Test scope extraction from symmetric key tokens."""
|
|
token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
scopes=["read", "write", "admin"],
|
|
)
|
|
|
|
access_token = await symmetric_provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert set(access_token.scopes) == {"read", "write", "admin"}
|
|
|
|
async def test_symmetric_token_expiration(
|
|
self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
|
|
):
|
|
"""Test expiration validation with symmetric key tokens."""
|
|
# Valid token
|
|
valid_token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
expires_in_seconds=3600, # 1 hour from now
|
|
)
|
|
access_token = await symmetric_provider.load_access_token(valid_token)
|
|
assert access_token is not None
|
|
|
|
# Expired token
|
|
expired_token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
expires_in_seconds=-3600, # 1 hour ago
|
|
)
|
|
access_token = await symmetric_provider.load_access_token(expired_token)
|
|
assert access_token is None
|
|
|
|
async def test_symmetric_token_invalid_signature(
|
|
self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
|
|
):
|
|
"""Test rejection of tokens with invalid signatures."""
|
|
# Create a token with a different secret
|
|
other_helper = SymmetricKeyHelper("different-secret-key")
|
|
token = other_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
access_token = await symmetric_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_symmetric_token_algorithm_mismatch(
|
|
self, symmetric_key_helper: SymmetricKeyHelper
|
|
):
|
|
"""Test that tokens with mismatched algorithms are rejected."""
|
|
# Create provider expecting HS256
|
|
provider = JWTVerifier(
|
|
public_key=symmetric_key_helper.secret,
|
|
issuer="https://test.example.com",
|
|
algorithm="HS256",
|
|
)
|
|
|
|
# Create token with HS512
|
|
token = symmetric_key_helper.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
algorithm="HS512",
|
|
)
|
|
|
|
# Should fail because provider expects HS256
|
|
access_token = await provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
|
|
class TestBearerTokenJWKS:
|
|
"""Tests for JWKS URI functionality."""
|
|
|
|
@pytest.fixture
|
|
def jwks_provider(self, rsa_key_pair: RSAKeyPair) -> JWTVerifier:
|
|
"""Provider configured with JWKS URI."""
|
|
return JWTVerifier(
|
|
jwks_uri="https://test.example.com/.well-known/jwks.json",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
@pytest.fixture
|
|
def mock_jwks_data(self, rsa_key_pair: RSAKeyPair) -> JWKSData:
|
|
"""Create mock JWKS data from RSA key pair."""
|
|
from authlib.jose import JsonWebKey
|
|
|
|
# Create JWK from the RSA public key
|
|
jwk = JsonWebKey.import_key(rsa_key_pair.public_key) # type: ignore
|
|
jwk_data: JWKData = jwk.as_dict() # type: ignore
|
|
jwk_data["kid"] = "test-key-1"
|
|
jwk_data["alg"] = "RS256"
|
|
|
|
return {"keys": [jwk_data]}
|
|
|
|
async def test_jwks_token_validation(
|
|
self,
|
|
rsa_key_pair: RSAKeyPair,
|
|
jwks_provider: JWTVerifier,
|
|
mock_jwks_data: JWKSData,
|
|
httpx_mock: HTTPXMock,
|
|
):
|
|
"""Test token validation using JWKS URI."""
|
|
httpx_mock.add_response(
|
|
url="https://test.example.com/.well-known/jwks.json",
|
|
json=mock_jwks_data,
|
|
)
|
|
|
|
username = "test-user"
|
|
issuer = "https://test.example.com"
|
|
audience = "https://api.example.com"
|
|
|
|
token = rsa_key_pair.create_token(
|
|
subject=username,
|
|
issuer=issuer,
|
|
audience=audience,
|
|
)
|
|
|
|
access_token = await jwks_provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert access_token.client_id == username
|
|
|
|
# ensure the raw claims are present - #1398
|
|
assert access_token.claims.get("sub") == username
|
|
assert access_token.claims.get("iss") == issuer
|
|
assert access_token.claims.get("aud") == audience
|
|
|
|
async def test_jwks_token_validation_with_invalid_key(
|
|
self,
|
|
rsa_key_pair: RSAKeyPair,
|
|
jwks_provider: JWTVerifier,
|
|
mock_jwks_data: JWKSData,
|
|
httpx_mock: HTTPXMock,
|
|
):
|
|
httpx_mock.add_response(
|
|
url="https://test.example.com/.well-known/jwks.json",
|
|
json=mock_jwks_data,
|
|
)
|
|
token = RSAKeyPair.generate().create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
access_token = await jwks_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_jwks_token_validation_with_kid(
|
|
self,
|
|
rsa_key_pair: RSAKeyPair,
|
|
jwks_provider: JWTVerifier,
|
|
mock_jwks_data: JWKSData,
|
|
httpx_mock: HTTPXMock,
|
|
):
|
|
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
|
|
httpx_mock.add_response(
|
|
url="https://test.example.com/.well-known/jwks.json",
|
|
json=mock_jwks_data,
|
|
)
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
kid="test-key-1",
|
|
)
|
|
|
|
access_token = await jwks_provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert access_token.client_id == "test-user"
|
|
|
|
async def test_jwks_token_validation_with_kid_and_no_kid_in_token(
|
|
self,
|
|
rsa_key_pair: RSAKeyPair,
|
|
jwks_provider: JWTVerifier,
|
|
mock_jwks_data: JWKSData,
|
|
httpx_mock: HTTPXMock,
|
|
):
|
|
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
|
|
httpx_mock.add_response(
|
|
url="https://test.example.com/.well-known/jwks.json",
|
|
json=mock_jwks_data,
|
|
)
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
access_token = await jwks_provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert access_token.client_id == "test-user"
|
|
|
|
async def test_jwks_token_validation_with_no_kid_and_kid_in_jwks(
|
|
self,
|
|
rsa_key_pair: RSAKeyPair,
|
|
jwks_provider: JWTVerifier,
|
|
mock_jwks_data: JWKSData,
|
|
httpx_mock: HTTPXMock,
|
|
):
|
|
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
|
|
httpx_mock.add_response(
|
|
url="https://test.example.com/.well-known/jwks.json",
|
|
json=mock_jwks_data,
|
|
)
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
access_token = await jwks_provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert access_token.client_id == "test-user"
|
|
|
|
async def test_jwks_token_validation_with_kid_mismatch(
|
|
self,
|
|
rsa_key_pair: RSAKeyPair,
|
|
jwks_provider: JWTVerifier,
|
|
mock_jwks_data: JWKSData,
|
|
httpx_mock: HTTPXMock,
|
|
):
|
|
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
|
|
httpx_mock.add_response(
|
|
url="https://test.example.com/.well-known/jwks.json",
|
|
json=mock_jwks_data,
|
|
)
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
kid="test-key-2",
|
|
)
|
|
|
|
access_token = await jwks_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_jwks_token_validation_with_multiple_keys_and_no_kid_in_token(
|
|
self,
|
|
rsa_key_pair: RSAKeyPair,
|
|
jwks_provider: JWTVerifier,
|
|
mock_jwks_data: JWKSData,
|
|
httpx_mock: HTTPXMock,
|
|
):
|
|
mock_jwks_data["keys"] = [
|
|
{
|
|
"kid": "test-key-1",
|
|
"alg": "RS256",
|
|
},
|
|
{
|
|
"kid": "test-key-2",
|
|
"alg": "RS256",
|
|
},
|
|
]
|
|
|
|
httpx_mock.add_response(
|
|
url="https://test.example.com/.well-known/jwks.json",
|
|
json=mock_jwks_data,
|
|
)
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
access_token = await jwks_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
|
|
class TestBearerToken:
|
|
def test_initialization_with_public_key(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test provider initialization with public key."""
|
|
provider = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key, issuer="https://test.example.com"
|
|
)
|
|
|
|
assert provider.issuer == "https://test.example.com"
|
|
assert provider.public_key is not None
|
|
assert provider.jwks_uri is None
|
|
|
|
def test_initialization_with_jwks_uri(self):
|
|
"""Test provider initialization with JWKS URI."""
|
|
provider = JWTVerifier(
|
|
jwks_uri="https://test.example.com/.well-known/jwks.json",
|
|
issuer="https://test.example.com",
|
|
)
|
|
|
|
assert provider.issuer == "https://test.example.com"
|
|
assert provider.jwks_uri == "https://test.example.com/.well-known/jwks.json"
|
|
assert provider.public_key is None
|
|
|
|
def test_initialization_requires_key_or_uri(self):
|
|
"""Test that either public_key or jwks_uri is required."""
|
|
with pytest.raises(
|
|
ValueError, match="Either public_key or jwks_uri must be provided"
|
|
):
|
|
JWTVerifier(issuer="https://test.example.com")
|
|
|
|
def test_initialization_rejects_both_key_and_uri(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test that both public_key and jwks_uri cannot be provided."""
|
|
with pytest.raises(
|
|
ValueError, match="Provide either public_key or jwks_uri, not both"
|
|
):
|
|
JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
jwks_uri="https://test.example.com/.well-known/jwks.json",
|
|
issuer="https://test.example.com",
|
|
)
|
|
|
|
async def test_valid_token_validation(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test validation of a valid token."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
scopes=["read", "write"],
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
|
|
assert access_token is not None
|
|
assert access_token.client_id == "test-user"
|
|
assert "read" in access_token.scopes
|
|
assert "write" in access_token.scopes
|
|
assert access_token.expires_at is not None
|
|
|
|
async def test_expired_token_rejection(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test rejection of expired tokens."""
|
|
token = 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
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_invalid_issuer_rejection(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test rejection of tokens with invalid issuer."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://evil.example.com", # Wrong issuer
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_invalid_audience_rejection(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test rejection of tokens with invalid audience."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://wrong-api.example.com", # Wrong audience
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_no_issuer_validation_when_none(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test that issuer validation is skipped when provider has no issuer configured."""
|
|
provider = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer=None, # No issuer validation
|
|
)
|
|
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user", issuer="https://any.example.com"
|
|
)
|
|
|
|
access_token = await provider.load_access_token(token)
|
|
assert access_token is not None
|
|
|
|
async def test_no_audience_validation_when_none(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test that audience validation is skipped when provider has no audience configured."""
|
|
provider = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.example.com",
|
|
audience=None, # No audience validation
|
|
)
|
|
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://any-api.example.com",
|
|
)
|
|
|
|
access_token = await provider.load_access_token(token)
|
|
assert access_token is not None
|
|
|
|
async def test_multiple_audiences_validation(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test validation with multiple audiences in token."""
|
|
provider = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
additional_claims={
|
|
"aud": ["https://api.example.com", "https://other-api.example.com"]
|
|
},
|
|
)
|
|
|
|
access_token = await provider.load_access_token(token)
|
|
assert access_token is not None
|
|
|
|
async def test_provider_with_multiple_expected_audiences(
|
|
self, rsa_key_pair: RSAKeyPair
|
|
):
|
|
"""Test provider configured with multiple expected audiences."""
|
|
provider = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.example.com",
|
|
audience=["https://api.example.com", "https://other-api.example.com"],
|
|
)
|
|
|
|
# Token with single audience that matches one of the expected
|
|
token1 = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
access_token1 = await provider.load_access_token(token1)
|
|
assert access_token1 is not None
|
|
|
|
# Token with multiple audiences, one of which matches
|
|
token2 = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
additional_claims={
|
|
"aud": ["https://api.example.com", "https://third-party.example.com"]
|
|
},
|
|
)
|
|
access_token2 = await provider.load_access_token(token2)
|
|
assert access_token2 is not None
|
|
|
|
# Token with audience that doesn't match any expected
|
|
token3 = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://wrong-api.example.com",
|
|
)
|
|
access_token3 = await provider.load_access_token(token3)
|
|
assert access_token3 is None
|
|
|
|
async def test_scope_extraction_string(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test scope extraction from space-separated string."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
scopes=["read", "write", "admin"],
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
|
|
assert access_token is not None
|
|
assert set(access_token.scopes) == {"read", "write", "admin"}
|
|
|
|
async def test_scope_extraction_list(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test scope extraction from list format."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
additional_claims={"scope": ["read", "write"]}, # List format
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
|
|
assert access_token is not None
|
|
assert set(access_token.scopes) == {"read", "write"}
|
|
|
|
async def test_no_scopes(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test token with no scopes."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
# No scopes
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
|
|
assert access_token is not None
|
|
assert access_token.scopes == []
|
|
|
|
async def test_scp_claim_extraction_string(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test scope extraction from 'scp' claim with space-separated string."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
additional_claims={"scp": "read write admin"}, # 'scp' claim as string
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
|
|
assert access_token is not None
|
|
assert set(access_token.scopes) == {"read", "write", "admin"}
|
|
|
|
async def test_scp_claim_extraction_list(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test scope extraction from 'scp' claim with list format."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
additional_claims={
|
|
"scp": ["read", "write", "admin"]
|
|
}, # 'scp' claim as list
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
|
|
assert access_token is not None
|
|
assert set(access_token.scopes) == {"read", "write", "admin"}
|
|
|
|
async def test_scope_precedence_over_scp(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test that 'scope' claim takes precedence over 'scp' claim when both are present."""
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
additional_claims={
|
|
"scope": "read write", # Standard OAuth2 claim
|
|
"scp": "admin delete", # Should be ignored when 'scope' is present
|
|
},
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
|
|
assert access_token is not None
|
|
assert set(access_token.scopes) == {"read", "write"} # Only 'scope' claim used
|
|
|
|
async def test_malformed_token_rejection(self, bearer_provider: JWTVerifier):
|
|
"""Test rejection of malformed tokens."""
|
|
malformed_tokens = [
|
|
"not.a.jwt",
|
|
"too.many.parts.here.invalid",
|
|
"invalid-token",
|
|
"",
|
|
"header.body", # Missing signature
|
|
]
|
|
|
|
for token in malformed_tokens:
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_invalid_signature_rejection(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test rejection of tokens with invalid signatures."""
|
|
# Create a token with a different key pair
|
|
other_key_pair = RSAKeyPair.generate()
|
|
token = other_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_client_id_fallback(
|
|
self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
|
|
):
|
|
"""Test client_id extraction with fallback logic."""
|
|
# Test with explicit client_id claim
|
|
token = rsa_key_pair.create_token(
|
|
subject="user123",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
additional_claims={"client_id": "app456"},
|
|
)
|
|
|
|
access_token = await bearer_provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert access_token.client_id == "app456" # Should prefer client_id over sub
|
|
|
|
async def test_string_issuer_validation(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test that string (non-URL) issuers are supported per RFC 7519."""
|
|
# Create provider with string issuer
|
|
provider = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="my-service", # String issuer, not a URL
|
|
)
|
|
|
|
# Create token with matching string issuer
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="my-service", # Same string issuer
|
|
)
|
|
|
|
access_token = await provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert access_token.client_id == "test-user"
|
|
|
|
async def test_string_issuer_mismatch_rejection(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test that mismatched string issuers are rejected."""
|
|
# Create provider with one string issuer
|
|
provider = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="my-service",
|
|
)
|
|
|
|
# Create token with different string issuer
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="other-service", # Different string issuer
|
|
)
|
|
|
|
access_token = await provider.load_access_token(token)
|
|
assert access_token is None
|
|
|
|
async def test_url_issuer_still_works(self, rsa_key_pair: RSAKeyPair):
|
|
"""Test that URL issuers still work after the fix."""
|
|
# Create provider with URL issuer
|
|
provider = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://my-auth-server.com", # URL issuer
|
|
)
|
|
|
|
# Create token with matching URL issuer
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://my-auth-server.com", # Same URL issuer
|
|
)
|
|
|
|
access_token = await provider.load_access_token(token)
|
|
assert access_token is not None
|
|
assert access_token.client_id == "test-user"
|
|
|
|
|
|
class TestFastMCPBearerAuth:
|
|
def test_bearer_auth(self):
|
|
mcp = FastMCP(
|
|
auth=JWTVerifier(issuer="https://test.example.com", public_key="abc")
|
|
)
|
|
assert isinstance(mcp.auth, JWTVerifier)
|
|
|
|
async def test_unauthorized_access(self, mcp_server_url: str):
|
|
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
|
async with Client(mcp_server_url) as client:
|
|
tools = await client.list_tools() # noqa: F841
|
|
assert isinstance(exc_info.value, httpx.HTTPStatusError)
|
|
assert exc_info.value.response.status_code == 401
|
|
assert "tools" not in locals()
|
|
|
|
async def test_authorized_access(self, mcp_server_url: str, bearer_token):
|
|
async with Client(mcp_server_url, auth=BearerAuth(bearer_token)) as client:
|
|
tools = await client.list_tools() # noqa: F841
|
|
assert tools
|
|
|
|
async def test_invalid_token_raises_401(self, mcp_server_url: str):
|
|
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
|
async with Client(mcp_server_url, auth=BearerAuth("invalid")) as client:
|
|
tools = await client.list_tools() # noqa: F841
|
|
assert isinstance(exc_info.value, httpx.HTTPStatusError)
|
|
assert exc_info.value.response.status_code == 401
|
|
assert "tools" not in locals()
|
|
|
|
async def test_expired_token(self, mcp_server_url: str, rsa_key_pair: RSAKeyPair):
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
expires_in_seconds=-3600,
|
|
)
|
|
|
|
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
|
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
|
|
tools = await client.list_tools() # noqa: F841
|
|
assert isinstance(exc_info.value, httpx.HTTPStatusError)
|
|
assert exc_info.value.response.status_code == 401
|
|
assert "tools" not in locals()
|
|
|
|
async def test_token_with_bad_signature(self, mcp_server_url: str):
|
|
rsa_key_pair = RSAKeyPair.generate()
|
|
token = rsa_key_pair.create_token()
|
|
|
|
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
|
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
|
|
tools = await client.list_tools() # noqa: F841
|
|
assert isinstance(exc_info.value, httpx.HTTPStatusError)
|
|
assert exc_info.value.response.status_code == 401
|
|
assert "tools" not in locals()
|
|
|
|
async def test_token_with_insufficient_scopes(self, rsa_key_pair: RSAKeyPair):
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
scopes=["read"],
|
|
)
|
|
|
|
server = create_mcp_server(
|
|
public_key=rsa_key_pair.public_key,
|
|
auth_kwargs=dict(required_scopes=["read", "write"]),
|
|
)
|
|
|
|
async with run_server_async(server, transport="http") as mcp_server_url:
|
|
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
|
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
|
|
tools = await client.list_tools() # noqa: F841
|
|
# JWTVerifier returns 401 when verify_token returns None (invalid token)
|
|
# This is correct behavior - when TokenVerifier.verify_token returns None,
|
|
# it indicates the token is invalid (not just insufficient permissions)
|
|
assert isinstance(exc_info.value, httpx.HTTPStatusError)
|
|
assert exc_info.value.response.status_code == 401
|
|
assert "tools" not in locals()
|
|
|
|
async def test_token_with_sufficient_scopes(self, rsa_key_pair: RSAKeyPair):
|
|
token = rsa_key_pair.create_token(
|
|
subject="test-user",
|
|
issuer="https://test.example.com",
|
|
audience="https://api.example.com",
|
|
scopes=["read", "write"],
|
|
)
|
|
|
|
server = create_mcp_server(
|
|
public_key=rsa_key_pair.public_key,
|
|
auth_kwargs=dict(required_scopes=["read", "write"]),
|
|
)
|
|
|
|
async with run_server_async(server, transport="http") as mcp_server_url:
|
|
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
|
|
tools = await client.list_tools()
|
|
assert tools
|
|
|
|
|
|
class TestJWTVerifierImport:
|
|
"""Test JWT token verifier can be imported and created."""
|
|
|
|
def test_jwt_verifier_requires_pyjwt(self):
|
|
"""Test that JWTVerifier raises helpful error without PyJWT."""
|
|
# Since PyJWT is likely installed in test environment, we'll just test construction
|
|
from fastmcp.server.auth.providers.jwt import JWTVerifier
|
|
|
|
# This should work if PyJWT is available
|
|
try:
|
|
verifier = JWTVerifier(public_key="dummy-key")
|
|
assert verifier.public_key == "dummy-key"
|
|
assert verifier.algorithm == "RS256"
|
|
except ImportError as e:
|
|
# If PyJWT not available, should get helpful error
|
|
assert "PyJWT is required" in str(e)
|