Improve validation for registration

Author: praboud-ant

src/mcp/server/auth/errors.py

@@ -6,6 +6,8 @@
 
 from typing import Dict
 
+from pydantic import ValidationError
+
 
 class OAuthError(Exception):
     """
@@ -143,3 +145,6 @@ class InsufficientScopeError(OAuthError):
     """
 
     error_code = "insufficient_scope"
+
+def stringify_pydantic_error(validation_error: ValidationError) -> str:
+    return "\n".join(f"{'.'.join(str(loc) for loc in e['loc'])}: {e['msg']}" for e in validation_error.errors())

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

@@ -6,103 +6,83 @@
 
 import secrets
 import time
-from typing import Callable
+from typing import Callable, Literal
 from uuid import uuid4
 
-from pydantic import ValidationError
+from pydantic import BaseModel, ValidationError
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 
 from mcp.server.auth.errors import (
     InvalidRequestError,
     OAuthError,
     ServerError,
+    stringify_pydantic_error,
 )
 from mcp.server.auth.json_response import PydanticJSONResponse
 from mcp.server.auth.provider import OAuthRegisteredClientsStore
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
 
+class ErrorResponse(BaseModel):
+    error: Literal["invalid_redirect_uri", "invalid_client_metadata", "invalid_software_statement", "unapproved_software_statement"]
+    error_description: str
+
 
 def create_registration_handler(
     clients_store: OAuthRegisteredClientsStore, client_secret_expiry_seconds: int | None
 ) -> Callable:
-    """
-    Create a handler for OAuth 2.0 Dynamic Client Registration.
-
-    Corresponds to clientRegistrationHandler in src/server/auth/handlers/register.ts
-
-    Args:
-        clients_store: The store for registered clients
-        client_secret_expiry_seconds: Optional expiry time for client secrets
-
-    Returns:
-        A Starlette endpoint handler function
-    """
-
     async def registration_handler(request: Request) -> Response:
-        """
-        Handler for the OAuth 2.0 Dynamic Client Registration endpoint.
-
-        Args:
-            request: The Starlette request
-
-        Returns:
-            JSON response with client information or error
-        """
+        # Implements dynamic client registration as defined in https://datatracker.ietf.org/doc/html/rfc7591#section-3.1
         try:
             # Parse request body as JSON
-            try:
-                body = await request.json()
-                client_metadata = OAuthClientMetadata.model_validate(body)
-            except ValidationError as e:
-                raise InvalidRequestError(f"Invalid client metadata: {str(e)}")
-
-            client_id = str(uuid4())
-            client_secret = None
-            if client_metadata.token_endpoint_auth_method != "none":
-                # cryptographically secure random 32-byte hex string
-                client_secret = secrets.token_hex(32)
-
-            client_id_issued_at = int(time.time())
-            client_secret_expires_at = (
-                client_id_issued_at + client_secret_expiry_seconds
-                if client_secret_expiry_seconds is not None
-                else None
-            )
-
-            client_info = OAuthClientInformationFull(
-                client_id=client_id,
-                client_id_issued_at=client_id_issued_at,
-                client_secret=client_secret,
-                client_secret_expires_at=client_secret_expires_at,
-                # passthrough information from the client request
-                redirect_uris=client_metadata.redirect_uris,
-                token_endpoint_auth_method=client_metadata.token_endpoint_auth_method,
-                grant_types=client_metadata.grant_types,
-                response_types=client_metadata.response_types,
-                client_name=client_metadata.client_name,
-                client_uri=client_metadata.client_uri,
-                logo_uri=client_metadata.logo_uri,
-                scope=client_metadata.scope,
-                contacts=client_metadata.contacts,
-                tos_uri=client_metadata.tos_uri,
-                policy_uri=client_metadata.policy_uri,
-                jwks_uri=client_metadata.jwks_uri,
-                jwks=client_metadata.jwks,
-                software_id=client_metadata.software_id,
-                software_version=client_metadata.software_version,
-            )
-            # Register client
-            client = await clients_store.register_client(client_info)
-            if not client:
-                raise ServerError("Failed to register client")
-
-            # Return client information
-            return PydanticJSONResponse(content=client, status_code=201)
-
-        except OAuthError as e:
-            # Handle OAuth errors
-            status_code = 500 if isinstance(e, ServerError) else 400
-            return JSONResponse(status_code=status_code, content=e.to_response_object())
+            body = await request.json()
+            client_metadata = OAuthClientMetadata.model_validate(body)
+        except ValidationError as validation_error:
+            return PydanticJSONResponse(content=ErrorResponse(
+                error="invalid_client_metadata",
+                error_description=stringify_pydantic_error(validation_error)
+            ), status_code=400)
+            raise InvalidRequestError(f"Invalid client metadata: {str(e)}")
+
+        client_id = str(uuid4())
+        client_secret = None
+        if client_metadata.token_endpoint_auth_method != "none":
+            # cryptographically secure random 32-byte hex string
+            client_secret = secrets.token_hex(32)
+
+        client_id_issued_at = int(time.time())
+        client_secret_expires_at = (
+            client_id_issued_at + client_secret_expiry_seconds
+            if client_secret_expiry_seconds is not None
+            else None
+        )
+
+        client_info = OAuthClientInformationFull(
+            client_id=client_id,
+            client_id_issued_at=client_id_issued_at,
+            client_secret=client_secret,
+            client_secret_expires_at=client_secret_expires_at,
+            # passthrough information from the client request
+            redirect_uris=client_metadata.redirect_uris,
+            token_endpoint_auth_method=client_metadata.token_endpoint_auth_method,
+            grant_types=client_metadata.grant_types,
+            response_types=client_metadata.response_types,
+            client_name=client_metadata.client_name,
+            client_uri=client_metadata.client_uri,
+            logo_uri=client_metadata.logo_uri,
+            scope=client_metadata.scope,
+            contacts=client_metadata.contacts,
+            tos_uri=client_metadata.tos_uri,
+            policy_uri=client_metadata.policy_uri,
+            jwks_uri=client_metadata.jwks_uri,
+            jwks=client_metadata.jwks,
+            software_id=client_metadata.software_id,
+            software_version=client_metadata.software_version,
+        )
+        # Register client
+        client = await clients_store.register_client(client_info)
+
+        # Return client information
+        return PydanticJSONResponse(content=client, status_code=201)
 
     return registration_handler

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

@@ -14,6 +14,7 @@
 
 from mcp.server.auth.errors import (
     InvalidRequestError,
+    stringify_pydantic_error,
 )
 from mcp.server.auth.json_response import PydanticJSONResponse
 from mcp.server.auth.middleware.client_auth import (
@@ -74,7 +75,7 @@ async def token_handler(request: Request):
         except ValidationError as validation_error:
             return response(TokenErrorResponse(
                 error="invalid_request",
-                error_description="\n".join(e['msg'] for e in validation_error.errors())
+                error_description=stringify_pydantic_error(validation_error)
 
             ))
         client_info = await client_authenticator(token_request)

src/mcp/server/auth/provider.py

@@ -72,15 +72,12 @@ async def get_client(self, client_id: str) -> Optional[OAuthClientInformationFul
 
     async def register_client(
         self, client_info: OAuthClientInformationFull
-    ) -> Optional[OAuthClientInformationFull]:
+    ) -> None:
         """
-        Registers a new client and returns client information.
+        Registers a new client
 
         Args:
-            metadata: The client metadata to register.
-
-        Returns:
-            The client information, or None if registration failed.
+            client_info: The client metadata to register.
         """
         ...
 

tests/server/fastmcp/auth/test_auth_integration.py

@@ -681,6 +681,68 @@ async def test_client_registration(
         #     client_info["client_id"]
         # ) is not None
 
+    @pytest.mark.anyio
+    async def test_client_registration_missing_required_fields(
+        self, test_client: httpx.AsyncClient
+    ):
+        """Test client registration with missing required fields."""
+        # Missing redirect_uris which is a required field
+        client_metadata = {
+            "client_name": "Test Client",
+            "client_uri": "https://client.example.com",
+        }
+
+        response = await test_client.post(
+            "/register",
+            json=client_metadata,
+        )
+        assert response.status_code == 400
+        error_data = response.json()
+        assert "error" in error_data
+        assert error_data["error"] == "invalid_client_metadata"
+        assert error_data["error_description"] == "redirect_uris: Field required"
+        
+    @pytest.mark.anyio
+    async def test_client_registration_invalid_uri(
+        self, test_client: httpx.AsyncClient
+    ):
+        """Test client registration with invalid URIs."""
+        # Invalid redirect_uri format
+        client_metadata = {
+            "redirect_uris": ["not-a-valid-uri"],
+            "client_name": "Test Client",
+        }
+
+        response = await test_client.post(
+            "/register",
+            json=client_metadata,
+        )
+        assert response.status_code == 400
+        error_data = response.json()
+        assert "error" in error_data
+        assert error_data["error"] == "invalid_client_metadata"
+        assert error_data["error_description"] == "redirect_uris.0: Input should be a valid URL, relative URL without a base"
+        
+    @pytest.mark.anyio
+    async def test_client_registration_empty_redirect_uris(
+        self, test_client: httpx.AsyncClient
+    ):
+        """Test client registration with empty redirect_uris array."""
+        client_metadata = {
+            "redirect_uris": [],  # Empty array
+            "client_name": "Test Client",
+        }
+
+        response = await test_client.post(
+            "/register",
+            json=client_metadata,
+        )
+        assert response.status_code == 400
+        error_data = response.json()
+        assert "error" in error_data
+        assert error_data["error"] == "invalid_client_metadata"
+        assert error_data["error_description"] == "redirect_uris: List should have at least 1 item after validation, not 0"
+        
     @pytest.mark.anyio
     async def test_authorize_form_post(
         self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider