From c99e0c63519eb5a4e74a93b7a2670403dcc234cf Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Mon, 19 Jan 2026 10:10:07 -0500
Subject: [PATCH] Refactor OAuthProxy into focused modules (#2935)
---
loq.toml | 4 +-
.../server/auth/oauth_proxy/__init__.py | 14 +
.../server/auth/oauth_proxy/consent.py | 361 ++++++++
src/fastmcp/server/auth/oauth_proxy/models.py | 178 ++++
.../{oauth_proxy.py => oauth_proxy/proxy.py} | 790 +-----------------
src/fastmcp/server/auth/oauth_proxy/ui.py | 277 ++++++
.../auth/test_github_provider_integration.py | 17 +-
tests/server/auth/test_oauth_consent_flow.py | 42 +-
tests/server/auth/test_oauth_proxy.py | 90 +-
.../test_oauth_proxy_redirect_validation.py | 3 +-
10 files changed, 890 insertions(+), 886 deletions(-)
create mode 100644 src/fastmcp/server/auth/oauth_proxy/__init__.py
create mode 100644 src/fastmcp/server/auth/oauth_proxy/consent.py
create mode 100644 src/fastmcp/server/auth/oauth_proxy/models.py
rename src/fastmcp/server/auth/{oauth_proxy.py => oauth_proxy/proxy.py} (69%)
create mode 100644 src/fastmcp/server/auth/oauth_proxy/ui.py
diff --git a/loq.toml b/loq.toml
index e3084de73..4c6d7e28b 100644
--- a/loq.toml
+++ b/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"
diff --git a/src/fastmcp/server/auth/oauth_proxy/__init__.py b/src/fastmcp/server/auth/oauth_proxy/__init__.py
new file mode 100644
index 000000000..5e9ff4315
--- /dev/null
+++ b/src/fastmcp/server/auth/oauth_proxy/__init__.py
@@ -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",
+]
diff --git a/src/fastmcp/server/auth/oauth_proxy/consent.py b/src/fastmcp/server/auth/oauth_proxy/consent.py
new file mode 100644
index 000000000..6f47a5da7
--- /dev/null
+++ b/src/fastmcp/server/auth/oauth_proxy/consent.py
@@ -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(
+ "Error
Invalid or expired transaction
", status_code=400
+ )
+
+ txn_model = await self._transaction_store.get(key=txn_id)
+ if not txn_model:
+ return create_secure_html_response(
+ "Error
Invalid or expired transaction
", 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(
+ "Error
Invalid or expired transaction
", status_code=400
+ )
+
+ txn_model = await self._transaction_store.get(key=txn_id)
+ if not txn_model:
+ return create_secure_html_response(
+ "Error
Invalid or expired transaction
", 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(
+ "Error
Invalid or expired consent token
", 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(
+ "Error
Invalid action
", status_code=400
+ )
diff --git a/src/fastmcp/server/auth/oauth_proxy/models.py b/src/fastmcp/server/auth/oauth_proxy/models.py
new file mode 100644
index 000000000..fe6c77941
--- /dev/null
+++ b/src/fastmcp/server/auth/oauth_proxy/models.py
@@ -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)
diff --git a/src/fastmcp/server/auth/oauth_proxy.py b/src/fastmcp/server/auth/oauth_proxy/proxy.py
similarity index 69%
rename from src/fastmcp/server/auth/oauth_proxy.py
rename to src/fastmcp/server/auth/oauth_proxy/proxy.py
index 14c80918a..f81f0884d 100644
--- a/src/fastmcp/server/auth/oauth_proxy.py
+++ b/src/fastmcp/server/auth/oauth_proxy/proxy.py
@@ -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'{server_name_escaped}'
- else:
- server_display = server_name_escaped
-
- # Build intro box with call-to-action
- intro_box = f"""
-
-
The application {client_display} wants to access the MCP server {server_display}. Please ensure you recognize the callback address below.
-
-
Credentials will be sent to:
-
{redirect_uri_escaped}
-
- Advanced Details
-
- {detail_rows_html}
-
-
- """
-
- # 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"""
-
- """
-
- # Build help link with tooltip (identical to current implementation)
- help_link = """
-
- """
-
- # Build the page content
- content = f"""
-
- {create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
-
Application Access Request
- {intro_box}
- {redirect_section}
- {advanced_details}
- {form}
-
- {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"""
-
-
{error_message_escaped}
-
-
{html_module.escape(label)}:
-
{html_module.escape(value)}
-
- Error Details
-
- {detail_rows_html}
-
-
- """
-
- # Build the page content
- content = f"""
-
- {create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
-
{html_module.escape(error_title)}
- {error_box}
- {details_section}
-
- """
-
- # 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(
- "Error
Invalid or expired transaction
", status_code=400
- )
-
- txn_model = await self._transaction_store.get(key=txn_id)
- if not txn_model:
- return create_secure_html_response(
- "Error
Invalid or expired transaction
", 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(
- "Error
Invalid or expired transaction
", status_code=400
- )
-
- txn_model = await self._transaction_store.get(key=txn_id)
- if not txn_model:
- return create_secure_html_response(
- "Error
Invalid or expired transaction
", 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(
- "Error
Invalid or expired consent token
", 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(
- "Error
Invalid action
", status_code=400
- )
diff --git a/src/fastmcp/server/auth/oauth_proxy/ui.py b/src/fastmcp/server/auth/oauth_proxy/ui.py
new file mode 100644
index 000000000..3bae1a11c
--- /dev/null
+++ b/src/fastmcp/server/auth/oauth_proxy/ui.py
@@ -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'{server_name_escaped}'
+ else:
+ server_display = server_name_escaped
+
+ # Build intro box with call-to-action
+ intro_box = f"""
+
+
The application {client_display} wants to access the MCP server {server_display}. Please ensure you recognize the callback address below.
+
+
Credentials will be sent to:
+
{redirect_uri_escaped}
+
+ Advanced Details
+
+ {detail_rows_html}
+
+
+ """
+
+ # 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"""
+
+ """
+
+ # Build help link with tooltip (identical to current implementation)
+ help_link = """
+
+ """
+
+ # Build the page content
+ content = f"""
+
+ {create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
+
Application Access Request
+ {intro_box}
+ {redirect_section}
+ {advanced_details}
+ {form}
+
+ {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"""
+
+
{error_message_escaped}
+
+
{html_module.escape(label)}:
+
{html_module.escape(value)}
+
+ Error Details
+
+ {detail_rows_html}
+
+
+ """
+
+ # Build the page content
+ content = f"""
+
+ {create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
+
{html_module.escape(error_title)}
+ {error_box}
+ {details_section}
+
+ """
+
+ # 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,
+ )
diff --git a/tests/integration_tests/auth/test_github_provider_integration.py b/tests/integration_tests/auth/test_github_provider_integration.py
index 8ad84dc65..3563dd093 100644
--- a/tests/integration_tests/auth/test_github_provider_integration.py
+++ b/tests/integration_tests/auth/test_github_provider_integration.py
@@ -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",
diff --git a/tests/server/auth/test_oauth_consent_flow.py b/tests/server/auth/test_oauth_consent_flow.py
index 87304935d..8ff6fc404 100644
--- a/tests/server/auth/test_oauth_consent_flow.py
+++ b/tests/server/auth/test_oauth_consent_flow.py
@@ -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,
diff --git a/tests/server/auth/test_oauth_proxy.py b/tests/server/auth/test_oauth_proxy.py
index 631bd61cc..68b87d2e3 100644
--- a/tests/server/auth/test_oauth_proxy.py
+++ b/tests/server/auth/test_oauth_proxy.py
@@ -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 ",
@@ -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"],
diff --git a/tests/server/auth/test_oauth_proxy_redirect_validation.py b/tests/server/auth/test_oauth_proxy_redirect_validation.py
index 4e00648a0..c91560c52 100644
--- a/tests/server/auth/test_oauth_proxy_redirect_validation.py
+++ b/tests/server/auth/test_oauth_proxy_redirect_validation.py
@@ -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):