fix(auth): allow public CIMD clients to use the jwt-bearer grant

docs/client/identity-assertion.md

@@ -85,7 +85,7 @@ The SDK can also *be* the authorization server: `create_auth_routes` returns the
 ```
 
 * `identity_assertion_enabled=True` gates everything. Off, which is the default, `/token` answers this grant with `unsupported_grant_type` even if you implemented the hook, and the metadata does not mention it. On, the metadata gains the `jwt-bearer` grant type and lists `urn:ietf:params:oauth:grant-profile:id-jag` in `authorization_grant_profiles_supported`, the field the extension uses to advertise support. (This SDK's client never reads it: it is provisioned for one issuer and simply asks.)
-* **`exchange_identity_assertion`** is the hook. Before it runs, the SDK has authenticated the client, refused public clients, and refused clients whose registration does not list the grant. You get an `IdentityAssertionParams` (the raw `assertion`, the requested `scopes` and `resource`) and return a plain `OAuthToken`.
+* **`exchange_identity_assertion`** is the hook. Before it runs, the SDK has authenticated the client, refused public clients (unless `is_metadata_document_client` reports the client was resolved from a Client ID Metadata Document and the ID-JAG's `client_id` claim names it), and refused clients whose registration does not list the grant. You get an `IdentityAssertionParams` (the raw `assertion`, the requested `scopes` and `resource`) and return a plain `OAuthToken`.
 * Dynamic client registration refuses this grant unconditionally, so `get_client` here serves a hand-provisioned client. An ID-JAG client cannot register itself into existence.
 * Half the class is refusals. `OAuthAuthorizationServerProvider` is the *whole* authorization server, so it also asks for the authorization-code flow; a server that signs users in as well implements those for real, and this one has exactly one door.
 

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

@@ -4,6 +4,7 @@
 from dataclasses import dataclass
 from typing import Annotated, Any, Literal
 
+import jwt
 from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, TypeAdapter, ValidationError
 from starlette.requests import Request
 
@@ -74,6 +75,13 @@ class TokenErrorResponse(BaseModel):
     error_uri: AnyHttpUrl | None = None
 
 
+def _unverified_client_id(assertion: str) -> object:
+    try:
+        return jwt.decode(assertion, options={"verify_signature": False}).get("client_id")
+    except jwt.PyJWTError:
+        return None
+
+
 # this is just an alias over OAuthToken; the only reason we do this
 # is to have some separation between the HTTP response type, and the
 # type returned by the provider
@@ -251,16 +259,28 @@ async def handle(self, request: Request):
                 # SEP-990 §5.1: only confidential clients may present an ID-JAG. ClientAuthenticator
                 # already rejects a secret-based method with no stored secret; this additionally
                 # rejects the public `none` method so an unauthenticated client never reaches the
-                # provider hook.
+                # provider hook. ext-auth §5 exempts a client identified by its Client ID Metadata
+                # Document, which cannot hold a secret.
                 if not client_info.client_secret:
-                    # RFC 6749 §5.2: the client authenticated but is not permitted this grant, so
-                    # unauthorized_client (not invalid_client, which is for failed authentication).
-                    return self.response(
-                        TokenErrorResponse(
-                            error="unauthorized_client",
-                            error_description="The JWT bearer grant requires a confidential client",
+                    if not await self.provider.is_metadata_document_client(client_info):
+                        # RFC 6749 §5.2: the client authenticated but is not permitted this grant, so
+                        # unauthorized_client (not invalid_client, which is for failed authentication).
+                        return self.response(
+                            TokenErrorResponse(
+                                error="unauthorized_client",
+                                error_description="The JWT bearer grant requires a confidential client",
+                            )
+                        )
+                    # With no client authentication, the ID-JAG must at least name the requesting
+                    # client. Reading the claim unverified is safe: a mismatch only rejects, and the
+                    # provider verifies the signature before issuing anything.
+                    if _unverified_client_id(token_request.assertion) != client_info.client_id:
+                        return self.response(
+                            TokenErrorResponse(
+                                error="invalid_grant",
+                                error_description="The assertion was not issued to this client",
+                            )
                         )
-                    )
 
                 params = IdentityAssertionParams(
                     assertion=token_request.assertion,

src/mcp/server/auth/provider.py

@@ -305,6 +305,22 @@ async def revoke_token(
             token: The token to revoke.
         """
 
+    async def is_metadata_document_client(self, client: OAuthClientInformationFull) -> bool:
+        """Reports whether ``client`` was resolved from a Client ID Metadata Document (CIMD).
+
+        The enterprise-managed authorization extension lets a client that is not pre-registered
+        use its CIMD URL as ``client_id`` and present an ID-JAG without client authentication.
+        Return True for such clients to let them use the jwt-bearer grant with the ``none`` auth
+        method. The default returns False, keeping the grant limited to confidential clients.
+
+        Args:
+            client: The client presenting the assertion.
+
+        Returns:
+            True if ``client`` was resolved from a Client ID Metadata Document.
+        """
+        return False
+
     async def exchange_identity_assertion(
         self,
         client: OAuthClientInformationFull,
@@ -324,19 +340,21 @@ async def exchange_identity_assertion(
         - require ``aud`` to identify this authorization server (its own issuer);
         - require a ``sub`` (RFC 7523 §3 makes it mandatory) identifying the end user;
         - reject replays - enforce ``exp``, and track ``jti`` for the assertion's lifetime;
-        - require the ID-JAG's ``client_id`` claim to match the authenticated ``client`` - do
-          NOT derive authorization from ``client.client_id`` alone, which for a confidential
-          client is authenticated but for any client is ultimately self-asserted in the request;
+        - require the ID-JAG's ``client_id`` claim to match ``client`` - do NOT derive
+          authorization from ``client.client_id`` alone, which for a confidential client is
+          authenticated but for any client is ultimately self-asserted in the request;
         - audience-restrict the issued access token to the resource named in the ID-JAG's
           ``resource`` claim, not merely ``params.resource`` (which the client controls);
         - derive the granted scopes from the ID-JAG and policy rather than granting
           ``params.scopes`` verbatim.
 
-        The handler guarantees ``client`` is confidential (it rejects clients without a stored
-        secret before calling this hook), but the ID-JAG remains the authoritative grant.
+        ``client`` is either an authenticated confidential client or, when
+        ``is_metadata_document_client`` returns True, a public client identified by its Client ID
+        Metadata Document. For the latter the handler has only checked the unverified ``client_id``
+        claim against ``client.client_id``; either way the ID-JAG remains the authoritative grant.
 
         Args:
-            client: The authenticated client presenting the assertion.
+            client: The client presenting the assertion.
             params: The validated jwt-bearer request parameters (the ID-JAG and indicators).
 
         Returns:

src/mcp/server/auth/routes.py

@@ -162,9 +162,12 @@ def build_metadata(
     # SEP-990 / ext-auth §6: support for the ID-JAG flow is advertised as a grant PROFILE, not as
     # the jwt-bearer grant type (which an AS might support for other purposes).
     authorization_grant_profiles_supported: list[str] | None = None
+    token_endpoint_auth_methods_supported = ["client_secret_post", "client_secret_basic"]
     if supports_identity_assertion:
         grant_types_supported.append(JWT_BEARER_GRANT_TYPE)
         authorization_grant_profiles_supported = [ID_JAG_GRANT_PROFILE]
+        # ext-auth §5: a Client ID Metadata Document client may present an ID-JAG unauthenticated.
+        token_endpoint_auth_methods_supported.append("none")
 
     # Create metadata
     metadata = OAuthMetadata(
@@ -175,7 +178,7 @@ def build_metadata(
         response_types_supported=["code"],
         response_modes_supported=None,
         grant_types_supported=grant_types_supported,
-        token_endpoint_auth_methods_supported=["client_secret_post", "client_secret_basic"],
+        token_endpoint_auth_methods_supported=token_endpoint_auth_methods_supported,
         token_endpoint_auth_signing_alg_values_supported=None,
         service_documentation=service_documentation_url,
         ui_locales_supported=None,

tests/server/auth/test_identity_assertion.py

@@ -4,6 +4,7 @@
 import time
 
 import httpx2
+import jwt
 import pytest
 from httpx2 import ASGITransport
 from pydantic import AnyHttpUrl
@@ -26,6 +27,12 @@
 VALID_ASSERTION = "valid-id-jag"
 CONFIDENTIAL_CLIENT_ID = "enterprise-client"
 CONFIDENTIAL_CLIENT_SECRET = "enterprise-secret"
+CIMD_CLIENT_ID = "https://client.example.com/client.json"
+IDP_KEY = "idp-signing-key-for-tests-only-0123456789"
+
+
+def id_jag(client_id: str) -> str:
+    return jwt.encode({"client_id": client_id, "sub": "assertion-user"}, IDP_KEY, algorithm="HS256")
 
 
 class IdentityAssertionProvider(OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken]):
@@ -35,6 +42,9 @@ def __init__(self) -> None:
         self.clients: dict[str, OAuthClientInformationFull] = {}
         self.tokens: dict[str, AccessToken] = {}
         self.last_params: IdentityAssertionParams | None = None
+        self.last_client: OAuthClientInformationFull | None = None
+        self.metadata_document_client_ids: set[str] = set()
+        self.valid_assertions = {VALID_ASSERTION}
 
     async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
         return self.clients.get(client_id)
@@ -70,12 +80,16 @@ async def load_access_token(self, token: str) -> AccessToken | None:
     async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
         raise NotImplementedError
 
+    async def is_metadata_document_client(self, client: OAuthClientInformationFull) -> bool:
+        return client.client_id in self.metadata_document_client_ids
+
     async def exchange_identity_assertion(
         self, client: OAuthClientInformationFull, params: IdentityAssertionParams
     ) -> OAuthToken:
         self.last_params = params
+        self.last_client = client
         # Stand-in for RFC 7523 §3 / SEP-990 §5.1 assertion validation.
-        if params.assertion != VALID_ASSERTION:
+        if params.assertion not in self.valid_assertions:
             raise TokenError(error="invalid_grant", error_description="assertion is not valid")
         assert client.client_id is not None
         scopes = params.scopes or ["mcp"]
@@ -146,8 +160,8 @@ def test_build_metadata_advertises_id_jag_profile_when_enabled():
     )
     assert JWT_BEARER_GRANT_TYPE in (enabled.grant_types_supported or [])
     assert enabled.authorization_grant_profiles_supported == [ID_JAG_GRANT_PROFILE]
-    # The grant is confidential-only, so the `none` auth method is NOT advertised.
-    assert "none" not in (enabled.token_endpoint_auth_methods_supported or [])
+    # A Client ID Metadata Document client may present an ID-JAG with the `none` auth method.
+    assert enabled.token_endpoint_auth_methods_supported == ["client_secret_post", "client_secret_basic", "none"]
 
     disabled = build_metadata(
         AnyHttpUrl("https://auth.example.com"),
@@ -157,6 +171,7 @@ def test_build_metadata_advertises_id_jag_profile_when_enabled():
     )
     assert JWT_BEARER_GRANT_TYPE not in (disabled.grant_types_supported or [])
     assert disabled.authorization_grant_profiles_supported is None
+    assert disabled.token_endpoint_auth_methods_supported == ["client_secret_post", "client_secret_basic"]
 
 
 @pytest.mark.anyio
@@ -219,7 +234,9 @@ async def test_identity_assertion_rejected_when_disabled(provider: IdentityAsser
 async def test_identity_assertion_rejects_public_client(
     client: httpx2.AsyncClient, provider: IdentityAssertionProvider
 ):
-    """A public (auth method 'none') client cannot use the grant, even if it presents a valid assertion."""
+    """A public client not resolved from a metadata document cannot use the grant, even with an ID-JAG naming it."""
+    assertion = id_jag("public-client")
+    provider.valid_assertions.add(assertion)
     provider.clients["public-client"] = OAuthClientInformationFull(
         client_id="public-client",
         redirect_uris=None,
@@ -230,7 +247,7 @@ async def test_identity_assertion_rejects_public_client(
 
     response = await client.post(
         "/token",
-        data={"grant_type": JWT_BEARER_GRANT_TYPE, "client_id": "public-client", "assertion": VALID_ASSERTION},
+        data={"grant_type": JWT_BEARER_GRANT_TYPE, "client_id": "public-client", "assertion": assertion},
     )
 
     assert response.status_code == 400
@@ -397,3 +414,134 @@ async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
     with pytest.raises(TokenError) as excinfo:
         await bare.exchange_identity_assertion(client_info, params)
     assert excinfo.value.error == "unsupported_grant_type"
+    assert await bare.is_metadata_document_client(client_info) is False
+
+
+def register_cimd_client(provider: IdentityAssertionProvider, grant_types: list[str]) -> OAuthClientInformationFull:
+    client_info = OAuthClientInformationFull(
+        client_id=CIMD_CLIENT_ID,
+        redirect_uris=None,
+        grant_types=grant_types,
+        token_endpoint_auth_method="none",
+        scope="mcp",
+    )
+    provider.clients[CIMD_CLIENT_ID] = client_info
+    provider.metadata_document_client_ids.add(CIMD_CLIENT_ID)
+    return client_info
+
+
+@pytest.mark.anyio
+async def test_metadata_document_client_exchanges_id_jag_without_client_authentication(
+    client: httpx2.AsyncClient, provider: IdentityAssertionProvider
+):
+    """A public CIMD client may present an ID-JAG naming it with no client authentication (ext-auth §5)."""
+    cimd_client = register_cimd_client(provider, [JWT_BEARER_GRANT_TYPE])
+    assertion = id_jag(CIMD_CLIENT_ID)
+    provider.valid_assertions.add(assertion)
+
+    response = await client.post(
+        "/token",
+        data={
+            "grant_type": JWT_BEARER_GRANT_TYPE,
+            "client_id": CIMD_CLIENT_ID,
+            "assertion": assertion,
+            "resource": "https://mcp.example.com/mcp",
+        },
+    )
+
+    assert response.status_code == 200, response.content
+    assert provider.last_client is cimd_client
+    assert provider.last_params is not None
+    assert provider.last_params.assertion == assertion
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "assertion",
+    [
+        pytest.param(id_jag("https://other.example.com/client.json"), id="names-another-client"),
+        pytest.param(jwt.encode({"sub": "assertion-user"}, IDP_KEY, algorithm="HS256"), id="no-client-id-claim"),
+        pytest.param("not-a-jwt", id="malformed"),
+    ],
+)
+async def test_metadata_document_client_rejects_id_jag_not_issued_to_it(
+    client: httpx2.AsyncClient, provider: IdentityAssertionProvider, assertion: str
+):
+    """Without client authentication, an ID-JAG whose `client_id` claim is not the requester is invalid_grant."""
+    register_cimd_client(provider, [JWT_BEARER_GRANT_TYPE])
+    provider.valid_assertions.add(assertion)
+
+    response = await client.post(
+        "/token",
+        data={"grant_type": JWT_BEARER_GRANT_TYPE, "client_id": CIMD_CLIENT_ID, "assertion": assertion},
+    )
+
+    assert response.status_code == 400
+    assert response.json() == {
+        "error": "invalid_grant",
+        "error_description": "The assertion was not issued to this client",
+    }
+    assert provider.last_params is None
+
+
+@pytest.mark.anyio
+async def test_metadata_document_client_without_the_grant_is_rejected(
+    client: httpx2.AsyncClient, provider: IdentityAssertionProvider
+):
+    """A CIMD client whose metadata does not list the jwt-bearer grant is refused it."""
+    register_cimd_client(provider, ["authorization_code"])
+
+    response = await client.post(
+        "/token",
+        data={"grant_type": JWT_BEARER_GRANT_TYPE, "client_id": CIMD_CLIENT_ID, "assertion": id_jag(CIMD_CLIENT_ID)},
+    )
+
+    assert response.status_code == 400
+    assert response.json()["error"] == "unsupported_grant_type"
+    assert provider.last_params is None
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("secret", [pytest.param("wrong-secret", id="wrong"), pytest.param(None, id="missing")])
+async def test_confidential_client_with_bad_secret_is_invalid_client(
+    client: httpx2.AsyncClient, provider: IdentityAssertionProvider, secret: str | None
+):
+    """A confidential client still has to authenticate; the metadata-document path does not apply to it."""
+    form = assertion_form()
+    if secret is None:
+        del form["client_secret"]
+    else:
+        form["client_secret"] = secret
+
+    response = await client.post("/token", data=form)
+
+    assert response.status_code == 401
+    assert response.json()["error"] == "invalid_client"
+    assert provider.last_params is None
+
+
+@pytest.mark.anyio
+async def test_metadata_document_client_rejected_when_identity_assertion_disabled(provider: IdentityAssertionProvider):
+    """With the grant disabled, a CIMD client is refused like every other client."""
+    register_cimd_client(provider, [JWT_BEARER_GRANT_TYPE])
+    routes = create_auth_routes(
+        provider,
+        issuer_url=AnyHttpUrl("https://auth.example.com"),
+        client_registration_options=ClientRegistrationOptions(enabled=True, valid_scopes=["mcp"]),
+        revocation_options=RevocationOptions(enabled=False),
+        identity_assertion_enabled=False,
+    )
+    transport = ASGITransport(app=Starlette(routes=routes))
+    async with httpx2.AsyncClient(transport=transport, base_url="https://auth.example.com") as http:
+        response = await http.post(
+            "/token",
+            data={
+                "grant_type": JWT_BEARER_GRANT_TYPE,
+                "client_id": CIMD_CLIENT_ID,
+                "assertion": id_jag(CIMD_CLIENT_ID),
+            },
+        )
+
+    assert response.status_code == 400
+    assert response.json()["error"] == "unsupported_grant_type"
+    assert provider.last_params is None