Skip to content
Merged
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
52 changes: 51 additions & 1 deletion src/late/mcp/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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.

Expand All @@ -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,
Expand Down
39 changes: 39 additions & 0 deletions tests/test_mcp_protected_resource_metadata.py
Original file line number Diff line number Diff line change
@@ -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]
Loading