diff --git a/src/mcp/server/auth/handlers/revoke.py b/src/mcp/server/auth/handlers/revoke.py index 4efd154001..3b3ada14c2 100644 --- a/src/mcp/server/auth/handlers/revoke.py +++ b/src/mcp/server/auth/handlers/revoke.py @@ -24,7 +24,7 @@ class RevocationRequest(BaseModel): class RevocationErrorResponse(BaseModel): - error: Literal["invalid_request", "unauthorized_client"] + error: Literal["invalid_request", "invalid_client", "unauthorized_client"] error_description: str | None = None @@ -37,13 +37,17 @@ async def handle(self, request: Request) -> Response: """Handler for the OAuth 2.0 Token Revocation endpoint.""" try: client = await self.client_authenticator.authenticate_request(request) - except AuthenticationError as e: # pragma: no cover + except AuthenticationError as e: return PydanticJSONResponse( status_code=401, content=RevocationErrorResponse( - error="unauthorized_client", + error="invalid_client", error_description=e.message, ), + headers={ + "Cache-Control": "no-store", + "Pragma": "no-cache", + }, ) try: diff --git a/src/mcp/server/auth/handlers/token.py b/src/mcp/server/auth/handlers/token.py index 00df115573..653480bc95 100644 --- a/src/mcp/server/auth/handlers/token.py +++ b/src/mcp/server/auth/handlers/token.py @@ -117,10 +117,31 @@ async def handle(self, request: Request): }, ) + form_data = await request.form() + form_dict = dict(form_data) + + # RFC 6749 §5.2: If the authorization grant type is not supported by the + # authorization server, return unsupported_grant_type. + supported_grant_types: set[str] = {"authorization_code", "refresh_token"} + if self.identity_assertion_enabled: + supported_grant_types.add("urn:ietf:params:oauth:grant-type:jwt-bearer") + + grant_type = form_dict.get("grant_type") + if grant_type and grant_type not in supported_grant_types: + if grant_type == "urn:ietf:params:oauth:grant-type:jwt-bearer": + desc = "The JWT bearer grant is not supported by this authorization server" + else: + desc = f"Unsupported grant type: {grant_type}" + return self.response( + TokenErrorResponse( + error="unsupported_grant_type", + error_description=desc, + ) + ) + try: - form_data = await request.form() # TODO(Marcelo): Can someone check if this `dict()` wrapper is necessary? - token_request = token_request_adapter.validate_python(dict(form_data)) + token_request = token_request_adapter.validate_python(form_dict) except ValidationError as validation_error: return self.response( TokenErrorResponse( @@ -130,10 +151,12 @@ async def handle(self, request: Request): ) if token_request.grant_type not in client_info.grant_types: + # RFC 6749 §5.2: unauthorized_client: The authenticated client is not + # authorized to use this authorization grant type. return self.response( TokenErrorResponse( - error="unsupported_grant_type", - error_description=(f"Unsupported grant type (supported grant types are {client_info.grant_types})"), + error="unauthorized_client", + error_description=(f"Client is not authorized to use grant type '{token_request.grant_type}'"), ) ) diff --git a/tests/server/auth/test_error_handling.py b/tests/server/auth/test_error_handling.py index f13f23ea33..6a8f78394a 100644 --- a/tests/server/auth/test_error_handling.py +++ b/tests/server/auth/test_error_handling.py @@ -291,6 +291,79 @@ async def test_token_error_handling_refresh_token( assert data["error_description"] == "The requested scope is invalid" +@pytest.mark.anyio +async def test_token_unsupported_grant_type_returns_unsupported_grant_type( + client: httpx2.AsyncClient, + registered_client: dict[str, Any], +): + """RFC 6749 §5.2: Server must return unsupported_grant_type when the grant type is not supported.""" + client_id = registered_client["client_id"] + client_secret = registered_client["client_secret"] + + response = await client.post( + "/token", + data={ + "grant_type": "client_credentials", + "client_id": client_id, + "client_secret": client_secret, + }, + ) + assert response.status_code == 400 + data = response.json() + assert data["error"] == "unsupported_grant_type" + assert "client_credentials" in data["error_description"] + + +@pytest.mark.anyio +async def test_token_unauthorized_client_grant_type_returns_unauthorized_client( + client: httpx2.AsyncClient, + registered_client: dict[str, Any], + oauth_provider: MockOAuthProvider, +): + """RFC 6749 §5.2: Server must return unauthorized_client when the client is not authorized for grant type.""" + client_id = registered_client["client_id"] + client_secret = registered_client["client_secret"] + + # Modify the registered client to only allow 'authorization_code' + stored_client = oauth_provider.clients[client_id] + oauth_provider.clients[client_id] = stored_client.model_copy(update={"grant_types": ["authorization_code"]}) + + response = await client.post( + "/token", + data={ + "grant_type": "refresh_token", + "refresh_token": "dummy_refresh_token", + "client_id": client_id, + "client_secret": client_secret, + }, + ) + assert response.status_code == 400 + data = response.json() + assert data["error"] == "unauthorized_client" + + +@pytest.mark.anyio +async def test_revoke_invalid_client_authentication_returns_401( + client: httpx2.AsyncClient, + registered_client: dict[str, Any], +): + """RFC 7009 §2.2.1 & RFC 6749 §5.2: Revocation endpoint returns 401 invalid_client on auth failure.""" + client_id = registered_client["client_id"] + + # Wrong client_secret + response = await client.post( + "/revoke", + data={ + "token": "some_token", + "client_id": client_id, + "client_secret": "wrong_secret", + }, + ) + assert response.status_code == 401 + data = response.json() + assert data["error"] == "invalid_client" + + _FORM = "application/x-www-form-urlencoded" diff --git a/tests/server/auth/test_identity_assertion.py b/tests/server/auth/test_identity_assertion.py index 9ce12cb8fb..00e196a9dd 100644 --- a/tests/server/auth/test_identity_assertion.py +++ b/tests/server/auth/test_identity_assertion.py @@ -309,7 +309,7 @@ async def test_client_without_the_grant_registered_is_rejected( ) assert response.status_code == 400 - assert response.json()["error"] == "unsupported_grant_type" + assert response.json()["error"] == "unauthorized_client" assert provider.last_params is None