Add client_secret_basic authentication support #1334
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Filter by extension
Viewed files
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| Expand Up | @@ -91,30 +91,33 @@ def response(self, obj: TokenSuccessResponse | TokenErrorResponse): | |
| ) | ||
|
|
||
| async def handle(self, request: Request): | ||
| try: | ||
| client_info = await self.client_authenticator.authenticate_request(request) | ||
| except AuthenticationError as e: | ||
| # Authentication failures should return 401 | ||
| return PydanticJSONResponse( | ||
| content=TokenErrorResponse( | ||
| error="unauthorized_client", | ||
| error_description=e.message, | ||
| ), | ||
| status_code=401, | ||
| headers={ | ||
| "Cache-Control": "no-store", | ||
| "Pragma": "no-cache", | ||
| }, | ||
| ) | ||
|
|
||
| try: | ||
| form_data = await request.form() | ||
|
Comment thread
Copy link
Copy Markdown
Contributor
There was a problem hiding this comment. Are we not reading the request.form() twice here? (once in authenticate_request, and here again? (Think starlette might complain about this) Might wanna push the form data (maybe other request fields, e.g. auth header) to the authenticator method?
Sorry, something went wrong.
Copy link
Copy Markdown
Contributor
Author
There was a problem hiding this comment. My understanding is that the current version of starlette caches calls to form(), json(), and body(), so it is safe to call them multiple times. For example, https://github.com/Kludex/starlette/blob/main/starlette/requests.py#L254-L287 . But I am not a starlette expert and I might be misunderstanding. There seems to be an edge case My first implementation parsed out the form in the request handler and passed it into the authenticate method, as you suggest, but I thought it felt clunky to duplicate the code that parsed "form data and auth header" across both the token and revoke endpoints. It’s not the end of the world, but it turns # Current version
async def handle(self, request: Request):
try:
client_info = await self.client_authenticator.authenticate_request(request)
into something like this # Parse out form and auth_header
async def handle(self, request: Request):
try:
form_data = await request.form()
except Exception:
return self.response(
TokenErrorResponse(
error="invalid_request",
error_description="Unable to parse request body",
)
)
auth_header = request.headers.get("Authorization")
try:
client_info = await self.client_authenticator.authenticate_request(form_data, auth_header)
Or I suppose we could handle invalid for data in the authenticate method: # Parse out form and auth_header, handle form error in client_authenticator
async def handle(self, request: Request):
form_data = None
try:
form_data = await request.form()
except Exception:
pass
auth_header = request.headers.get("Authorization")
try:
client_info = await self.client_authenticator.authenticate_request(form_data, auth_header)
Anyway, I defer to the maintainers. If you would like me to switch to one of the above implementations, I would be happy to do so.
Sorry, something went wrong. |
||
| token_request = TokenRequest.model_validate(dict(form_data)).root | ||
| except ValidationError as validation_error: # pragma: no cover | ||
| return self.response( | ||
| TokenErrorResponse( | ||
| error="invalid_request", | ||
| error_description=stringify_pydantic_error(validation_error), | ||
| ) | ||
| ) | ||
|
|
||
| if token_request.grant_type not in client_info.grant_types: # pragma: no cover | ||
| return self.response( | ||
| TokenErrorResponse( | ||
| Expand Down | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,11 @@ | ||
| import base64 | ||
| import binascii | ||
| import hmac | ||
| import time | ||
| from typing import Any | ||
| from urllib.parse import unquote | ||
|
|
||
| from starlette.requests import Request | ||
|
|
||
| from mcp.server.auth.provider import OAuthAuthorizationServerProvider | ||
| from mcp.shared.auth import OAuthClientInformationFull | ||
| Expand Down Expand Up | @@ -30,19 +36,77 @@ def __init__(self, provider: OAuthAuthorizationServerProvider[Any, Any, Any]): | |
| """ | ||
| self.provider = provider | ||
|
|
||
| async def authenticate_request(self, request: Request) -> OAuthClientInformationFull: | ||
| """ | ||
| Authenticate a client from an HTTP request. | ||
|
|
||
| Extracts client credentials from the appropriate location based on the | ||
| client's registered authentication method and validates them. | ||
|
|
||
| Args: | ||
| request: The HTTP request containing client credentials | ||
|
|
||
| Returns: | ||
| The authenticated client information | ||
|
|
||
| Raises: | ||
| AuthenticationError: If authentication fails | ||
| """ | ||
| form_data = await request.form() | ||
| client_id = form_data.get("client_id") | ||
| if not client_id: | ||
| raise AuthenticationError("Missing client_id") | ||
|
|
||
| client = await self.provider.get_client(str(client_id)) | ||
| if not client: | ||
| raise AuthenticationError("Invalid client_id") # pragma: no cover | ||
|
|
||
| request_client_secret: str | None = None | ||
| auth_header = request.headers.get("Authorization", "") | ||
|
|
||
| if client.token_endpoint_auth_method == "client_secret_basic": | ||
| if not auth_header.startswith("Basic "): | ||
| raise AuthenticationError("Missing or invalid Basic authentication in Authorization header") | ||
|
|
||
| try: | ||
| encoded_credentials = auth_header[6:] # Remove "Basic " prefix | ||
| decoded = base64.b64decode(encoded_credentials).decode("utf-8") | ||
| if ":" not in decoded: | ||
| raise ValueError("Invalid Basic auth format") | ||
| basic_client_id, request_client_secret = decoded.split(":", 1) | ||
|
Comment thread
Copy link
Copy Markdown
Contributor
There was a problem hiding this comment. We should probably urldecode both parts, as per RFC 6749 Section 2.3.1
Sorry, something went wrong.
Copy link
Copy Markdown
Contributor
Author
There was a problem hiding this comment. Good catch. Thank you.
Sorry, something went wrong. |
||
|
|
||
| # URL-decode both parts per RFC 6749 Section 2.3.1 | ||
| basic_client_id = unquote(basic_client_id) | ||
| request_client_secret = unquote(request_client_secret) | ||
|
|
||
| if basic_client_id != client_id: | ||
| raise AuthenticationError("Client ID mismatch in Basic auth") | ||
| except (ValueError, UnicodeDecodeError, binascii.Error): | ||
| raise AuthenticationError("Invalid Basic authentication header") | ||
|
|
||
| elif client.token_endpoint_auth_method == "client_secret_post": | ||
| raw_form_data = form_data.get("client_secret") | ||
| # form_data.get() can return a UploadFile or None, so we need to check if it's a string | ||
| if isinstance(raw_form_data, str): | ||
| request_client_secret = str(raw_form_data) | ||
|
|
||
| elif client.token_endpoint_auth_method == "none": | ||
| request_client_secret = None | ||
| else: | ||
| raise AuthenticationError( # pragma: no cover | ||
| f"Unsupported auth method: {client.token_endpoint_auth_method}" | ||
| ) | ||
|
|
||
| # If client from the store expects a secret, validate that the request provides | ||
| # that secret | ||
| if client.client_secret: # pragma: no branch | ||
| if not request_client_secret: | ||
| raise AuthenticationError("Client secret is required") # pragma: no cover | ||
|
|
||
| # hmac.compare_digest requires that both arguments are either bytes or a `str` containing | ||
| # only ASCII characters. Since we do not control `request_client_secret`, we encode both | ||
| # arguments to bytes. | ||
| if not hmac.compare_digest(client.client_secret.encode(), request_client_secret.encode()): | ||
| raise AuthenticationError("Invalid client_secret") # pragma: no cover | ||
|
|
||
| if client.client_secret_expires_at and client.client_secret_expires_at < int(time.time()): | ||
| Expand Down | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityI am torn on whether or not we should allow auto-selecting "none". It seems possibly like bad security to allow that, but I suppose if the server allows it then it is ok?
I suppose ideally we should allow the user to pick a list of auth methods they want to allow to be auto-configured, but I am not sure anyone cares enough to want to use it.
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.