mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
966 lines
39 KiB
Python
966 lines
39 KiB
Python
"""Tests for enhanced OAuth error responses.
|
|
|
|
This test suite covers:
|
|
1. Enhanced authorization handler (HTML and JSON error pages)
|
|
2. Enhanced middleware (better error messages)
|
|
3. Content negotiation
|
|
4. Server branding in error pages
|
|
"""
|
|
|
|
import asyncio
|
|
from urllib.parse import parse_qs, quote, urlparse
|
|
|
|
import pytest
|
|
from key_value.aio.stores.memory import MemoryStore
|
|
from mcp.shared.auth import OAuthClientInformationFull
|
|
from pydantic import AnyHttpUrl, AnyUrl
|
|
from starlette.applications import Starlette
|
|
from starlette.testclient import TestClient
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import RemoteAuthProvider, TokenVerifier
|
|
from fastmcp.server.auth.auth import AccessToken
|
|
from fastmcp.server.auth.oauth_proxy import OAuthProxy
|
|
from fastmcp.server.auth.providers.jwt import JWTVerifier
|
|
from fastmcp.server.http import create_streamable_http_app
|
|
|
|
|
|
class _UnderScopedTokenVerifier(TokenVerifier):
|
|
def __init__(self, required_scopes: list[str]):
|
|
super().__init__(required_scopes=required_scopes)
|
|
|
|
async def verify_token(self, token: str) -> AccessToken:
|
|
return AccessToken(token=token, client_id="test-client", scopes=["other"])
|
|
|
|
|
|
class _UnderScopedOAuthProxy(OAuthProxy):
|
|
async def verify_token(self, token: str) -> AccessToken:
|
|
return AccessToken(token=token, client_id="test-client", scopes=["other"])
|
|
|
|
|
|
class _DirectClientRedirectOAuthProxy(OAuthProxy):
|
|
"""Proxy whose `authorize()` bypasses consent/upstream entirely and
|
|
redirects straight back to the client with a `code` — the pattern used
|
|
by providers (or tests) that short-circuit the standard
|
|
consent -> upstream IdP -> callback flow. OAuthProxy's own `authorize()`
|
|
never does this itself, but a subclass legitimately can, and
|
|
`AuthorizationHandler.handle()` must still attach `iss` to whatever
|
|
redirect comes back.
|
|
|
|
Appends its own `code`/`state` with `&` rather than an unconditional
|
|
`?` so this still produces a well-formed URL when `redirect_uri` is a
|
|
registered redirect that already carries its own query string (e.g. a
|
|
client-supplied `iss`)."""
|
|
|
|
async def authorize(self, client, params): # type: ignore[override]
|
|
separator = "&" if "?" in str(params.redirect_uri) else "?"
|
|
return (
|
|
f"{params.redirect_uri}{separator}code=test-auth-code&state={params.state}"
|
|
)
|
|
|
|
|
|
class _DirectClientRedirectWithIssOAuthProxy(OAuthProxy):
|
|
"""Like `_DirectClientRedirectOAuthProxy`, but the provider's
|
|
`authorize()` override already put its own `iss` on the redirect —
|
|
simulating a provider that is itself RFC 9207-aware (or, when
|
|
`redirect_iss` doesn't match this server's issuer, a provider bug).
|
|
`response_kind` selects whether the redirect looks like a success
|
|
(`code`) or error (`error`) response; `AuthorizationHandler.handle()`
|
|
must not duplicate `iss` on either."""
|
|
|
|
def __init__(
|
|
self,
|
|
*args,
|
|
redirect_iss: str,
|
|
response_kind: str = "code",
|
|
**kwargs,
|
|
):
|
|
super().__init__(*args, **kwargs)
|
|
self._redirect_iss = redirect_iss
|
|
self._response_kind = response_kind
|
|
|
|
async def authorize(self, client, params): # type: ignore[override]
|
|
payload = (
|
|
f"code=test-auth-code&state={params.state}"
|
|
if self._response_kind == "code"
|
|
else f"error=access_denied&state={params.state}"
|
|
)
|
|
iss = quote(self._redirect_iss, safe="")
|
|
return f"{params.redirect_uri}?{payload}&iss={iss}"
|
|
|
|
|
|
class TestEnhancedAuthorizationHandler:
|
|
"""Tests for enhanced authorization handler error responses."""
|
|
|
|
@pytest.fixture
|
|
def oauth_proxy(self, rsa_key_pair):
|
|
"""Create OAuth proxy for testing."""
|
|
return OAuthProxy(
|
|
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
|
|
upstream_token_endpoint="https://github.com/login/oauth/access_token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
),
|
|
base_url="https://myserver.com",
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
)
|
|
|
|
def test_unregistered_client_returns_html_for_browser(self, oauth_proxy):
|
|
"""Test that unregistered client returns styled HTML for browser requests."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "unregistered-client-id",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
},
|
|
headers={"Accept": "text/html"},
|
|
)
|
|
|
|
# Should return 400 with HTML content
|
|
assert response.status_code == 400
|
|
assert "text/html" in response.headers["content-type"]
|
|
|
|
# HTML should contain error message
|
|
html = response.text
|
|
assert "Client Not Registered" in html
|
|
assert "unregistered-client-id" in html
|
|
assert "To fix this" in html
|
|
assert "Close this browser window" in html
|
|
assert "Clear authentication tokens" in html
|
|
|
|
# Should have Link header for registration endpoint
|
|
assert "Link" in response.headers
|
|
assert "/register" in response.headers["Link"]
|
|
|
|
def test_unregistered_client_returns_json_for_api(self, oauth_proxy):
|
|
"""Test that unregistered client returns enhanced JSON for API clients."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "unregistered-client-id",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
},
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
|
|
# Should return 400 with JSON content
|
|
assert response.status_code == 400
|
|
assert "application/json" in response.headers["content-type"]
|
|
|
|
# JSON should have enhanced error response
|
|
data = response.json()
|
|
assert data["error"] == "invalid_request"
|
|
assert "unregistered-client-id" in data["error_description"]
|
|
assert data["state"] == "test-state"
|
|
|
|
# Should include registration endpoint hints
|
|
assert "registration_endpoint" in data
|
|
assert data["registration_endpoint"] == "https://myserver.com/register"
|
|
assert "authorization_server_metadata" in data
|
|
|
|
# Should have Link header
|
|
assert "Link" in response.headers
|
|
assert "/register" in response.headers["Link"]
|
|
|
|
def test_successful_authorization_not_enhanced(self, oauth_proxy):
|
|
"""Test that successful authorizations are not modified by enhancement."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
# Register a valid client first
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="valid-client",
|
|
client_secret="valid-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
)
|
|
|
|
# Need to register synchronously
|
|
asyncio.run(oauth_proxy.register_client(client_info))
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "valid-client",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
},
|
|
headers={"Accept": "text/html"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
# Should redirect to consent page (302), not return error
|
|
assert response.status_code == 302
|
|
assert "/consent" in response.headers["location"]
|
|
|
|
def test_redirect_error_includes_proxy_issuer(self, oauth_proxy):
|
|
"""Authorization error redirects should include RFC 9207 issuer."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="valid-client",
|
|
client_secret="valid-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
scope="read",
|
|
)
|
|
|
|
asyncio.run(oauth_proxy.register_client(client_info))
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "valid-client",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
"scope": "admin",
|
|
},
|
|
headers={"Accept": "text/html"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
query_params = parse_qs(urlparse(response.headers["location"]).query)
|
|
assert query_params["error"] == ["invalid_scope"]
|
|
assert query_params["state"] == ["test-state"]
|
|
assert query_params["iss"] == ["https://myserver.com/"]
|
|
|
|
def test_redirect_error_matches_path_base_url_metadata_issuer(self, rsa_key_pair):
|
|
"""Authorization error redirects should match the metadata issuer exactly."""
|
|
oauth_proxy = OAuthProxy(
|
|
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
|
|
upstream_token_endpoint="https://github.com/login/oauth/access_token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
),
|
|
base_url="https://proxy.example.com/oauth",
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
)
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="valid-client",
|
|
client_secret="valid-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
scope="read",
|
|
)
|
|
|
|
asyncio.run(oauth_proxy.register_client(client_info))
|
|
|
|
with TestClient(app) as client:
|
|
metadata_response = client.get("/.well-known/oauth-authorization-server")
|
|
metadata = metadata_response.json()
|
|
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "valid-client",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
"scope": "admin",
|
|
},
|
|
headers={"Accept": "text/html"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert metadata["issuer"] == "https://proxy.example.com/oauth"
|
|
assert response.status_code == 302
|
|
query_params = parse_qs(urlparse(response.headers["location"]).query)
|
|
assert query_params["error"] == ["invalid_scope"]
|
|
assert query_params["state"] == ["test-state"]
|
|
assert query_params["iss"] == [metadata["issuer"]]
|
|
|
|
def test_success_redirect_from_authorize_override_includes_issuer(
|
|
self, rsa_key_pair
|
|
):
|
|
"""RFC 9207 regression: a `code` redirect returned directly by
|
|
`authorize()` (bypassing consent/upstream) must carry `iss` too, not
|
|
just `error` redirects.
|
|
|
|
`AuthorizationHandler.handle()` previously only attached `iss` when
|
|
the SDK's redirect contained an `error` parameter. The base
|
|
`OAuthProxy.authorize()` never redirects straight to the client, so
|
|
this gap was invisible until a provider override (or a test mock,
|
|
like the GitHub provider integration test) returned the client
|
|
redirect directly — at which point the server was advertising
|
|
`authorization_response_iss_parameter_supported: true` while
|
|
silently breaking RFC 9207-aware clients on this path.
|
|
"""
|
|
oauth_proxy = _DirectClientRedirectOAuthProxy(
|
|
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
|
|
upstream_token_endpoint="https://github.com/login/oauth/access_token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
),
|
|
base_url="https://myserver.com",
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
)
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="valid-client",
|
|
client_secret="valid-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
scope="read",
|
|
)
|
|
asyncio.run(oauth_proxy.register_client(client_info))
|
|
|
|
with TestClient(app) as client:
|
|
metadata = client.get("/.well-known/oauth-authorization-server").json()
|
|
assert metadata["authorization_response_iss_parameter_supported"] is True
|
|
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "valid-client",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
query_params = parse_qs(urlparse(response.headers["location"]).query)
|
|
assert query_params["code"] == ["test-auth-code"]
|
|
assert query_params["state"] == ["test-state"]
|
|
assert query_params["iss"] == [metadata["issuer"]]
|
|
|
|
def test_success_redirect_does_not_duplicate_iss_already_in_redirect_uri(
|
|
self, rsa_key_pair
|
|
):
|
|
"""RFC 9207 P2 regression: a registered redirect_uri may already
|
|
carry its own `iss` query parameter — distinct from the provider
|
|
adding one itself (covered by
|
|
`test_success_redirect_with_matching_iss_not_duplicated` below).
|
|
`AuthorizationHandler.handle()` must still land on exactly one
|
|
`iss` (the canonical value), with every other query byte on the
|
|
registered URI preserved untouched.
|
|
"""
|
|
oauth_proxy = _DirectClientRedirectOAuthProxy(
|
|
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
|
|
upstream_token_endpoint="https://github.com/login/oauth/access_token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
),
|
|
base_url="https://myserver.com",
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
)
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
client_redirect_uri = (
|
|
"http://localhost:12345/callback?iss=tenant&flag&sig=%FF%FE"
|
|
)
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="valid-client",
|
|
client_secret="valid-secret",
|
|
redirect_uris=[AnyUrl(client_redirect_uri)],
|
|
scope="read",
|
|
)
|
|
asyncio.run(oauth_proxy.register_client(client_info))
|
|
|
|
with TestClient(app) as client:
|
|
metadata = client.get("/.well-known/oauth-authorization-server").json()
|
|
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "valid-client",
|
|
"redirect_uri": client_redirect_uri,
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
location = response.headers["location"]
|
|
query = urlparse(location).query
|
|
query_params = parse_qs(query)
|
|
assert query_params["code"] == ["test-auth-code"]
|
|
assert query_params["state"] == ["test-state"]
|
|
# Exactly one `iss`, corrected to the canonical value -- a
|
|
# duplicate would make this list have length 2.
|
|
assert query_params["iss"] == [metadata["issuer"]]
|
|
# Other query bytes from the registered redirect_uri survive
|
|
# byte-for-byte.
|
|
assert "flag" in query
|
|
assert "sig=%FF%FE" in query
|
|
|
|
def test_success_redirect_with_matching_iss_not_duplicated(self, rsa_key_pair):
|
|
"""If a provider's `authorize()` override already stamped the
|
|
correct `iss` on its redirect, `handle()` must not append a second
|
|
one — RFC 6749 §3.1 forbids a response parameter appearing twice.
|
|
"""
|
|
oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
|
|
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
|
|
upstream_token_endpoint="https://github.com/login/oauth/access_token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
),
|
|
base_url="https://myserver.com",
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
redirect_iss="https://myserver.com/",
|
|
)
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="valid-client",
|
|
client_secret="valid-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
scope="read",
|
|
)
|
|
asyncio.run(oauth_proxy.register_client(client_info))
|
|
|
|
with TestClient(app) as client:
|
|
metadata = client.get("/.well-known/oauth-authorization-server").json()
|
|
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "valid-client",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
query_params = parse_qs(urlparse(response.headers["location"]).query)
|
|
# Exactly one `iss` (a duplicate would make this list have length 2).
|
|
assert query_params["iss"] == [metadata["issuer"]]
|
|
|
|
def test_success_redirect_with_mismatched_iss_is_corrected(self, rsa_key_pair):
|
|
"""A provider's `authorize()` override can put an `iss` on its
|
|
redirect that doesn't match what this server advertises in its own
|
|
discovery document (`self._issuer`). An RFC 9207 client validates
|
|
`iss` against that document, so the mismatched value is already
|
|
unusable to a spec-compliant client. `handle()` corrects it to the
|
|
canonical value rather than leaving the broken value in place or
|
|
appending a second `iss` (which RFC 6749 §3.1 forbids outright).
|
|
|
|
This is a deliberate policy choice, not the only defensible one —
|
|
see the comment in `AuthorizationHandler.handle()` for the
|
|
reasoning, and update this test if that policy changes.
|
|
"""
|
|
oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
|
|
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
|
|
upstream_token_endpoint="https://github.com/login/oauth/access_token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
),
|
|
base_url="https://myserver.com",
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
redirect_iss="https://wrong-issuer.example.com/",
|
|
)
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="valid-client",
|
|
client_secret="valid-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
scope="read",
|
|
)
|
|
asyncio.run(oauth_proxy.register_client(client_info))
|
|
|
|
with TestClient(app) as client:
|
|
metadata = client.get("/.well-known/oauth-authorization-server").json()
|
|
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "valid-client",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
query_params = parse_qs(urlparse(response.headers["location"]).query)
|
|
# Exactly one `iss`, corrected to the canonical value rather than
|
|
# left mismatched or duplicated.
|
|
assert query_params["iss"] == [metadata["issuer"]]
|
|
assert query_params["iss"] != ["https://wrong-issuer.example.com/"]
|
|
|
|
def test_error_redirect_with_existing_iss_not_duplicated(self, rsa_key_pair):
|
|
"""The duplication guard applies to error redirects too, not just
|
|
success ones — a provider override can construct an `error`
|
|
redirect that already carries `iss`.
|
|
"""
|
|
oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
|
|
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
|
|
upstream_token_endpoint="https://github.com/login/oauth/access_token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
),
|
|
base_url="https://myserver.com",
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
redirect_iss="https://myserver.com/",
|
|
response_kind="error",
|
|
)
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
client_info = OAuthClientInformationFull(
|
|
client_id="valid-client",
|
|
client_secret="valid-secret",
|
|
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
|
scope="read",
|
|
)
|
|
asyncio.run(oauth_proxy.register_client(client_info))
|
|
|
|
with TestClient(app) as client:
|
|
metadata = client.get("/.well-known/oauth-authorization-server").json()
|
|
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "valid-client",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
"state": "test-state",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
query_params = parse_qs(urlparse(response.headers["location"]).query)
|
|
assert query_params["error"] == ["access_denied"]
|
|
assert query_params["iss"] == [metadata["issuer"]]
|
|
|
|
def test_html_error_includes_server_branding(self, oauth_proxy):
|
|
"""Test that HTML error page includes server branding from FastMCP instance."""
|
|
from mcp_types import Icon
|
|
|
|
# Create FastMCP server with custom branding
|
|
mcp = FastMCP(
|
|
"My Custom Server",
|
|
icons=[Icon(src="https://example.com/icon.png", mime_type="image/png")],
|
|
)
|
|
|
|
# Create app with OAuth routes
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
# Attach FastMCP instance to app state (same as done in http.py)
|
|
app.state.fastmcp_server = mcp
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "unregistered-client-id",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
},
|
|
headers={"Accept": "text/html"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
html = response.text
|
|
|
|
# Should include custom server icon
|
|
assert "https://example.com/icon.png" in html
|
|
|
|
|
|
class TestEnhancedRequireAuthMiddleware:
|
|
"""Tests for enhanced authentication middleware error messages."""
|
|
|
|
@staticmethod
|
|
def create_scoped_app(
|
|
required_scopes: list[str],
|
|
scopes_supported: list[str],
|
|
challenge_scopes: list[str] | None = None,
|
|
) -> Starlette:
|
|
auth = RemoteAuthProvider(
|
|
token_verifier=_UnderScopedTokenVerifier(required_scopes),
|
|
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
|
base_url="http://localhost:8000",
|
|
scopes_supported=scopes_supported,
|
|
challenge_scopes=challenge_scopes,
|
|
)
|
|
return FastMCP("Test Server", auth=auth).http_app()
|
|
|
|
@staticmethod
|
|
def create_oauth_app() -> Starlette:
|
|
from key_value.aio.stores.memory import MemoryStore
|
|
|
|
auth = _UnderScopedOAuthProxy(
|
|
upstream_authorization_endpoint="https://auth.example.com/authorize",
|
|
upstream_token_endpoint="https://auth.example.com/token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=_UnderScopedTokenVerifier(["openid"]),
|
|
base_url="http://localhost:8000",
|
|
valid_scopes=["openid", "email", "calendar"],
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
)
|
|
return FastMCP("Test Server", auth=auth).http_app()
|
|
|
|
@pytest.fixture
|
|
def jwt_verifier(self, rsa_key_pair):
|
|
"""Create JWT verifier for testing."""
|
|
return JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
)
|
|
|
|
def test_missing_auth_no_error_attribute(self, jwt_verifier):
|
|
"""Test that missing auth returns 401 without error attribute (RFC 6750 §3.1)."""
|
|
server = FastMCP("Test Server")
|
|
|
|
@server.tool
|
|
def test_tool() -> str:
|
|
return "test"
|
|
|
|
app = create_streamable_http_app(
|
|
server=server,
|
|
streamable_http_path="/mcp",
|
|
auth=jwt_verifier,
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
# Request without Authorization header
|
|
response = client.post("/mcp")
|
|
|
|
assert response.status_code == 401
|
|
assert "www-authenticate" in response.headers
|
|
|
|
# Per RFC 6750 §3.1: no error attribute when auth is missing
|
|
www_auth = response.headers["www-authenticate"]
|
|
assert "error=" not in www_auth
|
|
assert response.content == b""
|
|
|
|
def test_missing_auth_challenge_includes_supported_scopes(self):
|
|
app = self.create_scoped_app(
|
|
required_scopes=["read"],
|
|
scopes_supported=["api://client-id/read"],
|
|
challenge_scopes=["api://client-id/read"],
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post("/mcp")
|
|
|
|
assert response.status_code == 401
|
|
assert response.headers["www-authenticate"] == (
|
|
'Bearer scope="api://client-id/read", '
|
|
'resource_metadata="http://localhost:8000/'
|
|
'.well-known/oauth-protected-resource/mcp"'
|
|
)
|
|
|
|
def test_insufficient_scope_challenge_includes_supported_scopes(self):
|
|
app = self.create_scoped_app(
|
|
required_scopes=["read"],
|
|
scopes_supported=["api://client-id/read"],
|
|
challenge_scopes=["api://client-id/read"],
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post("/mcp", headers={"Authorization": "Bearer narrow"})
|
|
|
|
assert response.status_code == 403
|
|
assert response.headers["www-authenticate"] == (
|
|
'Bearer error="insufficient_scope", '
|
|
'error_description="Required scope: read", '
|
|
'scope="api://client-id/read", '
|
|
'resource_metadata="http://localhost:8000/'
|
|
'.well-known/oauth-protected-resource/mcp"'
|
|
)
|
|
|
|
def test_missing_auth_challenge_uses_required_scope_with_empty_catalog(self):
|
|
app = self.create_scoped_app(required_scopes=["read"], scopes_supported=[])
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post("/mcp")
|
|
|
|
assert response.status_code == 401
|
|
assert response.headers["www-authenticate"] == (
|
|
'Bearer scope="read", resource_metadata="http://localhost:8000/'
|
|
'.well-known/oauth-protected-resource/mcp"'
|
|
)
|
|
|
|
def test_remote_missing_auth_challenge_excludes_optional_catalog_scopes(self):
|
|
app = self.create_scoped_app(
|
|
required_scopes=["read"],
|
|
scopes_supported=["read", "admin"],
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post("/mcp")
|
|
metadata = client.get("/.well-known/oauth-protected-resource/mcp").json()
|
|
|
|
assert response.status_code == 401
|
|
assert 'scope="read"' in response.headers["www-authenticate"]
|
|
assert "admin" not in response.headers["www-authenticate"]
|
|
assert metadata["scopes_supported"] == ["read", "admin"]
|
|
|
|
def test_remote_insufficient_scope_challenge_excludes_optional_catalog_scopes(
|
|
self,
|
|
):
|
|
app = self.create_scoped_app(
|
|
required_scopes=["read"],
|
|
scopes_supported=["read", "admin"],
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post("/mcp", headers={"Authorization": "Bearer narrow"})
|
|
|
|
assert response.status_code == 403
|
|
assert 'scope="read"' in response.headers["www-authenticate"]
|
|
assert "admin" not in response.headers["www-authenticate"]
|
|
|
|
def test_oauth_missing_auth_challenge_excludes_optional_scopes(self):
|
|
app = self.create_oauth_app()
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post("/mcp")
|
|
metadata = client.get("/.well-known/oauth-protected-resource/mcp").json()
|
|
|
|
assert response.status_code == 401
|
|
assert 'scope="openid"' in response.headers["www-authenticate"]
|
|
assert "email" not in response.headers["www-authenticate"]
|
|
assert "calendar" not in response.headers["www-authenticate"]
|
|
assert metadata["scopes_supported"] == ["openid", "email", "calendar"]
|
|
|
|
def test_oauth_insufficient_scope_challenge_excludes_optional_scopes(self):
|
|
app = self.create_oauth_app()
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post("/mcp", headers={"Authorization": "Bearer narrow"})
|
|
|
|
assert response.status_code == 403
|
|
assert 'scope="openid"' in response.headers["www-authenticate"]
|
|
assert "email" not in response.headers["www-authenticate"]
|
|
assert "calendar" not in response.headers["www-authenticate"]
|
|
|
|
def test_invalid_token_enhanced_error_message(self, jwt_verifier):
|
|
"""Test that invalid_token errors have enhanced error messages."""
|
|
server = FastMCP("Test Server")
|
|
|
|
@server.tool
|
|
def test_tool() -> str:
|
|
return "test"
|
|
|
|
app = create_streamable_http_app(
|
|
server=server,
|
|
streamable_http_path="/mcp",
|
|
auth=jwt_verifier,
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
# Request WITH an invalid Authorization header
|
|
response = client.post(
|
|
"/mcp", headers={"Authorization": "Bearer invalid-token"}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert "www-authenticate" in response.headers
|
|
|
|
# Check enhanced error message
|
|
data = response.json()
|
|
assert data["error"] == "invalid_token"
|
|
# Should have enhanced description with resolution steps
|
|
assert "clear authentication tokens" in data["error_description"]
|
|
assert "automatically re-register" in data["error_description"]
|
|
|
|
def test_invalid_token_www_authenticate_header_format(self, jwt_verifier):
|
|
"""Test that invalid token WWW-Authenticate header includes error attribute."""
|
|
server = FastMCP("Test Server")
|
|
app = create_streamable_http_app(
|
|
server=server,
|
|
streamable_http_path="/mcp",
|
|
auth=jwt_verifier,
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
# Request WITH an invalid token
|
|
response = client.post(
|
|
"/mcp", headers={"Authorization": "Bearer invalid-token"}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
www_auth = response.headers["www-authenticate"]
|
|
|
|
# Should follow Bearer challenge format with error
|
|
assert www_auth.startswith("Bearer ")
|
|
assert 'error="invalid_token"' in www_auth
|
|
assert "error_description=" in www_auth
|
|
|
|
def test_insufficient_scope_not_enhanced(self, rsa_key_pair):
|
|
"""Test that insufficient_scope errors are not modified."""
|
|
# Create a valid token with wrong scopes
|
|
jwt_verifier = JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
)
|
|
|
|
server = FastMCP("Test Server")
|
|
|
|
@server.tool
|
|
def test_tool() -> str:
|
|
return "test"
|
|
|
|
app = create_streamable_http_app(
|
|
server=server,
|
|
streamable_http_path="/mcp",
|
|
auth=jwt_verifier,
|
|
)
|
|
|
|
# Note: Testing insufficient_scope would require mocking the verifier
|
|
# to return a token with wrong scopes. For now, we verify the middleware
|
|
# is properly in place by checking it rejects unauthenticated requests.
|
|
with TestClient(app) as client:
|
|
response = client.post("/mcp")
|
|
# Without a valid token, we get invalid_token
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestContentNegotiation:
|
|
"""Tests for content negotiation in error responses."""
|
|
|
|
@pytest.fixture
|
|
def oauth_proxy(self, rsa_key_pair):
|
|
"""Create OAuth proxy for testing."""
|
|
return OAuthProxy(
|
|
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
|
|
upstream_token_endpoint="https://github.com/login/oauth/access_token",
|
|
upstream_client_id="test-client-id",
|
|
upstream_client_secret="test-client-secret",
|
|
token_verifier=JWTVerifier(
|
|
public_key=rsa_key_pair.public_key,
|
|
issuer="https://test.com",
|
|
audience="https://test.com",
|
|
base_url="https://test.com",
|
|
),
|
|
base_url="https://myserver.com",
|
|
jwt_signing_key="test-secret",
|
|
client_storage=MemoryStore(),
|
|
)
|
|
|
|
def test_html_preferred_when_both_accepted(self, oauth_proxy):
|
|
"""Test that HTML is preferred when both text/html and application/json are accepted."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "unregistered-client-id",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
},
|
|
headers={"Accept": "text/html,application/json"},
|
|
)
|
|
|
|
# Should prefer HTML
|
|
assert response.status_code == 400
|
|
assert "text/html" in response.headers["content-type"]
|
|
|
|
def test_json_when_only_json_accepted(self, oauth_proxy):
|
|
"""Test that JSON is returned when only application/json is accepted."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "unregistered-client-id",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
},
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "application/json" in response.headers["content-type"]
|
|
|
|
def test_json_when_no_accept_header(self, oauth_proxy):
|
|
"""Test that JSON is returned when no Accept header is provided."""
|
|
app = Starlette(routes=oauth_proxy.get_routes())
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/authorize",
|
|
params={
|
|
"client_id": "unregistered-client-id",
|
|
"redirect_uri": "http://localhost:12345/callback",
|
|
"response_type": "code",
|
|
"code_challenge": "test-challenge",
|
|
},
|
|
)
|
|
|
|
# Without Accept header, should return JSON (API default)
|
|
assert response.status_code == 400
|
|
assert "application/json" in response.headers["content-type"]
|