Validate scopes + provide default

Author: praboud-ant

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

@@ -11,6 +11,7 @@
 from mcp.server.auth.errors import stringify_pydantic_error
 from mcp.server.auth.json_response import PydanticJSONResponse
 from mcp.server.auth.provider import OAuthRegisteredClientsStore
+from mcp.server.auth.settings import ClientRegistrationOptions
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
 
 
@@ -33,14 +34,16 @@ class RegistrationErrorResponse(BaseModel):
 @dataclass
 class RegistrationHandler:
     clients_store: OAuthRegisteredClientsStore
-    client_secret_expiry_seconds: int | None
+    options: ClientRegistrationOptions
 
     async def handle(self, request: Request) -> Response:
         # Implements dynamic client registration as defined in https://datatracker.ietf.org/doc/html/rfc7591#section-3.1
         try:
             # Parse request body as JSON
             body = await request.json()
             client_metadata = OAuthClientMetadata.model_validate(body)
+
+            # Scope validation is handled below
         except ValidationError as validation_error:
             return PydanticJSONResponse(
                 content=RegistrationErrorResponse(
@@ -56,10 +59,27 @@ async def handle(self, request: Request) -> Response:
             # cryptographically secure random 32-byte hex string
             client_secret = secrets.token_hex(32)
 
+        if client_metadata.scope is None and self.options.default_scopes is not None:
+            client_metadata.scope = " ".join(self.options.default_scopes)
+        elif (
+            client_metadata.scope is not None and self.options.valid_scopes is not None
+        ):
+            requested_scopes = set(client_metadata.scope.split())
+            valid_scopes = set(self.options.valid_scopes)
+            if not requested_scopes.issubset(valid_scopes):
+                return PydanticJSONResponse(
+                    content=RegistrationErrorResponse(
+                        error="invalid_client_metadata",
+                        error_description="Requested scopes are not valid: "
+                        f"{', '.join(requested_scopes - valid_scopes)}",
+                    ),
+                    status_code=400,
+                )
+
         client_id_issued_at = int(time.time())
         client_secret_expires_at = (
-            client_id_issued_at + self.client_secret_expiry_seconds
-            if self.client_secret_expiry_seconds is not None
+            client_id_issued_at + self.options.client_secret_expiry_seconds
+            if self.options.client_secret_expiry_seconds is not None
             else None
         )
 

src/mcp/server/auth/routes.py

@@ -1,6 +1,6 @@
 from typing import Callable
 
-from pydantic import AnyHttpUrl, BaseModel, Field
+from pydantic import AnyHttpUrl
 from starlette.routing import Route
 
 from mcp.server.auth.handlers.authorize import AuthorizationHandler
@@ -10,30 +10,10 @@
 from mcp.server.auth.handlers.token import TokenHandler
 from mcp.server.auth.middleware.client_auth import ClientAuthenticator
 from mcp.server.auth.provider import OAuthServerProvider
+from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
 from mcp.shared.auth import OAuthMetadata
 
 
-class ClientRegistrationOptions(BaseModel):
-    enabled: bool = False
-    client_secret_expiry_seconds: int | None = None
-
-
-class RevocationOptions(BaseModel):
-    enabled: bool = False
-
-
-class AuthSettings(BaseModel):
-    issuer_url: AnyHttpUrl = Field(
-        ...,
-        description="URL advertised as OAuth issuer; this should be the URL the server "
-        "is reachable at",
-    )
-    service_documentation_url: AnyHttpUrl | None = None
-    client_registration_options: ClientRegistrationOptions | None = None
-    revocation_options: RevocationOptions | None = None
-    required_scopes: list[str] | None = None
-
-
 def validate_issuer_url(url: AnyHttpUrl):
     """
     Validate that the issuer URL meets OAuth 2.0 requirements.
@@ -109,7 +89,7 @@ def create_auth_routes(
     if client_registration_options.enabled:
         registration_handler = RegistrationHandler(
             provider.clients_store,
-            client_secret_expiry_seconds=client_registration_options.client_secret_expiry_seconds,
+            options=client_registration_options,
         )
         routes.append(
             Route(

src/mcp/server/auth/settings.py

@@ -0,0 +1,24 @@
+from pydantic import AnyHttpUrl, BaseModel, Field
+
+
+class ClientRegistrationOptions(BaseModel):
+    enabled: bool = False
+    client_secret_expiry_seconds: int | None = None
+    valid_scopes: list[str] | None = None
+    default_scopes: list[str] | None = None
+
+
+class RevocationOptions(BaseModel):
+    enabled: bool = False
+
+
+class AuthSettings(BaseModel):
+    issuer_url: AnyHttpUrl = Field(
+        ...,
+        description="URL advertised as OAuth issuer; this should be the URL the server "
+        "is reachable at",
+    )
+    service_documentation_url: AnyHttpUrl | None = None
+    client_registration_options: ClientRegistrationOptions | None = None
+    revocation_options: RevocationOptions | None = None
+    required_scopes: list[str] | None = None

src/mcp/server/fastmcp/server.py

@@ -30,7 +30,7 @@
     RequireAuthMiddleware,
 )
 from mcp.server.auth.provider import OAuthServerProvider
-from mcp.server.auth.routes import (
+from mcp.server.auth.settings import (
     AuthSettings,
 )
 from mcp.server.fastmcp.exceptions import ResourceError

tests/server/fastmcp/auth/test_auth_integration.py

@@ -27,11 +27,11 @@
     construct_redirect_uri,
 )
 from mcp.server.auth.routes import (
-    AuthSettings,
     ClientRegistrationOptions,
     RevocationOptions,
     create_auth_routes,
 )
+from mcp.server.auth.settings import AuthSettings
 from mcp.server.fastmcp import FastMCP
 from mcp.shared.auth import (
     OAuthClientInformationFull,
@@ -226,7 +226,11 @@ def auth_app(mock_oauth_provider):
         mock_oauth_provider,
         AnyHttpUrl("https://auth.example.com"),
         AnyHttpUrl("https://docs.example.com"),
-        client_registration_options=ClientRegistrationOptions(enabled=True),
+        client_registration_options=ClientRegistrationOptions(
+            enabled=True,
+            valid_scopes=["read", "write", "profile"],
+            default_scopes=["read", "write"],
+        ),
         revocation_options=RevocationOptions(enabled=True),
     )
 
@@ -946,6 +950,57 @@ async def test_revoke_with_malformed_token(self, test_client, registered_client)
         assert error_response["error"] == "invalid_request"
         assert "token_type_hint" in error_response["error_description"]
 
+    @pytest.mark.anyio
+    async def test_client_registration_disallowed_scopes(
+        self, test_client: httpx.AsyncClient
+    ):
+        """Test client registration with scopes that are not allowed."""
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Test Client",
+            "scope": "read write profile admin",  # 'admin' is not in valid_scopes
+        }
+
+        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 "scope" in error_data["error_description"]
+        assert "admin" in error_data["error_description"]
+
+    @pytest.mark.anyio
+    async def test_client_registration_default_scopes(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider
+    ):
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Test Client",
+            # No scope specified
+        }
+
+        response = await test_client.post(
+            "/register",
+            json=client_metadata,
+        )
+        assert response.status_code == 201
+        client_info = response.json()
+
+        # Verify client was registered successfully
+        assert client_info["scope"] == "read write"
+
+        # Retrieve the client from the store to verify default scopes
+        registered_client = await mock_oauth_provider.clients_store.get_client(
+            client_info["client_id"]
+        )
+        assert registered_client is not None
+
+        # Check that default scopes were applied
+        assert registered_client.scope == "read write"
+
 
 class TestFastMCPWithAuth:
     """Test FastMCP server with authentication."""