Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions src/mcp/server/auth/handlers/revoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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:
Expand Down
31 changes: 27 additions & 4 deletions src/mcp/server/auth/handlers/token.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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}'"),
)
)

Expand Down
73 changes: 73 additions & 0 deletions tests/server/auth/test_error_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"


Expand Down
2 changes: 1 addition & 1 deletion tests/server/auth/test_identity_assertion.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
Loading