diff --git a/src/late/mcp/auth.py b/src/late/mcp/auth.py index 9753a92f..e355d03c 100644 --- a/src/late/mcp/auth.py +++ b/src/late/mcp/auth.py @@ -10,8 +10,12 @@ import httpx from fastmcp.server.auth import AccessToken, RemoteAuthProvider, TokenVerifier +from mcp.server.auth.routes import build_resource_metadata_url, cors_middleware +from mcp.shared.auth import ProtectedResourceMetadata from starlette.authentication import AuthenticationError from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.routing import Route from late.mcp.constants import ( DOCS_URL, @@ -240,6 +244,52 @@ def _grant(self, token: str) -> AccessToken: return AccessToken(token=token, client_id="zernio", scopes=list(OAUTH_SCOPES)) +class VerbatimIssuerRemoteAuthProvider(RemoteAuthProvider): + """RemoteAuthProvider whose metadata names the authorization server verbatim. + + The SDK model types `authorization_servers` as AnyHttpUrl, and pydantic + renders a bare origin with a trailing slash ("https://zernio.com/"). MCP + SDK 2.x clients (Hermes Agent 0.21+) compare that string byte for byte with + the `issuer` in zernio.com/.well-known/oauth-authorization-server + ("https://zernio.com", SEP-2468) and abort the whole OAuth flow on mismatch. + Upstream fixed the model in mcp 2.0 (url_preserve_empty_path), which + fastmcp 3.x cannot resolve to, so the document is rendered here instead. + """ + + def get_routes(self, mcp_path: str | None = None) -> list[Route]: + self.set_mcp_path(mcp_path) + resource_url = self._get_resource_url(mcp_path) + if not resource_url: + return [] + metadata = ProtectedResourceMetadata( + resource=resource_url, + authorization_servers=self.authorization_servers, + scopes_supported=( + self._scopes_supported + if self._scopes_supported is not None + else self.token_verifier.scopes_supported + ), + resource_name=self.resource_name, + resource_documentation=self.resource_documentation, + ) + payload = metadata.model_dump(mode="json", exclude_none=True) + payload["authorization_servers"] = [str(s) for s in self.authorization_servers] + + async def handle(_request: Request) -> JSONResponse: + return JSONResponse( + payload, headers={"Cache-Control": "public, max-age=3600"} + ) + + path = urlparse(str(build_resource_metadata_url(resource_url))).path + return [ + Route( + path, + endpoint=cors_middleware(handle, ["GET", "OPTIONS"]), + methods=["GET", "OPTIONS"], + ) + ] + + def build_auth_provider(scopes: list[str] | None = None) -> RemoteAuthProvider: """Build the FastMCP resource-server auth provider. @@ -253,7 +303,7 @@ def build_auth_provider(scopes: list[str] | None = None) -> RemoteAuthProvider: request every advertised scope, so a surface that only publishes posts must not advertise ads or messaging scopes on its consent screen. """ - return RemoteAuthProvider( + return VerbatimIssuerRemoteAuthProvider( token_verifier=ZernioTokenVerifier(), authorization_servers=[OAUTH_AUTHORIZATION_SERVER], base_url=MCP_PUBLIC_URL, diff --git a/tests/test_mcp_protected_resource_metadata.py b/tests/test_mcp_protected_resource_metadata.py new file mode 100644 index 00000000..2c775a85 --- /dev/null +++ b/tests/test_mcp_protected_resource_metadata.py @@ -0,0 +1,39 @@ +"""Protected-resource metadata (RFC 9728) served by the MCP HTTP server.""" + +import httpx +from starlette.applications import Starlette +from starlette.testclient import TestClient + +from late.mcp.auth import build_auth_provider +from late.mcp.constants import OAUTH_AUTHORIZATION_SERVER + + +def _metadata(scopes: list[str] | None = None) -> dict: + app = Starlette(routes=build_auth_provider(scopes).get_routes("/mcp")) + response = TestClient(app).get("/.well-known/oauth-protected-resource/mcp") + assert response.status_code == 200 + return response.json() + + +def test_authorization_server_is_the_issuer_string_byte_for_byte() -> None: + # MCP SDK 2.x clients (Hermes Agent 0.21+) reject "https://zernio.com/" against + # the issuer "https://zernio.com" with "Authorization server metadata issuer mismatch". + assert _metadata()["authorization_servers"] == [OAUTH_AUTHORIZATION_SERVER] + assert not OAUTH_AUTHORIZATION_SERVER.endswith("/") + + +def test_metadata_keeps_the_resource_scopes_and_narrowing() -> None: + body = _metadata() + assert body["resource"] == "https://mcp.zernio.com/mcp" + assert body["bearer_methods_supported"] == ["header"] + assert "posts:write" in body["scopes_supported"] + assert body["resource_name"] + assert _metadata(["posts:write"])["scopes_supported"] == ["posts:write"] + + +def test_authorization_server_matches_the_live_issuer() -> None: + issuer = httpx.get( + f"{OAUTH_AUTHORIZATION_SERVER}/.well-known/oauth-authorization-server", + timeout=10, + ).json()["issuer"] + assert _metadata()["authorization_servers"] == [issuer]