1"""OAuth 2.1 resource-server support for MCP endpoints (RFC 9728).
  2
  3`plain.mcp` stays agnostic about who issues tokens — implement
  4`authenticate_token` to validate the bearer against your authorization server
  5(for example `plain.oauthserver.validate_access_token`). This is the
  6server side of the flow an MCP client like Claude's custom connector runs:
  7an unauthenticated request gets a 401 whose `WWW-Authenticate` header points
  8at the protected-resource metadata, which names the authorization server.
  9"""
 10
 11from __future__ import annotations
 12
 13from dataclasses import dataclass, field
 14from typing import Any
 15from urllib.parse import urlparse, urlunparse
 16
 17from plain.http import JsonResponse, Request, Response
 18from plain.views import View
 19
 20from .exceptions import MCPUnauthorized
 21
 22_WELL_KNOWN_PRM = "/.well-known/oauth-protected-resource"
 23
 24
 25def _canonical_resource(url: str) -> str:
 26    """Canonical audience identifier: lowercase scheme + host, no trailing slash.
 27
 28    Keeps the advertised `resource` and the validated audience byte-identical
 29    regardless of trailing-slash config or client-side normalization (RFC 8707).
 30    """
 31    parsed = urlparse(url)
 32    return urlunparse(
 33        parsed._replace(
 34            scheme=parsed.scheme.lower(),
 35            netloc=parsed.netloc.lower(),
 36            path=parsed.path.rstrip("/"),
 37        )
 38    )
 39
 40
 41@dataclass
 42class TokenInfo:
 43    """Who a validated bearer token belongs to, and what it may do."""
 44
 45    user: Any
 46    scopes: frozenset[str] = field(default_factory=frozenset)
 47
 48
 49def _bearer_token(request: Request) -> str:
 50    header = request.headers.get("Authorization", "")
 51    scheme, _, value = header.partition(" ")
 52    # RFC 7235: the auth-scheme is matched case-insensitively.
 53    return value.strip() if scheme.lower() == "bearer" else ""
 54
 55
 56class OAuthResourceServer:
 57    """Mixin for an `MCPView` that authenticates requests with OAuth bearer tokens.
 58
 59    Subclass alongside `MCPView` and implement `authenticate_token`:
 60
 61        class AppMCP(OAuthResourceServer, MCPView):
 62            name = "myapp"
 63            tools = [...]
 64
 65            def authenticate_token(self, token):
 66                at = validate_access_token(token, resource=self.oauth_resource)
 67                return TokenInfo(at.user, at.scopes) if at else None
 68
 69    Name the authorization server on the companion `MCPProtectedResourceView`,
 70    not here. On success `self.user` and `self.scopes` are set for tools to read
 71    via `self.mcp`. On failure the request gets a 401 with an RFC 9728
 72    `WWW-Authenticate` challenge pointing at the protected-resource metadata.
 73    """
 74
 75    user: Any = None
 76    scopes: frozenset[str] = frozenset()
 77
 78    # Provided by the MCPView this mixin is combined with.
 79    request: Request
 80
 81    def authenticate_token(self, token: str) -> TokenInfo | None:
 82        raise NotImplementedError
 83
 84    def before_request(self) -> None:
 85        token = _bearer_token(self.request)
 86        info = self.authenticate_token(token) if token else None
 87        if info is None:
 88            # RFC 6750: a token that was supplied but rejected is invalid_token;
 89            # a missing token just gets the bare discovery challenge.
 90            error = "invalid_token" if token else None
 91            raise MCPUnauthorized(
 92                "Authentication required", www_authenticate=self._challenge(error)
 93            )
 94        self.user = info.user
 95        self.scopes = info.scopes
 96
 97    @property
 98    def oauth_resource(self) -> str:
 99        """The canonical URI of this MCP endpoint — the token audience."""
100        return _canonical_resource(self.request.build_absolute_uri(self.request.path))
101
102    def _challenge(self, error: str | None = None) -> str:
103        metadata_url = self.request.build_absolute_uri(
104            _WELL_KNOWN_PRM + self.request.path
105        )
106        challenge = f'Bearer resource_metadata="{metadata_url}"'
107        if error:
108            challenge += f', error="{error}"'
109        return challenge
110
111
112class MCPProtectedResourceView(View):
113    """Serves RFC 9728 protected-resource metadata for an MCP endpoint.
114
115    Mount at `.well-known/oauth-protected-resource/<mcp-path>` (the path the
116    `OAuthResourceServer` challenge points to). `authorization_servers` defaults
117    to this app's own origin — set it only when an external IdP issues the
118    tokens. The `resource` is derived from the request path, so one view can sit
119    in front of any MCP mount point.
120    """
121
122    authorization_servers: list[str] = []
123    oauth_scopes_supported: list[str] = []
124
125    def get(self) -> Response:
126        resource_path = self.request.path.removeprefix(_WELL_KNOWN_PRM) or "/"
127        # Same app issues the tokens by default; override for an external IdP.
128        origin = f"{self.request.scheme}://{self.request.host}"
129        servers = self.authorization_servers or [origin]
130        metadata: dict[str, Any] = {
131            "resource": _canonical_resource(
132                self.request.build_absolute_uri(resource_path)
133            ),
134            "authorization_servers": list(servers),
135            "bearer_methods_supported": ["header"],
136        }
137        if self.oauth_scopes_supported:
138            metadata["scopes_supported"] = list(self.oauth_scopes_supported)
139        return JsonResponse(metadata)