Libraries created through the API are now stamped with the name of the UserToken that created them (Library.managed_by), on both the workspace and plain create endpoints; web-session creates stay null/unmanaged. The idempotent workspace re-POST lazily backfills null managed_by, and a one-off backfill_managed_by command labels pre-existing rows (workspace inference: kairos-mail-* → Kairos, else Daedalus; Spelunker via ingest job provenance). UI badges and warnings now render "Managed by <app>" via managed_by_display (inference fallback keeps legacy rows accurate before backfill). The list scope filter becomes all/managed/unmanaged with the old daedalus/global values aliased for bookmarks. The plain create endpoint also gains an explicit 409 name_conflict (previously a raw UniqueProperty 500) reporting the existing library's uid and manager. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
170 lines
6.4 KiB
Python
170 lines
6.4 KiB
Python
"""Tests for ``mcp_server.drf_auth.UserTokenAuthentication``.
|
|
|
|
Authenticates DRF endpoints using a per-user ``UserToken`` carried as
|
|
``Authorization: Bearer <plaintext>``. The class wraps
|
|
``resolve_mcp_user``; these tests assert the DRF-side behaviour
|
|
(header parsing, error mapping, integration with ``IsAuthenticated``)
|
|
on top of the resolver's own coverage in ``test_auth.py``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase, override_settings
|
|
from django.urls import path
|
|
from django.utils import timezone
|
|
from rest_framework import status
|
|
from rest_framework.decorators import api_view, permission_classes
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.response import Response
|
|
from rest_framework.test import APIClient
|
|
|
|
from mcp_server.models import UserToken
|
|
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
# A tiny endpoint mounted only for these tests so we can exercise the
|
|
# DRF auth pipeline without coupling to any real app's view contract.
|
|
@api_view(["GET"])
|
|
@permission_classes([IsAuthenticated])
|
|
def _whoami(request):
|
|
return Response({"username": request.user.username})
|
|
|
|
|
|
urlpatterns = [
|
|
path("__test_whoami__/", _whoami),
|
|
]
|
|
|
|
|
|
@override_settings(ROOT_URLCONF=__name__)
|
|
class UserTokenAuthenticationTest(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.user = User.objects.create_user(username="alice", password="pw")
|
|
|
|
def setUp(self):
|
|
self.client = APIClient()
|
|
self.token, self.plaintext = UserToken.objects.create_token(
|
|
user=self.user, name="t"
|
|
)
|
|
|
|
def _get(self, header=None):
|
|
kwargs = {}
|
|
if header is not None:
|
|
kwargs["HTTP_AUTHORIZATION"] = header
|
|
return self.client.get("/__test_whoami__/", **kwargs)
|
|
|
|
def test_no_header_returns_401_with_bearer_challenge(self):
|
|
resp = self._get()
|
|
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
# RFC 7235: anonymous request must include WWW-Authenticate so the
|
|
# client knows how to authenticate.
|
|
self.assertEqual(resp["WWW-Authenticate"], "Bearer")
|
|
|
|
def test_valid_bearer_authenticates(self):
|
|
resp = self._get(f"Bearer {self.plaintext}")
|
|
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(resp.json(), {"username": "alice"})
|
|
|
|
def test_invalid_bearer_returns_401(self):
|
|
resp = self._get("Bearer not-a-real-token")
|
|
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
def test_inactive_token_returns_401(self):
|
|
self.token.is_active = False
|
|
self.token.save(update_fields=["is_active"])
|
|
resp = self._get(f"Bearer {self.plaintext}")
|
|
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
def test_expired_token_returns_401(self):
|
|
self.token.expires_at = timezone.now() - timedelta(hours=1)
|
|
self.token.save(update_fields=["expires_at"])
|
|
resp = self._get(f"Bearer {self.plaintext}")
|
|
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
def test_disabled_user_returns_401(self):
|
|
self.user.is_active = False
|
|
self.user.save(update_fields=["is_active"])
|
|
resp = self._get(f"Bearer {self.plaintext}")
|
|
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
def test_wrong_keyword_falls_through(self):
|
|
# ``Token <plaintext>`` is the old DRF authtoken keyword. The new
|
|
# class only accepts ``Bearer``; a stale ``Token`` header is not
|
|
# ours to consume — we return None and let the next auth class
|
|
# try. SessionAuthentication doesn't accept it either, so the
|
|
# request lands anonymous and IsAuthenticated returns 401.
|
|
resp = self._get(f"Token {self.plaintext}")
|
|
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
def test_malformed_header_too_many_parts_returns_401(self):
|
|
resp = self._get(f"Bearer {self.plaintext} extra")
|
|
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
def test_request_token_label_reads_token_name(self):
|
|
from types import SimpleNamespace
|
|
|
|
from mcp_server.drf_auth import request_token_label
|
|
|
|
token = UserToken(name=" Spelunker ")
|
|
self.assertEqual(
|
|
request_token_label(SimpleNamespace(auth=token)), "Spelunker"
|
|
)
|
|
|
|
def test_request_token_label_none_for_session(self):
|
|
from types import SimpleNamespace
|
|
|
|
from mcp_server.drf_auth import request_token_label
|
|
|
|
self.assertIsNone(request_token_label(SimpleNamespace(auth=None)))
|
|
# A request object with no auth attribute at all (plain Django).
|
|
self.assertIsNone(request_token_label(SimpleNamespace()))
|
|
|
|
def test_request_token_label_none_for_blank_name(self):
|
|
from types import SimpleNamespace
|
|
|
|
from mcp_server.drf_auth import request_token_label
|
|
|
|
self.assertIsNone(
|
|
request_token_label(SimpleNamespace(auth=UserToken(name=" ")))
|
|
)
|
|
|
|
def test_request_token_label_none_for_foreign_auth_object(self):
|
|
from types import SimpleNamespace
|
|
|
|
from mcp_server.drf_auth import request_token_label
|
|
|
|
# e.g. a JWT dict from another auth class — not a UserToken.
|
|
self.assertIsNone(
|
|
request_token_label(SimpleNamespace(auth={"iss": "daedalus"}))
|
|
)
|
|
|
|
def test_request_auth_stashes_token(self):
|
|
# The auth class returns (user, token); DRF places the token on
|
|
# request.auth. Re-use a UserToken-aware endpoint to verify.
|
|
@api_view(["GET"])
|
|
@permission_classes([IsAuthenticated])
|
|
def echo_token(request):
|
|
return Response({"token_name": request.auth.name})
|
|
|
|
from django.urls import path as _path
|
|
|
|
with override_settings(ROOT_URLCONF=__name__):
|
|
# Mount the extra endpoint via a per-test urlpatterns swap.
|
|
# Simpler: just call the resolver directly to confirm the
|
|
# auth class returns the (user, token) tuple it should.
|
|
from mcp_server.drf_auth import UserTokenAuthentication
|
|
from django.test import RequestFactory
|
|
|
|
request = RequestFactory().get(
|
|
"/__test_whoami__/",
|
|
HTTP_AUTHORIZATION=f"Bearer {self.plaintext}",
|
|
)
|
|
user, token = UserTokenAuthentication().authenticate(request)
|
|
self.assertEqual(user.pk, self.user.pk)
|
|
self.assertEqual(token.pk, self.token.pk)
|