Robert Helewka
93639188d3
- Rename MCPToken to UserToken across models, views, and tests - Update URL names from mcp-token-* to token-* - Add Daedalus/Pallas integration design doc (v2) - Switch docker-compose to build local mnemosyne:local image via shared build config instead of pulling from git.helu.ca
55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
"""DRF authentication class backed by :class:`mcp_server.models.UserToken`.
|
|
|
|
Wraps :func:`mcp_server.auth.resolve_mcp_user` so a single verification
|
|
routine serves both surfaces:
|
|
|
|
* the FastMCP middleware on ``/mcp/`` (via ``MCPAuthMiddleware``); and
|
|
* the Django REST surface on ``/library/api/*`` and
|
|
``/mcp_server/api/teams/*`` (via this class).
|
|
|
|
Scope: this class authenticates the caller — it does *not* honour the
|
|
token's ``allowed_libraries`` / ``allowed_tools`` fields. Those apply
|
|
only to the MCP tool surface. On the REST endpoints, access is gated by
|
|
``Team.owner`` and ``Library.owner_username`` rather than per-token
|
|
scope claims; treating ``allowed_libraries`` as authoritative there
|
|
would either force Daedalus to mint an effectively-unrestricted token
|
|
(redundant with the user identity) or invent a per-endpoint scope
|
|
mapping with no natural shape.
|
|
|
|
The accepted header is ``Authorization: Bearer <plaintext>``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from rest_framework import authentication, exceptions
|
|
|
|
from .auth import MCPAuthError, resolve_mcp_user
|
|
|
|
|
|
class UserTokenAuthentication(authentication.BaseAuthentication):
|
|
"""Authenticate DRF requests with a ``UserToken`` bearer."""
|
|
|
|
keyword = "Bearer"
|
|
|
|
def authenticate(self, request):
|
|
header = authentication.get_authorization_header(request).decode("iso-8859-1")
|
|
if not header:
|
|
return None
|
|
parts = header.split()
|
|
if len(parts) < 2 or parts[0] != self.keyword:
|
|
# Not our scheme. Let other authenticators try.
|
|
return None
|
|
if len(parts) > 2:
|
|
raise exceptions.AuthenticationFailed(
|
|
"Invalid Authorization header: too many components."
|
|
)
|
|
|
|
try:
|
|
user, token = resolve_mcp_user(parts[1])
|
|
except MCPAuthError as exc:
|
|
raise exceptions.AuthenticationFailed(str(exc))
|
|
return user, token
|
|
|
|
def authenticate_header(self, request):
|
|
return self.keyword
|