feat: rework auth model with UserToken and Daedalus/Pallas integration
- 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
This commit is contained in:
@@ -3,7 +3,7 @@
|
||||
Covers all three credential types described in
|
||||
``docs/DAEDALUS_PALLAS_INTEGRATION_v1.md`` §3.2:
|
||||
|
||||
1. Opaque :class:`~mcp_server.models.MCPToken` — ``resolve_mcp_user``
|
||||
1. Opaque :class:`~mcp_server.models.UserToken` — ``resolve_mcp_user``
|
||||
+ ``MCPAuthMiddleware`` opaque branch.
|
||||
2. Per-turn JWT (``iss=daedalus``, legacy) — ``resolve_mcp_jwt`` normal
|
||||
path, ``_remember_jti`` replay cache, ``claims["libs"]``-derived
|
||||
@@ -42,7 +42,7 @@ from mcp_server.auth import (
|
||||
)
|
||||
from mcp_server.models import (
|
||||
MCPSigningKey,
|
||||
MCPToken,
|
||||
UserToken,
|
||||
Team,
|
||||
TeamWorkspaceAssignment,
|
||||
)
|
||||
@@ -51,7 +51,7 @@ User = get_user_model()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Opaque MCPToken
|
||||
# Opaque UserToken
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@@ -62,7 +62,7 @@ class ResolveMCPUserTest(TestCase):
|
||||
self.user = User.objects.create_user(
|
||||
username="bob", email="bob@example.com", password="pw"
|
||||
)
|
||||
self.token, self.plaintext = MCPToken.objects.create_token(
|
||||
self.token, self.plaintext = UserToken.objects.create_token(
|
||||
user=self.user, name="t"
|
||||
)
|
||||
|
||||
@@ -105,7 +105,7 @@ class ResolveMCPUserTest(TestCase):
|
||||
|
||||
plaintext = self.plaintext
|
||||
with connection.cursor() as cur:
|
||||
cur.execute("SELECT * FROM mcp_server_mcptoken")
|
||||
cur.execute("SELECT * FROM mcp_server_usertoken")
|
||||
rows = cur.fetchall()
|
||||
for row in rows:
|
||||
for value in row:
|
||||
@@ -190,7 +190,7 @@ class LooksLikeJWTTest(TestCase):
|
||||
self.assertFalse(looks_like_jwt("!!!.bbb.ccc"))
|
||||
|
||||
def test_opaque_token_rejected(self):
|
||||
# Real ``MCPToken.create_token`` plaintext is 48-byte base64, often
|
||||
# Real ``UserToken.create_token`` plaintext is 48-byte base64, often
|
||||
# contains dashes but never two dots.
|
||||
self.assertFalse(
|
||||
looks_like_jwt("CxGb3rThJ7_4jUGl0q2_fakey_fakey_fakey_fakey_fakey")
|
||||
@@ -506,3 +506,37 @@ class ResolveJWTActorTest(TestCase):
|
||||
claims = {"typ": "team", "team_id": uuid.uuid4()}
|
||||
with self.assertRaises(MCPAuthError):
|
||||
_resolve_jwt_actor(claims)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module-level import guards (regression tests)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class AuthModuleImportsTest(TestCase):
|
||||
"""Pin the imports the runtime depends on.
|
||||
|
||||
These are regressions waiting to happen: a quick `grep` doesn't
|
||||
catch a missing import when the consuming code is only reached via
|
||||
a runtime path (FastMCP middleware, async tool dispatch, …) that
|
||||
the test suite doesn't exercise end-to-end.
|
||||
|
||||
Add a check here whenever production fails with a
|
||||
``NameError: name 'X' is not defined`` that the test suite missed.
|
||||
"""
|
||||
|
||||
def test_settings_is_importable(self):
|
||||
"""``MCPAuthMiddleware.on_call_tool`` reads
|
||||
``settings.MCP_REQUIRE_AUTH`` on every tool call (including
|
||||
unauthenticated ``get_health`` polls from Pallas). Removing the
|
||||
``from django.conf import settings`` import — as happened during
|
||||
the v2 token-consolidation cleanup — surfaces as
|
||||
``NameError: name 'settings' is not defined`` for *every* MCP
|
||||
client. Keep this import alive.
|
||||
"""
|
||||
from django.conf import settings as dj_settings
|
||||
|
||||
from mcp_server import auth as auth_module
|
||||
|
||||
self.assertTrue(hasattr(auth_module, "settings"))
|
||||
self.assertIs(auth_module.settings, dj_settings)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Tests for the create_mcp_token management command."""
|
||||
"""Tests for the create_user_token management command."""
|
||||
|
||||
from io import StringIO
|
||||
|
||||
@@ -7,12 +7,12 @@ from django.core.management import call_command
|
||||
from django.core.management.base import CommandError
|
||||
from django.test import TestCase
|
||||
|
||||
from mcp_server.models import MCPToken
|
||||
from mcp_server.models import UserToken
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class CreateMCPTokenCommandTest(TestCase):
|
||||
class CreateUserTokenCommandTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(
|
||||
username="carol", email="carol@example.com", password="pw"
|
||||
@@ -20,33 +20,33 @@ class CreateMCPTokenCommandTest(TestCase):
|
||||
|
||||
def test_create_basic_token(self):
|
||||
out = StringIO()
|
||||
call_command("create_mcp_token", user="carol@example.com", name="CLI", stdout=out)
|
||||
self.assertEqual(MCPToken.objects.count(), 1)
|
||||
call_command("create_user_token", user="carol@example.com", name="CLI", stdout=out)
|
||||
self.assertEqual(UserToken.objects.count(), 1)
|
||||
self.assertIn("CLI", out.getvalue())
|
||||
|
||||
def test_lookup_by_username(self):
|
||||
out = StringIO()
|
||||
call_command("create_mcp_token", user="carol", name="CLI2", stdout=out)
|
||||
self.assertEqual(MCPToken.objects.count(), 1)
|
||||
call_command("create_user_token", user="carol", name="CLI2", stdout=out)
|
||||
self.assertEqual(UserToken.objects.count(), 1)
|
||||
|
||||
def test_unknown_user_raises(self):
|
||||
with self.assertRaises(CommandError):
|
||||
call_command("create_mcp_token", user="nobody@x.com", name="x")
|
||||
call_command("create_user_token", user="nobody@x.com", name="x")
|
||||
|
||||
def test_inactive_user_raises(self):
|
||||
self.user.is_active = False
|
||||
self.user.save()
|
||||
with self.assertRaises(CommandError):
|
||||
call_command("create_mcp_token", user="carol", name="x")
|
||||
call_command("create_user_token", user="carol", name="x")
|
||||
|
||||
def test_tool_whitelist_parsed(self):
|
||||
out = StringIO()
|
||||
call_command(
|
||||
"create_mcp_token",
|
||||
"create_user_token",
|
||||
user="carol",
|
||||
name="Restricted",
|
||||
tools="search,get_chunk",
|
||||
stdout=out,
|
||||
)
|
||||
token = MCPToken.objects.get(name="Restricted")
|
||||
token = UserToken.objects.get(name="Restricted")
|
||||
self.assertEqual(token.allowed_tools, ["search", "get_chunk"])
|
||||
|
||||
131
mnemosyne/mcp_server/tests/test_drf_auth.py
Normal file
131
mnemosyne/mcp_server/tests/test_drf_auth.py
Normal file
@@ -0,0 +1,131 @@
|
||||
"""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_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)
|
||||
@@ -3,24 +3,24 @@
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase
|
||||
|
||||
from mcp_server.forms import MCPTokenCreateForm, MCPTokenEditForm
|
||||
from mcp_server.models import MCPToken
|
||||
from mcp_server.forms import UserTokenCreateForm, UserTokenEditForm
|
||||
from mcp_server.models import UserToken
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class CreateFormTest(TestCase):
|
||||
def test_required_fields(self):
|
||||
form = MCPTokenCreateForm(data={})
|
||||
form = UserTokenCreateForm(data={})
|
||||
self.assertFalse(form.is_valid())
|
||||
self.assertIn("name", form.errors)
|
||||
|
||||
def test_name_only_is_valid(self):
|
||||
form = MCPTokenCreateForm(data={"name": "Test"})
|
||||
form = UserTokenCreateForm(data={"name": "Test"})
|
||||
self.assertTrue(form.is_valid(), form.errors)
|
||||
|
||||
def test_tool_choices_match_registered_tools(self):
|
||||
form = MCPTokenCreateForm()
|
||||
form = UserTokenCreateForm()
|
||||
choices = {value for value, _ in form.fields["allowed_tools"].choices}
|
||||
# These five must always be present per the FastMCP server.
|
||||
for expected in {"search", "get_chunk", "list_libraries", "list_collections", "list_items"}:
|
||||
@@ -30,16 +30,16 @@ class CreateFormTest(TestCase):
|
||||
class EditFormTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(username="alice", password="pw")
|
||||
self.token, _ = MCPToken.objects.create_token(
|
||||
self.token, _ = UserToken.objects.create_token(
|
||||
user=self.user, name="t", allowed_tools=["search"]
|
||||
)
|
||||
|
||||
def test_initial_allowed_tools_populated(self):
|
||||
form = MCPTokenEditForm(instance=self.token)
|
||||
form = UserTokenEditForm(instance=self.token)
|
||||
self.assertEqual(form.fields["allowed_tools"].initial, ["search"])
|
||||
|
||||
def test_save_updates_metadata(self):
|
||||
form = MCPTokenEditForm(
|
||||
form = UserTokenEditForm(
|
||||
data={
|
||||
"name": "Renamed",
|
||||
"is_active": False,
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
"""Tests for the Team / LibraryMembership / TeamWorkspaceAssignment models.
|
||||
|
||||
``MCPToken``'s hash-at-rest semantics live in ``test_token.py``; this
|
||||
``UserToken``'s hash-at-rest semantics live in ``test_token.py``; this
|
||||
module exercises the new Phase 2 tables introduced by
|
||||
``docs/DAEDALUS_PALLAS_INTEGRATION_v1.md`` §4 plus the
|
||||
``allowed_libraries`` JSONField attached to the existing
|
||||
:class:`~mcp_server.models.MCPToken`.
|
||||
:class:`~mcp_server.models.UserToken`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -18,7 +18,7 @@ from django.test import TestCase
|
||||
from mcp_server.models import (
|
||||
LibraryMembership,
|
||||
MCPSigningKey,
|
||||
MCPToken,
|
||||
UserToken,
|
||||
Team,
|
||||
TeamWorkspaceAssignment,
|
||||
)
|
||||
@@ -92,26 +92,26 @@ class LibraryMembershipTest(TestCase):
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MCPToken.allowed_libraries
|
||||
# UserToken.allowed_libraries
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class MCPTokenAllowedLibrariesTest(TestCase):
|
||||
class UserTokenAllowedLibrariesTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(username="u", password="pw")
|
||||
|
||||
def test_defaults_to_empty_list(self):
|
||||
token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
self.assertEqual(token.allowed_libraries, [])
|
||||
|
||||
def test_create_token_accepts_allowed_libraries(self):
|
||||
token, _ = MCPToken.objects.create_token(
|
||||
token, _ = UserToken.objects.create_token(
|
||||
user=self.user, name="t", allowed_libraries=["lib-a", "lib-b"]
|
||||
)
|
||||
self.assertEqual(token.allowed_libraries, ["lib-a", "lib-b"])
|
||||
|
||||
def test_allowed_libraries_round_trips(self):
|
||||
token, _ = MCPToken.objects.create_token(
|
||||
token, _ = UserToken.objects.create_token(
|
||||
user=self.user,
|
||||
name="t",
|
||||
allowed_libraries=["lib-a", "lib-b", "lib-c"],
|
||||
|
||||
@@ -385,13 +385,23 @@ class TeamRotateTest(_AuthenticatedAPITest):
|
||||
kwargs={"team_id": self.team.id},
|
||||
)
|
||||
|
||||
def test_unknown_team_returns_404(self):
|
||||
def test_rotate_upserts_missing_team(self):
|
||||
# Rotate is upsert-on-missing: if no Team row exists for this
|
||||
# id, create one owned by the caller and mint its first JWT.
|
||||
# Eliminates the create-before-rotate ordering trap Daedalus hit
|
||||
# in production.
|
||||
new_id = uuid.uuid4()
|
||||
url = reverse(
|
||||
"mcp-server-api:team-rotate",
|
||||
kwargs={"team_id": uuid.uuid4()},
|
||||
kwargs={"team_id": new_id},
|
||||
)
|
||||
resp = self.client.post(url)
|
||||
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertIn("jwt", resp.data)
|
||||
team = Team.objects.get(pk=new_id)
|
||||
self.assertEqual(team.owner_id, self.user.id)
|
||||
self.assertTrue(team.active)
|
||||
self.assertIsNotNone(team.active_jti)
|
||||
|
||||
def test_rotate_returns_new_jwt_and_changes_active_jti(self):
|
||||
before = self.team.active_jti
|
||||
@@ -417,10 +427,16 @@ class TeamRotateTest(_AuthenticatedAPITest):
|
||||
resp.status_code, status.HTTP_503_SERVICE_UNAVAILABLE
|
||||
)
|
||||
|
||||
def test_rotate_by_non_owner_returns_404(self):
|
||||
def test_rotate_by_non_owner_returns_409(self):
|
||||
# The team row exists under Alice; Bob rotating it must not
|
||||
# upsert (that would silently steal the id) and must not 404
|
||||
# (would tell Bob the id is free). 409 is the right answer.
|
||||
before = self.team.active_jti
|
||||
before_owner = self.team.owner_id
|
||||
self.client.force_authenticate(user=self.other_user)
|
||||
resp = self.client.post(self.url)
|
||||
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
|
||||
self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT)
|
||||
# Alice's team is untouched.
|
||||
self.team.refresh_from_db()
|
||||
self.assertEqual(self.team.active_jti, before)
|
||||
self.assertEqual(self.team.owner_id, before_owner)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Tests for the MCPToken model."""
|
||||
"""Tests for the UserToken model."""
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
@@ -6,19 +6,19 @@ from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase
|
||||
from django.utils import timezone
|
||||
|
||||
from mcp_server.models import MCPToken, hash_token
|
||||
from mcp_server.models import UserToken, hash_token
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class MCPTokenModelTest(TestCase):
|
||||
class UserTokenModelTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(
|
||||
username="alice", email="alice@example.com", password="pw"
|
||||
)
|
||||
|
||||
def test_create_token_returns_plaintext_and_stores_hash(self):
|
||||
token, plaintext = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
token, plaintext = UserToken.objects.create_token(user=self.user, name="t")
|
||||
self.assertTrue(plaintext)
|
||||
self.assertGreater(len(plaintext), 20)
|
||||
# Database stores hash, not plaintext
|
||||
@@ -29,21 +29,21 @@ class MCPTokenModelTest(TestCase):
|
||||
def test_token_hash_never_equals_plaintext(self):
|
||||
# Regression guard: if anyone ever wires plaintext back into token_hash,
|
||||
# this fails.
|
||||
token, plaintext = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
token, plaintext = UserToken.objects.create_token(user=self.user, name="t")
|
||||
self.assertNotIn(plaintext, token.token_hash)
|
||||
|
||||
def test_active_token_is_valid(self):
|
||||
token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
self.assertTrue(token.is_valid)
|
||||
|
||||
def test_inactive_token_not_valid(self):
|
||||
token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
token.is_active = False
|
||||
token.save()
|
||||
self.assertFalse(token.is_valid)
|
||||
|
||||
def test_expired_token_not_valid(self):
|
||||
token, _ = MCPToken.objects.create_token(
|
||||
token, _ = UserToken.objects.create_token(
|
||||
user=self.user,
|
||||
name="t",
|
||||
expires_at=timezone.now() - timedelta(hours=1),
|
||||
@@ -51,27 +51,27 @@ class MCPTokenModelTest(TestCase):
|
||||
self.assertFalse(token.is_valid)
|
||||
|
||||
def test_unrestricted_permits_all(self):
|
||||
token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
self.assertTrue(token.can_use_tool("anything"))
|
||||
|
||||
def test_tool_whitelist(self):
|
||||
token, _ = MCPToken.objects.create_token(
|
||||
token, _ = UserToken.objects.create_token(
|
||||
user=self.user, name="t", allowed_tools=["search"]
|
||||
)
|
||||
self.assertTrue(token.can_use_tool("search"))
|
||||
self.assertFalse(token.can_use_tool("get_chunk"))
|
||||
|
||||
def test_record_usage(self):
|
||||
token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
self.assertIsNone(token.last_used_at)
|
||||
token.record_usage()
|
||||
token.refresh_from_db()
|
||||
self.assertIsNotNone(token.last_used_at)
|
||||
|
||||
def test_masked_token_is_hash_prefix(self):
|
||||
token, plaintext = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
token, plaintext = UserToken.objects.create_token(user=self.user, name="t")
|
||||
masked = token.get_masked_token()
|
||||
self.assertTrue(masked.startswith("mcp_…"))
|
||||
self.assertTrue(masked.startswith("tok_…"))
|
||||
self.assertIn(token.token_hash[:8], masked)
|
||||
# Plaintext must never leak through the masked display
|
||||
self.assertNotIn(plaintext, masked)
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
"""View tests for the MCP token self-service dashboard."""
|
||||
"""View tests for the per-user API token self-service dashboard."""
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase
|
||||
from django.urls import reverse
|
||||
|
||||
from mcp_server.models import MCPToken
|
||||
from mcp_server.models import UserToken
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
@@ -14,7 +14,7 @@ class TokenListViewTest(TestCase):
|
||||
self.user = User.objects.create_user(
|
||||
username="alice", email="alice@example.com", password="pw"
|
||||
)
|
||||
self.url = reverse("mcp_server:mcp-token-list")
|
||||
self.url = reverse("mcp_server:token-list")
|
||||
|
||||
def test_login_required(self):
|
||||
resp = self.client.get(self.url)
|
||||
@@ -23,8 +23,8 @@ class TokenListViewTest(TestCase):
|
||||
|
||||
def test_list_shows_only_own_tokens(self):
|
||||
other = User.objects.create_user(username="bob", password="pw")
|
||||
MCPToken.objects.create_token(user=self.user, name="mine")
|
||||
MCPToken.objects.create_token(user=other, name="theirs")
|
||||
UserToken.objects.create_token(user=self.user, name="mine")
|
||||
UserToken.objects.create_token(user=other, name="theirs")
|
||||
self.client.force_login(self.user)
|
||||
resp = self.client.get(self.url)
|
||||
self.assertContains(resp, "mine")
|
||||
@@ -33,19 +33,19 @@ class TokenListViewTest(TestCase):
|
||||
def test_empty_state(self):
|
||||
self.client.force_login(self.user)
|
||||
resp = self.client.get(self.url)
|
||||
self.assertContains(resp, "No MCP tokens yet.")
|
||||
self.assertContains(resp, "No API tokens yet.")
|
||||
|
||||
|
||||
class TokenCreateViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(username="alice", password="pw")
|
||||
self.client.force_login(self.user)
|
||||
self.url = reverse("mcp_server:mcp-token-create")
|
||||
self.url = reverse("mcp_server:token-create")
|
||||
|
||||
def test_get_renders_form(self):
|
||||
resp = self.client.get(self.url)
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
self.assertContains(resp, "Generate MCP Token")
|
||||
self.assertContains(resp, "Generate API Token")
|
||||
|
||||
def test_post_creates_token_and_shows_plaintext_once(self):
|
||||
resp = self.client.post(self.url, {"name": "Claude Desktop"})
|
||||
@@ -53,7 +53,7 @@ class TokenCreateViewTest(TestCase):
|
||||
self.assertContains(resp, "Save this token now")
|
||||
# Pull the created row, verify the response contained a plaintext that
|
||||
# is NOT what we stored.
|
||||
token = MCPToken.objects.get(user=self.user, name="Claude Desktop")
|
||||
token = UserToken.objects.get(user=self.user, name="Claude Desktop")
|
||||
self.assertNotContains(resp, token.token_hash) # hash is not what we display
|
||||
# And the detail page never renders the plaintext.
|
||||
body = resp.content.decode()
|
||||
@@ -71,35 +71,35 @@ class TokenCreateViewTest(TestCase):
|
||||
self.assertEqual(hash_token(plaintext), token.token_hash)
|
||||
# Detail page must NOT contain the plaintext.
|
||||
detail_resp = self.client.get(
|
||||
reverse("mcp_server:mcp-token-detail", args=[token.pk])
|
||||
reverse("mcp_server:token-detail", args=[token.pk])
|
||||
)
|
||||
self.assertNotContains(detail_resp, plaintext)
|
||||
|
||||
def test_post_invalid_renders_form_again(self):
|
||||
resp = self.client.post(self.url, {"name": ""})
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
self.assertContains(resp, "Generate MCP Token")
|
||||
self.assertEqual(MCPToken.objects.count(), 0)
|
||||
self.assertContains(resp, "Generate API Token")
|
||||
self.assertEqual(UserToken.objects.count(), 0)
|
||||
|
||||
|
||||
class TokenDetailViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(username="alice", password="pw")
|
||||
self.client.force_login(self.user)
|
||||
self.token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
self.token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
|
||||
def test_renders_token(self):
|
||||
resp = self.client.get(
|
||||
reverse("mcp_server:mcp-token-detail", args=[self.token.pk])
|
||||
reverse("mcp_server:token-detail", args=[self.token.pk])
|
||||
)
|
||||
self.assertContains(resp, self.token.name)
|
||||
self.assertContains(resp, self.token.get_masked_token())
|
||||
|
||||
def test_cannot_view_other_users_token(self):
|
||||
other = User.objects.create_user(username="bob", password="pw")
|
||||
other_token, _ = MCPToken.objects.create_token(user=other, name="theirs")
|
||||
other_token, _ = UserToken.objects.create_token(user=other, name="theirs")
|
||||
resp = self.client.get(
|
||||
reverse("mcp_server:mcp-token-detail", args=[other_token.pk])
|
||||
reverse("mcp_server:token-detail", args=[other_token.pk])
|
||||
)
|
||||
self.assertEqual(resp.status_code, 404)
|
||||
|
||||
@@ -108,11 +108,11 @@ class TokenEditViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(username="alice", password="pw")
|
||||
self.client.force_login(self.user)
|
||||
self.token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
self.token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
|
||||
def test_post_updates_metadata(self):
|
||||
resp = self.client.post(
|
||||
reverse("mcp_server:mcp-token-edit", args=[self.token.pk]),
|
||||
reverse("mcp_server:token-edit", args=[self.token.pk]),
|
||||
{
|
||||
"name": "Renamed",
|
||||
"is_active": "on",
|
||||
@@ -127,9 +127,9 @@ class TokenEditViewTest(TestCase):
|
||||
|
||||
def test_cannot_edit_other_users_token(self):
|
||||
other = User.objects.create_user(username="bob", password="pw")
|
||||
other_token, _ = MCPToken.objects.create_token(user=other, name="theirs")
|
||||
other_token, _ = UserToken.objects.create_token(user=other, name="theirs")
|
||||
resp = self.client.post(
|
||||
reverse("mcp_server:mcp-token-edit", args=[other_token.pk]),
|
||||
reverse("mcp_server:token-edit", args=[other_token.pk]),
|
||||
{"name": "hacked"},
|
||||
)
|
||||
self.assertEqual(resp.status_code, 404)
|
||||
@@ -139,19 +139,19 @@ class TokenRevokeViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(username="alice", password="pw")
|
||||
self.client.force_login(self.user)
|
||||
self.token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
self.token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
|
||||
def test_revoke_sets_inactive_keeps_row(self):
|
||||
url = reverse("mcp_server:mcp-token-revoke", args=[self.token.pk])
|
||||
url = reverse("mcp_server:token-revoke", args=[self.token.pk])
|
||||
resp = self.client.post(url)
|
||||
self.assertEqual(resp.status_code, 302)
|
||||
self.token.refresh_from_db()
|
||||
self.assertFalse(self.token.is_active)
|
||||
# Row still exists for audit trail.
|
||||
self.assertTrue(MCPToken.objects.filter(pk=self.token.pk).exists())
|
||||
self.assertTrue(UserToken.objects.filter(pk=self.token.pk).exists())
|
||||
|
||||
def test_get_not_allowed(self):
|
||||
url = reverse("mcp_server:mcp-token-revoke", args=[self.token.pk])
|
||||
url = reverse("mcp_server:token-revoke", args=[self.token.pk])
|
||||
resp = self.client.get(url)
|
||||
self.assertEqual(resp.status_code, 405)
|
||||
|
||||
@@ -160,18 +160,18 @@ class TokenDeleteViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(username="alice", password="pw")
|
||||
self.client.force_login(self.user)
|
||||
self.token, _ = MCPToken.objects.create_token(user=self.user, name="t")
|
||||
self.token, _ = UserToken.objects.create_token(user=self.user, name="t")
|
||||
|
||||
def test_delete_removes_row(self):
|
||||
url = reverse("mcp_server:mcp-token-delete", args=[self.token.pk])
|
||||
url = reverse("mcp_server:token-delete", args=[self.token.pk])
|
||||
resp = self.client.post(url)
|
||||
self.assertEqual(resp.status_code, 302)
|
||||
self.assertFalse(MCPToken.objects.filter(pk=self.token.pk).exists())
|
||||
self.assertFalse(UserToken.objects.filter(pk=self.token.pk).exists())
|
||||
|
||||
def test_cannot_delete_other_users_token(self):
|
||||
other = User.objects.create_user(username="bob", password="pw")
|
||||
other_token, _ = MCPToken.objects.create_token(user=other, name="theirs")
|
||||
url = reverse("mcp_server:mcp-token-delete", args=[other_token.pk])
|
||||
other_token, _ = UserToken.objects.create_token(user=other, name="theirs")
|
||||
url = reverse("mcp_server:token-delete", args=[other_token.pk])
|
||||
resp = self.client.post(url)
|
||||
self.assertEqual(resp.status_code, 404)
|
||||
self.assertTrue(MCPToken.objects.filter(pk=other_token.pk).exists())
|
||||
self.assertTrue(UserToken.objects.filter(pk=other_token.pk).exists())
|
||||
|
||||
Reference in New Issue
Block a user