mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Refactor OAuthProxy into focused modules (#2935)
This commit is contained in:
Jeremiah Lowin
GitHub
parent
352ff82eab
commit
c99e0c6351
10 changed files with 890 additions and 886 deletions
4
loq.toml
4
loq.toml
|
|
@ -75,8 +75,8 @@ path = "tests/utilities/test_json_schema_type.py"
|
|||
max_lines = 1584
|
||||
|
||||
[[rules]]
|
||||
path = "src/fastmcp/server/auth/oauth_proxy.py"
|
||||
max_lines = 2282
|
||||
path = "src/fastmcp/server/auth/oauth_proxy/proxy.py"
|
||||
max_lines = 1600
|
||||
|
||||
[[rules]]
|
||||
path = "tests/server/test_dependencies.py"
|
||||
|
|
|
|||
14
src/fastmcp/server/auth/oauth_proxy/__init__.py
Normal file
14
src/fastmcp/server/auth/oauth_proxy/__init__.py
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
"""OAuth Proxy Provider for FastMCP.
|
||||
|
||||
This package provides OAuth proxy functionality split across multiple modules:
|
||||
- models: Pydantic models and constants
|
||||
- ui: HTML generation functions
|
||||
- consent: Consent management mixin
|
||||
- proxy: Main OAuthProxy class
|
||||
"""
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy.proxy import OAuthProxy
|
||||
|
||||
__all__ = [
|
||||
"OAuthProxy",
|
||||
]
|
||||
361
src/fastmcp/server/auth/oauth_proxy/consent.py
Normal file
361
src/fastmcp/server/auth/oauth_proxy/consent.py
Normal file
|
|
@ -0,0 +1,361 @@
|
|||
"""OAuth Proxy Consent Management.
|
||||
|
||||
This module contains consent management functionality for the OAuth proxy.
|
||||
The ConsentMixin class provides methods for handling user consent flows,
|
||||
cookie management, and consent page rendering.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import secrets
|
||||
import time
|
||||
from base64 import urlsafe_b64encode
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from urllib.parse import urlencode, urlparse
|
||||
|
||||
from pydantic import AnyUrl
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import HTMLResponse, RedirectResponse
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy.ui import create_consent_html
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.ui import create_secure_html_response
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.server.auth.oauth_proxy.proxy import OAuthProxy
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class ConsentMixin:
|
||||
"""Mixin class providing consent management functionality for OAuthProxy.
|
||||
|
||||
This mixin contains all methods related to:
|
||||
- Cookie signing and verification
|
||||
- Consent page rendering
|
||||
- Consent approval/denial handling
|
||||
- URI normalization for consent tracking
|
||||
"""
|
||||
|
||||
def _normalize_uri(self, uri: str) -> str:
|
||||
"""Normalize a URI to a canonical form for consent tracking."""
|
||||
parsed = urlparse(uri)
|
||||
path = parsed.path or ""
|
||||
normalized = f"{parsed.scheme.lower()}://{parsed.netloc.lower()}{path}"
|
||||
if normalized.endswith("/") and len(path) > 1:
|
||||
normalized = normalized[:-1]
|
||||
return normalized
|
||||
|
||||
def _make_client_key(self, client_id: str, redirect_uri: str | AnyUrl) -> str:
|
||||
"""Create a stable key for consent tracking from client_id and redirect_uri."""
|
||||
normalized = self._normalize_uri(str(redirect_uri))
|
||||
return f"{client_id}:{normalized}"
|
||||
|
||||
def _cookie_name(self: OAuthProxy, base_name: str) -> str:
|
||||
"""Return secure cookie name for HTTPS, fallback for HTTP development."""
|
||||
if self._is_https:
|
||||
return f"__Host-{base_name}"
|
||||
return f"__{base_name}"
|
||||
|
||||
def _sign_cookie(self: OAuthProxy, payload: str) -> str:
|
||||
"""Sign a cookie payload with HMAC-SHA256.
|
||||
|
||||
Returns: base64(payload).base64(signature)
|
||||
"""
|
||||
# Use upstream client secret as signing key
|
||||
key = self._upstream_client_secret.get_secret_value().encode()
|
||||
signature = hmac.new(key, payload.encode(), hashlib.sha256).digest()
|
||||
signature_b64 = base64.b64encode(signature).decode()
|
||||
return f"{payload}.{signature_b64}"
|
||||
|
||||
def _verify_cookie(self: OAuthProxy, signed_value: str) -> str | None:
|
||||
"""Verify and extract payload from signed cookie.
|
||||
|
||||
Returns: payload if signature valid, None otherwise
|
||||
"""
|
||||
try:
|
||||
if "." not in signed_value:
|
||||
return None
|
||||
payload, signature_b64 = signed_value.rsplit(".", 1)
|
||||
|
||||
# Verify signature
|
||||
key = self._upstream_client_secret.get_secret_value().encode()
|
||||
expected_sig = hmac.new(key, payload.encode(), hashlib.sha256).digest()
|
||||
provided_sig = base64.b64decode(signature_b64.encode())
|
||||
|
||||
# Constant-time comparison
|
||||
if not hmac.compare_digest(expected_sig, provided_sig):
|
||||
return None
|
||||
|
||||
return payload
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _decode_list_cookie(
|
||||
self: OAuthProxy, request: Request, base_name: str
|
||||
) -> list[str]:
|
||||
"""Decode and verify a signed base64-encoded JSON list from cookie. Returns [] if missing/invalid."""
|
||||
# Prefer secure name, but also check non-secure variant for dev
|
||||
secure_name = self._cookie_name(base_name)
|
||||
raw = request.cookies.get(secure_name) or request.cookies.get(f"__{base_name}")
|
||||
if not raw:
|
||||
return []
|
||||
try:
|
||||
# Verify signature
|
||||
payload = self._verify_cookie(raw)
|
||||
if not payload:
|
||||
logger.debug("Cookie signature verification failed for %s", secure_name)
|
||||
return []
|
||||
|
||||
# Decode payload
|
||||
data = base64.b64decode(payload.encode())
|
||||
value = json.loads(data.decode())
|
||||
if isinstance(value, list):
|
||||
return [str(x) for x in value]
|
||||
except Exception:
|
||||
logger.debug("Failed to decode cookie %s; treating as empty", secure_name)
|
||||
return []
|
||||
|
||||
def _encode_list_cookie(self: OAuthProxy, values: list[str]) -> str:
|
||||
"""Encode values to base64 and sign with HMAC.
|
||||
|
||||
Returns: signed cookie value (payload.signature)
|
||||
"""
|
||||
payload = json.dumps(values, separators=(",", ":")).encode()
|
||||
payload_b64 = base64.b64encode(payload).decode()
|
||||
return self._sign_cookie(payload_b64)
|
||||
|
||||
def _set_list_cookie(
|
||||
self: OAuthProxy,
|
||||
response: HTMLResponse | RedirectResponse,
|
||||
base_name: str,
|
||||
value_b64: str,
|
||||
max_age: int,
|
||||
) -> None:
|
||||
name = self._cookie_name(base_name)
|
||||
response.set_cookie(
|
||||
name,
|
||||
value_b64,
|
||||
max_age=max_age,
|
||||
secure=self._is_https,
|
||||
httponly=True,
|
||||
samesite="lax",
|
||||
path="/",
|
||||
)
|
||||
|
||||
def _build_upstream_authorize_url(
|
||||
self: OAuthProxy, txn_id: str, transaction: dict[str, Any]
|
||||
) -> str:
|
||||
"""Construct the upstream IdP authorization URL using stored transaction data."""
|
||||
query_params: dict[str, Any] = {
|
||||
"response_type": "code",
|
||||
"client_id": self._upstream_client_id,
|
||||
"redirect_uri": f"{str(self.base_url).rstrip('/')}{self._redirect_path}",
|
||||
"state": txn_id,
|
||||
}
|
||||
|
||||
scopes_to_use = transaction.get("scopes") or self.required_scopes or []
|
||||
if scopes_to_use:
|
||||
query_params["scope"] = " ".join(scopes_to_use)
|
||||
|
||||
# If PKCE forwarding was enabled, include the proxy challenge
|
||||
proxy_code_verifier = transaction.get("proxy_code_verifier")
|
||||
if proxy_code_verifier:
|
||||
challenge_bytes = hashlib.sha256(proxy_code_verifier.encode()).digest()
|
||||
proxy_code_challenge = (
|
||||
urlsafe_b64encode(challenge_bytes).decode().rstrip("=")
|
||||
)
|
||||
query_params["code_challenge"] = proxy_code_challenge
|
||||
query_params["code_challenge_method"] = "S256"
|
||||
|
||||
# Forward resource indicator if present in transaction
|
||||
if resource := transaction.get("resource"):
|
||||
query_params["resource"] = resource
|
||||
|
||||
# Extra configured parameters
|
||||
if self._extra_authorize_params:
|
||||
query_params.update(self._extra_authorize_params)
|
||||
|
||||
separator = "&" if "?" in self._upstream_authorization_endpoint else "?"
|
||||
return f"{self._upstream_authorization_endpoint}{separator}{urlencode(query_params)}"
|
||||
|
||||
async def _handle_consent(
|
||||
self: OAuthProxy, request: Request
|
||||
) -> HTMLResponse | RedirectResponse:
|
||||
"""Handle consent page - dispatch to GET or POST handler based on method."""
|
||||
if request.method == "POST":
|
||||
return await self._submit_consent(request)
|
||||
return await self._show_consent_page(request)
|
||||
|
||||
async def _show_consent_page(
|
||||
self: OAuthProxy, request: Request
|
||||
) -> HTMLResponse | RedirectResponse:
|
||||
"""Display consent page or auto-approve/deny based on cookies."""
|
||||
from fastmcp.server.server import FastMCP
|
||||
|
||||
txn_id = request.query_params.get("txn_id")
|
||||
if not txn_id:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired transaction</p>", status_code=400
|
||||
)
|
||||
|
||||
txn_model = await self._transaction_store.get(key=txn_id)
|
||||
if not txn_model:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired transaction</p>", status_code=400
|
||||
)
|
||||
|
||||
txn = txn_model.model_dump()
|
||||
client_key = self._make_client_key(txn["client_id"], txn["client_redirect_uri"])
|
||||
|
||||
approved = set(self._decode_list_cookie(request, "MCP_APPROVED_CLIENTS"))
|
||||
denied = set(self._decode_list_cookie(request, "MCP_DENIED_CLIENTS"))
|
||||
|
||||
if client_key in approved:
|
||||
upstream_url = self._build_upstream_authorize_url(txn_id, txn)
|
||||
return RedirectResponse(url=upstream_url, status_code=302)
|
||||
|
||||
if client_key in denied:
|
||||
callback_params = {
|
||||
"error": "access_denied",
|
||||
"state": txn.get("client_state") or "",
|
||||
}
|
||||
sep = "&" if "?" in txn["client_redirect_uri"] else "?"
|
||||
return RedirectResponse(
|
||||
url=f"{txn['client_redirect_uri']}{sep}{urlencode(callback_params)}",
|
||||
status_code=302,
|
||||
)
|
||||
|
||||
# Need consent: issue CSRF token and show HTML
|
||||
csrf_token = secrets.token_urlsafe(32)
|
||||
csrf_expires_at = time.time() + 15 * 60
|
||||
|
||||
# Update transaction with CSRF token
|
||||
txn_model.csrf_token = csrf_token
|
||||
txn_model.csrf_expires_at = csrf_expires_at
|
||||
await self._transaction_store.put(
|
||||
key=txn_id, value=txn_model, ttl=15 * 60
|
||||
) # Auto-expire after 15 minutes
|
||||
|
||||
# Update dict for use in HTML generation
|
||||
txn["csrf_token"] = csrf_token
|
||||
txn["csrf_expires_at"] = csrf_expires_at
|
||||
|
||||
# Load client to get client_name if available
|
||||
client = await self.get_client(txn["client_id"])
|
||||
client_name = getattr(client, "client_name", None) if client else None
|
||||
|
||||
# Extract server metadata from app state
|
||||
fastmcp = getattr(request.app.state, "fastmcp_server", None)
|
||||
|
||||
if isinstance(fastmcp, FastMCP):
|
||||
server_name = fastmcp.name
|
||||
icons = fastmcp.icons
|
||||
server_icon_url = icons[0].src if icons else None
|
||||
server_website_url = fastmcp.website_url
|
||||
else:
|
||||
server_name = None
|
||||
server_icon_url = None
|
||||
server_website_url = None
|
||||
|
||||
html = create_consent_html(
|
||||
client_id=txn["client_id"],
|
||||
redirect_uri=txn["client_redirect_uri"],
|
||||
scopes=txn.get("scopes") or [],
|
||||
txn_id=txn_id,
|
||||
csrf_token=csrf_token,
|
||||
client_name=client_name,
|
||||
server_name=server_name,
|
||||
server_icon_url=server_icon_url,
|
||||
server_website_url=server_website_url,
|
||||
csp_policy=self._consent_csp_policy,
|
||||
)
|
||||
response = create_secure_html_response(html)
|
||||
# Store CSRF in cookie with short lifetime
|
||||
self._set_list_cookie(
|
||||
response,
|
||||
"MCP_CONSENT_STATE",
|
||||
self._encode_list_cookie([csrf_token]),
|
||||
max_age=15 * 60,
|
||||
)
|
||||
return response
|
||||
|
||||
async def _submit_consent(
|
||||
self: OAuthProxy, request: Request
|
||||
) -> RedirectResponse | HTMLResponse:
|
||||
"""Handle consent approval/denial, set cookies, and redirect appropriately."""
|
||||
form = await request.form()
|
||||
txn_id = str(form.get("txn_id", ""))
|
||||
action = str(form.get("action", ""))
|
||||
csrf_token = str(form.get("csrf_token", ""))
|
||||
|
||||
if not txn_id:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired transaction</p>", status_code=400
|
||||
)
|
||||
|
||||
txn_model = await self._transaction_store.get(key=txn_id)
|
||||
if not txn_model:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired transaction</p>", status_code=400
|
||||
)
|
||||
|
||||
txn = txn_model.model_dump()
|
||||
expected_csrf = txn.get("csrf_token")
|
||||
expires_at = float(txn.get("csrf_expires_at") or 0)
|
||||
|
||||
if not expected_csrf or csrf_token != expected_csrf or time.time() > expires_at:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired consent token</p>", status_code=400
|
||||
)
|
||||
|
||||
client_key = self._make_client_key(txn["client_id"], txn["client_redirect_uri"])
|
||||
|
||||
if action == "approve":
|
||||
approved = set(self._decode_list_cookie(request, "MCP_APPROVED_CLIENTS"))
|
||||
if client_key not in approved:
|
||||
approved.add(client_key)
|
||||
approved_b64 = self._encode_list_cookie(sorted(approved))
|
||||
|
||||
upstream_url = self._build_upstream_authorize_url(txn_id, txn)
|
||||
response = RedirectResponse(url=upstream_url, status_code=302)
|
||||
self._set_list_cookie(
|
||||
response, "MCP_APPROVED_CLIENTS", approved_b64, max_age=365 * 24 * 3600
|
||||
)
|
||||
# Clear CSRF cookie by setting empty short-lived value
|
||||
self._set_list_cookie(
|
||||
response, "MCP_CONSENT_STATE", self._encode_list_cookie([]), max_age=60
|
||||
)
|
||||
return response
|
||||
|
||||
elif action == "deny":
|
||||
denied = set(self._decode_list_cookie(request, "MCP_DENIED_CLIENTS"))
|
||||
if client_key not in denied:
|
||||
denied.add(client_key)
|
||||
denied_b64 = self._encode_list_cookie(sorted(denied))
|
||||
|
||||
callback_params = {
|
||||
"error": "access_denied",
|
||||
"state": txn.get("client_state") or "",
|
||||
}
|
||||
sep = "&" if "?" in txn["client_redirect_uri"] else "?"
|
||||
client_callback_url = (
|
||||
f"{txn['client_redirect_uri']}{sep}{urlencode(callback_params)}"
|
||||
)
|
||||
response = RedirectResponse(url=client_callback_url, status_code=302)
|
||||
self._set_list_cookie(
|
||||
response, "MCP_DENIED_CLIENTS", denied_b64, max_age=365 * 24 * 3600
|
||||
)
|
||||
self._set_list_cookie(
|
||||
response, "MCP_CONSENT_STATE", self._encode_list_cookie([]), max_age=60
|
||||
)
|
||||
return response
|
||||
|
||||
else:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid action</p>", status_code=400
|
||||
)
|
||||
178
src/fastmcp/server/auth/oauth_proxy/models.py
Normal file
178
src/fastmcp/server/auth/oauth_proxy/models.py
Normal file
|
|
@ -0,0 +1,178 @@
|
|||
"""OAuth Proxy Models and Constants.
|
||||
|
||||
This module contains all Pydantic models and constants used by the OAuth proxy.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from typing import Any, Final
|
||||
|
||||
from mcp.shared.auth import OAuthClientInformationFull
|
||||
from pydantic import AnyUrl, BaseModel, Field
|
||||
|
||||
from fastmcp.server.auth.redirect_validation import validate_redirect_uri
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Constants
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
# Default token expiration times
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS: Final[int] = 60 * 60 # 1 hour
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_NO_REFRESH_SECONDS: Final[int] = (
|
||||
60 * 60 * 24 * 365
|
||||
) # 1 year
|
||||
DEFAULT_AUTH_CODE_EXPIRY_SECONDS: Final[int] = 5 * 60 # 5 minutes
|
||||
|
||||
# HTTP client timeout
|
||||
HTTP_TIMEOUT_SECONDS: Final[int] = 30
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Pydantic Models
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
|
||||
class OAuthTransaction(BaseModel):
|
||||
"""OAuth transaction state for consent flow.
|
||||
|
||||
Stored server-side to track active authorization flows with client context.
|
||||
Includes CSRF tokens for consent protection per MCP security best practices.
|
||||
"""
|
||||
|
||||
txn_id: str
|
||||
client_id: str
|
||||
client_redirect_uri: str
|
||||
client_state: str
|
||||
code_challenge: str | None
|
||||
code_challenge_method: str
|
||||
scopes: list[str]
|
||||
created_at: float
|
||||
resource: str | None = None
|
||||
proxy_code_verifier: str | None = None
|
||||
csrf_token: str | None = None
|
||||
csrf_expires_at: float | None = None
|
||||
|
||||
|
||||
class ClientCode(BaseModel):
|
||||
"""Client authorization code with PKCE and upstream tokens.
|
||||
|
||||
Stored server-side after upstream IdP callback. Contains the upstream
|
||||
tokens bound to the client's PKCE challenge for secure token exchange.
|
||||
"""
|
||||
|
||||
code: str
|
||||
client_id: str
|
||||
redirect_uri: str
|
||||
code_challenge: str | None
|
||||
code_challenge_method: str
|
||||
scopes: list[str]
|
||||
idp_tokens: dict[str, Any]
|
||||
expires_at: float
|
||||
created_at: float
|
||||
|
||||
|
||||
class UpstreamTokenSet(BaseModel):
|
||||
"""Stored upstream OAuth tokens from identity provider.
|
||||
|
||||
These tokens are obtained from the upstream provider (Google, GitHub, etc.)
|
||||
and stored in plaintext within this model. Encryption is handled transparently
|
||||
at the storage layer via FernetEncryptionWrapper. Tokens are never exposed to MCP clients.
|
||||
"""
|
||||
|
||||
upstream_token_id: str # Unique ID for this token set
|
||||
access_token: str # Upstream access token
|
||||
refresh_token: str | None # Upstream refresh token
|
||||
refresh_token_expires_at: (
|
||||
float | None
|
||||
) # Unix timestamp when refresh token expires (if known)
|
||||
expires_at: float # Unix timestamp when access token expires
|
||||
token_type: str # Usually "Bearer"
|
||||
scope: str # Space-separated scopes
|
||||
client_id: str # MCP client this is bound to
|
||||
created_at: float # Unix timestamp
|
||||
raw_token_data: dict[str, Any] = Field(default_factory=dict) # Full token response
|
||||
|
||||
|
||||
class JTIMapping(BaseModel):
|
||||
"""Maps FastMCP token JTI to upstream token ID.
|
||||
|
||||
This allows stateless JWT validation while still being able to look up
|
||||
the corresponding upstream token when tools need to access upstream APIs.
|
||||
"""
|
||||
|
||||
jti: str # JWT ID from FastMCP-issued token
|
||||
upstream_token_id: str # References UpstreamTokenSet
|
||||
created_at: float # Unix timestamp
|
||||
|
||||
|
||||
class RefreshTokenMetadata(BaseModel):
|
||||
"""Metadata for a refresh token, stored keyed by token hash.
|
||||
|
||||
We store only metadata (not the token itself) for security - if storage
|
||||
is compromised, attackers get hashes they can't reverse into usable tokens.
|
||||
"""
|
||||
|
||||
client_id: str
|
||||
scopes: list[str]
|
||||
expires_at: int | None = None
|
||||
created_at: float
|
||||
|
||||
|
||||
def _hash_token(token: str) -> str:
|
||||
"""Hash a token for secure storage lookup.
|
||||
|
||||
Uses SHA-256 to create a one-way hash. The original token cannot be
|
||||
recovered from the hash, providing defense in depth if storage is compromised.
|
||||
"""
|
||||
return hashlib.sha256(token.encode()).hexdigest()
|
||||
|
||||
|
||||
class ProxyDCRClient(OAuthClientInformationFull):
|
||||
"""Client for DCR proxy with configurable redirect URI validation.
|
||||
|
||||
This special client class is critical for the OAuth proxy to work correctly
|
||||
with Dynamic Client Registration (DCR). Here's why it exists:
|
||||
|
||||
Problem:
|
||||
--------
|
||||
When MCP clients use OAuth, they dynamically register with random localhost
|
||||
ports (e.g., http://localhost:55454/callback). The OAuth proxy needs to:
|
||||
1. Accept these dynamic redirect URIs from clients based on configured patterns
|
||||
2. Use its own fixed redirect URI with the upstream provider (Google, GitHub, etc.)
|
||||
3. Forward the authorization code back to the client's dynamic URI
|
||||
|
||||
Solution:
|
||||
---------
|
||||
This class validates redirect URIs against configurable patterns,
|
||||
while the proxy internally uses its own fixed redirect URI with the upstream
|
||||
provider. This allows the flow to work even when clients reconnect with
|
||||
different ports or when tokens are cached.
|
||||
|
||||
Without proper validation, clients could get "Redirect URI not registered" errors
|
||||
when trying to authenticate with cached tokens, or security vulnerabilities could
|
||||
arise from accepting arbitrary redirect URIs.
|
||||
"""
|
||||
|
||||
allowed_redirect_uri_patterns: list[str] | None = Field(default=None)
|
||||
client_name: str | None = Field(default=None)
|
||||
|
||||
def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
|
||||
"""Validate redirect URI against allowed patterns.
|
||||
|
||||
Since we're acting as a proxy and clients register dynamically,
|
||||
we validate their redirect URIs against configurable patterns.
|
||||
This is essential for cached token scenarios where the client may
|
||||
reconnect with a different port.
|
||||
"""
|
||||
if redirect_uri is not None:
|
||||
# Validate against allowed patterns
|
||||
if validate_redirect_uri(
|
||||
redirect_uri=redirect_uri,
|
||||
allowed_patterns=self.allowed_redirect_uri_patterns,
|
||||
):
|
||||
return redirect_uri
|
||||
# Fall back to normal validation if not in allowed patterns
|
||||
return super().validate_redirect_uri(redirect_uri)
|
||||
# If no redirect_uri provided, use default behavior
|
||||
return super().validate_redirect_uri(redirect_uri)
|
||||
|
|
@ -18,15 +18,12 @@ production use with enterprise identity providers.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import secrets
|
||||
import time
|
||||
from base64 import urlsafe_b64encode
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
from urllib.parse import urlencode, urlparse
|
||||
from typing import Any
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import httpx
|
||||
from authlib.common.security import generate_token
|
||||
|
|
@ -48,7 +45,7 @@ from mcp.server.auth.settings import (
|
|||
RevocationOptions,
|
||||
)
|
||||
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
|
||||
from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, SecretStr
|
||||
from pydantic import AnyHttpUrl, AnyUrl, SecretStr
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import HTMLResponse, RedirectResponse
|
||||
from starlette.routing import Route
|
||||
|
|
@ -61,457 +58,27 @@ from fastmcp.server.auth.jwt_issuer import (
|
|||
JWTIssuer,
|
||||
derive_jwt_key,
|
||||
)
|
||||
from fastmcp.server.auth.redirect_validation import (
|
||||
validate_redirect_uri,
|
||||
from fastmcp.server.auth.oauth_proxy.consent import ConsentMixin
|
||||
from fastmcp.server.auth.oauth_proxy.models import (
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_NO_REFRESH_SECONDS,
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS,
|
||||
DEFAULT_AUTH_CODE_EXPIRY_SECONDS,
|
||||
HTTP_TIMEOUT_SECONDS,
|
||||
ClientCode,
|
||||
JTIMapping,
|
||||
OAuthTransaction,
|
||||
ProxyDCRClient,
|
||||
RefreshTokenMetadata,
|
||||
UpstreamTokenSet,
|
||||
_hash_token,
|
||||
)
|
||||
from fastmcp.server.auth.oauth_proxy.ui import create_error_html
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.ui import (
|
||||
BUTTON_STYLES,
|
||||
DETAIL_BOX_STYLES,
|
||||
DETAILS_STYLES,
|
||||
INFO_BOX_STYLES,
|
||||
REDIRECT_SECTION_STYLES,
|
||||
TOOLTIP_STYLES,
|
||||
create_logo,
|
||||
create_page,
|
||||
create_secure_html_response,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Constants
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
# Default token expiration times
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS: Final[int] = 60 * 60 # 1 hour
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_NO_REFRESH_SECONDS: Final[int] = (
|
||||
60 * 60 * 24 * 365
|
||||
) # 1 year
|
||||
DEFAULT_AUTH_CODE_EXPIRY_SECONDS: Final[int] = 5 * 60 # 5 minutes
|
||||
|
||||
# HTTP client timeout
|
||||
HTTP_TIMEOUT_SECONDS: Final[int] = 30
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Pydantic Models
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
|
||||
class OAuthTransaction(BaseModel):
|
||||
"""OAuth transaction state for consent flow.
|
||||
|
||||
Stored server-side to track active authorization flows with client context.
|
||||
Includes CSRF tokens for consent protection per MCP security best practices.
|
||||
"""
|
||||
|
||||
txn_id: str
|
||||
client_id: str
|
||||
client_redirect_uri: str
|
||||
client_state: str
|
||||
code_challenge: str | None
|
||||
code_challenge_method: str
|
||||
scopes: list[str]
|
||||
created_at: float
|
||||
resource: str | None = None
|
||||
proxy_code_verifier: str | None = None
|
||||
csrf_token: str | None = None
|
||||
csrf_expires_at: float | None = None
|
||||
|
||||
|
||||
class ClientCode(BaseModel):
|
||||
"""Client authorization code with PKCE and upstream tokens.
|
||||
|
||||
Stored server-side after upstream IdP callback. Contains the upstream
|
||||
tokens bound to the client's PKCE challenge for secure token exchange.
|
||||
"""
|
||||
|
||||
code: str
|
||||
client_id: str
|
||||
redirect_uri: str
|
||||
code_challenge: str | None
|
||||
code_challenge_method: str
|
||||
scopes: list[str]
|
||||
idp_tokens: dict[str, Any]
|
||||
expires_at: float
|
||||
created_at: float
|
||||
|
||||
|
||||
class UpstreamTokenSet(BaseModel):
|
||||
"""Stored upstream OAuth tokens from identity provider.
|
||||
|
||||
These tokens are obtained from the upstream provider (Google, GitHub, etc.)
|
||||
and stored in plaintext within this model. Encryption is handled transparently
|
||||
at the storage layer via FernetEncryptionWrapper. Tokens are never exposed to MCP clients.
|
||||
"""
|
||||
|
||||
upstream_token_id: str # Unique ID for this token set
|
||||
access_token: str # Upstream access token
|
||||
refresh_token: str | None # Upstream refresh token
|
||||
refresh_token_expires_at: (
|
||||
float | None
|
||||
) # Unix timestamp when refresh token expires (if known)
|
||||
expires_at: float # Unix timestamp when access token expires
|
||||
token_type: str # Usually "Bearer"
|
||||
scope: str # Space-separated scopes
|
||||
client_id: str # MCP client this is bound to
|
||||
created_at: float # Unix timestamp
|
||||
raw_token_data: dict[str, Any] = Field(default_factory=dict) # Full token response
|
||||
|
||||
|
||||
class JTIMapping(BaseModel):
|
||||
"""Maps FastMCP token JTI to upstream token ID.
|
||||
|
||||
This allows stateless JWT validation while still being able to look up
|
||||
the corresponding upstream token when tools need to access upstream APIs.
|
||||
"""
|
||||
|
||||
jti: str # JWT ID from FastMCP-issued token
|
||||
upstream_token_id: str # References UpstreamTokenSet
|
||||
created_at: float # Unix timestamp
|
||||
|
||||
|
||||
class RefreshTokenMetadata(BaseModel):
|
||||
"""Metadata for a refresh token, stored keyed by token hash.
|
||||
|
||||
We store only metadata (not the token itself) for security - if storage
|
||||
is compromised, attackers get hashes they can't reverse into usable tokens.
|
||||
"""
|
||||
|
||||
client_id: str
|
||||
scopes: list[str]
|
||||
expires_at: int | None = None
|
||||
created_at: float
|
||||
|
||||
|
||||
def _hash_token(token: str) -> str:
|
||||
"""Hash a token for secure storage lookup.
|
||||
|
||||
Uses SHA-256 to create a one-way hash. The original token cannot be
|
||||
recovered from the hash, providing defense in depth if storage is compromised.
|
||||
"""
|
||||
return hashlib.sha256(token.encode()).hexdigest()
|
||||
|
||||
|
||||
class ProxyDCRClient(OAuthClientInformationFull):
|
||||
"""Client for DCR proxy with configurable redirect URI validation.
|
||||
|
||||
This special client class is critical for the OAuth proxy to work correctly
|
||||
with Dynamic Client Registration (DCR). Here's why it exists:
|
||||
|
||||
Problem:
|
||||
--------
|
||||
When MCP clients use OAuth, they dynamically register with random localhost
|
||||
ports (e.g., http://localhost:55454/callback). The OAuth proxy needs to:
|
||||
1. Accept these dynamic redirect URIs from clients based on configured patterns
|
||||
2. Use its own fixed redirect URI with the upstream provider (Google, GitHub, etc.)
|
||||
3. Forward the authorization code back to the client's dynamic URI
|
||||
|
||||
Solution:
|
||||
---------
|
||||
This class validates redirect URIs against configurable patterns,
|
||||
while the proxy internally uses its own fixed redirect URI with the upstream
|
||||
provider. This allows the flow to work even when clients reconnect with
|
||||
different ports or when tokens are cached.
|
||||
|
||||
Without proper validation, clients could get "Redirect URI not registered" errors
|
||||
when trying to authenticate with cached tokens, or security vulnerabilities could
|
||||
arise from accepting arbitrary redirect URIs.
|
||||
"""
|
||||
|
||||
allowed_redirect_uri_patterns: list[str] | None = Field(default=None)
|
||||
client_name: str | None = Field(default=None)
|
||||
|
||||
def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
|
||||
"""Validate redirect URI against allowed patterns.
|
||||
|
||||
Since we're acting as a proxy and clients register dynamically,
|
||||
we validate their redirect URIs against configurable patterns.
|
||||
This is essential for cached token scenarios where the client may
|
||||
reconnect with a different port.
|
||||
"""
|
||||
if redirect_uri is not None:
|
||||
# Validate against allowed patterns
|
||||
if validate_redirect_uri(
|
||||
redirect_uri=redirect_uri,
|
||||
allowed_patterns=self.allowed_redirect_uri_patterns,
|
||||
):
|
||||
return redirect_uri
|
||||
# Fall back to normal validation if not in allowed patterns
|
||||
return super().validate_redirect_uri(redirect_uri)
|
||||
# If no redirect_uri provided, use default behavior
|
||||
return super().validate_redirect_uri(redirect_uri)
|
||||
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Helper Functions
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
|
||||
def create_consent_html(
|
||||
client_id: str,
|
||||
redirect_uri: str,
|
||||
scopes: list[str],
|
||||
txn_id: str,
|
||||
csrf_token: str,
|
||||
client_name: str | None = None,
|
||||
title: str = "Application Access Request",
|
||||
server_name: str | None = None,
|
||||
server_icon_url: str | None = None,
|
||||
server_website_url: str | None = None,
|
||||
client_website_url: str | None = None,
|
||||
csp_policy: str | None = None,
|
||||
) -> str:
|
||||
"""Create a styled HTML consent page for OAuth authorization requests.
|
||||
|
||||
Args:
|
||||
csp_policy: Content Security Policy override.
|
||||
If None, uses the built-in CSP policy with appropriate directives.
|
||||
If empty string "", disables CSP entirely (no meta tag is rendered).
|
||||
If a non-empty string, uses that as the CSP policy value.
|
||||
"""
|
||||
import html as html_module
|
||||
|
||||
client_display = html_module.escape(client_name or client_id)
|
||||
server_name_escaped = html_module.escape(server_name or "FastMCP")
|
||||
|
||||
# Make server name a hyperlink if website URL is available
|
||||
if server_website_url:
|
||||
website_url_escaped = html_module.escape(server_website_url)
|
||||
server_display = f'<a href="{website_url_escaped}" target="_blank" rel="noopener noreferrer" class="server-name-link">{server_name_escaped}</a>'
|
||||
else:
|
||||
server_display = server_name_escaped
|
||||
|
||||
# Build intro box with call-to-action
|
||||
intro_box = f"""
|
||||
<div class="info-box">
|
||||
<p>The application <strong>{client_display}</strong> wants to access the MCP server <strong>{server_display}</strong>. Please ensure you recognize the callback address below.</p>
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Build redirect URI section (yellow box, centered)
|
||||
redirect_uri_escaped = html_module.escape(redirect_uri)
|
||||
redirect_section = f"""
|
||||
<div class="redirect-section">
|
||||
<span class="label">Credentials will be sent to:</span>
|
||||
<div class="value">{redirect_uri_escaped}</div>
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Build advanced details with collapsible section
|
||||
detail_rows = [
|
||||
("Application Name", html_module.escape(client_name or client_id)),
|
||||
("Application Website", html_module.escape(client_website_url or "N/A")),
|
||||
("Application ID", client_id),
|
||||
("Redirect URI", redirect_uri_escaped),
|
||||
(
|
||||
"Requested Scopes",
|
||||
", ".join(html_module.escape(s) for s in scopes) if scopes else "None",
|
||||
),
|
||||
]
|
||||
|
||||
detail_rows_html = "\n".join(
|
||||
[
|
||||
f"""
|
||||
<div class="detail-row">
|
||||
<div class="detail-label">{label}:</div>
|
||||
<div class="detail-value">{value}</div>
|
||||
</div>
|
||||
"""
|
||||
for label, value in detail_rows
|
||||
]
|
||||
)
|
||||
|
||||
advanced_details = f"""
|
||||
<details>
|
||||
<summary>Advanced Details</summary>
|
||||
<div class="detail-box">
|
||||
{detail_rows_html}
|
||||
</div>
|
||||
</details>
|
||||
"""
|
||||
|
||||
# Build form with buttons
|
||||
# Use empty action to submit to current URL (/consent or /mcp/consent)
|
||||
# The POST handler is registered at the same path as GET
|
||||
form = f"""
|
||||
<form id="consentForm" method="POST" action="">
|
||||
<input type="hidden" name="txn_id" value="{txn_id}" />
|
||||
<input type="hidden" name="csrf_token" value="{csrf_token}" />
|
||||
<input type="hidden" name="submit" value="true" />
|
||||
<div class="button-group">
|
||||
<button type="submit" name="action" value="approve" class="btn-approve">Allow Access</button>
|
||||
<button type="submit" name="action" value="deny" class="btn-deny">Deny</button>
|
||||
</div>
|
||||
</form>
|
||||
"""
|
||||
|
||||
# Build help link with tooltip (identical to current implementation)
|
||||
help_link = """
|
||||
<div class="help-link-container">
|
||||
<span class="help-link">
|
||||
Why am I seeing this?
|
||||
<span class="tooltip">
|
||||
This FastMCP server requires your consent to allow a new client
|
||||
to connect. This protects you from <a
|
||||
href="https://modelcontextprotocol.io/specification/2025-06-18/basic/security_best_practices#confused-deputy-problem"
|
||||
target="_blank" class="tooltip-link">confused deputy
|
||||
attacks</a>, where malicious clients could impersonate you
|
||||
and steal access.<br><br>
|
||||
<a
|
||||
href="https://gofastmcp.com/servers/auth/oauth-proxy#confused-deputy-attacks"
|
||||
target="_blank" class="tooltip-link">Learn more about
|
||||
FastMCP security →</a>
|
||||
</span>
|
||||
</span>
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Build the page content
|
||||
content = f"""
|
||||
<div class="container">
|
||||
{create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
|
||||
<h1>Application Access Request</h1>
|
||||
{intro_box}
|
||||
{redirect_section}
|
||||
{advanced_details}
|
||||
{form}
|
||||
</div>
|
||||
{help_link}
|
||||
"""
|
||||
|
||||
# Additional styles needed for this page
|
||||
additional_styles = (
|
||||
INFO_BOX_STYLES
|
||||
+ REDIRECT_SECTION_STYLES
|
||||
+ DETAILS_STYLES
|
||||
+ DETAIL_BOX_STYLES
|
||||
+ BUTTON_STYLES
|
||||
+ TOOLTIP_STYLES
|
||||
)
|
||||
|
||||
# Determine CSP policy to use
|
||||
# If csp_policy is None, build the default CSP policy
|
||||
# If csp_policy is empty string, CSP will be disabled entirely in create_page
|
||||
# If csp_policy is a non-empty string, use it as-is
|
||||
if csp_policy is None:
|
||||
# Need to allow form-action for form submission
|
||||
# Chrome requires explicit scheme declarations in CSP form-action when redirect chains
|
||||
# end in custom protocol schemes (e.g., cursor://). Parse redirect_uri to include its scheme.
|
||||
parsed_redirect = urlparse(redirect_uri)
|
||||
redirect_scheme = parsed_redirect.scheme.lower()
|
||||
|
||||
# Build form-action directive with standard schemes plus custom protocol if present
|
||||
form_action_schemes = ["https:", "http:"]
|
||||
if redirect_scheme and redirect_scheme not in ("http", "https"):
|
||||
# Custom protocol scheme (e.g., cursor:, vscode:, etc.)
|
||||
form_action_schemes.append(f"{redirect_scheme}:")
|
||||
|
||||
form_action_directive = " ".join(form_action_schemes)
|
||||
csp_policy = f"default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'; form-action {form_action_directive}"
|
||||
|
||||
return create_page(
|
||||
content=content,
|
||||
title=title,
|
||||
additional_styles=additional_styles,
|
||||
csp_policy=csp_policy,
|
||||
)
|
||||
|
||||
|
||||
def create_error_html(
|
||||
error_title: str,
|
||||
error_message: str,
|
||||
error_details: dict[str, str] | None = None,
|
||||
server_name: str | None = None,
|
||||
server_icon_url: str | None = None,
|
||||
) -> str:
|
||||
"""Create a styled HTML error page for OAuth errors.
|
||||
|
||||
Args:
|
||||
error_title: The error title (e.g., "OAuth Error", "Authorization Failed")
|
||||
error_message: The main error message to display
|
||||
error_details: Optional dictionary of error details to show (e.g., `{"Error Code": "invalid_client"}`)
|
||||
server_name: Optional server name to display
|
||||
server_icon_url: Optional URL to server icon/logo
|
||||
|
||||
Returns:
|
||||
Complete HTML page as a string
|
||||
"""
|
||||
import html as html_module
|
||||
|
||||
error_message_escaped = html_module.escape(error_message)
|
||||
|
||||
# Build error message box
|
||||
error_box = f"""
|
||||
<div class="info-box error">
|
||||
<p>{error_message_escaped}</p>
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Build error details section if provided
|
||||
details_section = ""
|
||||
if error_details:
|
||||
detail_rows_html = "\n".join(
|
||||
[
|
||||
f"""
|
||||
<div class="detail-row">
|
||||
<div class="detail-label">{html_module.escape(label)}:</div>
|
||||
<div class="detail-value">{html_module.escape(value)}</div>
|
||||
</div>
|
||||
"""
|
||||
for label, value in error_details.items()
|
||||
]
|
||||
)
|
||||
|
||||
details_section = f"""
|
||||
<details>
|
||||
<summary>Error Details</summary>
|
||||
<div class="detail-box">
|
||||
{detail_rows_html}
|
||||
</div>
|
||||
</details>
|
||||
"""
|
||||
|
||||
# Build the page content
|
||||
content = f"""
|
||||
<div class="container">
|
||||
{create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
|
||||
<h1>{html_module.escape(error_title)}</h1>
|
||||
{error_box}
|
||||
{details_section}
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Additional styles needed for this page
|
||||
# Override .info-box.error to use normal text color instead of red
|
||||
additional_styles = (
|
||||
INFO_BOX_STYLES
|
||||
+ DETAILS_STYLES
|
||||
+ DETAIL_BOX_STYLES
|
||||
+ """
|
||||
.info-box.error {
|
||||
color: #111827;
|
||||
}
|
||||
"""
|
||||
)
|
||||
|
||||
# Simple CSP policy for error pages (no forms needed)
|
||||
csp_policy = "default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'"
|
||||
|
||||
return create_page(
|
||||
content=content,
|
||||
title=error_title,
|
||||
additional_styles=additional_styles,
|
||||
csp_policy=csp_policy,
|
||||
)
|
||||
|
||||
|
||||
class OAuthProxy(OAuthProvider):
|
||||
class OAuthProxy(OAuthProvider, ConsentMixin):
|
||||
"""OAuth provider that presents a DCR-compliant interface while proxying to non-DCR IDPs.
|
||||
|
||||
Purpose
|
||||
|
|
@ -1959,324 +1526,3 @@ class OAuthProxy(OAuthProvider):
|
|||
error_message="Internal server error during OAuth callback processing. Please try again.",
|
||||
)
|
||||
return HTMLResponse(content=html_content, status_code=500)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Consent Interstitial
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def _normalize_uri(self, uri: str) -> str:
|
||||
"""Normalize a URI to a canonical form for consent tracking."""
|
||||
parsed = urlparse(uri)
|
||||
path = parsed.path or ""
|
||||
normalized = f"{parsed.scheme.lower()}://{parsed.netloc.lower()}{path}"
|
||||
if normalized.endswith("/") and len(path) > 1:
|
||||
normalized = normalized[:-1]
|
||||
return normalized
|
||||
|
||||
def _make_client_key(self, client_id: str, redirect_uri: str | AnyUrl) -> str:
|
||||
"""Create a stable key for consent tracking from client_id and redirect_uri."""
|
||||
normalized = self._normalize_uri(str(redirect_uri))
|
||||
return f"{client_id}:{normalized}"
|
||||
|
||||
def _cookie_name(self, base_name: str) -> str:
|
||||
"""Return secure cookie name for HTTPS, fallback for HTTP development."""
|
||||
if self._is_https:
|
||||
return f"__Host-{base_name}"
|
||||
return f"__{base_name}"
|
||||
|
||||
def _sign_cookie(self, payload: str) -> str:
|
||||
"""Sign a cookie payload with HMAC-SHA256.
|
||||
|
||||
Returns: base64(payload).base64(signature)
|
||||
"""
|
||||
# Use upstream client secret as signing key
|
||||
key = self._upstream_client_secret.get_secret_value().encode()
|
||||
signature = hmac.new(key, payload.encode(), hashlib.sha256).digest()
|
||||
signature_b64 = base64.b64encode(signature).decode()
|
||||
return f"{payload}.{signature_b64}"
|
||||
|
||||
def _verify_cookie(self, signed_value: str) -> str | None:
|
||||
"""Verify and extract payload from signed cookie.
|
||||
|
||||
Returns: payload if signature valid, None otherwise
|
||||
"""
|
||||
try:
|
||||
if "." not in signed_value:
|
||||
return None
|
||||
payload, signature_b64 = signed_value.rsplit(".", 1)
|
||||
|
||||
# Verify signature
|
||||
key = self._upstream_client_secret.get_secret_value().encode()
|
||||
expected_sig = hmac.new(key, payload.encode(), hashlib.sha256).digest()
|
||||
provided_sig = base64.b64decode(signature_b64.encode())
|
||||
|
||||
# Constant-time comparison
|
||||
if not hmac.compare_digest(expected_sig, provided_sig):
|
||||
return None
|
||||
|
||||
return payload
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _decode_list_cookie(self, request: Request, base_name: str) -> list[str]:
|
||||
"""Decode and verify a signed base64-encoded JSON list from cookie. Returns [] if missing/invalid."""
|
||||
# Prefer secure name, but also check non-secure variant for dev
|
||||
secure_name = self._cookie_name(base_name)
|
||||
raw = request.cookies.get(secure_name) or request.cookies.get(f"__{base_name}")
|
||||
if not raw:
|
||||
return []
|
||||
try:
|
||||
# Verify signature
|
||||
payload = self._verify_cookie(raw)
|
||||
if not payload:
|
||||
logger.debug("Cookie signature verification failed for %s", secure_name)
|
||||
return []
|
||||
|
||||
# Decode payload
|
||||
data = base64.b64decode(payload.encode())
|
||||
value = json.loads(data.decode())
|
||||
if isinstance(value, list):
|
||||
return [str(x) for x in value]
|
||||
except Exception:
|
||||
logger.debug("Failed to decode cookie %s; treating as empty", secure_name)
|
||||
return []
|
||||
|
||||
def _encode_list_cookie(self, values: list[str]) -> str:
|
||||
"""Encode values to base64 and sign with HMAC.
|
||||
|
||||
Returns: signed cookie value (payload.signature)
|
||||
"""
|
||||
payload = json.dumps(values, separators=(",", ":")).encode()
|
||||
payload_b64 = base64.b64encode(payload).decode()
|
||||
return self._sign_cookie(payload_b64)
|
||||
|
||||
def _set_list_cookie(
|
||||
self,
|
||||
response: HTMLResponse | RedirectResponse,
|
||||
base_name: str,
|
||||
value_b64: str,
|
||||
max_age: int,
|
||||
) -> None:
|
||||
name = self._cookie_name(base_name)
|
||||
response.set_cookie(
|
||||
name,
|
||||
value_b64,
|
||||
max_age=max_age,
|
||||
secure=self._is_https,
|
||||
httponly=True,
|
||||
samesite="lax",
|
||||
path="/",
|
||||
)
|
||||
|
||||
def _build_upstream_authorize_url(
|
||||
self, txn_id: str, transaction: dict[str, Any]
|
||||
) -> str:
|
||||
"""Construct the upstream IdP authorization URL using stored transaction data."""
|
||||
query_params: dict[str, Any] = {
|
||||
"response_type": "code",
|
||||
"client_id": self._upstream_client_id,
|
||||
"redirect_uri": f"{str(self.base_url).rstrip('/')}{self._redirect_path}",
|
||||
"state": txn_id,
|
||||
}
|
||||
|
||||
scopes_to_use = transaction.get("scopes") or self.required_scopes or []
|
||||
if scopes_to_use:
|
||||
query_params["scope"] = " ".join(scopes_to_use)
|
||||
|
||||
# If PKCE forwarding was enabled, include the proxy challenge
|
||||
proxy_code_verifier = transaction.get("proxy_code_verifier")
|
||||
if proxy_code_verifier:
|
||||
challenge_bytes = hashlib.sha256(proxy_code_verifier.encode()).digest()
|
||||
proxy_code_challenge = (
|
||||
urlsafe_b64encode(challenge_bytes).decode().rstrip("=")
|
||||
)
|
||||
query_params["code_challenge"] = proxy_code_challenge
|
||||
query_params["code_challenge_method"] = "S256"
|
||||
|
||||
# Forward resource indicator if present in transaction
|
||||
if resource := transaction.get("resource"):
|
||||
query_params["resource"] = resource
|
||||
|
||||
# Extra configured parameters
|
||||
if self._extra_authorize_params:
|
||||
query_params.update(self._extra_authorize_params)
|
||||
|
||||
separator = "&" if "?" in self._upstream_authorization_endpoint else "?"
|
||||
return f"{self._upstream_authorization_endpoint}{separator}{urlencode(query_params)}"
|
||||
|
||||
async def _handle_consent(
|
||||
self, request: Request
|
||||
) -> HTMLResponse | RedirectResponse:
|
||||
"""Handle consent page - dispatch to GET or POST handler based on method."""
|
||||
if request.method == "POST":
|
||||
return await self._submit_consent(request)
|
||||
return await self._show_consent_page(request)
|
||||
|
||||
async def _show_consent_page(
|
||||
self, request: Request
|
||||
) -> HTMLResponse | RedirectResponse:
|
||||
"""Display consent page or auto-approve/deny based on cookies."""
|
||||
from fastmcp.server.server import FastMCP
|
||||
|
||||
txn_id = request.query_params.get("txn_id")
|
||||
if not txn_id:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired transaction</p>", status_code=400
|
||||
)
|
||||
|
||||
txn_model = await self._transaction_store.get(key=txn_id)
|
||||
if not txn_model:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired transaction</p>", status_code=400
|
||||
)
|
||||
|
||||
txn = txn_model.model_dump()
|
||||
client_key = self._make_client_key(txn["client_id"], txn["client_redirect_uri"])
|
||||
|
||||
approved = set(self._decode_list_cookie(request, "MCP_APPROVED_CLIENTS"))
|
||||
denied = set(self._decode_list_cookie(request, "MCP_DENIED_CLIENTS"))
|
||||
|
||||
if client_key in approved:
|
||||
upstream_url = self._build_upstream_authorize_url(txn_id, txn)
|
||||
return RedirectResponse(url=upstream_url, status_code=302)
|
||||
|
||||
if client_key in denied:
|
||||
callback_params = {
|
||||
"error": "access_denied",
|
||||
"state": txn.get("client_state") or "",
|
||||
}
|
||||
sep = "&" if "?" in txn["client_redirect_uri"] else "?"
|
||||
return RedirectResponse(
|
||||
url=f"{txn['client_redirect_uri']}{sep}{urlencode(callback_params)}",
|
||||
status_code=302,
|
||||
)
|
||||
|
||||
# Need consent: issue CSRF token and show HTML
|
||||
csrf_token = secrets.token_urlsafe(32)
|
||||
csrf_expires_at = time.time() + 15 * 60
|
||||
|
||||
# Update transaction with CSRF token
|
||||
txn_model.csrf_token = csrf_token
|
||||
txn_model.csrf_expires_at = csrf_expires_at
|
||||
await self._transaction_store.put(
|
||||
key=txn_id, value=txn_model, ttl=15 * 60
|
||||
) # Auto-expire after 15 minutes
|
||||
|
||||
# Update dict for use in HTML generation
|
||||
txn["csrf_token"] = csrf_token
|
||||
txn["csrf_expires_at"] = csrf_expires_at
|
||||
|
||||
# Load client to get client_name if available
|
||||
client = await self.get_client(txn["client_id"])
|
||||
client_name = getattr(client, "client_name", None) if client else None
|
||||
|
||||
# Extract server metadata from app state
|
||||
fastmcp = getattr(request.app.state, "fastmcp_server", None)
|
||||
|
||||
if isinstance(fastmcp, FastMCP):
|
||||
server_name = fastmcp.name
|
||||
icons = fastmcp.icons
|
||||
server_icon_url = icons[0].src if icons else None
|
||||
server_website_url = fastmcp.website_url
|
||||
else:
|
||||
server_name = None
|
||||
server_icon_url = None
|
||||
server_website_url = None
|
||||
|
||||
html = create_consent_html(
|
||||
client_id=txn["client_id"],
|
||||
redirect_uri=txn["client_redirect_uri"],
|
||||
scopes=txn.get("scopes") or [],
|
||||
txn_id=txn_id,
|
||||
csrf_token=csrf_token,
|
||||
client_name=client_name,
|
||||
server_name=server_name,
|
||||
server_icon_url=server_icon_url,
|
||||
server_website_url=server_website_url,
|
||||
csp_policy=self._consent_csp_policy,
|
||||
)
|
||||
response = create_secure_html_response(html)
|
||||
# Store CSRF in cookie with short lifetime
|
||||
self._set_list_cookie(
|
||||
response,
|
||||
"MCP_CONSENT_STATE",
|
||||
self._encode_list_cookie([csrf_token]),
|
||||
max_age=15 * 60,
|
||||
)
|
||||
return response
|
||||
|
||||
async def _submit_consent(
|
||||
self, request: Request
|
||||
) -> RedirectResponse | HTMLResponse:
|
||||
"""Handle consent approval/denial, set cookies, and redirect appropriately."""
|
||||
form = await request.form()
|
||||
txn_id = str(form.get("txn_id", ""))
|
||||
action = str(form.get("action", ""))
|
||||
csrf_token = str(form.get("csrf_token", ""))
|
||||
|
||||
if not txn_id:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired transaction</p>", status_code=400
|
||||
)
|
||||
|
||||
txn_model = await self._transaction_store.get(key=txn_id)
|
||||
if not txn_model:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired transaction</p>", status_code=400
|
||||
)
|
||||
|
||||
txn = txn_model.model_dump()
|
||||
expected_csrf = txn.get("csrf_token")
|
||||
expires_at = float(txn.get("csrf_expires_at") or 0)
|
||||
|
||||
if not expected_csrf or csrf_token != expected_csrf or time.time() > expires_at:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid or expired consent token</p>", status_code=400
|
||||
)
|
||||
|
||||
client_key = self._make_client_key(txn["client_id"], txn["client_redirect_uri"])
|
||||
|
||||
if action == "approve":
|
||||
approved = set(self._decode_list_cookie(request, "MCP_APPROVED_CLIENTS"))
|
||||
if client_key not in approved:
|
||||
approved.add(client_key)
|
||||
approved_b64 = self._encode_list_cookie(sorted(approved))
|
||||
|
||||
upstream_url = self._build_upstream_authorize_url(txn_id, txn)
|
||||
response = RedirectResponse(url=upstream_url, status_code=302)
|
||||
self._set_list_cookie(
|
||||
response, "MCP_APPROVED_CLIENTS", approved_b64, max_age=365 * 24 * 3600
|
||||
)
|
||||
# Clear CSRF cookie by setting empty short-lived value
|
||||
self._set_list_cookie(
|
||||
response, "MCP_CONSENT_STATE", self._encode_list_cookie([]), max_age=60
|
||||
)
|
||||
return response
|
||||
|
||||
elif action == "deny":
|
||||
denied = set(self._decode_list_cookie(request, "MCP_DENIED_CLIENTS"))
|
||||
if client_key not in denied:
|
||||
denied.add(client_key)
|
||||
denied_b64 = self._encode_list_cookie(sorted(denied))
|
||||
|
||||
callback_params = {
|
||||
"error": "access_denied",
|
||||
"state": txn.get("client_state") or "",
|
||||
}
|
||||
sep = "&" if "?" in txn["client_redirect_uri"] else "?"
|
||||
client_callback_url = (
|
||||
f"{txn['client_redirect_uri']}{sep}{urlencode(callback_params)}"
|
||||
)
|
||||
response = RedirectResponse(url=client_callback_url, status_code=302)
|
||||
self._set_list_cookie(
|
||||
response, "MCP_DENIED_CLIENTS", denied_b64, max_age=365 * 24 * 3600
|
||||
)
|
||||
self._set_list_cookie(
|
||||
response, "MCP_CONSENT_STATE", self._encode_list_cookie([]), max_age=60
|
||||
)
|
||||
return response
|
||||
|
||||
else:
|
||||
return create_secure_html_response(
|
||||
"<h1>Error</h1><p>Invalid action</p>", status_code=400
|
||||
)
|
||||
277
src/fastmcp/server/auth/oauth_proxy/ui.py
Normal file
277
src/fastmcp/server/auth/oauth_proxy/ui.py
Normal file
|
|
@ -0,0 +1,277 @@
|
|||
"""OAuth Proxy UI Generation Functions.
|
||||
|
||||
This module contains HTML generation functions for consent and error pages.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from fastmcp.utilities.ui import (
|
||||
BUTTON_STYLES,
|
||||
DETAIL_BOX_STYLES,
|
||||
DETAILS_STYLES,
|
||||
INFO_BOX_STYLES,
|
||||
REDIRECT_SECTION_STYLES,
|
||||
TOOLTIP_STYLES,
|
||||
create_logo,
|
||||
create_page,
|
||||
)
|
||||
|
||||
|
||||
def create_consent_html(
|
||||
client_id: str,
|
||||
redirect_uri: str,
|
||||
scopes: list[str],
|
||||
txn_id: str,
|
||||
csrf_token: str,
|
||||
client_name: str | None = None,
|
||||
title: str = "Application Access Request",
|
||||
server_name: str | None = None,
|
||||
server_icon_url: str | None = None,
|
||||
server_website_url: str | None = None,
|
||||
client_website_url: str | None = None,
|
||||
csp_policy: str | None = None,
|
||||
) -> str:
|
||||
"""Create a styled HTML consent page for OAuth authorization requests.
|
||||
|
||||
Args:
|
||||
csp_policy: Content Security Policy override.
|
||||
If None, uses the built-in CSP policy with appropriate directives.
|
||||
If empty string "", disables CSP entirely (no meta tag is rendered).
|
||||
If a non-empty string, uses that as the CSP policy value.
|
||||
"""
|
||||
import html as html_module
|
||||
|
||||
client_display = html_module.escape(client_name or client_id)
|
||||
server_name_escaped = html_module.escape(server_name or "FastMCP")
|
||||
|
||||
# Make server name a hyperlink if website URL is available
|
||||
if server_website_url:
|
||||
website_url_escaped = html_module.escape(server_website_url)
|
||||
server_display = f'<a href="{website_url_escaped}" target="_blank" rel="noopener noreferrer" class="server-name-link">{server_name_escaped}</a>'
|
||||
else:
|
||||
server_display = server_name_escaped
|
||||
|
||||
# Build intro box with call-to-action
|
||||
intro_box = f"""
|
||||
<div class="info-box">
|
||||
<p>The application <strong>{client_display}</strong> wants to access the MCP server <strong>{server_display}</strong>. Please ensure you recognize the callback address below.</p>
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Build redirect URI section (yellow box, centered)
|
||||
redirect_uri_escaped = html_module.escape(redirect_uri)
|
||||
redirect_section = f"""
|
||||
<div class="redirect-section">
|
||||
<span class="label">Credentials will be sent to:</span>
|
||||
<div class="value">{redirect_uri_escaped}</div>
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Build advanced details with collapsible section
|
||||
detail_rows = [
|
||||
("Application Name", html_module.escape(client_name or client_id)),
|
||||
("Application Website", html_module.escape(client_website_url or "N/A")),
|
||||
("Application ID", client_id),
|
||||
("Redirect URI", redirect_uri_escaped),
|
||||
(
|
||||
"Requested Scopes",
|
||||
", ".join(html_module.escape(s) for s in scopes) if scopes else "None",
|
||||
),
|
||||
]
|
||||
|
||||
detail_rows_html = "\n".join(
|
||||
[
|
||||
f"""
|
||||
<div class="detail-row">
|
||||
<div class="detail-label">{label}:</div>
|
||||
<div class="detail-value">{value}</div>
|
||||
</div>
|
||||
"""
|
||||
for label, value in detail_rows
|
||||
]
|
||||
)
|
||||
|
||||
advanced_details = f"""
|
||||
<details>
|
||||
<summary>Advanced Details</summary>
|
||||
<div class="detail-box">
|
||||
{detail_rows_html}
|
||||
</div>
|
||||
</details>
|
||||
"""
|
||||
|
||||
# Build form with buttons
|
||||
# Use empty action to submit to current URL (/consent or /mcp/consent)
|
||||
# The POST handler is registered at the same path as GET
|
||||
form = f"""
|
||||
<form id="consentForm" method="POST" action="">
|
||||
<input type="hidden" name="txn_id" value="{txn_id}" />
|
||||
<input type="hidden" name="csrf_token" value="{csrf_token}" />
|
||||
<input type="hidden" name="submit" value="true" />
|
||||
<div class="button-group">
|
||||
<button type="submit" name="action" value="approve" class="btn-approve">Allow Access</button>
|
||||
<button type="submit" name="action" value="deny" class="btn-deny">Deny</button>
|
||||
</div>
|
||||
</form>
|
||||
"""
|
||||
|
||||
# Build help link with tooltip (identical to current implementation)
|
||||
help_link = """
|
||||
<div class="help-link-container">
|
||||
<span class="help-link">
|
||||
Why am I seeing this?
|
||||
<span class="tooltip">
|
||||
This FastMCP server requires your consent to allow a new client
|
||||
to connect. This protects you from <a
|
||||
href="https://modelcontextprotocol.io/specification/2025-06-18/basic/security_best_practices#confused-deputy-problem"
|
||||
target="_blank" class="tooltip-link">confused deputy
|
||||
attacks</a>, where malicious clients could impersonate you
|
||||
and steal access.<br><br>
|
||||
<a
|
||||
href="https://gofastmcp.com/servers/auth/oauth-proxy#confused-deputy-attacks"
|
||||
target="_blank" class="tooltip-link">Learn more about
|
||||
FastMCP security →</a>
|
||||
</span>
|
||||
</span>
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Build the page content
|
||||
content = f"""
|
||||
<div class="container">
|
||||
{create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
|
||||
<h1>Application Access Request</h1>
|
||||
{intro_box}
|
||||
{redirect_section}
|
||||
{advanced_details}
|
||||
{form}
|
||||
</div>
|
||||
{help_link}
|
||||
"""
|
||||
|
||||
# Additional styles needed for this page
|
||||
additional_styles = (
|
||||
INFO_BOX_STYLES
|
||||
+ REDIRECT_SECTION_STYLES
|
||||
+ DETAILS_STYLES
|
||||
+ DETAIL_BOX_STYLES
|
||||
+ BUTTON_STYLES
|
||||
+ TOOLTIP_STYLES
|
||||
)
|
||||
|
||||
# Determine CSP policy to use
|
||||
# If csp_policy is None, build the default CSP policy
|
||||
# If csp_policy is empty string, CSP will be disabled entirely in create_page
|
||||
# If csp_policy is a non-empty string, use it as-is
|
||||
if csp_policy is None:
|
||||
# Need to allow form-action for form submission
|
||||
# Chrome requires explicit scheme declarations in CSP form-action when redirect chains
|
||||
# end in custom protocol schemes (e.g., cursor://). Parse redirect_uri to include its scheme.
|
||||
parsed_redirect = urlparse(redirect_uri)
|
||||
redirect_scheme = parsed_redirect.scheme.lower()
|
||||
|
||||
# Build form-action directive with standard schemes plus custom protocol if present
|
||||
form_action_schemes = ["https:", "http:"]
|
||||
if redirect_scheme and redirect_scheme not in ("http", "https"):
|
||||
# Custom protocol scheme (e.g., cursor:, vscode:, etc.)
|
||||
form_action_schemes.append(f"{redirect_scheme}:")
|
||||
|
||||
form_action_directive = " ".join(form_action_schemes)
|
||||
csp_policy = f"default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'; form-action {form_action_directive}"
|
||||
|
||||
return create_page(
|
||||
content=content,
|
||||
title=title,
|
||||
additional_styles=additional_styles,
|
||||
csp_policy=csp_policy,
|
||||
)
|
||||
|
||||
|
||||
def create_error_html(
|
||||
error_title: str,
|
||||
error_message: str,
|
||||
error_details: dict[str, str] | None = None,
|
||||
server_name: str | None = None,
|
||||
server_icon_url: str | None = None,
|
||||
) -> str:
|
||||
"""Create a styled HTML error page for OAuth errors.
|
||||
|
||||
Args:
|
||||
error_title: The error title (e.g., "OAuth Error", "Authorization Failed")
|
||||
error_message: The main error message to display
|
||||
error_details: Optional dictionary of error details to show (e.g., `{"Error Code": "invalid_client"}`)
|
||||
server_name: Optional server name to display
|
||||
server_icon_url: Optional URL to server icon/logo
|
||||
|
||||
Returns:
|
||||
Complete HTML page as a string
|
||||
"""
|
||||
import html as html_module
|
||||
|
||||
error_message_escaped = html_module.escape(error_message)
|
||||
|
||||
# Build error message box
|
||||
error_box = f"""
|
||||
<div class="info-box error">
|
||||
<p>{error_message_escaped}</p>
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Build error details section if provided
|
||||
details_section = ""
|
||||
if error_details:
|
||||
detail_rows_html = "\n".join(
|
||||
[
|
||||
f"""
|
||||
<div class="detail-row">
|
||||
<div class="detail-label">{html_module.escape(label)}:</div>
|
||||
<div class="detail-value">{html_module.escape(value)}</div>
|
||||
</div>
|
||||
"""
|
||||
for label, value in error_details.items()
|
||||
]
|
||||
)
|
||||
|
||||
details_section = f"""
|
||||
<details>
|
||||
<summary>Error Details</summary>
|
||||
<div class="detail-box">
|
||||
{detail_rows_html}
|
||||
</div>
|
||||
</details>
|
||||
"""
|
||||
|
||||
# Build the page content
|
||||
content = f"""
|
||||
<div class="container">
|
||||
{create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
|
||||
<h1>{html_module.escape(error_title)}</h1>
|
||||
{error_box}
|
||||
{details_section}
|
||||
</div>
|
||||
"""
|
||||
|
||||
# Additional styles needed for this page
|
||||
# Override .info-box.error to use normal text color instead of red
|
||||
additional_styles = (
|
||||
INFO_BOX_STYLES
|
||||
+ DETAILS_STYLES
|
||||
+ DETAIL_BOX_STYLES
|
||||
+ """
|
||||
.info-box.error {
|
||||
color: #111827;
|
||||
}
|
||||
"""
|
||||
)
|
||||
|
||||
# Simple CSP policy for error pages (no forms needed)
|
||||
csp_policy = "default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'"
|
||||
|
||||
return create_page(
|
||||
content=content,
|
||||
title=error_title,
|
||||
additional_styles=additional_styles,
|
||||
csp_policy=csp_policy,
|
||||
)
|
||||
|
|
@ -12,14 +12,18 @@ with the following configuration:
|
|||
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
import time
|
||||
from collections.abc import AsyncGenerator
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
from urllib.parse import parse_qs, urlencode, urlparse
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.client import Client
|
||||
from fastmcp.server.auth.auth import AccessToken
|
||||
from fastmcp.server.auth.oauth_proxy.models import ClientCode
|
||||
from fastmcp.server.auth.providers.github import GitHubProvider
|
||||
from fastmcp.utilities.tests import HeadlessOAuth, run_server_async
|
||||
|
||||
|
|
@ -81,11 +85,6 @@ def create_github_server_with_mock_callback(base_url: str) -> FastMCP:
|
|||
# Mock the authorize method to return a fake code instead of redirecting to GitHub
|
||||
async def mock_authorize(client, params):
|
||||
# Instead of redirecting to GitHub, simulate an immediate callback
|
||||
import secrets
|
||||
import time
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import ClientCode
|
||||
|
||||
# Generate a fake authorization code
|
||||
fake_code = secrets.token_urlsafe(32)
|
||||
|
||||
|
|
@ -117,8 +116,6 @@ def create_github_server_with_mock_callback(base_url: str) -> FastMCP:
|
|||
"code": fake_code,
|
||||
"state": params.state,
|
||||
}
|
||||
from urllib.parse import urlencode
|
||||
|
||||
separator = "&" if "?" in str(params.redirect_uri) else "?"
|
||||
return f"{params.redirect_uri}{separator}{urlencode(callback_params)}"
|
||||
|
||||
|
|
@ -130,10 +127,6 @@ def create_github_server_with_mock_callback(base_url: str) -> FastMCP:
|
|||
async def mock_verify_token(token: str):
|
||||
if token.startswith("gho_mock_token_"):
|
||||
# Return a mock AccessToken for our fake tokens
|
||||
import time
|
||||
|
||||
from fastmcp.server.auth.auth import AccessToken
|
||||
|
||||
return AccessToken(
|
||||
token=token,
|
||||
client_id=FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID or "test-client",
|
||||
|
|
|
|||
|
|
@ -14,18 +14,22 @@ This test suite verifies:
|
|||
import re
|
||||
import secrets
|
||||
import time
|
||||
from unittest.mock import Mock
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
import pytest
|
||||
from key_value.aio.stores.memory import MemoryStore
|
||||
from mcp.server.auth.provider import AuthorizationParams
|
||||
from mcp.shared.auth import OAuthClientInformationFull
|
||||
from mcp.types import Icon
|
||||
from pydantic import AnyUrl
|
||||
from starlette.applications import Starlette
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from fastmcp.server.auth.auth import TokenVerifier
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.auth import AccessToken, TokenVerifier
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthProxy
|
||||
from fastmcp.server.auth.oauth_proxy.models import OAuthTransaction
|
||||
|
||||
|
||||
class MockTokenVerifier(TokenVerifier):
|
||||
|
|
@ -36,8 +40,6 @@ class MockTokenVerifier(TokenVerifier):
|
|||
|
||||
async def verify_token(self, token: str):
|
||||
"""Mock token verification."""
|
||||
from fastmcp.server.auth.auth import AccessToken
|
||||
|
||||
return AccessToken(
|
||||
token=token,
|
||||
client_id="mock-client",
|
||||
|
|
@ -53,8 +55,6 @@ class _Verifier(TokenVerifier):
|
|||
self.required_scopes = ["read"]
|
||||
|
||||
async def verify_token(self, token: str):
|
||||
from fastmcp.server.auth.auth import AccessToken
|
||||
|
||||
return AccessToken(
|
||||
token=token, client_id="c", scopes=self.required_scopes, expires_at=None
|
||||
)
|
||||
|
|
@ -504,8 +504,6 @@ class TestStoragePersistence:
|
|||
|
||||
async def test_storage_uses_pydantic_adapter(self, oauth_proxy_with_storage):
|
||||
"""Verify that PydanticAdapter serializes/deserializes correctly."""
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="pydantic-test-client",
|
||||
client_secret="test-secret",
|
||||
|
|
@ -668,9 +666,6 @@ class TestConsentPageServerIcon:
|
|||
|
||||
async def test_consent_screen_displays_server_icon(self):
|
||||
"""Test that consent screen shows server's custom icon when available."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create mock JWT verifier
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
|
|
@ -690,7 +685,6 @@ class TestConsentPageServerIcon:
|
|||
)
|
||||
|
||||
# Create FastMCP server with custom icon
|
||||
from mcp.types import Icon
|
||||
|
||||
server = FastMCP(
|
||||
name="My Custom Server",
|
||||
|
|
@ -711,7 +705,6 @@ class TestConsentPageServerIcon:
|
|||
await proxy.register_client(client_info)
|
||||
|
||||
# Create a transaction manually
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
|
|
@ -741,9 +734,6 @@ class TestConsentPageServerIcon:
|
|||
|
||||
async def test_consent_screen_falls_back_to_fastmcp_logo(self):
|
||||
"""Test that consent screen shows FastMCP logo when no server icon provided."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create mock JWT verifier
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
|
|
@ -777,7 +767,6 @@ class TestConsentPageServerIcon:
|
|||
await proxy.register_client(client_info)
|
||||
|
||||
# Create a transaction
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
|
|
@ -807,11 +796,6 @@ class TestConsentPageServerIcon:
|
|||
|
||||
async def test_consent_screen_escapes_server_name(self):
|
||||
"""Test that server name is properly HTML-escaped."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from mcp.types import Icon
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create mock JWT verifier
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
|
|
@ -849,7 +833,6 @@ class TestConsentPageServerIcon:
|
|||
await proxy.register_client(client_info)
|
||||
|
||||
# Create a transaction
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
|
|
@ -885,9 +868,6 @@ class TestConsentCSPPolicy:
|
|||
|
||||
async def test_default_csp_includes_form_action(self):
|
||||
"""Test that default CSP includes form-action directive."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
verifier.required_scopes = ["read"]
|
||||
|
|
@ -915,8 +895,6 @@ class TestConsentCSPPolicy:
|
|||
)
|
||||
await proxy.register_client(client_info)
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
txn_id=txn_id,
|
||||
|
|
@ -940,9 +918,6 @@ class TestConsentCSPPolicy:
|
|||
|
||||
async def test_empty_csp_disables_csp_meta_tag(self):
|
||||
"""Test that empty string CSP disables CSP meta tag entirely."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
verifier.required_scopes = ["read"]
|
||||
|
|
@ -971,8 +946,6 @@ class TestConsentCSPPolicy:
|
|||
)
|
||||
await proxy.register_client(client_info)
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
txn_id=txn_id,
|
||||
|
|
@ -995,9 +968,6 @@ class TestConsentCSPPolicy:
|
|||
|
||||
async def test_custom_csp_policy_is_used(self):
|
||||
"""Test that custom CSP policy is applied to consent page."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
verifier.required_scopes = ["read"]
|
||||
|
|
@ -1027,8 +997,6 @@ class TestConsentCSPPolicy:
|
|||
)
|
||||
await proxy.register_client(client_info)
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
txn_id=txn_id,
|
||||
|
|
|
|||
|
|
@ -18,16 +18,35 @@ from urllib.parse import parse_qs, urlencode, urlparse
|
|||
|
||||
import httpx
|
||||
import pytest
|
||||
from mcp.server.auth.provider import AuthorizationParams
|
||||
from mcp.server.auth.handlers.token import TokenErrorResponse
|
||||
from mcp.server.auth.handlers.token import TokenHandler as SDKTokenHandler
|
||||
from mcp.server.auth.provider import (
|
||||
AuthorizationCode,
|
||||
AuthorizationParams,
|
||||
AuthorizeError,
|
||||
)
|
||||
from mcp.shared.auth import OAuthClientInformationFull
|
||||
from pydantic import AnyUrl
|
||||
from starlette.applications import Starlette
|
||||
from starlette.responses import JSONResponse
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import HTMLResponse, JSONResponse
|
||||
from starlette.routing import Route
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.auth import AccessToken, RefreshToken, TokenVerifier
|
||||
from fastmcp.server.auth.auth import (
|
||||
AccessToken,
|
||||
RefreshToken,
|
||||
TokenHandler,
|
||||
TokenVerifier,
|
||||
)
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthProxy
|
||||
from fastmcp.server.auth.oauth_proxy.models import (
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_NO_REFRESH_SECONDS,
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS,
|
||||
ClientCode,
|
||||
)
|
||||
from fastmcp.server.auth.oauth_proxy.ui import create_error_html
|
||||
from fastmcp.server.auth.providers.jwt import JWTVerifier
|
||||
|
||||
# =============================================================================
|
||||
|
|
@ -651,7 +670,9 @@ class TestOAuthProxyTokenEndpointAuth:
|
|||
)
|
||||
|
||||
# Mock the upstream OAuth provider response
|
||||
with patch("fastmcp.server.auth.oauth_proxy.AsyncOAuth2Client") as MockClient:
|
||||
with patch(
|
||||
"fastmcp.server.auth.oauth_proxy.proxy.AsyncOAuth2Client"
|
||||
) as MockClient:
|
||||
mock_client = AsyncMock()
|
||||
|
||||
# Mock initial token exchange (authorization code flow)
|
||||
|
|
@ -679,8 +700,6 @@ class TestOAuthProxyTokenEndpointAuth:
|
|||
await proxy.register_client(client)
|
||||
|
||||
# Store client code that would be created during OAuth callback
|
||||
from fastmcp.server.auth.oauth_proxy import ClientCode
|
||||
|
||||
client_code = ClientCode(
|
||||
code="test-auth-code",
|
||||
client_id="test-client",
|
||||
|
|
@ -700,8 +719,6 @@ class TestOAuthProxyTokenEndpointAuth:
|
|||
await proxy._code_store.put(key=client_code.code, value=client_code)
|
||||
|
||||
# Exchange authorization code to get FastMCP tokens
|
||||
from mcp.server.auth.provider import AuthorizationCode
|
||||
|
||||
auth_code = AuthorizationCode(
|
||||
code="test-auth-code",
|
||||
scopes=["read"],
|
||||
|
|
@ -837,7 +854,9 @@ class TestOAuthProxyE2E:
|
|||
"scope": "read write",
|
||||
}
|
||||
|
||||
with patch("fastmcp.server.auth.oauth_proxy.AsyncOAuth2Client") as MockClient:
|
||||
with patch(
|
||||
"fastmcp.server.auth.oauth_proxy.proxy.AsyncOAuth2Client"
|
||||
) as MockClient:
|
||||
mock_client = AsyncMock()
|
||||
|
||||
# Mock initial token exchange to get FastMCP tokens
|
||||
|
|
@ -866,8 +885,6 @@ class TestOAuthProxyE2E:
|
|||
MockClient.return_value = mock_client
|
||||
|
||||
# Store client code that would be created during OAuth callback
|
||||
from fastmcp.server.auth.oauth_proxy import ClientCode
|
||||
|
||||
client_code = ClientCode(
|
||||
code="test-auth-code",
|
||||
client_id="test-client",
|
||||
|
|
@ -887,8 +904,6 @@ class TestOAuthProxyE2E:
|
|||
await proxy._code_store.put(key=client_code.code, value=client_code)
|
||||
|
||||
# Exchange authorization code to get FastMCP tokens
|
||||
from mcp.server.auth.provider import AuthorizationCode
|
||||
|
||||
auth_code = AuthorizationCode(
|
||||
code="test-auth-code",
|
||||
scopes=["read", "write"],
|
||||
|
|
@ -1201,9 +1216,6 @@ class TestParameterForwarding:
|
|||
|
||||
This aligns with OAuth 2.1 spec and enables Claude's automatic client re-registration.
|
||||
"""
|
||||
from starlette.applications import Starlette
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
proxy = OAuthProxy(
|
||||
upstream_authorization_endpoint="https://oauth.example.com/authorize",
|
||||
upstream_token_endpoint="https://oauth.example.com/token",
|
||||
|
|
@ -1254,12 +1266,6 @@ class TestTokenHandlerErrorTransformation:
|
|||
|
||||
async def test_transforms_client_auth_failure_to_invalid_client_401(self):
|
||||
"""Test that client authentication failures return invalid_client with 401."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from mcp.server.auth.handlers.token import TokenHandler as SDKTokenHandler
|
||||
|
||||
from fastmcp.server.auth.auth import TokenHandler
|
||||
|
||||
handler = TokenHandler(provider=Mock(), client_authenticator=Mock())
|
||||
|
||||
# Create a mock 401 response like the SDK returns for auth failures
|
||||
|
|
@ -1285,10 +1291,6 @@ class TestTokenHandlerErrorTransformation:
|
|||
|
||||
def test_does_not_transform_grant_type_unauthorized_to_invalid_client(self):
|
||||
"""Test that grant type authorization errors stay as unauthorized_client with 400."""
|
||||
from mcp.server.auth.handlers.token import TokenErrorResponse
|
||||
|
||||
from fastmcp.server.auth.auth import TokenHandler
|
||||
|
||||
handler = TokenHandler(provider=Mock(), client_authenticator=Mock())
|
||||
|
||||
# Simulate error from grant_type not in client_info.grant_types
|
||||
|
|
@ -1309,12 +1311,6 @@ class TestTokenHandlerErrorTransformation:
|
|||
Per MCP spec: "Invalid or expired tokens MUST receive a HTTP 401 response."
|
||||
The SDK incorrectly returns 400 for all TokenErrorResponse including invalid_grant.
|
||||
"""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from mcp.server.auth.handlers.token import TokenHandler as SDKTokenHandler
|
||||
|
||||
from fastmcp.server.auth.auth import TokenHandler
|
||||
|
||||
handler = TokenHandler(provider=Mock(), client_authenticator=Mock())
|
||||
|
||||
# Create a mock 400 response like the SDK returns for invalid_grant
|
||||
|
|
@ -1340,10 +1336,6 @@ class TestTokenHandlerErrorTransformation:
|
|||
|
||||
def test_does_not_transform_other_400_errors(self):
|
||||
"""Test that non-invalid_grant 400 errors pass through unchanged."""
|
||||
from mcp.server.auth.handlers.token import TokenErrorResponse
|
||||
|
||||
from fastmcp.server.auth.auth import TokenHandler
|
||||
|
||||
handler = TokenHandler(provider=Mock(), client_authenticator=Mock())
|
||||
|
||||
# Test with invalid_request error (should stay 400)
|
||||
|
|
@ -1364,7 +1356,6 @@ class TestErrorPageRendering:
|
|||
|
||||
def test_create_error_html_basic(self):
|
||||
"""Test basic error page generation."""
|
||||
from fastmcp.server.auth.oauth_proxy import create_error_html
|
||||
|
||||
html = create_error_html(
|
||||
error_title="Test Error",
|
||||
|
|
@ -1379,7 +1370,6 @@ class TestErrorPageRendering:
|
|||
|
||||
def test_create_error_html_with_details(self):
|
||||
"""Test error page with error details."""
|
||||
from fastmcp.server.auth.oauth_proxy import create_error_html
|
||||
|
||||
html = create_error_html(
|
||||
error_title="OAuth Error",
|
||||
|
|
@ -1399,7 +1389,6 @@ class TestErrorPageRendering:
|
|||
|
||||
def test_create_error_html_escapes_user_input(self):
|
||||
"""Test that error page properly escapes HTML in user input."""
|
||||
from fastmcp.server.auth.oauth_proxy import create_error_html
|
||||
|
||||
html = create_error_html(
|
||||
error_title="Error <script>alert('xss')</script>",
|
||||
|
|
@ -1415,14 +1404,6 @@ class TestErrorPageRendering:
|
|||
|
||||
async def test_callback_error_returns_html_page(self):
|
||||
"""Test that OAuth callback errors return styled HTML instead of data: URLs."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import HTMLResponse
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthProxy
|
||||
from fastmcp.server.auth.providers.jwt import JWTVerifier
|
||||
|
||||
# Create a minimal OAuth proxy
|
||||
provider = OAuthProxy(
|
||||
upstream_authorization_endpoint="https://idp.example.com/authorize",
|
||||
|
|
@ -1464,11 +1445,6 @@ class TestFallbackAccessTokenExpiry:
|
|||
|
||||
def test_default_constants(self):
|
||||
"""Verify the default expiry constants are set correctly."""
|
||||
from fastmcp.server.auth.oauth_proxy import (
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_NO_REFRESH_SECONDS,
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS,
|
||||
)
|
||||
|
||||
assert DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS == 60 * 60 # 1 hour
|
||||
assert (
|
||||
DEFAULT_ACCESS_TOKEN_EXPIRY_NO_REFRESH_SECONDS == 60 * 60 * 24 * 365
|
||||
|
|
@ -1531,7 +1507,6 @@ class TestResourceURLValidation:
|
|||
|
||||
async def test_authorize_rejects_mismatched_resource(self, proxy_with_resource_url):
|
||||
"""Test that authorization rejects requests with mismatched resource."""
|
||||
from mcp.server.auth.provider import AuthorizeError
|
||||
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
|
|
@ -1585,7 +1560,6 @@ class TestResourceURLValidation:
|
|||
self, proxy_with_resource_url
|
||||
):
|
||||
"""Test that old hardcoded /mcp path is rejected when server uses different path."""
|
||||
from mcp.server.auth.provider import AuthorizeError
|
||||
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
|
|
@ -1752,8 +1726,6 @@ class TestUpstreamTokenStorageTTL:
|
|||
but expires_in=28800 (8 hours). The upstream tokens should persist for
|
||||
8 hours (the access token lifetime), not 2 minutes.
|
||||
"""
|
||||
from fastmcp.server.auth.oauth_proxy import ClientCode
|
||||
|
||||
# Register client
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
|
|
@ -1783,8 +1755,6 @@ class TestUpstreamTokenStorageTTL:
|
|||
await proxy._code_store.put(key=client_code.code, value=client_code)
|
||||
|
||||
# Exchange the code
|
||||
from mcp.server.auth.provider import AuthorizationCode
|
||||
|
||||
auth_code = AuthorizationCode(
|
||||
code="test-auth-code",
|
||||
scopes=["read", "write"],
|
||||
|
|
@ -1834,8 +1804,6 @@ class TestUpstreamTokenStorageTTL:
|
|||
refresh_expires_in=32318 (9 hours). The upstream tokens should persist
|
||||
for 9 hours (the refresh token lifetime).
|
||||
"""
|
||||
from fastmcp.server.auth.oauth_proxy import ClientCode
|
||||
|
||||
# Register client
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
|
|
@ -1865,8 +1833,6 @@ class TestUpstreamTokenStorageTTL:
|
|||
await proxy._code_store.put(key=client_code.code, value=client_code)
|
||||
|
||||
# Exchange the code
|
||||
from mcp.server.auth.provider import AuthorizationCode
|
||||
|
||||
auth_code = AuthorizationCode(
|
||||
code="test-auth-code-2",
|
||||
scopes=["read", "write"],
|
||||
|
|
|
|||
|
|
@ -5,7 +5,8 @@ from mcp.shared.auth import InvalidRedirectUriError
|
|||
from pydantic import AnyUrl
|
||||
|
||||
from fastmcp.server.auth.auth import TokenVerifier
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthProxy, ProxyDCRClient
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthProxy
|
||||
from fastmcp.server.auth.oauth_proxy.models import ProxyDCRClient
|
||||
|
||||
|
||||
class MockTokenVerifier(TokenVerifier):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue