{client_display} is requesting access to {server_display}.
+Review the details below before approving.
+{error}: {request.query_params.get('error_description', 'Unknown error')}
", + status_code=302, + ) + + if not idp_code or not txn_id: + logger.error("IdP callback missing code or transaction ID") + return RedirectResponse( + url="data:text/html,Missing authorization code or transaction ID
", + status_code=302, + ) + + # Look up transaction data + transaction_model = await self._transaction_store.get(key=txn_id) + if not transaction_model: + logger.error("IdP callback with invalid transaction ID: %s", txn_id) + return RedirectResponse( + url="data:text/html,Invalid or expired transaction
", + status_code=302, + ) + transaction = transaction_model.model_dump() + + # Exchange IdP code for tokens (server-side) + oauth_client = AsyncOAuth2Client( + client_id=self._upstream_client_id, + client_secret=self._upstream_client_secret.get_secret_value(), + token_endpoint_auth_method=self._token_endpoint_auth_method, + timeout=HTTP_TIMEOUT_SECONDS, + ) + + try: + idp_redirect_uri = ( + f"{str(self.base_url).rstrip('/')}{self._redirect_path}" + ) + logger.debug( + f"Exchanging IdP code for tokens with redirect_uri: {idp_redirect_uri}" + ) + + # Build token exchange parameters + token_params = { + "url": self._upstream_token_endpoint, + "code": idp_code, + "redirect_uri": idp_redirect_uri, + } + + # Include proxy's code_verifier if we forwarded PKCE + proxy_code_verifier = transaction.get("proxy_code_verifier") + if proxy_code_verifier: + token_params["code_verifier"] = proxy_code_verifier + logger.debug( + "Including proxy code_verifier in token exchange for transaction %s", + txn_id, + ) + + # Add any extra token parameters configured for this proxy + if self._extra_token_params: + token_params.update(self._extra_token_params) + logger.debug( + "Adding extra token parameters for transaction %s: %s", + txn_id, + list(self._extra_token_params.keys()), + ) + + idp_tokens: dict[str, Any] = await oauth_client.fetch_token( + **token_params + ) # type: ignore[misc] + + logger.debug( + f"Successfully exchanged IdP code for tokens (transaction: {txn_id}, PKCE: {bool(proxy_code_verifier)})" + ) + + except Exception as e: + logger.error("IdP token exchange failed: %s", e) + # TODO: Forward error to client callback + return RedirectResponse( + url=f"data:text/html,Token exchange failed: {e}
", + status_code=302, + ) + + # Generate our own authorization code for the client + client_code = secrets.token_urlsafe(32) + code_expires_at = int(time.time() + DEFAULT_AUTH_CODE_EXPIRY_SECONDS) + + # Store client code with PKCE challenge and IdP tokens + await self._code_store.put( + key=client_code, + value=ClientCode( + code=client_code, + client_id=transaction["client_id"], + redirect_uri=transaction["client_redirect_uri"], + code_challenge=transaction["code_challenge"], + code_challenge_method=transaction["code_challenge_method"], + scopes=transaction["scopes"], + idp_tokens=idp_tokens, + expires_at=code_expires_at, + created_at=time.time(), + ), + ttl=DEFAULT_AUTH_CODE_EXPIRY_SECONDS, # Auto-expire after 5 minutes + ) + + # Clean up transaction + await self._transaction_store.delete(key=txn_id) + + # Build client callback URL with our code and original state + client_redirect_uri = transaction["client_redirect_uri"] + client_state = transaction["client_state"] + + callback_params = { + "code": client_code, + "state": client_state, + } + + # Add query parameters to client redirect URI + separator = "&" if "?" in client_redirect_uri else "?" + client_callback_url = ( + f"{client_redirect_uri}{separator}{urlencode(callback_params)}" + ) + + logger.debug(f"Forwarding to client callback for transaction {txn_id}") + + return RedirectResponse(url=client_callback_url, status_code=302) + + except Exception as e: + logger.error("Error in IdP callback handler: %s", e, exc_info=True) + return RedirectResponse( + url="data:text/html,Internal server error during IdP callback
", + status_code=302, + ) + + # ------------------------------------------------------------------------- + # 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 _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( + "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( + "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, + ) + 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( + "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( + "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( + "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( + "Invalid action
", status_code=400 + ) diff --git a/src/fastmcp/server/auth/oauth_proxy.py b/src/fastmcp/server/auth/oauth_proxy.py index 9729bcea2..197c7ef0f 100644 --- a/src/fastmcp/server/auth/oauth_proxy.py +++ b/src/fastmcp/server/auth/oauth_proxy.py @@ -1,2022 +1,24 @@ -"""OAuth Proxy Provider for FastMCP. +"""Backwards compatibility shim for oauth_proxy.py -This provider acts as a transparent proxy to an upstream OAuth Authorization Server, -handling Dynamic Client Registration locally while forwarding all other OAuth flows. -This enables authentication with upstream providers that don't support DCR or have -restricted client registration policies. - -Key features: -- Proxies authorization and token endpoints to upstream server -- Implements local Dynamic Client Registration with fixed upstream credentials -- Validates tokens using upstream JWKS -- Maintains minimal local state for bookkeeping -- Enhanced logging with request correlation - -This implementation is based on the OAuth 2.1 specification and is designed for -production use with enterprise identity providers. +The OauthProxy class has been moved to fastmcp.server.auth.oauth_dcr_proxy.OAuthDCRProxy +for better organization. This module provides a backwards-compatible import. """ -from __future__ import annotations +import warnings -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 +import fastmcp +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy as OAuthProxy -import httpx -from authlib.common.security import generate_token -from authlib.integrations.httpx_client import AsyncOAuth2Client -from key_value.aio.adapters.pydantic import PydanticAdapter -from key_value.aio.protocols import AsyncKeyValue -from key_value.aio.stores.memory import MemoryStore -from mcp.server.auth.handlers.token import TokenErrorResponse, TokenSuccessResponse -from mcp.server.auth.handlers.token import TokenHandler as _SDKTokenHandler -from mcp.server.auth.json_response import PydanticJSONResponse -from mcp.server.auth.middleware.client_auth import ClientAuthenticator -from mcp.server.auth.provider import ( - AccessToken, - AuthorizationCode, - AuthorizationParams, - RefreshToken, - TokenError, -) -from mcp.server.auth.routes import cors_middleware -from mcp.server.auth.settings import ( - ClientRegistrationOptions, - RevocationOptions, -) -from mcp.shared.auth import OAuthClientInformationFull, OAuthToken -from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, SecretStr -from starlette.requests import Request -from starlette.responses import HTMLResponse, RedirectResponse -from starlette.routing import Route +# Re-export for backwards compatibility +__all__ = ["OAuthProxy"] -from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier -from fastmcp.server.auth.jwt_issuer import ( - JWTIssuer, - TokenEncryption, -) -from fastmcp.server.auth.redirect_validation import ( - validate_redirect_uri, -) -from fastmcp.utilities.logging import get_logger -from fastmcp.utilities.ui import ( - BUTTON_STYLES, - DETAIL_BOX_STYLES, - INFO_BOX_STYLES, - TOOLTIP_STYLES, - create_detail_box, - 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_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 are stored encrypted at rest. They are never exposed to MCP clients. - """ - - upstream_token_id: str # Unique ID for this token set - access_token: bytes # Encrypted upstream access token - refresh_token: bytes | None # Encrypted 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 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 = "Authorization Consent", - server_name: str | None = None, - server_icon_url: str | None = None, - server_website_url: str | None = None, -) -> str: - """Create a styled HTML consent page for OAuth authorization requests.""" - # Format scopes for display - scopes_display = ", ".join(scopes) if scopes else "None" - - # Build warning box with client name if available - 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 - - warning_box = f""" -{client_display} is requesting access to {server_display}.
-Review the details below before approving.
-{error}: {request.query_params.get('error_description', 'Unknown error')}
", - status_code=302, - ) - - if not idp_code or not txn_id: - logger.error("IdP callback missing code or transaction ID") - return RedirectResponse( - url="data:text/html,Missing authorization code or transaction ID
", - status_code=302, - ) - - # Look up transaction data - transaction_model = await self._transaction_store.get(key=txn_id) - if not transaction_model: - logger.error("IdP callback with invalid transaction ID: %s", txn_id) - return RedirectResponse( - url="data:text/html,Invalid or expired transaction
", - status_code=302, - ) - transaction = transaction_model.model_dump() - - # Exchange IdP code for tokens (server-side) - oauth_client = AsyncOAuth2Client( - client_id=self._upstream_client_id, - client_secret=self._upstream_client_secret.get_secret_value(), - token_endpoint_auth_method=self._token_endpoint_auth_method, - timeout=HTTP_TIMEOUT_SECONDS, - ) - - try: - idp_redirect_uri = ( - f"{str(self.base_url).rstrip('/')}{self._redirect_path}" - ) - logger.debug( - f"Exchanging IdP code for tokens with redirect_uri: {idp_redirect_uri}" - ) - - # Build token exchange parameters - token_params = { - "url": self._upstream_token_endpoint, - "code": idp_code, - "redirect_uri": idp_redirect_uri, - } - - # Include proxy's code_verifier if we forwarded PKCE - proxy_code_verifier = transaction.get("proxy_code_verifier") - if proxy_code_verifier: - token_params["code_verifier"] = proxy_code_verifier - logger.debug( - "Including proxy code_verifier in token exchange for transaction %s", - txn_id, - ) - - # Add any extra token parameters configured for this proxy - if self._extra_token_params: - token_params.update(self._extra_token_params) - logger.debug( - "Adding extra token parameters for transaction %s: %s", - txn_id, - list(self._extra_token_params.keys()), - ) - - idp_tokens: dict[str, Any] = await oauth_client.fetch_token( - **token_params - ) # type: ignore[misc] - - logger.debug( - f"Successfully exchanged IdP code for tokens (transaction: {txn_id}, PKCE: {bool(proxy_code_verifier)})" - ) - - except Exception as e: - logger.error("IdP token exchange failed: %s", e) - # TODO: Forward error to client callback - return RedirectResponse( - url=f"data:text/html,Token exchange failed: {e}
", - status_code=302, - ) - - # Generate our own authorization code for the client - client_code = secrets.token_urlsafe(32) - code_expires_at = int(time.time() + DEFAULT_AUTH_CODE_EXPIRY_SECONDS) - - # Store client code with PKCE challenge and IdP tokens - await self._code_store.put( - key=client_code, - value=ClientCode( - code=client_code, - client_id=transaction["client_id"], - redirect_uri=transaction["client_redirect_uri"], - code_challenge=transaction["code_challenge"], - code_challenge_method=transaction["code_challenge_method"], - scopes=transaction["scopes"], - idp_tokens=idp_tokens, - expires_at=code_expires_at, - created_at=time.time(), - ), - ttl=DEFAULT_AUTH_CODE_EXPIRY_SECONDS, # Auto-expire after 5 minutes - ) - - # Clean up transaction - await self._transaction_store.delete(key=txn_id) - - # Build client callback URL with our code and original state - client_redirect_uri = transaction["client_redirect_uri"] - client_state = transaction["client_state"] - - callback_params = { - "code": client_code, - "state": client_state, - } - - # Add query parameters to client redirect URI - separator = "&" if "?" in client_redirect_uri else "?" - client_callback_url = ( - f"{client_redirect_uri}{separator}{urlencode(callback_params)}" - ) - - logger.debug(f"Forwarding to client callback for transaction {txn_id}") - - return RedirectResponse(url=client_callback_url, status_code=302) - - except Exception as e: - logger.error("Error in IdP callback handler: %s", e, exc_info=True) - return RedirectResponse( - url="data:text/html,Internal server error during IdP callback
", - status_code=302, - ) - - # ------------------------------------------------------------------------- - # 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 _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( - "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( - "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, - ) - 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( - "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( - "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( - "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( - "Invalid action
", status_code=400 - ) diff --git a/src/fastmcp/server/auth/oidc_proxy.py b/src/fastmcp/server/auth/oidc_proxy.py index 6d5c82fe8..7084d1b02 100644 --- a/src/fastmcp/server/auth/oidc_proxy.py +++ b/src/fastmcp/server/auth/oidc_proxy.py @@ -17,7 +17,7 @@ from pydantic import AnyHttpUrl, BaseModel, model_validator from typing_extensions import Self from fastmcp.server.auth import TokenVerifier -from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy from fastmcp.server.auth.providers.jwt import JWTVerifier from fastmcp.utilities.logging import get_logger @@ -169,7 +169,7 @@ class OIDCConfiguration(BaseModel): raise -class OIDCProxy(OAuthProxy): +class OIDCProxy(OAuthDCRProxy): """OAuth provider that wraps OAuthProxy to provide configuration via an OIDC configuration URL. This provider makes it easier to add OAuth protection for any upstream provider diff --git a/src/fastmcp/server/auth/providers/azure.py b/src/fastmcp/server/auth/providers/azure.py index 93ad94182..1e1e69c32 100644 --- a/src/fastmcp/server/auth/providers/azure.py +++ b/src/fastmcp/server/auth/providers/azure.py @@ -12,7 +12,7 @@ from key_value.aio.protocols import AsyncKeyValue from pydantic import SecretStr, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict -from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy from fastmcp.server.auth.providers.jwt import JWTVerifier from fastmcp.settings import ENV_FILE from fastmcp.utilities.auth import parse_scopes @@ -57,7 +57,7 @@ class AzureProviderSettings(BaseSettings): return parse_scopes(v) -class AzureProvider(OAuthProxy): +class AzureProvider(OAuthDCRProxy): """Azure (Microsoft Entra) OAuth provider for FastMCP. This provider implements Azure/Microsoft Entra ID authentication using the diff --git a/src/fastmcp/server/auth/providers/github.py b/src/fastmcp/server/auth/providers/github.py index d34bf041d..83ccd42d4 100644 --- a/src/fastmcp/server/auth/providers/github.py +++ b/src/fastmcp/server/auth/providers/github.py @@ -28,7 +28,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict from fastmcp.server.auth import TokenVerifier from fastmcp.server.auth.auth import AccessToken -from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy from fastmcp.settings import ENV_FILE from fastmcp.utilities.auth import parse_scopes from fastmcp.utilities.logging import get_logger @@ -166,7 +166,7 @@ class GitHubTokenVerifier(TokenVerifier): return None -class GitHubProvider(OAuthProxy): +class GitHubProvider(OAuthDCRProxy): """Complete GitHub OAuth provider for FastMCP. This provider makes it trivial to add GitHub OAuth protection to any diff --git a/src/fastmcp/server/auth/providers/google.py b/src/fastmcp/server/auth/providers/google.py index 1d925a3d0..8c2c89cfb 100644 --- a/src/fastmcp/server/auth/providers/google.py +++ b/src/fastmcp/server/auth/providers/google.py @@ -30,7 +30,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict from fastmcp.server.auth import TokenVerifier from fastmcp.server.auth.auth import AccessToken -from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy from fastmcp.settings import ENV_FILE from fastmcp.utilities.auth import parse_scopes from fastmcp.utilities.logging import get_logger @@ -182,7 +182,7 @@ class GoogleTokenVerifier(TokenVerifier): return None -class GoogleProvider(OAuthProxy): +class GoogleProvider(OAuthDCRProxy): """Complete Google OAuth provider for FastMCP. This provider makes it trivial to add Google OAuth protection to any diff --git a/src/fastmcp/server/auth/providers/workos.py b/src/fastmcp/server/auth/providers/workos.py index 52b6c62d2..bd88c5c98 100644 --- a/src/fastmcp/server/auth/providers/workos.py +++ b/src/fastmcp/server/auth/providers/workos.py @@ -18,7 +18,7 @@ from starlette.responses import JSONResponse from starlette.routing import Route from fastmcp.server.auth import AccessToken, RemoteAuthProvider, TokenVerifier -from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy from fastmcp.server.auth.providers.jwt import JWTVerifier from fastmcp.settings import ENV_FILE from fastmcp.utilities.auth import parse_scopes @@ -125,7 +125,7 @@ class WorkOSTokenVerifier(TokenVerifier): return None -class WorkOSProvider(OAuthProxy): +class WorkOSProvider(OAuthDCRProxy): """Complete WorkOS OAuth provider for FastMCP. This provider implements WorkOS AuthKit OAuth using the OAuth Proxy pattern. diff --git a/tests/integration_tests/auth/test_github_provider_integration.py b/tests/integration_tests/auth/test_github_provider_integration.py index b03798c62..b4071082a 100644 --- a/tests/integration_tests/auth/test_github_provider_integration.py +++ b/tests/integration_tests/auth/test_github_provider_integration.py @@ -82,7 +82,7 @@ def create_github_server_with_mock_callback(base_url: str) -> FastMCP: import secrets import time - from fastmcp.server.auth.oauth_proxy import ClientCode + from fastmcp.server.auth.oauth_dcr_proxy import ClientCode # Generate a fake authorization code fake_code = secrets.token_urlsafe(32) diff --git a/tests/server/auth/oauth_dcr_proxy/__init__.py b/tests/server/auth/oauth_dcr_proxy/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/server/auth/test_oauth_consent_flow.py b/tests/server/auth/oauth_dcr_proxy/test_oauth_consent_flow.py similarity index 98% rename from tests/server/auth/test_oauth_consent_flow.py rename to tests/server/auth/oauth_dcr_proxy/test_oauth_consent_flow.py index 648bcf990..e44fd53f8 100644 --- a/tests/server/auth/test_oauth_consent_flow.py +++ b/tests/server/auth/oauth_dcr_proxy/test_oauth_consent_flow.py @@ -25,7 +25,7 @@ from starlette.applications import Starlette from starlette.testclient import TestClient from fastmcp.server.auth.auth import TokenVerifier -from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy class MockTokenVerifier(TokenVerifier): @@ -69,7 +69,7 @@ def storage(): @pytest.fixture def oauth_proxy_with_storage(storage): """Create OAuth proxy with explicit storage backend.""" - return OAuthProxy( + return OAuthDCRProxy( upstream_authorization_endpoint="https://github.com/login/oauth/authorize", upstream_token_endpoint="https://github.com/login/oauth/access_token", upstream_client_id="test-upstream-client", @@ -84,7 +84,7 @@ def oauth_proxy_with_storage(storage): @pytest.fixture def oauth_proxy_https(): """OAuthProxy configured with HTTPS base_url for __Host- cookies.""" - return OAuthProxy( + return OAuthDCRProxy( upstream_authorization_endpoint="https://github.com/login/oauth/authorize", upstream_token_endpoint="https://github.com/login/oauth/access_token", upstream_client_id="client-id", @@ -96,7 +96,7 @@ def oauth_proxy_https(): async def _start_flow( - proxy: OAuthProxy, client_id: str, redirect: str + proxy: OAuthDCRProxy, client_id: str, redirect: str ) -> tuple[str, str]: """Register client and start auth; returns (txn_id, consent_url).""" await proxy.register_client( @@ -503,7 +503,7 @@ 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 + from fastmcp.server.auth.oauth_dcr_proxy import OAuthTransaction client = OAuthClientInformationFull( client_id="pydantic-test-client", @@ -674,7 +674,7 @@ class TestConsentPageServerIcon: verifier.verify_token = Mock(return_value=None) # Create OAuthProxy - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -705,7 +705,7 @@ class TestConsentPageServerIcon: await proxy.register_client(client_info) # Create a transaction manually - from fastmcp.server.auth.oauth_proxy import OAuthTransaction + from fastmcp.server.auth.oauth_dcr_proxy import OAuthTransaction txn_id = "test-txn-id" transaction = OAuthTransaction( @@ -745,7 +745,7 @@ class TestConsentPageServerIcon: verifier.verify_token = Mock(return_value=None) # Create OAuthProxy - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -769,7 +769,7 @@ class TestConsentPageServerIcon: await proxy.register_client(client_info) # Create a transaction - from fastmcp.server.auth.oauth_proxy import OAuthTransaction + from fastmcp.server.auth.oauth_dcr_proxy import OAuthTransaction txn_id = "test-txn-id" transaction = OAuthTransaction( @@ -811,7 +811,7 @@ class TestConsentPageServerIcon: verifier.verify_token = Mock(return_value=None) # Create OAuthProxy - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -839,7 +839,7 @@ class TestConsentPageServerIcon: await proxy.register_client(client_info) # Create a transaction - from fastmcp.server.auth.oauth_proxy import OAuthTransaction + from fastmcp.server.auth.oauth_dcr_proxy import OAuthTransaction txn_id = "test-txn-id" transaction = OAuthTransaction( diff --git a/tests/server/auth/oauth_dcr_proxy/test_oauth_proxy.py b/tests/server/auth/oauth_dcr_proxy/test_oauth_proxy.py new file mode 100644 index 000000000..89b4ac63b --- /dev/null +++ b/tests/server/auth/oauth_dcr_proxy/test_oauth_proxy.py @@ -0,0 +1,1297 @@ +"""Comprehensive tests for OAuth Proxy Provider functionality. + +This test suite covers: +1. Initialization and configuration +2. Client registration (DCR) +3. Authorization flow +4. Token management +5. PKCE forwarding +6. Token endpoint authentication methods +7. E2E testing with mock OAuth provider +""" + +import asyncio +import secrets +import time +from unittest.mock import AsyncMock, Mock, patch +from urllib.parse import parse_qs, urlencode, urlparse + +import httpx +import pytest +from mcp.server.auth.provider import AuthorizationParams +from mcp.shared.auth import OAuthClientInformationFull +from pydantic import AnyUrl +from starlette.applications import Starlette +from starlette.responses import JSONResponse +from starlette.routing import Route + +from fastmcp import FastMCP +from fastmcp.server.auth.auth import AccessToken, RefreshToken, TokenVerifier +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy +from fastmcp.server.auth.providers.jwt import JWTVerifier + +# ============================================================================= +# Mock OAuth Provider for E2E Testing +# ============================================================================= + + +class MockOAuthProvider: + """Mock OAuth provider for testing OAuth proxy E2E flows. + + This provider simulates a complete OAuth server without requiring: + - Real authentication credentials + - Browser automation + - Network calls to external services + """ + + def __init__(self, port: int = 0): + self.port = port + self.base_url = f"http://localhost:{port}" + self.app = None + self.server = None + + # Storage for OAuth state + self.authorization_codes = {} + self.access_tokens = {} + self.refresh_tokens = {} + self.revoked_tokens = set() + + # Tracking for assertions + self.authorize_called = False + self.token_called = False + self.refresh_called = False + self.revoke_called = False + + # Configuration + self.require_pkce = False + self.token_endpoint_auth_method = "client_secret_basic" + + @property + def authorize_endpoint(self) -> str: + return f"{self.base_url}/authorize" + + @property + def token_endpoint(self) -> str: + return f"{self.base_url}/token" + + @property + def revocation_endpoint(self) -> str: + return f"{self.base_url}/revoke" + + def create_app(self) -> Starlette: + """Create the mock OAuth server application.""" + return Starlette( + routes=[ + Route("/authorize", self.handle_authorize), + Route("/token", self.handle_token, methods=["POST"]), + Route("/revoke", self.handle_revoke, methods=["POST"]), + ] + ) + + async def handle_authorize(self, request): + """Handle authorization requests.""" + self.authorize_called = True + query = dict(request.query_params) + + # Validate PKCE if required + if self.require_pkce and "code_challenge" not in query: + return JSONResponse( + {"error": "invalid_request", "error_description": "PKCE required"}, + status_code=400, + ) + + # Generate authorization code + code = secrets.token_urlsafe(32) + self.authorization_codes[code] = { + "client_id": query.get("client_id"), + "redirect_uri": query.get("redirect_uri"), + "state": query.get("state"), + "code_challenge": query.get("code_challenge"), + "code_challenge_method": query.get("code_challenge_method", "S256"), + "scope": query.get("scope"), + "created_at": time.time(), + } + + # Redirect back to callback + redirect_uri = query["redirect_uri"] + params = {"code": code} + if query.get("state"): + params["state"] = query["state"] + + redirect_url = f"{redirect_uri}?{urlencode(params)}" + return JSONResponse( + content={}, status_code=302, headers={"Location": redirect_url} + ) + + async def handle_token(self, request): + """Handle token requests.""" + self.token_called = True + form = await request.form() + grant_type = form.get("grant_type") + + if grant_type == "authorization_code": + code = form.get("code") + if code not in self.authorization_codes: + return JSONResponse( + {"error": "invalid_grant", "error_description": "Invalid code"}, + status_code=400, + ) + + # Validate PKCE if it was used + auth_data = self.authorization_codes[code] + if auth_data.get("code_challenge"): + verifier = form.get("code_verifier") + if not verifier: + return JSONResponse( + { + "error": "invalid_request", + "error_description": "Missing code_verifier", + }, + status_code=400, + ) + # In a real implementation, we'd validate the verifier + + # Generate tokens + access_token = f"mock_access_{secrets.token_hex(16)}" + refresh_token = f"mock_refresh_{secrets.token_hex(16)}" + + self.access_tokens[access_token] = { + "client_id": auth_data["client_id"], + "scope": auth_data.get("scope"), + "expires_at": time.time() + 3600, + } + self.refresh_tokens[refresh_token] = { + "client_id": auth_data["client_id"], + "scope": auth_data.get("scope"), + } + + # Clean up used code + del self.authorization_codes[code] + + return JSONResponse( + { + "access_token": access_token, + "token_type": "Bearer", + "expires_in": 3600, + "refresh_token": refresh_token, + "scope": auth_data.get("scope"), + } + ) + + elif grant_type == "refresh_token": + self.refresh_called = True + refresh_token = form.get("refresh_token") + + if refresh_token not in self.refresh_tokens: + return JSONResponse( + { + "error": "invalid_grant", + "error_description": "Invalid refresh token", + }, + status_code=400, + ) + + # Generate new access token + new_access = f"mock_access_{secrets.token_hex(16)}" + token_data = self.refresh_tokens[refresh_token] + + self.access_tokens[new_access] = { + "client_id": token_data["client_id"], + "scope": token_data.get("scope"), + "expires_at": time.time() + 3600, + } + + return JSONResponse( + { + "access_token": new_access, + "token_type": "Bearer", + "expires_in": 3600, + "refresh_token": refresh_token, # Same refresh token + "scope": token_data.get("scope"), + } + ) + + return JSONResponse({"error": "unsupported_grant_type"}, status_code=400) + + async def handle_revoke(self, request): + """Handle token revocation.""" + self.revoke_called = True + form = await request.form() + token = form.get("token") + + if token: + self.revoked_tokens.add(token) + # Remove from active tokens + self.access_tokens.pop(token, None) + self.refresh_tokens.pop(token, None) + + return JSONResponse({}) + + async def start(self): + """Start the mock OAuth server.""" + import socket + + from uvicorn import Config, Server + + self.app = self.create_app() + + # If port is 0, find an available port + if self.port == 0: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + s.listen(1) + self.port = s.getsockname()[1] + + self.base_url = f"http://localhost:{self.port}" + config = Config( + self.app, + host="localhost", + port=self.port, + log_level="error", + ws="websockets-sansio", + ) + self.server = Server(config) + + # Start server in background + asyncio.create_task(self.server.serve()) + + # Wait for server to be ready + await asyncio.sleep(0.05) + + async def stop(self): + """Stop the mock OAuth server.""" + if self.server: + self.server.should_exit = True + await asyncio.sleep(0.01) + + def reset(self): + """Reset all state for next test.""" + self.authorization_codes.clear() + self.access_tokens.clear() + self.refresh_tokens.clear() + self.revoked_tokens.clear() + self.authorize_called = False + self.token_called = False + self.refresh_called = False + self.revoke_called = False + + +class MockTokenVerifier(TokenVerifier): + """Mock token verifier for testing.""" + + def __init__(self, required_scopes=None): + self.required_scopes = required_scopes or ["read", "write"] + self.verify_called = False + + async def verify_token(self, token: str) -> AccessToken: + """Mock token verification.""" + self.verify_called = True + return AccessToken( + token=token, + client_id="mock-client", + scopes=self.required_scopes, + expires_at=int(time.time() + 3600), + ) + + +# ============================================================================= +# Test Fixtures +# ============================================================================= + + +@pytest.fixture +def jwt_verifier(): + """Create a mock JWT verifier for testing.""" + verifier = Mock(spec=JWTVerifier) + verifier.required_scopes = ["read", "write"] + verifier.verify_token = Mock(return_value=None) + return verifier + + +@pytest.fixture +def oauth_proxy(jwt_verifier): + """Create a standard OAuthProxy instance for testing.""" + return OAuthDCRProxy( + upstream_authorization_endpoint="https://github.com/login/oauth/authorize", + upstream_token_endpoint="https://github.com/login/oauth/access_token", + upstream_client_id="test-client-id", + upstream_client_secret="test-client-secret", + token_verifier=jwt_verifier, + base_url="https://myserver.com", + redirect_path="/auth/callback", + ) + + +@pytest.fixture +async def mock_oauth_provider(): + """Create and start a mock OAuth provider.""" + provider = MockOAuthProvider() + await provider.start() + yield provider + await provider.stop() + + +# ============================================================================= +# Test Classes +# ============================================================================= + + +class TestOAuthProxyInitialization: + """Tests for OAuth proxy initialization and configuration.""" + + def test_basic_initialization(self, jwt_verifier): + """Test basic proxy initialization with required parameters.""" + proxy = OAuthDCRProxy( + upstream_authorization_endpoint="https://auth.example.com/authorize", + upstream_token_endpoint="https://auth.example.com/token", + upstream_client_id="client-123", + upstream_client_secret="secret-456", + token_verifier=jwt_verifier, + base_url="https://api.example.com", + ) + + assert ( + proxy._upstream_authorization_endpoint + == "https://auth.example.com/authorize" + ) + assert proxy._upstream_token_endpoint == "https://auth.example.com/token" + assert proxy._upstream_client_id == "client-123" + assert proxy._upstream_client_secret.get_secret_value() == "secret-456" + assert str(proxy.base_url) == "https://api.example.com/" + + def test_all_optional_parameters(self, jwt_verifier): + """Test initialization with all optional parameters.""" + proxy = OAuthDCRProxy( + upstream_authorization_endpoint="https://auth.example.com/authorize", + upstream_token_endpoint="https://auth.example.com/token", + upstream_client_id="client-123", + upstream_client_secret="secret-456", + upstream_revocation_endpoint="https://auth.example.com/revoke", + token_verifier=jwt_verifier, + base_url="https://api.example.com", + redirect_path="/custom/callback", + issuer_url="https://issuer.example.com", + service_documentation_url="https://docs.example.com", + allowed_client_redirect_uris=["http://localhost:*"], + valid_scopes=["custom", "scopes"], + forward_pkce=False, + token_endpoint_auth_method="client_secret_post", + ) + + assert proxy._upstream_revocation_endpoint == "https://auth.example.com/revoke" + assert proxy._redirect_path == "/custom/callback" + assert proxy._forward_pkce is False + assert proxy._token_endpoint_auth_method == "client_secret_post" + assert proxy.client_registration_options is not None + assert proxy.client_registration_options.valid_scopes == ["custom", "scopes"] + + def test_redirect_path_normalization(self, jwt_verifier): + """Test that redirect_path is normalized with leading slash.""" + proxy = OAuthDCRProxy( + upstream_authorization_endpoint="https://auth.com/authorize", + upstream_token_endpoint="https://auth.com/token", + upstream_client_id="client", + upstream_client_secret="secret", + token_verifier=jwt_verifier, + base_url="https://api.com", + redirect_path="auth/callback", # No leading slash + ) + assert proxy._redirect_path == "/auth/callback" + + +class TestOAuthProxyClientRegistration: + """Tests for OAuth proxy client registration (DCR).""" + + async def test_register_client(self, oauth_proxy): + """Test client registration creates ProxyDCRClient.""" + client_info = OAuthClientInformationFull( + client_id="original-client", + client_secret="original-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + await oauth_proxy.register_client(client_info) + + # Client should be retrievable with original credentials + stored = await oauth_proxy.get_client("original-client") + assert stored is not None + assert stored.client_id == "original-client" + assert stored.client_secret == "original-secret" + + async def test_get_registered_client(self, oauth_proxy): + """Test retrieving a registered client.""" + client_info = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:8080/callback")], + ) + await oauth_proxy.register_client(client_info) + + retrieved = await oauth_proxy.get_client("test-client") + assert retrieved is not None + assert retrieved.client_id == "test-client" + + async def test_get_unregistered_client_returns_none(self, oauth_proxy): + """Test that unregistered clients return None.""" + client = await oauth_proxy.get_client("unknown-client") + assert client is None + + +class TestOAuthProxyAuthorization: + """Tests for OAuth proxy authorization flow.""" + + async def test_authorize_creates_transaction(self, oauth_proxy): + """Test that authorize creates transaction and redirects to consent.""" + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:54321/callback")], + ) + + # Register client first (required for consent flow) + await oauth_proxy.register_client(client) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:54321/callback"), + redirect_uri_provided_explicitly=True, + state="client-state-123", + code_challenge="challenge-abc", + code_challenge_method="S256", + scopes=["read", "write"], + ) + + redirect_url = await oauth_proxy.authorize(client, params) + + # Parse the redirect URL + parsed = urlparse(redirect_url) + query_params = parse_qs(parsed.query) + + # Should redirect to consent page + assert "/consent" in redirect_url + assert "txn_id" in query_params + + # Verify transaction was stored with correct data + txn_id = query_params["txn_id"][0] + transaction = await oauth_proxy._transaction_store.get(key=txn_id) + assert transaction is not None + assert transaction.client_id == "test-client" + assert transaction.code_challenge == "challenge-abc" + assert transaction.client_state == "client-state-123" + assert transaction.scopes == ["read", "write"] + + +class TestOAuthProxyPKCE: + """Tests for OAuth proxy PKCE forwarding.""" + + @pytest.fixture + def proxy_with_pkce(self, jwt_verifier): + return OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="upstream-client", + upstream_client_secret="upstream-secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + forward_pkce=True, + ) + + @pytest.fixture + def proxy_without_pkce(self, jwt_verifier): + return OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="upstream-client", + upstream_client_secret="upstream-secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + forward_pkce=False, + ) + + async def test_pkce_forwarding_enabled(self, proxy_with_pkce): + """Test that proxy generates and forwards its own PKCE.""" + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy_with_pkce.register_client(client) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="client_challenge", + scopes=["read"], + ) + + redirect_url = await proxy_with_pkce.authorize(client, params) + query_params = parse_qs(urlparse(redirect_url).query) + + # Should redirect to consent page + assert "/consent" in redirect_url + assert "txn_id" in query_params + + # Transaction should store both challenges + txn_id = query_params["txn_id"][0] + transaction = await proxy_with_pkce._transaction_store.get(key=txn_id) + assert transaction is not None + assert transaction.code_challenge == "client_challenge" # Client's + assert transaction.proxy_code_verifier is not None # Proxy's verifier + # Proxy code challenge is computed from verifier when building upstream URL + # Just verify the verifier exists and is different from client's challenge + assert len(transaction.proxy_code_verifier) > 0 + + async def test_pkce_forwarding_disabled(self, proxy_without_pkce): + """Test that PKCE is not forwarded when disabled.""" + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy_without_pkce.register_client(client) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="client_challenge", + scopes=["read"], + ) + + redirect_url = await proxy_without_pkce.authorize(client, params) + query_params = parse_qs(urlparse(redirect_url).query) + + # Should redirect to consent page + assert "/consent" in redirect_url + assert "txn_id" in query_params + + # Client's challenge still stored, but no proxy PKCE + txn_id = query_params["txn_id"][0] + transaction = await proxy_without_pkce._transaction_store.get(key=txn_id) + assert transaction is not None + assert transaction.code_challenge == "client_challenge" + assert transaction.proxy_code_verifier is None # No proxy PKCE when disabled + + +class TestOAuthProxyTokenEndpointAuth: + """Tests for token endpoint authentication methods.""" + + def test_token_auth_method_initialization(self, jwt_verifier): + """Test different token endpoint auth methods.""" + # client_secret_post + proxy_post = OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="client", + upstream_client_secret="secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + token_endpoint_auth_method="client_secret_post", + ) + assert proxy_post._token_endpoint_auth_method == "client_secret_post" + + # client_secret_basic (default) + proxy_basic = OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="client", + upstream_client_secret="secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + token_endpoint_auth_method="client_secret_basic", + ) + assert proxy_basic._token_endpoint_auth_method == "client_secret_basic" + + # None (use authlib default) + proxy_default = OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="client", + upstream_client_secret="secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + ) + assert proxy_default._token_endpoint_auth_method is None + + async def test_token_auth_method_passed_to_client(self, jwt_verifier): + """Test that auth method is passed to AsyncOAuth2Client.""" + proxy = OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="client-id", + upstream_client_secret="client-secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + token_endpoint_auth_method="client_secret_post", + ) + + # First, create a valid FastMCP token via full OAuth flow + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Mock the upstream OAuth provider response + with patch( + "fastmcp.server.auth.oauth_dcr_proxy.AsyncOAuth2Client" + ) as MockClient: + mock_client = AsyncMock() + + # Mock initial token exchange (authorization code flow) + mock_client.fetch_token = AsyncMock( + return_value={ + "access_token": "upstream-access-token", + "refresh_token": "upstream-refresh-token", + "expires_in": 3600, + "token_type": "Bearer", + } + ) + + # Mock token refresh + mock_client.refresh_token = AsyncMock( + return_value={ + "access_token": "new-upstream-token", + "refresh_token": "new-upstream-refresh", + "expires_in": 3600, + "token_type": "Bearer", + } + ) + MockClient.return_value = mock_client + + # Register client and do initial OAuth flow to get valid FastMCP tokens + await proxy.register_client(client) + + # Store client code that would be created during OAuth callback + from fastmcp.server.auth.oauth_dcr_proxy import ClientCode + + client_code = ClientCode( + code="test-auth-code", + client_id="test-client", + redirect_uri="http://localhost:12345/callback", + code_challenge="", + code_challenge_method="S256", + scopes=["read"], + idp_tokens={ + "access_token": "upstream-access-token", + "refresh_token": "upstream-refresh-token", + "expires_in": 3600, + "token_type": "Bearer", + }, + expires_at=time.time() + 300, + created_at=time.time(), + ) + 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"], + expires_at=time.time() + 300, + client_id="test-client", + code_challenge="", + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + ) + result = await proxy.exchange_authorization_code( + client=client, + authorization_code=auth_code, + ) + + # Now test refresh with the valid FastMCP refresh token + assert result.refresh_token is not None + fastmcp_refresh = RefreshToken( + token=result.refresh_token, + client_id="test-client", + scopes=["read"], + expires_at=None, + ) + + # Reset mock to check refresh call + MockClient.reset_mock() + mock_client.refresh_token = AsyncMock( + return_value={ + "access_token": "new-upstream-token-2", + "refresh_token": "new-upstream-refresh-2", + "expires_in": 3600, + "token_type": "Bearer", + } + ) + MockClient.return_value = mock_client + + await proxy.exchange_refresh_token(client, fastmcp_refresh, ["read"]) + + # Verify auth method was passed to OAuth client + MockClient.assert_called_with( + client_id="client-id", + client_secret="client-secret", + token_endpoint_auth_method="client_secret_post", + timeout=30.0, + ) + + +class TestOAuthProxyE2E: + """End-to-end tests using mock OAuth provider.""" + + async def test_full_oauth_flow_with_mock_provider(self, mock_oauth_provider): + """Test complete OAuth flow with mock provider.""" + # Create proxy pointing to mock provider + proxy = OAuthDCRProxy( + upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, + upstream_token_endpoint=mock_oauth_provider.token_endpoint, + upstream_client_id="mock-client", + upstream_client_secret="mock-secret", + token_verifier=MockTokenVerifier(), + base_url="http://localhost:8000", + ) + + # Create FastMCP server with proxy + server = FastMCP("Test Server", auth=proxy) + + @server.tool + def protected_tool() -> str: + return "Protected data" + + # Start authorization flow + client_info = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy.register_client(client_info) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="", # Empty string for no PKCE + scopes=["read"], + ) + + # Get authorization URL (now returns consent redirect) + auth_url = await proxy.authorize(client_info, params) + + # Should redirect to consent page + assert "/consent" in auth_url + query_params = parse_qs(urlparse(auth_url).query) + assert "txn_id" in query_params + + # Verify transaction was created with correct configuration + txn_id = query_params["txn_id"][0] + transaction = await proxy._transaction_store.get(key=txn_id) + assert transaction is not None + assert transaction.client_id == "test-client" + assert transaction.scopes == ["read"] + # Transaction ID itself is used as upstream state parameter + assert transaction.txn_id == txn_id + + async def test_token_refresh_with_mock_provider(self, mock_oauth_provider): + """Test token refresh flow with mock provider.""" + proxy = OAuthDCRProxy( + upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, + upstream_token_endpoint=mock_oauth_provider.token_endpoint, + upstream_client_id="mock-client", + upstream_client_secret="mock-secret", + token_verifier=MockTokenVerifier(), + base_url="http://localhost:8000", + ) + + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy.register_client(client) + + # Set up initial upstream tokens in mock provider + upstream_refresh_token = "mock_refresh_initial" + mock_oauth_provider.refresh_tokens[upstream_refresh_token] = { + "client_id": "mock-client", + "scope": "read write", + } + + with patch( + "fastmcp.server.auth.oauth_dcr_proxy.AsyncOAuth2Client" + ) as MockClient: + mock_client = AsyncMock() + + # Mock initial token exchange to get FastMCP tokens + mock_client.fetch_token = AsyncMock( + return_value={ + "access_token": "upstream-access-initial", + "refresh_token": upstream_refresh_token, + "expires_in": 3600, + "token_type": "Bearer", + } + ) + + # Configure mock to call real provider for refresh + async def mock_refresh(*args, **kwargs): + async with httpx.AsyncClient() as http: + response = await http.post( + mock_oauth_provider.token_endpoint, + data={ + "grant_type": "refresh_token", + "refresh_token": upstream_refresh_token, + }, + ) + return response.json() + + mock_client.refresh_token = mock_refresh + MockClient.return_value = mock_client + + # Store client code that would be created during OAuth callback + from fastmcp.server.auth.oauth_dcr_proxy import ClientCode + + client_code = ClientCode( + code="test-auth-code", + client_id="test-client", + redirect_uri="http://localhost:12345/callback", + code_challenge="", + code_challenge_method="S256", + scopes=["read", "write"], + idp_tokens={ + "access_token": "upstream-access-initial", + "refresh_token": upstream_refresh_token, + "expires_in": 3600, + "token_type": "Bearer", + }, + expires_at=time.time() + 300, + created_at=time.time(), + ) + 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"], + expires_at=time.time() + 300, + client_id="test-client", + code_challenge="", + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + ) + initial_result = await proxy.exchange_authorization_code( + client=client, + authorization_code=auth_code, + ) + + # Now test refresh with the valid FastMCP refresh token + assert initial_result.refresh_token is not None + fastmcp_refresh = RefreshToken( + token=initial_result.refresh_token, + client_id="test-client", + scopes=["read"], + expires_at=None, + ) + + result = await proxy.exchange_refresh_token( + client, fastmcp_refresh, ["read"] + ) + + # Should return new FastMCP tokens (not upstream tokens) + assert result.access_token != "upstream-access-initial" + # FastMCP tokens are JWTs (have 3 segments) + assert len(result.access_token.split(".")) == 3 + assert mock_oauth_provider.refresh_called + + async def test_pkce_validation_with_mock_provider(self, mock_oauth_provider): + """Test PKCE validation with mock provider.""" + mock_oauth_provider.require_pkce = True + + proxy = OAuthDCRProxy( + upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, + upstream_token_endpoint=mock_oauth_provider.token_endpoint, + upstream_client_id="mock-client", + upstream_client_secret="mock-secret", + token_verifier=MockTokenVerifier(), + base_url="http://localhost:8000", + forward_pkce=True, # Enable PKCE forwarding + ) + + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy.register_client(client) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="client_challenge_value", + code_challenge_method="S256", + scopes=["read"], + ) + + # Start authorization with PKCE + auth_url = await proxy.authorize(client, params) + query_params = parse_qs(urlparse(auth_url).query) + + # Should redirect to consent page + assert "/consent" in auth_url + assert "txn_id" in query_params + + # Transaction should have proxy's PKCE verifier (different from client's) + txn_id = query_params["txn_id"][0] + transaction = await proxy._transaction_store.get(key=txn_id) + assert transaction is not None + assert ( + transaction.code_challenge == "client_challenge_value" + ) # Client's challenge + assert transaction.proxy_code_verifier is not None # Proxy generated its own + # Proxy code challenge is computed from verifier when needed + assert len(transaction.proxy_code_verifier) > 0 + + +class TestParameterForwarding: + """Tests for forwarding custom parameters to upstream OAuth provider.""" + + @pytest.fixture + def proxy_with_extra_params(self, jwt_verifier): + """Create OAuthProxy with extra parameters configured.""" + return OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="upstream-client", + upstream_client_secret="upstream-secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + extra_authorize_params={"audience": "https://api.example.com"}, + extra_token_params={"audience": "https://api.example.com"}, + ) + + @pytest.fixture + def proxy_without_extra_params(self, jwt_verifier): + """Create OAuthProxy without extra parameters.""" + return OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="upstream-client", + upstream_client_secret="upstream-secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + ) + + async def test_resource_parameter_forwarding(self, proxy_without_extra_params): + """Test that RFC 8707 resource parameter is forwarded from client request.""" + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy_without_extra_params.register_client(client) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="client_challenge", + scopes=["read"], + resource="https://api.example.com/v1", # RFC 8707 resource indicator + ) + + redirect_url = await proxy_without_extra_params.authorize(client, params) + query_params = parse_qs(urlparse(redirect_url).query) + + # Should redirect to consent page + assert "/consent" in redirect_url + assert "txn_id" in query_params + + # Resource parameter should be stored in transaction for upstream forwarding + txn_id = query_params["txn_id"][0] + transaction = await proxy_without_extra_params._transaction_store.get( + key=txn_id + ) + assert transaction is not None + assert transaction.resource == "https://api.example.com/v1" + + async def test_extra_authorize_params(self, proxy_with_extra_params): + """Test that extra authorization parameters are included.""" + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy_with_extra_params.register_client(client) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="client_challenge", + scopes=["read"], + ) + + redirect_url = await proxy_with_extra_params.authorize(client, params) + query_params = parse_qs(urlparse(redirect_url).query) + + # Should redirect to consent page + assert "/consent" in redirect_url + assert "txn_id" in query_params + + # Extra audience parameter is configured at proxy level (not per-transaction) + txn_id = query_params["txn_id"][0] + transaction = await proxy_with_extra_params._transaction_store.get(key=txn_id) + assert transaction is not None + # Verify proxy has extra params configured + assert ( + proxy_with_extra_params._extra_authorize_params.get("audience") + == "https://api.example.com" + ) + + async def test_resource_and_extra_params_together(self, proxy_with_extra_params): + """Test that both resource and extra params can be used together.""" + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy_with_extra_params.register_client(client) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="client_challenge", + scopes=["read"], + resource="https://resource.example.com", # Client-specified resource + ) + + redirect_url = await proxy_with_extra_params.authorize(client, params) + query_params = parse_qs(urlparse(redirect_url).query) + + # Should redirect to consent page + assert "/consent" in redirect_url + assert "txn_id" in query_params + + # Resource stored in transaction, extra params configured at proxy level + txn_id = query_params["txn_id"][0] + transaction = await proxy_with_extra_params._transaction_store.get(key=txn_id) + assert transaction is not None + assert transaction.resource == "https://resource.example.com" + assert ( + proxy_with_extra_params._extra_authorize_params.get("audience") + == "https://api.example.com" + ) + + async def test_no_extra_params_when_not_configured( + self, proxy_without_extra_params + ): + """Test that no extra params are added when not configured.""" + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="client_challenge", + scopes=["read"], + # No resource parameter + ) + + redirect_url = await proxy_without_extra_params.authorize(client, params) + query_params = parse_qs(urlparse(redirect_url).query) + + # No audience parameter should be present (not configured) + assert "audience" not in query_params + # No resource parameter should be present (not provided by client) + assert "resource" not in query_params + + async def test_multiple_extra_params(self, jwt_verifier): + """Test multiple extra parameters can be configured and forwarded.""" + proxy = OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="upstream-client", + upstream_client_secret="upstream-secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + extra_authorize_params={ + "audience": "https://api.example.com", + "prompt": "consent", + "max_age": "3600", + }, + ) + + client = OAuthClientInformationFull( + client_id="test-client", + client_secret="test-secret", + redirect_uris=[AnyUrl("http://localhost:12345/callback")], + ) + + # Register client first + await proxy.register_client(client) + + params = AuthorizationParams( + redirect_uri=AnyUrl("http://localhost:12345/callback"), + redirect_uri_provided_explicitly=True, + state="client-state", + code_challenge="client_challenge", + scopes=["read"], + ) + + redirect_url = await proxy.authorize(client, params) + query_params = parse_qs(urlparse(redirect_url).query) + + # Should redirect to consent page + assert "/consent" in redirect_url + assert "txn_id" in query_params + + # All extra parameters configured at proxy level + txn_id = query_params["txn_id"][0] + transaction = await proxy._transaction_store.get(key=txn_id) + assert transaction is not None + # Verify proxy has all extra params configured + assert ( + proxy._extra_authorize_params.get("audience") == "https://api.example.com" + ) + assert proxy._extra_authorize_params.get("prompt") == "consent" + assert proxy._extra_authorize_params.get("max_age") == "3600" + + async def test_token_endpoint_invalid_client_error(self, jwt_verifier): + """Test that invalid client_id returns OAuth 2.1 compliant error response. + + When a client ID is not found during token exchange, the proxy should: + 1. Return HTTP 401 status code + 2. Use 'invalid_client' error code instead of 'unauthorized_client' + + 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 = OAuthDCRProxy( + upstream_authorization_endpoint="https://oauth.example.com/authorize", + upstream_token_endpoint="https://oauth.example.com/token", + upstream_client_id="upstream-client", + upstream_client_secret="upstream-secret", + token_verifier=jwt_verifier, + base_url="https://proxy.example.com", + ) + + # Create a test app with OAuth routes + app = Starlette(routes=proxy.get_routes()) + + # Test the token endpoint with an invalid (non-existent) client_id + with TestClient(app) as client: + response = client.post( + "/token", + data={ + "grant_type": "authorization_code", + "code": "test-auth-code", + "client_id": "non-existent-client-id", + "code_verifier": "test-code-verifier", + "redirect_uri": "http://localhost:12345/callback", + }, + headers={ + "Content-Type": "application/x-www-form-urlencoded", + }, + ) + + # Verify OAuth 2.1 compliant error response + assert response.status_code == 401, ( + f"Expected 401 but got {response.status_code}" + ) + + error_data = response.json() + assert error_data["error"] == "invalid_client", ( + f"Expected 'invalid_client' but got '{error_data.get('error')}'" + ) + assert "Invalid client_id" in error_data["error_description"] + + # Verify proper cache headers are set + assert response.headers.get("Cache-Control") == "no-store" + assert response.headers.get("Pragma") == "no-cache" + + +class TestTokenHandlerErrorTransformation: + """Tests for TokenHandler's OAuth 2.1 compliant error transformation.""" + + def test_transforms_client_auth_failure_to_invalid_client_401(self): + """Test that client authentication failures return invalid_client with 401.""" + from mcp.server.auth.handlers.token import TokenErrorResponse + + from fastmcp.server.auth.oauth_dcr_proxy import TokenHandler + + handler = TokenHandler(provider=Mock(), client_authenticator=Mock()) + + # Simulate error from ClientAuthenticator.authenticate() failure + error_response = TokenErrorResponse( + error="unauthorized_client", + error_description="Invalid client_id 'test-client-id'", + ) + + response = handler.response(error_response) + + # Should transform to OAuth 2.1 compliant response + assert response.status_code == 401 + assert b'"error":"invalid_client"' in response.body + assert ( + b'"error_description":"Invalid client_id \'test-client-id\'"' + in response.body + ) + + 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.oauth_dcr_proxy import TokenHandler + + handler = TokenHandler(provider=Mock(), client_authenticator=Mock()) + + # Simulate error from grant_type not in client_info.grant_types + error_response = TokenErrorResponse( + error="unauthorized_client", + error_description="Client not authorized for this grant type", + ) + + response = handler.response(error_response) + + # Should NOT transform - keep as 400 unauthorized_client + assert response.status_code == 400 + assert b'"error":"unauthorized_client"' in response.body + + def test_does_not_transform_other_errors(self): + """Test that other error types pass through unchanged.""" + from mcp.server.auth.handlers.token import TokenErrorResponse + + from fastmcp.server.auth.oauth_dcr_proxy import TokenHandler + + handler = TokenHandler(provider=Mock(), client_authenticator=Mock()) + + error_response = TokenErrorResponse( + error="invalid_grant", + error_description="Authorization code has expired", + ) + + response = handler.response(error_response) + + # Should pass through unchanged + assert response.status_code == 400 + assert b'"error":"invalid_grant"' in response.body diff --git a/tests/server/auth/test_oauth_proxy_redirect_validation.py b/tests/server/auth/oauth_dcr_proxy/test_oauth_proxy_redirect_validation.py similarity index 97% rename from tests/server/auth/test_oauth_proxy_redirect_validation.py rename to tests/server/auth/oauth_dcr_proxy/test_oauth_proxy_redirect_validation.py index 4ef54db67..f687f8237 100644 --- a/tests/server/auth/test_oauth_proxy_redirect_validation.py +++ b/tests/server/auth/oauth_dcr_proxy/test_oauth_proxy_redirect_validation.py @@ -5,7 +5,7 @@ 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_dcr_proxy import OAuthDCRProxy, ProxyDCRClient class MockTokenVerifier(TokenVerifier): @@ -103,7 +103,7 @@ class TestOAuthProxyRedirectValidation: def test_proxy_default_allows_all(self): """Test that OAuth proxy defaults to allowing all URIs for DCR compatibility.""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://auth.example.com/authorize", upstream_token_endpoint="https://auth.example.com/token", upstream_client_id="test-client", @@ -119,7 +119,7 @@ class TestOAuthProxyRedirectValidation: """Test OAuth proxy with custom redirect patterns.""" custom_patterns = ["http://localhost:*", "https://*.myapp.com/*"] - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://auth.example.com/authorize", upstream_token_endpoint="https://auth.example.com/token", upstream_client_id="test-client", @@ -133,7 +133,7 @@ class TestOAuthProxyRedirectValidation: def test_proxy_empty_list_validation(self): """Test OAuth proxy with empty list (allow none).""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://auth.example.com/authorize", upstream_token_endpoint="https://auth.example.com/token", upstream_client_id="test-client", @@ -149,7 +149,7 @@ class TestOAuthProxyRedirectValidation: """Test that registered clients use the configured patterns.""" custom_patterns = ["https://app.example.com/*"] - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://auth.example.com/authorize", upstream_token_endpoint="https://auth.example.com/token", upstream_client_id="test-client", @@ -181,7 +181,7 @@ class TestOAuthProxyRedirectValidation: """Test that unregistered clients return None.""" custom_patterns = ["http://localhost:*", "http://127.0.0.1:*"] - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://auth.example.com/authorize", upstream_token_endpoint="https://auth.example.com/token", upstream_client_id="test-client", diff --git a/tests/server/auth/test_oauth_proxy_storage.py b/tests/server/auth/oauth_dcr_proxy/test_oauth_proxy_storage.py similarity index 97% rename from tests/server/auth/test_oauth_proxy_storage.py rename to tests/server/auth/oauth_dcr_proxy/test_oauth_proxy_storage.py index 629708427..e2e57ad36 100644 --- a/tests/server/auth/test_oauth_proxy_storage.py +++ b/tests/server/auth/oauth_dcr_proxy/test_oauth_proxy_storage.py @@ -12,7 +12,7 @@ from key_value.aio.stores.memory import MemoryStore from mcp.shared.auth import OAuthClientInformationFull from pydantic import AnyUrl -from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy class TestOAuthProxyStorage: @@ -39,9 +39,9 @@ class TestOAuthProxyStorage: """Create in-memory storage for testing.""" return MemoryStore() - def create_proxy(self, jwt_verifier, storage=None) -> OAuthProxy: + def create_proxy(self, jwt_verifier, storage=None) -> OAuthDCRProxy: """Create an OAuth proxy with specified storage.""" - return OAuthProxy( + return OAuthDCRProxy( upstream_authorization_endpoint="https://github.com/login/oauth/authorize", upstream_token_endpoint="https://github.com/login/oauth/access_token", upstream_client_id="test-client-id", @@ -109,7 +109,7 @@ class TestOAuthProxyStorage: self, jwt_verifier, temp_storage ): """Test that ProxyDCRClient is created with redirect URI patterns.""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://github.com/login/oauth/authorize", upstream_token_endpoint="https://github.com/login/oauth/access_token", upstream_client_id="test-client-id", diff --git a/tests/server/auth/test_oidc_proxy.py b/tests/server/auth/oauth_dcr_proxy/test_oidc_proxy.py similarity index 100% rename from tests/server/auth/test_oidc_proxy.py rename to tests/server/auth/oauth_dcr_proxy/test_oidc_proxy.py diff --git a/tests/server/auth/test_oauth_proxy.py b/tests/server/auth/test_oauth_proxy.py index 59cb0a38a..89b4ac63b 100644 --- a/tests/server/auth/test_oauth_proxy.py +++ b/tests/server/auth/test_oauth_proxy.py @@ -27,7 +27,7 @@ from starlette.routing import Route from fastmcp import FastMCP from fastmcp.server.auth.auth import AccessToken, RefreshToken, TokenVerifier -from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy from fastmcp.server.auth.providers.jwt import JWTVerifier # ============================================================================= @@ -311,7 +311,7 @@ def jwt_verifier(): @pytest.fixture def oauth_proxy(jwt_verifier): """Create a standard OAuthProxy instance for testing.""" - return OAuthProxy( + return OAuthDCRProxy( upstream_authorization_endpoint="https://github.com/login/oauth/authorize", upstream_token_endpoint="https://github.com/login/oauth/access_token", upstream_client_id="test-client-id", @@ -341,7 +341,7 @@ class TestOAuthProxyInitialization: def test_basic_initialization(self, jwt_verifier): """Test basic proxy initialization with required parameters.""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://auth.example.com/authorize", upstream_token_endpoint="https://auth.example.com/token", upstream_client_id="client-123", @@ -361,7 +361,7 @@ class TestOAuthProxyInitialization: def test_all_optional_parameters(self, jwt_verifier): """Test initialization with all optional parameters.""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://auth.example.com/authorize", upstream_token_endpoint="https://auth.example.com/token", upstream_client_id="client-123", @@ -387,7 +387,7 @@ class TestOAuthProxyInitialization: def test_redirect_path_normalization(self, jwt_verifier): """Test that redirect_path is normalized with leading slash.""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://auth.com/authorize", upstream_token_endpoint="https://auth.com/token", upstream_client_id="client", @@ -485,7 +485,7 @@ class TestOAuthProxyPKCE: @pytest.fixture def proxy_with_pkce(self, jwt_verifier): - return OAuthProxy( + return OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -497,7 +497,7 @@ class TestOAuthProxyPKCE: @pytest.fixture def proxy_without_pkce(self, jwt_verifier): - return OAuthProxy( + return OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -583,7 +583,7 @@ class TestOAuthProxyTokenEndpointAuth: def test_token_auth_method_initialization(self, jwt_verifier): """Test different token endpoint auth methods.""" # client_secret_post - proxy_post = OAuthProxy( + proxy_post = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="client", @@ -595,7 +595,7 @@ class TestOAuthProxyTokenEndpointAuth: assert proxy_post._token_endpoint_auth_method == "client_secret_post" # client_secret_basic (default) - proxy_basic = OAuthProxy( + proxy_basic = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="client", @@ -607,7 +607,7 @@ class TestOAuthProxyTokenEndpointAuth: assert proxy_basic._token_endpoint_auth_method == "client_secret_basic" # None (use authlib default) - proxy_default = OAuthProxy( + proxy_default = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="client", @@ -619,7 +619,7 @@ class TestOAuthProxyTokenEndpointAuth: async def test_token_auth_method_passed_to_client(self, jwt_verifier): """Test that auth method is passed to AsyncOAuth2Client.""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="client-id", @@ -637,7 +637,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_dcr_proxy.AsyncOAuth2Client" + ) as MockClient: mock_client = AsyncMock() # Mock initial token exchange (authorization code flow) @@ -665,7 +667,7 @@ 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 + from fastmcp.server.auth.oauth_dcr_proxy import ClientCode client_code = ClientCode( code="test-auth-code", @@ -740,7 +742,7 @@ class TestOAuthProxyE2E: async def test_full_oauth_flow_with_mock_provider(self, mock_oauth_provider): """Test complete OAuth flow with mock provider.""" # Create proxy pointing to mock provider - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, upstream_token_endpoint=mock_oauth_provider.token_endpoint, upstream_client_id="mock-client", @@ -793,7 +795,7 @@ class TestOAuthProxyE2E: async def test_token_refresh_with_mock_provider(self, mock_oauth_provider): """Test token refresh flow with mock provider.""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, upstream_token_endpoint=mock_oauth_provider.token_endpoint, upstream_client_id="mock-client", @@ -818,7 +820,9 @@ class TestOAuthProxyE2E: "scope": "read write", } - with patch("fastmcp.server.auth.oauth_proxy.AsyncOAuth2Client") as MockClient: + with patch( + "fastmcp.server.auth.oauth_dcr_proxy.AsyncOAuth2Client" + ) as MockClient: mock_client = AsyncMock() # Mock initial token exchange to get FastMCP tokens @@ -847,7 +851,7 @@ 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 + from fastmcp.server.auth.oauth_dcr_proxy import ClientCode client_code = ClientCode( code="test-auth-code", @@ -907,7 +911,7 @@ class TestOAuthProxyE2E: """Test PKCE validation with mock provider.""" mock_oauth_provider.require_pkce = True - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, upstream_token_endpoint=mock_oauth_provider.token_endpoint, upstream_client_id="mock-client", @@ -961,7 +965,7 @@ class TestParameterForwarding: @pytest.fixture def proxy_with_extra_params(self, jwt_verifier): """Create OAuthProxy with extra parameters configured.""" - return OAuthProxy( + return OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -975,7 +979,7 @@ class TestParameterForwarding: @pytest.fixture def proxy_without_extra_params(self, jwt_verifier): """Create OAuthProxy without extra parameters.""" - return OAuthProxy( + return OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -1121,7 +1125,7 @@ class TestParameterForwarding: async def test_multiple_extra_params(self, jwt_verifier): """Test multiple extra parameters can be configured and forwarded.""" - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -1182,7 +1186,7 @@ class TestParameterForwarding: from starlette.applications import Starlette from starlette.testclient import TestClient - proxy = OAuthProxy( + proxy = OAuthDCRProxy( upstream_authorization_endpoint="https://oauth.example.com/authorize", upstream_token_endpoint="https://oauth.example.com/token", upstream_client_id="upstream-client", @@ -1233,7 +1237,7 @@ class TestTokenHandlerErrorTransformation: """Test that client authentication failures return invalid_client with 401.""" from mcp.server.auth.handlers.token import TokenErrorResponse - from fastmcp.server.auth.oauth_proxy import TokenHandler + from fastmcp.server.auth.oauth_dcr_proxy import TokenHandler handler = TokenHandler(provider=Mock(), client_authenticator=Mock()) @@ -1257,7 +1261,7 @@ class TestTokenHandlerErrorTransformation: """Test that grant type authorization errors stay as unauthorized_client with 400.""" from mcp.server.auth.handlers.token import TokenErrorResponse - from fastmcp.server.auth.oauth_proxy import TokenHandler + from fastmcp.server.auth.oauth_dcr_proxy import TokenHandler handler = TokenHandler(provider=Mock(), client_authenticator=Mock()) @@ -1277,7 +1281,7 @@ class TestTokenHandlerErrorTransformation: """Test that other error types pass through unchanged.""" from mcp.server.auth.handlers.token import TokenErrorResponse - from fastmcp.server.auth.oauth_proxy import TokenHandler + from fastmcp.server.auth.oauth_dcr_proxy import TokenHandler handler = TokenHandler(provider=Mock(), client_authenticator=Mock())