diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py index 680cbfd022..62fcd57855 100644 --- a/src/mcp/client/auth/oauth2.py +++ b/src/mcp/client/auth/oauth2.py @@ -155,7 +155,7 @@ def get_resource_url(self) -> str: # If PRM provides a resource that's a valid parent, use it if self.protected_resource_metadata and self.protected_resource_metadata.resource: - prm_resource = str(self.protected_resource_metadata.resource) + prm_resource = self.protected_resource_metadata.resource_str if check_resource_allowed(requested_resource=resource, configured_resource=prm_resource): resource = prm_resource @@ -280,7 +280,7 @@ def __init__( async def _validate_resource_match(self, prm: ProtectedResourceMetadata) -> None: """Validate that PRM resource matches the server URL per RFC 8707.""" - prm_resource = str(prm.resource) if prm.resource else None + prm_resource = prm.resource_str if prm.resource else None if not prm_resource: return # pragma: no cover default_resource = resource_url_from_server_url(self.context.server_url) diff --git a/src/mcp/shared/auth.py b/src/mcp/shared/auth.py index 59cf2f5723..1fcf21a724 100644 --- a/src/mcp/shared/auth.py +++ b/src/mcp/shared/auth.py @@ -1,6 +1,23 @@ -from typing import Any, Literal - -from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, field_validator +from typing import Any, Literal, cast + +from pydantic import ( + AnyHttpUrl, + AnyUrl, + BaseModel, + ConfigDict, + Field, + PrivateAttr, + ValidatorFunctionWrapHandler, + field_validator, + model_validator, +) + +# `url_preserve_empty_path` (pydantic >= 2.12) keeps a path-less URL from gaining a trailing slash when +# parsed from the wire, so issuer/resource identifiers round-trip as transmitted (RFC 3986 ยง6.2.1 simple +# string comparison). Older pydantic ignores unknown config keys at runtime; the cast keeps the 2.11 type +# stubs (which do not know the key) quiet. See `ProtectedResourceMetadata.resource_str` for the +# version-independent path used for the RFC 8707 `resource` parameter. +_PRESERVE_EMPTY_PATH = cast(ConfigDict, {"url_preserve_empty_path": True}) class OAuthToken(BaseModel): @@ -41,6 +58,8 @@ class OAuthClientMetadata(BaseModel): for the full specification. """ + model_config = _PRESERVE_EMPTY_PATH + redirect_uris: list[AnyUrl] | None = Field(..., min_length=1) # supported auth methods for the token endpoint token_endpoint_auth_method: ( @@ -132,6 +151,8 @@ class OAuthMetadata(BaseModel): See https://datatracker.ietf.org/doc/html/rfc8414#section-2 """ + model_config = _PRESERVE_EMPTY_PATH + issuer: AnyHttpUrl authorization_endpoint: AnyHttpUrl token_endpoint: AnyHttpUrl @@ -162,8 +183,31 @@ class ProtectedResourceMetadata(BaseModel): See https://datatracker.ietf.org/doc/html/rfc9728#section-2 """ + model_config = _PRESERVE_EMPTY_PATH + resource: AnyHttpUrl authorization_servers: list[AnyHttpUrl] = Field(..., min_length=1) + # The `resource` value exactly as received. `url_preserve_empty_path` only takes effect on + # pydantic >= 2.12; on older pydantic a path-less URL still renders with a trailing slash, which + # breaks the byte-exact RFC 8707 `resource` parameter. Kept alongside the parsed URL so callers + # can echo the server's identifier verbatim regardless of pydantic version. + _resource_raw: str | None = PrivateAttr(default=None) + + @model_validator(mode="wrap") + @classmethod + def _capture_raw_resource(cls, data: Any, handler: ValidatorFunctionWrapHandler) -> "ProtectedResourceMetadata": + raw: Any = cast(dict[str, Any], data).get("resource") if isinstance(data, dict) else None + model = cast("ProtectedResourceMetadata", handler(data)) + if isinstance(raw, str): + model._resource_raw = raw + return model + + @property + def resource_str(self) -> str: + """The resource identifier as a string, exactly as the server published it when parsed from + JSON/dict input (RFC 8707 requires clients to send it byte-for-byte); otherwise the rendered URL.""" + return self._resource_raw if self._resource_raw is not None else str(self.resource) + jwks_uri: AnyHttpUrl | None = None scopes_supported: list[str] | None = None bearer_methods_supported: list[str] | None = Field(default=["header"]) # MCP only supports header method diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py index 7c3e5bd60c..b9e35031a7 100644 --- a/tests/client/test_auth.py +++ b/tests/client/test_auth.py @@ -522,10 +522,11 @@ async def test_handle_metadata_response_success(self, oauth_provider: OAuthClien }""" response = httpx.Response(200, content=content) - # Should set metadata + # Should set metadata. On pydantic >= 2.12 the empty path is preserved (no trailing slash); + # older pydantic ignores `url_preserve_empty_path` and still normalises to ".../". await oauth_provider._handle_oauth_metadata_response(response) assert oauth_provider.context.oauth_metadata is not None - assert str(oauth_provider.context.oauth_metadata.issuer) == "https://auth.example.com/" + assert str(oauth_provider.context.oauth_metadata.issuer).rstrip("/") == "https://auth.example.com" @pytest.mark.anyio async def test_prioritize_www_auth_scope_over_prm( @@ -2212,6 +2213,28 @@ async def test_get_resource_url_falls_back_when_prm_mismatches( assert provider.context.get_resource_url() == "https://api.example.com/v1/mcp" +@pytest.mark.anyio +async def test_get_resource_url_echoes_pathless_prm_resource_verbatim( + client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage +) -> None: + """RFC 8707: the `resource` parameter is the PRM `resource` byte-for-byte. A path-less identifier + parsed from the wire must not gain a trailing slash, on any supported pydantic version.""" + provider = OAuthClientProvider( + server_url="https://api.example.com/mcp", + client_metadata=client_metadata, + storage=mock_storage, + ) + provider._initialized = True + + prm = ProtectedResourceMetadata.model_validate_json( + '{"resource": "https://api.example.com", "authorization_servers": ["https://auth.example.com"]}' + ) + assert prm.resource_str == "https://api.example.com" + provider.context.protected_resource_metadata = prm + + assert provider.context.get_resource_url() == "https://api.example.com" + + def _prepare_full_flow(provider: OAuthClientProvider, client_info: OAuthClientInformationFull | None) -> list[str]: """Reset `provider` for a full flow with `client_info` as the stored registration, and wire a redirect/callback pair that echoes the `state` of the last authorization URL it was sent to.