mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-17 02:59:11 +02:00
269c9c9f46* feat: add TokenCache utility and caching to GitHubTokenVerifier Extract the caching machinery from IntrospectionTokenVerifier into a shared TokenCache class in fastmcp.utilities.token_cache, then wire it into both IntrospectionTokenVerifier and GitHubTokenVerifier. * Remove dead constant, validate negative cache params * Fix overwrite eviction bug, skip cache on scope lookup failure
336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""Tests for PropelAuthProvider."""
|
|
|
|
from typing import cast
|
|
from unittest.mock import AsyncMock
|
|
|
|
import httpx
|
|
import pytest
|
|
from pydantic import SecretStr
|
|
|
|
from fastmcp import Client, FastMCP
|
|
from fastmcp.server.auth import AccessToken
|
|
from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
|
|
from fastmcp.server.auth.providers.propelauth import (
|
|
PropelAuthProvider,
|
|
PropelAuthTokenIntrospectionOverrides,
|
|
)
|
|
from fastmcp.utilities.tests import run_server_async
|
|
|
|
|
|
class TestPropelAuthProvider:
|
|
"""Test PropelAuth's auth provider."""
|
|
|
|
def test_init_with_only_required_params(self):
|
|
"""Test PropelAuthProvider initialization with only required params."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
)
|
|
|
|
# Verify the provider is configured correctly
|
|
assert len(provider.authorization_servers) == 1
|
|
assert (
|
|
str(provider.authorization_servers[0])
|
|
== "https://auth.example.com/oauth/2.1"
|
|
)
|
|
assert str(provider.base_url) == "https://example.com/"
|
|
|
|
# Verify token verifier is configured correctly
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert (
|
|
provider.token_verifier.introspection_url
|
|
== "https://auth.example.com/oauth/2.1/introspect"
|
|
)
|
|
assert provider.token_verifier.client_id == "client_id_123"
|
|
assert provider.token_verifier.client_secret == "client_secret_123"
|
|
|
|
def test_auth_url_trailing_slash_normalization(self):
|
|
"""Test that trailing slash on auth_url is stripped before building URLs."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com/",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert len(provider.authorization_servers) == 1
|
|
assert (
|
|
str(provider.authorization_servers[0])
|
|
== "https://auth.example.com/oauth/2.1"
|
|
)
|
|
assert (
|
|
provider.token_verifier.introspection_url
|
|
== "https://auth.example.com/oauth/2.1/introspect"
|
|
)
|
|
|
|
def test_required_scopes_passed_to_verifier(self):
|
|
"""Test that required_scopes are passed through to the token verifier."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
required_scopes=["read", "write"],
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert provider.token_verifier.required_scopes == ["read", "write"]
|
|
|
|
def test_introspection_client_secret_as_secret_str(self):
|
|
"""Test that SecretStr client_secret is unwrapped correctly."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret=SecretStr("my_secret"),
|
|
base_url="https://example.com",
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert provider.token_verifier.client_secret == "my_secret"
|
|
|
|
def test_authorization_servers_configuration(self):
|
|
"""Test that authorization_servers contains the correct PropelAuth URL."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.propelauth.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
)
|
|
|
|
assert len(provider.authorization_servers) == 1
|
|
assert (
|
|
str(provider.authorization_servers[0])
|
|
== "https://auth.propelauth.com/oauth/2.1"
|
|
)
|
|
|
|
def test_token_introspection_overrides_timeout(self):
|
|
"""Test that timeout_seconds override is passed to the verifier."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
token_introspection_overrides={"timeout_seconds": 30},
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert provider.token_verifier.timeout_seconds == 30
|
|
|
|
def test_token_introspection_overrides_cache(self):
|
|
"""Test that cache overrides are passed to the verifier."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
token_introspection_overrides={
|
|
"cache_ttl_seconds": 300,
|
|
"max_cache_size": 500,
|
|
},
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert provider.token_verifier._cache._ttl == 300
|
|
assert provider.token_verifier._cache._max_size == 500
|
|
|
|
def test_token_introspection_overrides_http_client(self):
|
|
"""Test that http_client override is passed to the verifier."""
|
|
client = httpx.AsyncClient()
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
token_introspection_overrides={"http_client": client},
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert provider.token_verifier._http_client is client
|
|
|
|
def test_token_introspection_overrides_ignores_unknown_keys(self):
|
|
"""Test that unknown override keys are silently ignored."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
# This won't typecheck without casting, since it shouldn't be allowed
|
|
token_introspection_overrides=cast(
|
|
PropelAuthTokenIntrospectionOverrides, {"unknown_key": "value"}
|
|
),
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert provider.token_verifier.timeout_seconds == 10
|
|
|
|
def test_token_introspection_overrides_ignores_disallowed_known_keys(self):
|
|
"""Test that known IntrospectionTokenVerifier keys not in the allow list are ignored."""
|
|
provider = PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
# This won't typecheck without casting, since it shouldn't be allowed
|
|
token_introspection_overrides=cast(
|
|
PropelAuthTokenIntrospectionOverrides, {"client_id": "sneaky_override"}
|
|
),
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
|
|
assert provider.token_verifier.client_id == "client_id_123"
|
|
|
|
|
|
class TestPropelAuthResourceChecking:
|
|
"""Test audience (aud) checking when resource is configured."""
|
|
|
|
def _make_provider(self, resource: str | None = None) -> PropelAuthProvider:
|
|
return PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="https://example.com",
|
|
resource=resource,
|
|
)
|
|
|
|
def _make_access_token(self, aud: str) -> AccessToken:
|
|
return AccessToken(
|
|
token="test-token",
|
|
client_id="client_id_123",
|
|
scopes=[],
|
|
claims={"active": True, "sub": "user-1", "aud": aud},
|
|
)
|
|
|
|
async def test_no_resource_skips_aud_check(self, monkeypatch: pytest.MonkeyPatch):
|
|
"""When resource is not configured, tokens are accepted without aud checking."""
|
|
provider = self._make_provider(resource=None)
|
|
token = self._make_access_token(aud="https://anything.example.com")
|
|
monkeypatch.setattr(
|
|
provider.token_verifier, "verify_token", AsyncMock(return_value=token)
|
|
)
|
|
|
|
result = await provider.verify_token("test-token")
|
|
assert result is token
|
|
|
|
async def test_aud_matches_resource(self, monkeypatch: pytest.MonkeyPatch):
|
|
"""Token is accepted when aud matches the configured resource."""
|
|
provider = self._make_provider(resource="https://api.example.com/mcp")
|
|
token = self._make_access_token(aud="https://api.example.com/mcp")
|
|
monkeypatch.setattr(
|
|
provider.token_verifier, "verify_token", AsyncMock(return_value=token)
|
|
)
|
|
|
|
result = await provider.verify_token("test-token")
|
|
assert result is token
|
|
|
|
async def test_aud_does_not_match_resource(self, monkeypatch: pytest.MonkeyPatch):
|
|
"""Token is rejected when aud doesn't match the configured resource."""
|
|
provider = self._make_provider(resource="https://api.example.com/mcp")
|
|
token = self._make_access_token(aud="https://other-server.example.com/mcp")
|
|
monkeypatch.setattr(
|
|
provider.token_verifier, "verify_token", AsyncMock(return_value=token)
|
|
)
|
|
|
|
result = await provider.verify_token("test-token")
|
|
assert result is None
|
|
|
|
async def test_inner_verifier_returns_none(self, monkeypatch: pytest.MonkeyPatch):
|
|
"""When the inner verifier rejects the token, None is returned without aud checking."""
|
|
provider = self._make_provider(resource="https://api.example.com/mcp")
|
|
monkeypatch.setattr(
|
|
provider.token_verifier, "verify_token", AsyncMock(return_value=None)
|
|
)
|
|
|
|
result = await provider.verify_token("test-token")
|
|
assert result is None
|
|
|
|
|
|
@pytest.fixture
|
|
async def mcp_server_url():
|
|
"""Start MCP server with PropelAuth authentication."""
|
|
mcp = FastMCP(
|
|
auth=PropelAuthProvider(
|
|
auth_url="https://auth.example.com",
|
|
introspection_client_id="client_id_123",
|
|
introspection_client_secret="client_secret_123",
|
|
base_url="http://localhost:4321",
|
|
)
|
|
)
|
|
|
|
@mcp.tool
|
|
def add(a: int, b: int) -> int:
|
|
return a + b
|
|
|
|
async with run_server_async(mcp, transport="http") as url:
|
|
yield url
|
|
|
|
|
|
class TestPropelAuthProviderIntegration:
|
|
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_metadata_route_forwards_propelauth_response(
|
|
self,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
mcp_server_url: str,
|
|
) -> None:
|
|
"""Ensure PropelAuth metadata route proxies upstream JSON."""
|
|
|
|
metadata_payload = {
|
|
"issuer": "https://auth.example.com",
|
|
"token_endpoint": "https://auth.example.com/oauth/2.1/token",
|
|
"authorization_endpoint": "https://auth.example.com/oauth/2.1/authorize",
|
|
}
|
|
|
|
class DummyResponse:
|
|
status_code = 200
|
|
|
|
def __init__(self, data: dict[str, str]):
|
|
self._data = data
|
|
|
|
def json(self):
|
|
return self._data
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
class DummyAsyncClient:
|
|
last_url: str | None = None
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
async def get(self, url: str):
|
|
DummyAsyncClient.last_url = url
|
|
return DummyResponse(metadata_payload)
|
|
|
|
real_httpx_client = httpx.AsyncClient
|
|
|
|
monkeypatch.setattr(
|
|
"fastmcp.server.auth.providers.propelauth.httpx.AsyncClient",
|
|
DummyAsyncClient,
|
|
)
|
|
|
|
base_url = mcp_server_url.rsplit("/mcp", 1)[0]
|
|
async with real_httpx_client() as client:
|
|
response = await client.get(
|
|
f"{base_url}/.well-known/oauth-authorization-server"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == metadata_payload
|
|
assert (
|
|
DummyAsyncClient.last_url
|
|
== "https://auth.example.com/.well-known/oauth-authorization-server/oauth/2.1"
|
|
)
|