From 3718ad98ef290884bc22470c91e2dd3ef755e39b Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Fri, 2 May 2025 15:49:24 -0400
Subject: [PATCH 1/3] Add auth support
---
.github/workflows/run-tests.yml | 2 +-
README.md | 2 +-
docs/servers/fastmcp.mdx | 36 +-
justfile | 2 +-
src/fastmcp/server/server.py | 161 +++-
src/fastmcp/settings.py | 7 +
tests/server/test_auth_integration.py | 1263 +++++++++++++++++++++++++
7 files changed, 1460 insertions(+), 13 deletions(-)
create mode 100644 tests/server/test_auth_integration.py
diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml
index f65e20a0d..db1e14e2c 100644
--- a/.github/workflows/run-tests.yml
+++ b/.github/workflows/run-tests.yml
@@ -59,4 +59,4 @@ jobs:
uv pip install pyreadline3
- name: Run tests
- run: uv run pytest -vv
+ run: uv run --frozen pytest -vv
diff --git a/README.md b/README.md
index 31275b4c1..96002750d 100644
--- a/README.md
+++ b/README.md
@@ -760,7 +760,7 @@ Contributions make the open-source community vibrant! We welcome improvements an
Run the test suite:
```bash
-uv run pytest -vv
+uv run --frozen pytest -vv
```
#### Formatting & Linting
diff --git a/docs/servers/fastmcp.mdx b/docs/servers/fastmcp.mdx
index 9294ddb2a..2f378a66f 100644
--- a/docs/servers/fastmcp.mdx
+++ b/docs/servers/fastmcp.mdx
@@ -292,4 +292,38 @@ print(mcp.settings.on_duplicate_tools) # Output: "error"
- **`on_duplicate_resources`**: How to handle duplicate resource registrations
- **`on_duplicate_prompts`**: How to handle duplicate prompt registrations
-All of these can be configured directly as parameters when creating the `FastMCP` instance.
\ No newline at end of file
+All of these can be configured directly as parameters when creating the `FastMCP` instance.
+
+## Authentication
+
+
+
+FastMCP inherits support for OAuth 2.0 authentication from the MCP protocol, allowing servers to protect their tools and resources behind authentication.
+
+### OAuth 2.0 Support
+
+The `mcp.server.auth` module implements an OAuth 2.0 server interface that servers can use by providing an implementation of the `OAuthServerProvider` protocol.
+
+```python
+from fastmcp import FastMCP
+from mcp.server.auth.settings import RevocationOptions, ClientRegistrationOptions, AuthSettings
+
+
+# Create a server with authentication
+mcp = FastMCP(
+ name="SecureApp",
+ auth_provider=MyOAuthServerProvider(),
+ auth=AuthSettings(
+ issuer_url="https://myapp.com",
+ revocation_options=RevocationOptions(
+ enabled=True,
+ ),
+ client_registration_options=ClientRegistrationOptions(
+ enabled=True,
+ valid_scopes=["myscope", "myotherscope"],
+ default_scopes=["myscope"],
+ ),
+ required_scopes=["myscope"],
+ ),
+)
+```
diff --git a/justfile b/justfile
index 9fd3b0651..18bad5aff 100644
--- a/justfile
+++ b/justfile
@@ -6,4 +6,4 @@ test: build
# Run pyright on all files
typecheck:
- uv run pyright
\ No newline at end of file
+ uv run --frozen pyright
\ No newline at end of file
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 3563727f4..dfb71c1f9 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -15,6 +15,12 @@ from typing import TYPE_CHECKING, Any, Generic, Literal
import anyio
import httpx
import uvicorn
+from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
+from mcp.server.auth.middleware.bearer_auth import (
+ BearerAuthBackend,
+ RequireAuthMiddleware,
+)
+from mcp.server.auth.provider import OAuthAuthorizationServerProvider
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.lowlevel.server import LifespanResultT
from mcp.server.lowlevel.server import Server as MCPServer
@@ -36,8 +42,12 @@ from mcp.types import ResourceTemplate as MCPResourceTemplate
from mcp.types import Tool as MCPTool
from pydantic.networks import AnyUrl
from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.requests import Request
+from starlette.responses import Response
from starlette.routing import Mount, Route
+from starlette.types import Receive, Scope, Send
import fastmcp
import fastmcp.settings
@@ -184,6 +194,8 @@ class FastMCP(Generic[LifespanResultT]):
self,
name: str | None = None,
instructions: str | None = None,
+ auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any]
+ | None = None,
lifespan: (
Callable[
[FastMCP[LifespanResultT]],
@@ -221,6 +233,15 @@ class FastMCP(Generic[LifespanResultT]):
self._prompt_manager = PromptManager(
duplicate_behavior=self.settings.on_duplicate_prompts
)
+
+ if (self.settings.auth is not None) != (auth_server_provider is not None):
+ # TODO: after we support separate authorization servers (see
+ raise ValueError(
+ "settings.auth must be specified if and only if auth_server_provider "
+ "is specified"
+ )
+ self._auth_server_provider = auth_server_provider
+ self._custom_starlette_routes: list[Route] = []
self.dependencies = self.settings.dependencies
# Set up MCP protocol handlers
@@ -340,6 +361,50 @@ class FastMCP(Generic[LifespanResultT]):
self._cache.set("prompts", prompts)
return prompts
+ def custom_route(
+ self,
+ path: str,
+ methods: list[str],
+ name: str | None = None,
+ include_in_schema: bool = True,
+ ):
+ """
+ Decorator to register a custom HTTP route on the FastMCP server.
+
+ Allows adding arbitrary HTTP endpoints outside the standard MCP protocol,
+ which can be useful for OAuth callbacks, health checks, or admin APIs.
+ The handler function must be an async function that accepts a Starlette
+ Request and returns a Response.
+
+ Args:
+ path: URL path for the route (e.g., "/oauth/callback")
+ methods: List of HTTP methods to support (e.g., ["GET", "POST"])
+ name: Optional name for the route (to reference this route with
+ Starlette's reverse URL lookup feature)
+ include_in_schema: Whether to include in OpenAPI schema, defaults to True
+
+ Example:
+ @server.custom_route("/health", methods=["GET"])
+ async def health_check(request: Request) -> Response:
+ return JSONResponse({"status": "ok"})
+ """
+
+ def decorator(
+ func: Callable[[Request], Awaitable[Response]],
+ ) -> Callable[[Request], Awaitable[Response]]:
+ self._custom_starlette_routes.append(
+ Route(
+ path,
+ endpoint=func,
+ methods=methods,
+ name=name,
+ include_in_schema=include_in_schema,
+ )
+ )
+ return func
+
+ return decorator
+
async def _mcp_list_tools(self) -> list[MCPTool]:
"""
List all available tools, in the format expected by the low-level MCP
@@ -770,26 +835,104 @@ class FastMCP(Generic[LifespanResultT]):
def sse_app(self) -> Starlette:
"""Return an instance of the SSE server app."""
+ from starlette.middleware import Middleware
+ from starlette.routing import Mount, Route
+
+ # Set up auth context and dependencies
+
sse = SseServerTransport(self.settings.message_path)
- async def handle_sse(request: Request) -> None:
+ async def handle_sse(scope: Scope, receive: Receive, send: Send):
+ # Add client ID from auth context into request context if available
+
async with sse.connect_sse(
- request.scope,
- request.receive,
- request._send, # type: ignore[reportPrivateUsage]
+ scope,
+ receive,
+ send,
) as streams:
await self._mcp_server.run(
streams[0],
streams[1],
self._mcp_server.create_initialization_options(),
)
+ return Response()
+ # Create routes
+ routes: list[Route | Mount] = []
+ middleware: list[Middleware] = []
+ required_scopes = []
+
+ # Add auth endpoints if auth provider is configured
+ if self._auth_server_provider:
+ assert self.settings.auth
+ from mcp.server.auth.routes import create_auth_routes
+
+ required_scopes = self.settings.auth.required_scopes or []
+
+ middleware = [
+ # extract auth info from request (but do not require it)
+ Middleware(
+ AuthenticationMiddleware,
+ backend=BearerAuthBackend(
+ provider=self._auth_server_provider,
+ ),
+ ),
+ # Add the auth context middleware to store
+ # authenticated user in a contextvar
+ Middleware(AuthContextMiddleware),
+ ]
+ routes.extend(
+ create_auth_routes(
+ provider=self._auth_server_provider,
+ issuer_url=self.settings.auth.issuer_url,
+ service_documentation_url=self.settings.auth.service_documentation_url,
+ client_registration_options=self.settings.auth.client_registration_options,
+ revocation_options=self.settings.auth.revocation_options,
+ )
+ )
+
+ # When auth is not configured, we shouldn't require auth
+ if self._auth_server_provider:
+ # Auth is enabled, wrap the endpoints with RequireAuthMiddleware
+ routes.append(
+ Route(
+ self.settings.sse_path,
+ endpoint=RequireAuthMiddleware(handle_sse, required_scopes),
+ methods=["GET"],
+ )
+ )
+ routes.append(
+ Mount(
+ self.settings.message_path,
+ app=RequireAuthMiddleware(sse.handle_post_message, required_scopes),
+ )
+ )
+ else:
+ # Auth is disabled, no need for RequireAuthMiddleware
+ # Since handle_sse is an ASGI app, we need to create a compatible endpoint
+ async def sse_endpoint(request: Request) -> None:
+ # Convert the Starlette request to ASGI parameters
+ await handle_sse(request.scope, request.receive, request._send) # type: ignore[reportPrivateUsage]
+
+ routes.append(
+ Route(
+ self.settings.sse_path,
+ endpoint=sse_endpoint,
+ methods=["GET"],
+ )
+ )
+ routes.append(
+ Mount(
+ self.settings.message_path,
+ app=sse.handle_post_message,
+ )
+ )
+ # mount these routes last, so they have the lowest route matching precedence
+ routes.extend(self._custom_starlette_routes)
+
+ # Create Starlette app with routes and middleware
return Starlette(
- debug=self.settings.debug,
- routes=[
- Route(self.settings.sse_path, endpoint=handle_sse),
- Mount(self.settings.message_path, app=sse.handle_post_message),
- ],
+ debug=self.settings.debug, routes=routes, middleware=middleware
)
def mount(
diff --git a/src/fastmcp/settings.py b/src/fastmcp/settings.py
index 9d02b5d5a..30870e1c7 100644
--- a/src/fastmcp/settings.py
+++ b/src/fastmcp/settings.py
@@ -2,6 +2,7 @@ from __future__ import annotations as _annotations
from typing import TYPE_CHECKING, Literal
+from mcp.server.auth.settings import AuthSettings
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -20,6 +21,8 @@ class Settings(BaseSettings):
env_prefix="FASTMCP_",
env_file=".env",
extra="ignore",
+ env_nested_delimiter="__",
+ nested_model_default_partial_update=True,
)
test_mode: bool = False
@@ -37,6 +40,8 @@ class ServerSettings(BaseSettings):
env_prefix="FASTMCP_SERVER_",
env_file=".env",
extra="ignore",
+ env_nested_delimiter="__",
+ nested_model_default_partial_update=True,
)
log_level: LOG_LEVEL = Field(default_factory=lambda: Settings().log_level)
@@ -65,6 +70,8 @@ class ServerSettings(BaseSettings):
# cache settings (for checking mounted servers)
cache_expiration_seconds: float = 0
+ auth: AuthSettings | None = None
+
class ClientSettings(BaseSettings):
"""FastMCP client settings."""
diff --git a/tests/server/test_auth_integration.py b/tests/server/test_auth_integration.py
new file mode 100644
index 000000000..fb0330dcf
--- /dev/null
+++ b/tests/server/test_auth_integration.py
@@ -0,0 +1,1263 @@
+import base64
+import hashlib
+import secrets
+import time
+import unittest.mock
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+import pytest
+from mcp.server.auth.provider import (
+ AccessToken,
+ AuthorizationCode,
+ AuthorizationParams,
+ OAuthAuthorizationServerProvider,
+ RefreshToken,
+ construct_redirect_uri,
+)
+from mcp.server.auth.routes import (
+ create_auth_routes,
+)
+from mcp.server.auth.settings import (
+ ClientRegistrationOptions,
+ RevocationOptions,
+)
+from mcp.shared.auth import (
+ OAuthClientInformationFull,
+ OAuthToken,
+)
+from pydantic import AnyHttpUrl
+from starlette.applications import Starlette
+
+
+# Mock OAuth provider for testing
+class MockOAuthProvider(OAuthAuthorizationServerProvider):
+ def __init__(self):
+ self.clients = {}
+ self.auth_codes = {} # code -> {client_id, code_challenge, redirect_uri}
+ self.tokens = {} # token -> {client_id, scopes, expires_at}
+ self.refresh_tokens = {} # refresh_token -> access_token
+
+ async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
+ return self.clients.get(client_id)
+
+ async def register_client(self, client_info: OAuthClientInformationFull):
+ self.clients[client_info.client_id] = client_info
+
+ async def authorize(
+ self, client: OAuthClientInformationFull, params: AuthorizationParams
+ ) -> str:
+ # toy authorize implementation which just immediately generates an authorization
+ # code and completes the redirect
+ code = AuthorizationCode(
+ code=f"code_{int(time.time())}",
+ client_id=client.client_id,
+ code_challenge=params.code_challenge,
+ redirect_uri=params.redirect_uri,
+ redirect_uri_provided_explicitly=params.redirect_uri_provided_explicitly,
+ expires_at=time.time() + 300,
+ scopes=params.scopes or ["read", "write"],
+ )
+ self.auth_codes[code.code] = code
+
+ return construct_redirect_uri(
+ str(params.redirect_uri), code=code.code, state=params.state
+ )
+
+ async def load_authorization_code(
+ self, client: OAuthClientInformationFull, authorization_code: str
+ ) -> AuthorizationCode | None:
+ return self.auth_codes.get(authorization_code)
+
+ async def exchange_authorization_code(
+ self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
+ ) -> OAuthToken:
+ assert authorization_code.code in self.auth_codes
+
+ # Generate an access token and refresh token
+ access_token = f"access_{secrets.token_hex(32)}"
+ refresh_token = f"refresh_{secrets.token_hex(32)}"
+
+ # Store the tokens
+ self.tokens[access_token] = AccessToken(
+ token=access_token,
+ client_id=client.client_id,
+ scopes=authorization_code.scopes,
+ expires_at=int(time.time()) + 3600,
+ )
+
+ self.refresh_tokens[refresh_token] = access_token
+
+ # Remove the used code
+ del self.auth_codes[authorization_code.code]
+
+ return OAuthToken(
+ access_token=access_token,
+ token_type="bearer",
+ expires_in=3600,
+ scope="read write",
+ refresh_token=refresh_token,
+ )
+
+ async def load_refresh_token(
+ self, client: OAuthClientInformationFull, refresh_token: str
+ ) -> RefreshToken | None:
+ old_access_token = self.refresh_tokens.get(refresh_token)
+ if old_access_token is None:
+ return None
+ token_info = self.tokens.get(old_access_token)
+ if token_info is None:
+ return None
+
+ # Create a RefreshToken object that matches what is expected in later code
+ refresh_obj = RefreshToken(
+ token=refresh_token,
+ client_id=token_info.client_id,
+ scopes=token_info.scopes,
+ expires_at=token_info.expires_at,
+ )
+
+ return refresh_obj
+
+ async def exchange_refresh_token(
+ self,
+ client: OAuthClientInformationFull,
+ refresh_token: RefreshToken,
+ scopes: list[str],
+ ) -> OAuthToken:
+ # Check if refresh token exists
+ assert refresh_token.token in self.refresh_tokens
+
+ old_access_token = self.refresh_tokens[refresh_token.token]
+
+ # Check if the access token exists
+ assert old_access_token in self.tokens
+
+ # Check if the token was issued to this client
+ token_info = self.tokens[old_access_token]
+ assert token_info.client_id == client.client_id
+
+ # Generate a new access token and refresh token
+ new_access_token = f"access_{secrets.token_hex(32)}"
+ new_refresh_token = f"refresh_{secrets.token_hex(32)}"
+
+ # Store the new tokens
+ self.tokens[new_access_token] = AccessToken(
+ token=new_access_token,
+ client_id=client.client_id,
+ scopes=scopes or token_info.scopes,
+ expires_at=int(time.time()) + 3600,
+ )
+
+ self.refresh_tokens[new_refresh_token] = new_access_token
+
+ # Remove the old tokens
+ del self.refresh_tokens[refresh_token.token]
+ del self.tokens[old_access_token]
+
+ return OAuthToken(
+ access_token=new_access_token,
+ token_type="bearer",
+ expires_in=3600,
+ scope=" ".join(scopes) if scopes else " ".join(token_info.scopes),
+ refresh_token=new_refresh_token,
+ )
+
+ async def load_access_token(self, token: str) -> AccessToken | None:
+ token_info = self.tokens.get(token)
+
+ # Check if token is expired
+ # if token_info.expires_at < int(time.time()):
+ # raise InvalidTokenError("Access token has expired")
+
+ return token_info and AccessToken(
+ token=token,
+ client_id=token_info.client_id,
+ scopes=token_info.scopes,
+ expires_at=token_info.expires_at,
+ )
+
+ async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
+ match token:
+ case RefreshToken():
+ # Remove the refresh token
+ del self.refresh_tokens[token.token]
+
+ case AccessToken():
+ # Remove the access token
+ del self.tokens[token.token]
+
+ # Also remove any refresh tokens that point to this access token
+ for refresh_token, access_token in list(self.refresh_tokens.items()):
+ if access_token == token.token:
+ del self.refresh_tokens[refresh_token]
+
+
+@pytest.fixture
+def mock_oauth_provider():
+ return MockOAuthProvider()
+
+
+@pytest.fixture
+def auth_app(mock_oauth_provider):
+ # Create auth router
+ auth_routes = create_auth_routes(
+ mock_oauth_provider,
+ AnyHttpUrl("https://auth.example.com"),
+ AnyHttpUrl("https://docs.example.com"),
+ client_registration_options=ClientRegistrationOptions(
+ enabled=True,
+ valid_scopes=["read", "write", "profile"],
+ default_scopes=["read", "write"],
+ ),
+ revocation_options=RevocationOptions(enabled=True),
+ )
+
+ # Create Starlette app
+ app = Starlette(routes=auth_routes)
+
+ return app
+
+
+@pytest.fixture
+async def test_client(auth_app):
+ async with httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=auth_app), base_url="https://mcptest.com"
+ ) as client:
+ yield client
+
+
+@pytest.fixture
+async def registered_client(test_client: httpx.AsyncClient, request):
+ """Create and register a test client.
+
+ Parameters can be customized via indirect parameterization:
+ @pytest.mark.parametrize("registered_client",
+ [{"grant_types": ["authorization_code"]}],
+ indirect=True)
+ """
+ # Default client metadata
+ client_metadata = {
+ "redirect_uris": ["https://client.example.com/callback"],
+ "client_name": "Test Client",
+ "grant_types": ["authorization_code", "refresh_token"],
+ }
+
+ # Override with any parameters from the test
+ if hasattr(request, "param") and request.param:
+ client_metadata.update(request.param)
+
+ response = await test_client.post("/register", json=client_metadata)
+ assert response.status_code == 201, f"Failed to register client: {response.content}"
+
+ client_info = response.json()
+ return client_info
+
+
+@pytest.fixture
+def pkce_challenge():
+ """Create a PKCE challenge with code_verifier and code_challenge."""
+ code_verifier = "some_random_verifier_string"
+ code_challenge = (
+ base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest())
+ .decode()
+ .rstrip("=")
+ )
+
+ return {"code_verifier": code_verifier, "code_challenge": code_challenge}
+
+
+@pytest.fixture
+async def auth_code(test_client, registered_client, pkce_challenge, request):
+ """Get an authorization code.
+
+ Parameters can be customized via indirect parameterization:
+ @pytest.mark.parametrize("auth_code",
+ [{"redirect_uri": "https://client.example.com/other-callback"}],
+ indirect=True)
+ """
+ # Default authorize params
+ auth_params = {
+ "response_type": "code",
+ "client_id": registered_client["client_id"],
+ "redirect_uri": "https://client.example.com/callback",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "state": "test_state",
+ }
+
+ # Override with any parameters from the test
+ if hasattr(request, "param") and request.param:
+ auth_params.update(request.param)
+
+ response = await test_client.get("/authorize", params=auth_params)
+ assert response.status_code == 302, f"Failed to get auth code: {response.content}"
+
+ # Extract the authorization code
+ redirect_url = response.headers["location"]
+ parsed_url = urlparse(redirect_url)
+ query_params = parse_qs(parsed_url.query)
+
+ assert "code" in query_params, f"No code in response: {query_params}"
+ auth_code = query_params["code"][0]
+
+ return {
+ "code": auth_code,
+ "redirect_uri": auth_params["redirect_uri"],
+ "state": query_params.get("state", [None])[0],
+ }
+
+
+@pytest.fixture
+async def tokens(test_client, registered_client, auth_code, pkce_challenge, request):
+ """Exchange authorization code for tokens.
+
+ Parameters can be customized via indirect parameterization:
+ @pytest.mark.parametrize("tokens",
+ [{"code_verifier": "wrong_verifier"}],
+ indirect=True)
+ """
+ # Default token request params
+ token_params = {
+ "grant_type": "authorization_code",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "code": auth_code["code"],
+ "code_verifier": pkce_challenge["code_verifier"],
+ "redirect_uri": auth_code["redirect_uri"],
+ }
+
+ # Override with any parameters from the test
+ if hasattr(request, "param") and request.param:
+ token_params.update(request.param)
+
+ response = await test_client.post("/token", data=token_params)
+
+ # Don't assert success here since some tests will intentionally cause errors
+ return {
+ "response": response,
+ "params": token_params,
+ }
+
+
+class TestAuthEndpoints:
+ @pytest.mark.anyio
+ async def test_metadata_endpoint(self, test_client: httpx.AsyncClient):
+ """Test the OAuth 2.0 metadata endpoint."""
+ print("Sending request to metadata endpoint")
+ response = await test_client.get("/.well-known/oauth-authorization-server")
+ print(f"Got response: {response.status_code}")
+ if response.status_code != 200:
+ print(f"Response content: {response.content}")
+ assert response.status_code == 200
+
+ metadata = response.json()
+ assert metadata["issuer"] == "https://auth.example.com/"
+ assert (
+ metadata["authorization_endpoint"] == "https://auth.example.com/authorize"
+ )
+ assert metadata["token_endpoint"] == "https://auth.example.com/token"
+ assert metadata["registration_endpoint"] == "https://auth.example.com/register"
+ assert metadata["revocation_endpoint"] == "https://auth.example.com/revoke"
+ assert metadata["response_types_supported"] == ["code"]
+ assert metadata["code_challenge_methods_supported"] == ["S256"]
+ assert metadata["token_endpoint_auth_methods_supported"] == [
+ "client_secret_post"
+ ]
+ assert metadata["grant_types_supported"] == [
+ "authorization_code",
+ "refresh_token",
+ ]
+ assert metadata["service_documentation"] == "https://docs.example.com/"
+
+ @pytest.mark.anyio
+ async def test_token_validation_error(self, test_client: httpx.AsyncClient):
+ """Test token endpoint error - validation error."""
+ # Missing required fields
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "authorization_code",
+ # Missing code, code_verifier, client_id, etc.
+ },
+ )
+ error_response = response.json()
+ assert error_response["error"] == "invalid_request"
+ assert (
+ "error_description" in error_response
+ ) # Contains validation error messages
+
+ @pytest.mark.anyio
+ async def test_token_invalid_auth_code(
+ self, test_client, registered_client, pkce_challenge
+ ):
+ """Test token endpoint error - authorization code does not exist."""
+ # Try to use a non-existent authorization code
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "authorization_code",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "code": "non_existent_auth_code",
+ "code_verifier": pkce_challenge["code_verifier"],
+ "redirect_uri": "https://client.example.com/callback",
+ },
+ )
+ print(f"Status code: {response.status_code}")
+ print(f"Response body: {response.content}")
+ print(f"Response JSON: {response.json()}")
+ assert response.status_code == 400
+ error_response = response.json()
+ assert error_response["error"] == "invalid_grant"
+ assert (
+ "authorization code does not exist" in error_response["error_description"]
+ )
+
+ @pytest.mark.anyio
+ async def test_token_expired_auth_code(
+ self,
+ test_client,
+ registered_client,
+ auth_code,
+ pkce_challenge,
+ mock_oauth_provider,
+ ):
+ """Test token endpoint error - authorization code has expired."""
+ # Get the current time for our time mocking
+ current_time = time.time()
+
+ # Find the auth code object
+ code_value = auth_code["code"]
+ found_code = None
+ for code_obj in mock_oauth_provider.auth_codes.values():
+ if code_obj.code == code_value:
+ found_code = code_obj
+ break
+
+ assert found_code is not None
+
+ # Authorization codes are typically short-lived (5 minutes = 300 seconds)
+ # So we'll mock time to be 10 minutes (600 seconds) in the future
+ with unittest.mock.patch("time.time", return_value=current_time + 600):
+ # Try to use the expired authorization code
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "authorization_code",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "code": code_value,
+ "code_verifier": pkce_challenge["code_verifier"],
+ "redirect_uri": auth_code["redirect_uri"],
+ },
+ )
+ assert response.status_code == 400
+ error_response = response.json()
+ assert error_response["error"] == "invalid_grant"
+ assert (
+ "authorization code has expired" in error_response["error_description"]
+ )
+
+ @pytest.mark.anyio
+ @pytest.mark.parametrize(
+ "registered_client",
+ [
+ {
+ "redirect_uris": [
+ "https://client.example.com/callback",
+ "https://client.example.com/other-callback",
+ ]
+ }
+ ],
+ indirect=True,
+ )
+ async def test_token_redirect_uri_mismatch(
+ self, test_client, registered_client, auth_code, pkce_challenge
+ ):
+ """Test token endpoint error - redirect URI mismatch."""
+ # Try to use the code with a different redirect URI
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "authorization_code",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "code": auth_code["code"],
+ "code_verifier": pkce_challenge["code_verifier"],
+ # Different from the one used in /authorize
+ "redirect_uri": "https://client.example.com/other-callback",
+ },
+ )
+ assert response.status_code == 400
+ error_response = response.json()
+ assert error_response["error"] == "invalid_request"
+ assert "redirect_uri did not match" in error_response["error_description"]
+
+ @pytest.mark.anyio
+ async def test_token_code_verifier_mismatch(
+ self, test_client, registered_client, auth_code
+ ):
+ """Test token endpoint error - PKCE code verifier mismatch."""
+ # Try to use the code with an incorrect code verifier
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "authorization_code",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "code": auth_code["code"],
+ # Different from the one used to create challenge
+ "code_verifier": "incorrect_code_verifier",
+ "redirect_uri": auth_code["redirect_uri"],
+ },
+ )
+ assert response.status_code == 400
+ error_response = response.json()
+ assert error_response["error"] == "invalid_grant"
+ assert "incorrect code_verifier" in error_response["error_description"]
+
+ @pytest.mark.anyio
+ async def test_token_invalid_refresh_token(self, test_client, registered_client):
+ """Test token endpoint error - refresh token does not exist."""
+ # Try to use a non-existent refresh token
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "refresh_token",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "refresh_token": "non_existent_refresh_token",
+ },
+ )
+ assert response.status_code == 400
+ error_response = response.json()
+ assert error_response["error"] == "invalid_grant"
+ assert "refresh token does not exist" in error_response["error_description"]
+
+ @pytest.mark.anyio
+ async def test_token_expired_refresh_token(
+ self,
+ test_client,
+ registered_client,
+ auth_code,
+ pkce_challenge,
+ mock_oauth_provider,
+ ):
+ """Test token endpoint error - refresh token has expired."""
+ # Step 1: First, let's create a token and refresh token at the current time
+ current_time = time.time()
+
+ # Exchange authorization code for tokens normally
+ token_response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "authorization_code",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "code": auth_code["code"],
+ "code_verifier": pkce_challenge["code_verifier"],
+ "redirect_uri": auth_code["redirect_uri"],
+ },
+ )
+ assert token_response.status_code == 200
+ tokens = token_response.json()
+ refresh_token = tokens["refresh_token"]
+
+ # Step 2: Time travel forward 4 hours (tokens expire in 1 hour by default)
+ # Mock the time.time() function to return a value 4 hours in the future
+ with unittest.mock.patch(
+ "time.time", return_value=current_time + 14400
+ ): # 4 hours = 14400 seconds
+ # Try to use the refresh token which should now be considered expired
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "refresh_token",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "refresh_token": refresh_token,
+ },
+ )
+
+ # In the "future", the token should be considered expired
+ assert response.status_code == 400
+ error_response = response.json()
+ assert error_response["error"] == "invalid_grant"
+ assert "refresh token has expired" in error_response["error_description"]
+
+ @pytest.mark.anyio
+ async def test_token_invalid_scope(
+ self, test_client, registered_client, auth_code, pkce_challenge
+ ):
+ """Test token endpoint error - invalid scope in refresh token request."""
+ # Exchange authorization code for tokens
+ token_response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "authorization_code",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "code": auth_code["code"],
+ "code_verifier": pkce_challenge["code_verifier"],
+ "redirect_uri": auth_code["redirect_uri"],
+ },
+ )
+ assert token_response.status_code == 200
+
+ tokens = token_response.json()
+ refresh_token = tokens["refresh_token"]
+
+ # Try to use refresh token with an invalid scope
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "refresh_token",
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "refresh_token": refresh_token,
+ "scope": "read write invalid_scope", # Adding an invalid scope
+ },
+ )
+ assert response.status_code == 400
+ error_response = response.json()
+ assert error_response["error"] == "invalid_scope"
+ assert "cannot request scope" in error_response["error_description"]
+
+ @pytest.mark.anyio
+ async def test_client_registration(
+ self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider
+ ):
+ """Test client registration."""
+ client_metadata = {
+ "redirect_uris": ["https://client.example.com/callback"],
+ "client_name": "Test Client",
+ "client_uri": "https://client.example.com",
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 201, response.content
+
+ client_info = response.json()
+ assert "client_id" in client_info
+ assert "client_secret" in client_info
+ assert client_info["client_name"] == "Test Client"
+ assert client_info["redirect_uris"] == ["https://client.example.com/callback"]
+
+ # Verify that the client was registered
+ # assert await mock_oauth_provider.clients_store.get_client(
+ # client_info["client_id"]
+ # ) is not None
+
+ @pytest.mark.anyio
+ async def test_client_registration_missing_required_fields(
+ self, test_client: httpx.AsyncClient
+ ):
+ """Test client registration with missing required fields."""
+ # Missing redirect_uris which is a required field
+ client_metadata = {
+ "client_name": "Test Client",
+ "client_uri": "https://client.example.com",
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 400
+ error_data = response.json()
+ assert "error" in error_data
+ assert error_data["error"] == "invalid_client_metadata"
+ assert error_data["error_description"] == "redirect_uris: Field required"
+
+ @pytest.mark.anyio
+ async def test_client_registration_invalid_uri(
+ self, test_client: httpx.AsyncClient
+ ):
+ """Test client registration with invalid URIs."""
+ # Invalid redirect_uri format
+ client_metadata = {
+ "redirect_uris": ["not-a-valid-uri"],
+ "client_name": "Test Client",
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 400
+ error_data = response.json()
+ assert "error" in error_data
+ assert error_data["error"] == "invalid_client_metadata"
+ assert error_data["error_description"] == (
+ "redirect_uris.0: Input should be a valid URL, relative URL without a base"
+ )
+
+ @pytest.mark.anyio
+ async def test_client_registration_empty_redirect_uris(
+ self, test_client: httpx.AsyncClient
+ ):
+ """Test client registration with empty redirect_uris array."""
+ client_metadata = {
+ "redirect_uris": [], # Empty array
+ "client_name": "Test Client",
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 400
+ error_data = response.json()
+ assert "error" in error_data
+ assert error_data["error"] == "invalid_client_metadata"
+ assert (
+ error_data["error_description"]
+ == "redirect_uris: List should have at least 1 item after validation, not 0"
+ )
+
+ @pytest.mark.anyio
+ async def test_authorize_form_post(
+ self,
+ test_client: httpx.AsyncClient,
+ mock_oauth_provider: MockOAuthProvider,
+ pkce_challenge,
+ ):
+ """Test the authorization endpoint using POST with form-encoded data."""
+ # Register a client
+ client_metadata = {
+ "redirect_uris": ["https://client.example.com/callback"],
+ "client_name": "Test Client",
+ "grant_types": ["authorization_code", "refresh_token"],
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 201
+ client_info = response.json()
+
+ # Use POST with form-encoded data for authorization
+ response = await test_client.post(
+ "/authorize",
+ data={
+ "response_type": "code",
+ "client_id": client_info["client_id"],
+ "redirect_uri": "https://client.example.com/callback",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "state": "test_form_state",
+ },
+ )
+ assert response.status_code == 302
+
+ # Extract the authorization code from the redirect URL
+ redirect_url = response.headers["location"]
+ parsed_url = urlparse(redirect_url)
+ query_params = parse_qs(parsed_url.query)
+
+ assert "code" in query_params
+ assert query_params["state"][0] == "test_form_state"
+
+ @pytest.mark.anyio
+ async def test_authorization_get(
+ self,
+ test_client: httpx.AsyncClient,
+ mock_oauth_provider: MockOAuthProvider,
+ pkce_challenge,
+ ):
+ """Test the full authorization flow."""
+ # 1. Register a client
+ client_metadata = {
+ "redirect_uris": ["https://client.example.com/callback"],
+ "client_name": "Test Client",
+ "grant_types": ["authorization_code", "refresh_token"],
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 201
+ client_info = response.json()
+
+ # 2. Request authorization using GET with query params
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "code",
+ "client_id": client_info["client_id"],
+ "redirect_uri": "https://client.example.com/callback",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "state": "test_state",
+ },
+ )
+ assert response.status_code == 302
+
+ # 3. Extract the authorization code from the redirect URL
+ redirect_url = response.headers["location"]
+ parsed_url = urlparse(redirect_url)
+ query_params = parse_qs(parsed_url.query)
+
+ assert "code" in query_params
+ assert query_params["state"][0] == "test_state"
+ auth_code = query_params["code"][0]
+
+ # 4. Exchange the authorization code for tokens
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "authorization_code",
+ "client_id": client_info["client_id"],
+ "client_secret": client_info["client_secret"],
+ "code": auth_code,
+ "code_verifier": pkce_challenge["code_verifier"],
+ "redirect_uri": "https://client.example.com/callback",
+ },
+ )
+ assert response.status_code == 200
+
+ token_response = response.json()
+ assert "access_token" in token_response
+ assert "token_type" in token_response
+ assert "refresh_token" in token_response
+ assert "expires_in" in token_response
+ assert token_response["token_type"] == "bearer"
+
+ # 5. Verify the access token
+ access_token = token_response["access_token"]
+ refresh_token = token_response["refresh_token"]
+
+ # Create a test client with the token
+ auth_info = await mock_oauth_provider.load_access_token(access_token)
+ assert auth_info
+ assert auth_info.client_id == client_info["client_id"]
+ assert "read" in auth_info.scopes
+ assert "write" in auth_info.scopes
+
+ # 6. Refresh the token
+ response = await test_client.post(
+ "/token",
+ data={
+ "grant_type": "refresh_token",
+ "client_id": client_info["client_id"],
+ "client_secret": client_info["client_secret"],
+ "refresh_token": refresh_token,
+ "redirect_uri": "https://client.example.com/callback",
+ },
+ )
+ assert response.status_code == 200
+
+ new_token_response = response.json()
+ assert "access_token" in new_token_response
+ assert "refresh_token" in new_token_response
+ assert new_token_response["access_token"] != access_token
+ assert new_token_response["refresh_token"] != refresh_token
+
+ # 7. Revoke the token
+ response = await test_client.post(
+ "/revoke",
+ data={
+ "client_id": client_info["client_id"],
+ "client_secret": client_info["client_secret"],
+ "token": new_token_response["access_token"],
+ },
+ )
+ assert response.status_code == 200
+
+ # Verify that the token was revoked
+ assert (
+ await mock_oauth_provider.load_access_token(
+ new_token_response["access_token"]
+ )
+ is None
+ )
+
+ @pytest.mark.anyio
+ async def test_revoke_invalid_token(self, test_client, registered_client):
+ """Test revoking an invalid token."""
+ response = await test_client.post(
+ "/revoke",
+ data={
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "token": "invalid_token",
+ },
+ )
+ # per RFC, this should return 200 even if the token is invalid
+ assert response.status_code == 200
+
+ @pytest.mark.anyio
+ async def test_revoke_with_malformed_token(self, test_client, registered_client):
+ response = await test_client.post(
+ "/revoke",
+ data={
+ "client_id": registered_client["client_id"],
+ "client_secret": registered_client["client_secret"],
+ "token": 123,
+ "token_type_hint": "asdf",
+ },
+ )
+ assert response.status_code == 400
+ error_response = response.json()
+ assert error_response["error"] == "invalid_request"
+ assert "token_type_hint" in error_response["error_description"]
+
+ @pytest.mark.anyio
+ async def test_client_registration_disallowed_scopes(
+ self, test_client: httpx.AsyncClient
+ ):
+ """Test client registration with scopes that are not allowed."""
+ client_metadata = {
+ "redirect_uris": ["https://client.example.com/callback"],
+ "client_name": "Test Client",
+ "scope": "read write profile admin", # 'admin' is not in valid_scopes
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 400
+ error_data = response.json()
+ assert "error" in error_data
+ assert error_data["error"] == "invalid_client_metadata"
+ assert "scope" in error_data["error_description"]
+ assert "admin" in error_data["error_description"]
+
+ @pytest.mark.anyio
+ async def test_client_registration_default_scopes(
+ self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider
+ ):
+ client_metadata = {
+ "redirect_uris": ["https://client.example.com/callback"],
+ "client_name": "Test Client",
+ # No scope specified
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 201
+ client_info = response.json()
+
+ # Verify client was registered successfully
+ assert client_info["scope"] == "read write"
+
+ # Retrieve the client from the store to verify default scopes
+ registered_client = await mock_oauth_provider.get_client(
+ client_info["client_id"]
+ )
+ assert registered_client is not None
+
+ # Check that default scopes were applied
+ assert registered_client.scope == "read write"
+
+ @pytest.mark.anyio
+ async def test_client_registration_invalid_grant_type(
+ self, test_client: httpx.AsyncClient
+ ):
+ client_metadata = {
+ "redirect_uris": ["https://client.example.com/callback"],
+ "client_name": "Test Client",
+ "grant_types": ["authorization_code"],
+ }
+
+ response = await test_client.post(
+ "/register",
+ json=client_metadata,
+ )
+ assert response.status_code == 400
+ error_data = response.json()
+ assert "error" in error_data
+ assert error_data["error"] == "invalid_client_metadata"
+ assert (
+ error_data["error_description"]
+ == "grant_types must be authorization_code and refresh_token"
+ )
+
+
+class TestAuthorizeEndpointErrors:
+ """Test error handling in the OAuth authorization endpoint."""
+
+ @pytest.mark.anyio
+ async def test_authorize_missing_client_id(
+ self, test_client: httpx.AsyncClient, pkce_challenge
+ ):
+ """Test authorization endpoint with missing client_id.
+
+ According to the OAuth2.0 spec, if client_id is missing, the server should
+ inform the resource owner and NOT redirect.
+ """
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "code",
+ # Missing client_id
+ "redirect_uri": "https://client.example.com/callback",
+ "state": "test_state",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ },
+ )
+
+ # Should NOT redirect, should show an error page
+ assert response.status_code == 400
+ # The response should include an error message about missing client_id
+ assert "client_id" in response.text.lower()
+
+ @pytest.mark.anyio
+ async def test_authorize_invalid_client_id(
+ self, test_client: httpx.AsyncClient, pkce_challenge
+ ):
+ """Test authorization endpoint with invalid client_id.
+
+ According to the OAuth2.0 spec, if client_id is invalid, the server should
+ inform the resource owner and NOT redirect.
+ """
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "code",
+ "client_id": "invalid_client_id_that_does_not_exist",
+ "redirect_uri": "https://client.example.com/callback",
+ "state": "test_state",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ },
+ )
+
+ # Should NOT redirect, should show an error page
+ assert response.status_code == 400
+ # The response should include an error message about invalid client_id
+ assert "client" in response.text.lower()
+
+ @pytest.mark.anyio
+ async def test_authorize_missing_redirect_uri(
+ self, test_client: httpx.AsyncClient, registered_client, pkce_challenge
+ ):
+ """Test authorization endpoint with missing redirect_uri.
+
+ If client has only one registered redirect_uri, it can be omitted.
+ """
+
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "code",
+ "client_id": registered_client["client_id"],
+ # Missing redirect_uri
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "state": "test_state",
+ },
+ )
+
+ # Should redirect to the registered redirect_uri
+ assert response.status_code == 302, response.content
+ redirect_url = response.headers["location"]
+ assert redirect_url.startswith("https://client.example.com/callback")
+
+ @pytest.mark.anyio
+ async def test_authorize_invalid_redirect_uri(
+ self, test_client: httpx.AsyncClient, registered_client, pkce_challenge
+ ):
+ """Test authorization endpoint with invalid redirect_uri.
+
+ According to the OAuth2.0 spec, if redirect_uri is invalid or doesn't match,
+ the server should inform the resource owner and NOT redirect.
+ """
+
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "code",
+ "client_id": registered_client["client_id"],
+ # Non-matching URI
+ "redirect_uri": "https://attacker.example.com/callback",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "state": "test_state",
+ },
+ )
+
+ # Should NOT redirect, should show an error page
+ assert response.status_code == 400, response.content
+ # The response should include an error message about redirect_uri mismatch
+ assert "redirect" in response.text.lower()
+
+ @pytest.mark.anyio
+ @pytest.mark.parametrize(
+ "registered_client",
+ [
+ {
+ "redirect_uris": [
+ "https://client.example.com/callback",
+ "https://client.example.com/other-callback",
+ ]
+ }
+ ],
+ indirect=True,
+ )
+ async def test_authorize_missing_redirect_uri_multiple_registered(
+ self, test_client: httpx.AsyncClient, registered_client, pkce_challenge
+ ):
+ """Test endpoint with missing redirect_uri with multiple registered URIs.
+
+ If client has multiple registered redirect_uris, redirect_uri must be provided.
+ """
+
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "code",
+ "client_id": registered_client["client_id"],
+ # Missing redirect_uri
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "state": "test_state",
+ },
+ )
+
+ # Should NOT redirect, should return a 400 error
+ assert response.status_code == 400
+ # The response should include an error message about missing redirect_uri
+ assert "redirect_uri" in response.text.lower()
+
+ @pytest.mark.anyio
+ async def test_authorize_unsupported_response_type(
+ self, test_client: httpx.AsyncClient, registered_client, pkce_challenge
+ ):
+ """Test authorization endpoint with unsupported response_type.
+
+ According to the OAuth2.0 spec, for other errors like unsupported_response_type,
+ the server should redirect with error parameters.
+ """
+
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "token", # Unsupported (we only support "code")
+ "client_id": registered_client["client_id"],
+ "redirect_uri": "https://client.example.com/callback",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "state": "test_state",
+ },
+ )
+
+ # Should redirect with error parameters
+ assert response.status_code == 302
+ redirect_url = response.headers["location"]
+ parsed_url = urlparse(redirect_url)
+ query_params = parse_qs(parsed_url.query)
+
+ assert "error" in query_params
+ assert query_params["error"][0] == "unsupported_response_type"
+ # State should be preserved
+ assert "state" in query_params
+ assert query_params["state"][0] == "test_state"
+
+ @pytest.mark.anyio
+ async def test_authorize_missing_response_type(
+ self, test_client: httpx.AsyncClient, registered_client, pkce_challenge
+ ):
+ """Test authorization endpoint with missing response_type.
+
+ Missing required parameter should result in invalid_request error.
+ """
+
+ response = await test_client.get(
+ "/authorize",
+ params={
+ # Missing response_type
+ "client_id": registered_client["client_id"],
+ "redirect_uri": "https://client.example.com/callback",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "state": "test_state",
+ },
+ )
+
+ # Should redirect with error parameters
+ assert response.status_code == 302
+ redirect_url = response.headers["location"]
+ parsed_url = urlparse(redirect_url)
+ query_params = parse_qs(parsed_url.query)
+
+ assert "error" in query_params
+ assert query_params["error"][0] == "invalid_request"
+ # State should be preserved
+ assert "state" in query_params
+ assert query_params["state"][0] == "test_state"
+
+ @pytest.mark.anyio
+ async def test_authorize_missing_pkce_challenge(
+ self, test_client: httpx.AsyncClient, registered_client
+ ):
+ """Test authorization endpoint with missing PKCE code_challenge.
+
+ Missing PKCE parameters should result in invalid_request error.
+ """
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "code",
+ "client_id": registered_client["client_id"],
+ # Missing code_challenge
+ "state": "test_state",
+ # using default URL
+ },
+ )
+
+ # Should redirect with error parameters
+ assert response.status_code == 302
+ redirect_url = response.headers["location"]
+ parsed_url = urlparse(redirect_url)
+ query_params = parse_qs(parsed_url.query)
+
+ assert "error" in query_params
+ assert query_params["error"][0] == "invalid_request"
+ # State should be preserved
+ assert "state" in query_params
+ assert query_params["state"][0] == "test_state"
+
+ @pytest.mark.anyio
+ async def test_authorize_invalid_scope(
+ self, test_client: httpx.AsyncClient, registered_client, pkce_challenge
+ ):
+ """Test authorization endpoint with invalid scope.
+
+ Invalid scope should redirect with invalid_scope error.
+ """
+
+ response = await test_client.get(
+ "/authorize",
+ params={
+ "response_type": "code",
+ "client_id": registered_client["client_id"],
+ "redirect_uri": "https://client.example.com/callback",
+ "code_challenge": pkce_challenge["code_challenge"],
+ "code_challenge_method": "S256",
+ "scope": "invalid_scope_that_does_not_exist",
+ "state": "test_state",
+ },
+ )
+
+ # Should redirect with error parameters
+ assert response.status_code == 302
+ redirect_url = response.headers["location"]
+ parsed_url = urlparse(redirect_url)
+ query_params = parse_qs(parsed_url.query)
+
+ assert "error" in query_params
+ assert query_params["error"][0] == "invalid_scope"
+ # State should be preserved
+ assert "state" in query_params
+ assert query_params["state"][0] == "test_state"
From eb06a371e1b4776d2c5a75f266bfaa75075b0e6c Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Fri, 2 May 2025 15:50:01 -0400
Subject: [PATCH 2/3] Update fastmcp.mdx
---
docs/servers/fastmcp.mdx | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/servers/fastmcp.mdx b/docs/servers/fastmcp.mdx
index 2f378a66f..d9ca6c384 100644
--- a/docs/servers/fastmcp.mdx
+++ b/docs/servers/fastmcp.mdx
@@ -296,7 +296,7 @@ All of these can be configured directly as parameters when creating the `FastMCP
## Authentication
-
+
FastMCP inherits support for OAuth 2.0 authentication from the MCP protocol, allowing servers to protect their tools and resources behind authentication.
From 5f06c94076171887dfc77946fd27ea5a7230856f Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Fri, 2 May 2025 15:52:46 -0400
Subject: [PATCH 3/3] Update fastmcp.mdx
---
docs/servers/fastmcp.mdx | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/docs/servers/fastmcp.mdx b/docs/servers/fastmcp.mdx
index d9ca6c384..7c3c6da0f 100644
--- a/docs/servers/fastmcp.mdx
+++ b/docs/servers/fastmcp.mdx
@@ -306,7 +306,11 @@ The `mcp.server.auth` module implements an OAuth 2.0 server interface that serve
```python
from fastmcp import FastMCP
-from mcp.server.auth.settings import RevocationOptions, ClientRegistrationOptions, AuthSettings
+from mcp.server.auth.settings import (
+ RevocationOptions,
+ ClientRegistrationOptions,
+ AuthSettings,
+)
# Create a server with authentication