mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-21 13:04:18 +02:00
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Tests for Scalekit OAuth provider."""
|
|
|
|
import httpx2
|
|
import pytest
|
|
from mcp import MCPError
|
|
|
|
from fastmcp import Client, FastMCP
|
|
from fastmcp.client.transports import StreamableHttpTransport
|
|
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
|
|
from fastmcp.server.auth.providers.scalekit import ScalekitProvider
|
|
from fastmcp.utilities.tests import HeadlessOAuth, run_server_async
|
|
|
|
|
|
class TestScalekitProvider:
|
|
"""Test Scalekit OAuth provider functionality."""
|
|
|
|
def test_init_with_explicit_params(self):
|
|
"""Test ScalekitProvider initialization with explicit parameters."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://my-env.scalekit.com",
|
|
resource_id="sk_resource_456",
|
|
base_url="https://myserver.com/",
|
|
required_scopes=["read"],
|
|
)
|
|
|
|
assert provider.environment_url == "https://my-env.scalekit.com"
|
|
assert provider.resource_id == "sk_resource_456"
|
|
assert str(provider.base_url) == "https://myserver.com/"
|
|
assert provider.required_scopes == ["read"]
|
|
|
|
def test_init_with_mcp_url_only(self):
|
|
"""Allow legacy mcp_url parameter as base_url."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://legacy.scalekit.com",
|
|
resource_id="sk_resource_legacy",
|
|
mcp_url="https://legacy-app.com/",
|
|
)
|
|
|
|
assert str(provider.base_url) == "https://legacy-app.com/"
|
|
|
|
def test_init_prefers_base_url_over_mcp_url(self):
|
|
"""mcp_url should take precedence over base_url when both provided."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://my-env.scalekit.com",
|
|
resource_id="sk_resource_456",
|
|
base_url="https://preferred-base.com/",
|
|
mcp_url="https://unused-base.com/",
|
|
)
|
|
|
|
assert str(provider.base_url) == "https://preferred-base.com/"
|
|
|
|
def test_environment_variable_loading(self):
|
|
"""Test that environment variables are loaded correctly."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://test-env.scalekit.com",
|
|
resource_id="sk_resource_test_456",
|
|
base_url="http://test-server.com",
|
|
)
|
|
|
|
assert provider.environment_url == "https://test-env.scalekit.com"
|
|
assert provider.resource_id == "sk_resource_test_456"
|
|
assert str(provider.base_url) == "http://test-server.com/"
|
|
|
|
def test_accepts_client_id_argument(self):
|
|
"""client_id parameter should be accepted but ignored."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://my-env.scalekit.com",
|
|
resource_id="sk_resource_456",
|
|
base_url="https://myserver.com/",
|
|
client_id="client_123",
|
|
)
|
|
|
|
assert str(provider.base_url) == "https://myserver.com/"
|
|
|
|
def test_url_trailing_slash_handling(self):
|
|
"""Test that URLs handle trailing slashes correctly."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://my-env.scalekit.com/",
|
|
resource_id="sk_resource_456",
|
|
base_url="https://myserver.com/",
|
|
)
|
|
|
|
assert provider.environment_url == "https://my-env.scalekit.com"
|
|
assert str(provider.base_url) == "https://myserver.com/"
|
|
|
|
def test_jwt_verifier_configured_correctly(self):
|
|
"""Test that JWT verifier is configured correctly."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://my-env.scalekit.com",
|
|
resource_id="sk_resource_456",
|
|
base_url="https://myserver.com/",
|
|
)
|
|
|
|
# Check that JWT verifier uses the correct endpoints. Both the bare
|
|
# environment URL and the resource-scoped issuer are accepted so tokens
|
|
# from before and after Scalekit's issuer migration validate.
|
|
assert isinstance(provider.token_verifier, JWTVerifier)
|
|
assert provider.token_verifier.jwks_uri == "https://my-env.scalekit.com/keys"
|
|
assert provider.token_verifier.issuer == [
|
|
"https://my-env.scalekit.com",
|
|
"https://my-env.scalekit.com/resources/sk_resource_456",
|
|
]
|
|
assert provider.token_verifier.audience == "sk_resource_456"
|
|
|
|
def test_required_scopes_hooks_into_verifier(self):
|
|
"""Token verifier should enforce required scopes when provided."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://my-env.scalekit.com",
|
|
resource_id="sk_resource_456",
|
|
base_url="https://myserver.com/",
|
|
required_scopes=["read"],
|
|
)
|
|
|
|
assert isinstance(provider.token_verifier, JWTVerifier)
|
|
assert provider.token_verifier.required_scopes == ["read"]
|
|
|
|
def test_authorization_servers_configuration(self):
|
|
"""Test that authorization servers are configured correctly."""
|
|
provider = ScalekitProvider(
|
|
environment_url="https://my-env.scalekit.com",
|
|
resource_id="sk_resource_456",
|
|
base_url="https://myserver.com/",
|
|
)
|
|
|
|
assert len(provider.authorization_servers) == 1
|
|
assert (
|
|
str(provider.authorization_servers[0])
|
|
== "https://my-env.scalekit.com/resources/sk_resource_456"
|
|
)
|
|
|
|
|
|
class TestScalekitIssuerMigration:
|
|
"""Scalekit is migrating the `iss` claim from the bare environment URL to a
|
|
resource-scoped issuer. Tokens minted before and after the migration must
|
|
both validate against the same provider.
|
|
"""
|
|
|
|
ENV_URL = "https://my-env.scalekit.com"
|
|
RESOURCE_ID = "sk_resource_456"
|
|
|
|
@pytest.fixture
|
|
def key_pair(self) -> RSAKeyPair:
|
|
return RSAKeyPair.generate()
|
|
|
|
def _provider(self, key_pair: RSAKeyPair) -> ScalekitProvider:
|
|
provider = ScalekitProvider(
|
|
environment_url=self.ENV_URL,
|
|
resource_id=self.RESOURCE_ID,
|
|
base_url="https://myserver.com/",
|
|
)
|
|
# Verify against the test key instead of Scalekit's live JWKS endpoint.
|
|
assert isinstance(provider.token_verifier, JWTVerifier)
|
|
provider.token_verifier.public_key = key_pair.public_key
|
|
return provider
|
|
|
|
async def test_pre_migration_issuer_accepted(self, key_pair: RSAKeyPair):
|
|
"""The bare environment URL issuer (pre-migration) validates."""
|
|
provider = self._provider(key_pair)
|
|
token = key_pair.create_token(
|
|
issuer=self.ENV_URL,
|
|
audience=self.RESOURCE_ID,
|
|
)
|
|
|
|
assert await provider.token_verifier.verify_token(token) is not None
|
|
|
|
async def test_post_migration_issuer_accepted(self, key_pair: RSAKeyPair):
|
|
"""The resource-scoped issuer (post-migration) validates."""
|
|
provider = self._provider(key_pair)
|
|
token = key_pair.create_token(
|
|
issuer=f"{self.ENV_URL}/resources/{self.RESOURCE_ID}",
|
|
audience=self.RESOURCE_ID,
|
|
)
|
|
|
|
assert await provider.token_verifier.verify_token(token) is not None
|
|
|
|
async def test_unknown_issuer_rejected(self, key_pair: RSAKeyPair):
|
|
"""An issuer outside the accepted set is still rejected."""
|
|
provider = self._provider(key_pair)
|
|
token = key_pair.create_token(
|
|
issuer="https://evil.example.com",
|
|
audience=self.RESOURCE_ID,
|
|
)
|
|
|
|
assert await provider.token_verifier.verify_token(token) is None
|
|
|
|
|
|
@pytest.fixture
|
|
async def mcp_server_url():
|
|
"""Start Scalekit server."""
|
|
mcp = FastMCP(
|
|
auth=ScalekitProvider(
|
|
environment_url="https://test-env.scalekit.com",
|
|
resource_id="sk_resource_test_456",
|
|
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
|
|
|
|
|
|
@pytest.fixture
|
|
def client_with_headless_oauth(mcp_server_url: str) -> Client:
|
|
"""Client with headless OAuth that bypasses browser interaction."""
|
|
return Client(
|
|
transport=StreamableHttpTransport(mcp_server_url),
|
|
auth=HeadlessOAuth(mcp_url=mcp_server_url),
|
|
)
|
|
|
|
|
|
class TestScalekitProviderIntegration:
|
|
async def test_unauthorized_access(self, mcp_server_url: str):
|
|
# SDK v2 surfaces the server's 401 as a generic MCPError at the client
|
|
# boundary rather than re-raising httpx2.HTTPStatusError.
|
|
with pytest.raises(MCPError):
|
|
async with Client(mcp_server_url) as client:
|
|
tools = await client.list_tools() # noqa: F841
|
|
assert "tools" not in locals()
|
|
|
|
async def test_metadata_route_forwards_scalekit_response(
|
|
self,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
mcp_server_url: str,
|
|
) -> None:
|
|
"""Ensure Scalekit metadata route proxies upstream JSON."""
|
|
|
|
metadata_payload = {
|
|
"issuer": "https://test-env.scalekit.com",
|
|
"token_endpoint": "https://test-env.scalekit.com/token",
|
|
"authorization_endpoint": "https://test-env.scalekit.com/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 = httpx2.AsyncClient
|
|
|
|
monkeypatch.setattr(
|
|
"fastmcp.server.auth.providers.scalekit.httpx2.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://test-env.scalekit.com/.well-known/oauth-authorization-server/resources/sk_resource_test_456"
|
|
)
|