wip

Author: praboud-ant

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
     "sse-starlette>=1.6.1",
     "pydantic-settings>=2.5.2",
     "uvicorn>=0.23.1",
+    "fastapi",
 ]
 
 [project.optional-dependencies]
@@ -46,7 +47,7 @@ dev-dependencies = [
     "pytest>=8.3.4",
     "ruff>=0.8.5",
     "trio>=0.26.2",
-    "pytest-flakefinder>=1.1.0",
+    "pytest-flakefinder==1.1.0",
     "pytest-xdist>=3.6.1",
 ]
 

src/mcp/server/auth/__init__.py

@@ -0,0 +1,3 @@
+"""
+MCP OAuth server authorization components.
+"""

src/mcp/server/auth/errors.py

@@ -0,0 +1,135 @@
+"""
+OAuth error classes for MCP authorization.
+
+Corresponds to TypeScript file: src/server/auth/errors.ts
+"""
+
+from typing import Dict, Optional, Any
+
+
+class OAuthError(Exception):
+    """
+    Base class for all OAuth errors.
+    
+    Corresponds to OAuthError in src/server/auth/errors.ts
+    """
+    error_code: str = "server_error"
+    
+    def __init__(self, message: str):
+        super().__init__(message)
+        self.message = message
+        
+    def to_response_object(self) -> Dict[str, str]:
+        """Convert error to JSON response object."""
+        return {
+            "error": self.error_code,
+            "error_description": self.message
+        }
+
+
+class ServerError(OAuthError):
+    """
+    Server error.
+    
+    Corresponds to ServerError in src/server/auth/errors.ts
+    """
+    error_code = "server_error"
+
+
+class InvalidRequestError(OAuthError):
+    """
+    Invalid request error.
+    
+    Corresponds to InvalidRequestError in src/server/auth/errors.ts
+    """
+    error_code = "invalid_request"
+
+
+class InvalidClientError(OAuthError):
+    """
+    Invalid client error.
+    
+    Corresponds to InvalidClientError in src/server/auth/errors.ts
+    """
+    error_code = "invalid_client"
+
+
+class InvalidGrantError(OAuthError):
+    """
+    Invalid grant error.
+    
+    Corresponds to InvalidGrantError in src/server/auth/errors.ts
+    """
+    error_code = "invalid_grant"
+
+
+class UnauthorizedClientError(OAuthError):
+    """
+    Unauthorized client error.
+    
+    Corresponds to UnauthorizedClientError in src/server/auth/errors.ts
+    """
+    error_code = "unauthorized_client"
+
+
+class UnsupportedGrantTypeError(OAuthError):
+    """
+    Unsupported grant type error.
+    
+    Corresponds to UnsupportedGrantTypeError in src/server/auth/errors.ts
+    """
+    error_code = "unsupported_grant_type"
+
+
+class UnsupportedResponseTypeError(OAuthError):
+    """
+    Unsupported response type error.
+    
+    Corresponds to UnsupportedResponseTypeError in src/server/auth/errors.ts
+    """
+    error_code = "unsupported_response_type"
+
+
+class InvalidScopeError(OAuthError):
+    """
+    Invalid scope error.
+    
+    Corresponds to InvalidScopeError in src/server/auth/errors.ts
+    """
+    error_code = "invalid_scope"
+
+
+class AccessDeniedError(OAuthError):
+    """
+    Access denied error.
+    
+    Corresponds to AccessDeniedError in src/server/auth/errors.ts
+    """
+    error_code = "access_denied"
+
+
+class TemporarilyUnavailableError(OAuthError):
+    """
+    Temporarily unavailable error.
+    
+    Corresponds to TemporarilyUnavailableError in src/server/auth/errors.ts
+    """
+    error_code = "temporarily_unavailable"
+
+
+class InvalidTokenError(OAuthError):
+    """
+    Invalid token error.
+    
+    Corresponds to InvalidTokenError in src/server/auth/errors.ts
+    """
+    error_code = "invalid_token"
+
+
+class InsufficientScopeError(OAuthError):
+    """
+    Insufficient scope error.
+    
+    Corresponds to InsufficientScopeError in src/server/auth/errors.ts
+    """
+    error_code = "insufficient_scope"

src/mcp/server/auth/handlers/__init__.py

@@ -0,0 +1,3 @@
+"""
+Request handlers for MCP authorization endpoints.
+"""

src/mcp/server/auth/handlers/authorize.py

@@ -0,0 +1,150 @@
+"""
+Handler for OAuth 2.0 Authorization endpoint.
+
+Corresponds to TypeScript file: src/server/auth/handlers/authorize.ts
+"""
+
+import re
+from urllib.parse import urlparse, urlunparse, urlencode
+from typing import Any, Callable, Dict, List, Literal, Optional
+from urllib.parse import urlencode, parse_qs
+
+from fastapi import Request, Response
+from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, ValidationError
+from pydantic_core import Url
+from starlette.responses import JSONResponse, RedirectResponse
+
+from mcp.server.auth.errors import (
+    InvalidClientError, 
+    InvalidRequestError,
+    UnsupportedResponseTypeError,
+    ServerError,
+    OAuthError,
+)
+from mcp.server.auth.provider import AuthorizationParams, OAuthServerProvider
+from mcp.shared.auth import OAuthClientInformationFull
+
+
+class AuthorizationRequest(BaseModel):
+    """
+    Model for the authorization request parameters.
+    
+    Corresponds to request schema in authorizationHandler in src/server/auth/handlers/authorize.ts
+    """
+    client_id: str = Field(..., description="The client ID")
+    redirect_uri: AnyHttpUrl | None = Field(..., description="URL to redirect to after authorization")
+
+    response_type: Literal["code"] = Field(..., description="Must be 'code' for authorization code flow")
+    code_challenge: str = Field(..., description="PKCE code challenge")
+    code_challenge_method: Literal["S256"] = Field("S256", description="PKCE code challenge method")
+    state: Optional[str] = Field(None, description="Optional state parameter")
+    scope: Optional[str] = Field(None, description="Optional scope parameter")
+    
+    class Config:
+        extra = "ignore"
+
+def validate_scope(requested_scope: str | None, client: OAuthClientInformationFull) -> list[str] | None:
+    if requested_scope is None:
+        return None
+    requested_scopes = requested_scope.split(" ")
+    allowed_scopes = [] if client.scope is None else client.scope.split(" ")
+    for scope in requested_scopes:
+        if scope not in allowed_scopes:
+            raise InvalidRequestError(f"Client was not registered with scope {scope}")
+    return requested_scopes
+
+def validate_redirect_uri(auth_request: AuthorizationRequest, client: OAuthClientInformationFull) -> AnyHttpUrl:
+    if auth_request.redirect_uri is not None:
+        # Validate redirect_uri against client's registered redirect URIs
+        if auth_request.redirect_uri not in client.redirect_uris:
+            raise InvalidRequestError(
+                f"Redirect URI '{auth_request.redirect_uri}' not registered for client"
+            )
+        return auth_request.redirect_uri
+    elif len(client.redirect_uris) == 1:
+        return client.redirect_uris[0]
+    else:
+        raise InvalidRequestError("redirect_uri must be specified when client has multiple registered URIs")
+
+def create_authorization_handler(provider: OAuthServerProvider) -> Callable:
+    """
+    Create a handler for the OAuth 2.0 Authorization endpoint.
+    
+    Corresponds to authorizationHandler in src/server/auth/handlers/authorize.ts
+
+    """
+    
+    async def authorization_handler(request: Request) -> Response:
+        """
+        Handler for the OAuth 2.0 Authorization endpoint.
+        """
+        # Validate request parameters
+        try:
+            if request.method == "GET":
+                auth_request = AuthorizationRequest.model_validate(request.query_params)
+            else:
+                auth_request = AuthorizationRequest.model_validate_json(await request.body())
+        except ValidationError as e:
+            raise InvalidRequestError(str(e))
+        
+        # Get client information
+        try:
+            client = await provider.clients_store.get_client(auth_request.client_id)
+        except OAuthError as e:
+            # TODO: proper error rendering
+            raise InvalidClientError(str(e))
+        
+        if not client:
+            raise InvalidClientError(f"Client ID '{auth_request.client_id}' not found")
+        
+ 
+        # do validation which is dependent on the client configuration
+        redirect_uri = validate_redirect_uri(auth_request, client)
+        scopes = validate_scope(auth_request.scope, client)
+        
+        auth_params = AuthorizationParams(
+            state=auth_request.state,
+            scopes=scopes,
+            code_challenge=auth_request.code_challenge,
+            redirect_uri=redirect_uri,
+        )
+        
+        response = RedirectResponse(url="", status_code=302, headers={"Cache-Control": "no-store"})
+            
+        try:
+            # Let the provider handle the authorization flow
+            await provider.authorize(client, auth_params, response)
+            
+            return response
+        except Exception as e:
+            return RedirectResponse(
+                url=create_error_redirect(redirect_uri, e, auth_request.state),
+                status_code=302,
+                headers={"Cache-Control": "no-store"},
+                )
+    
+    return authorization_handler
+
+def create_error_redirect(redirect_uri: AnyUrl, error: Exception, state: Optional[str]) -> str:
+    parsed_uri = urlparse(str(redirect_uri))
+    if isinstance(error, OAuthError):
+        query_params = {
+            "error": error.error_code,
+            "error_description": str(error)
+        }
+    else:
+        query_params = {
+            "error": "internal_error",
+            "error_description": "An unknown error occurred"
+        }
+    # TODO: should we add error_uri?
+    # if error.error_uri:
+    #     query_params["error_uri"] = str(error.error_uri)
+    if state:
+        query_params["state"] = state
+    
+    new_query = urlencode(query_params)
+    if parsed_uri.query:
+        new_query = f"{parsed_uri.query}&{new_query}"
+    
+    return urlunparse(parsed_uri._replace(query=new_query))

src/mcp/server/auth/handlers/metadata.py

@@ -0,0 +1,43 @@
+"""
+Handler for OAuth 2.0 Authorization Server Metadata.
+
+Corresponds to TypeScript file: src/server/auth/handlers/metadata.ts
+"""
+
+from typing import Any, Callable, Dict, Optional
+from fastapi import Request, Response
+from starlette.responses import JSONResponse
+
+
+def create_metadata_handler(metadata: Dict[str, Any]) -> Callable:
+    """
+    Create a handler for OAuth 2.0 Authorization Server Metadata.
+    
+    Corresponds to metadataHandler in src/server/auth/handlers/metadata.ts
+    
+    Args:
+        metadata: The metadata to return in the response
+        
+    Returns:
+        A FastAPI route handler function
+    """
+    
+    async def metadata_handler(request: Request) -> Response:
+        """
+        Handler for the OAuth 2.0 Authorization Server Metadata endpoint.
+        
+        Args:
+            request: The FastAPI request
+            
+        Returns:
+            JSON response with the authorization server metadata
+        """
+        # Remove any None values from metadata
+        clean_metadata = {k: v for k, v in metadata.items() if v is not None}
+        
+        return JSONResponse(
+            content=clean_metadata,
+            headers={"Cache-Control": "public, max-age=3600"}  # Cache for 1 hour
+        )
+    
+    return metadata_handler