r/mnemosyne
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

docs: replace daedalus-service basic auth with per-user DRF tokens
All checks were successful
CVE Scan & Docker Build / security-scan (push) Successful in 56s
Details
CVE Scan & Docker Build / build-and-push (push) Successful in 3m30s
Details

Browse Source
This commit is contained in:
Robert Helewka
2026-05-22 22:59:59 -04:00
parent 7296b8c42f
commit 409da7d109
17 changed files with 364 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
mnemosyne/mcp_server/tests/test_auth.py
View File

@@ -34,6 +34,7 @@ from mcp_server.auth import (
MCPAuthError,
_libraries_for_team,
_remember_jti,
_resolve_jwt_actor,
_resolved_libraries_for_jwt,
looks_like_jwt,
resolve_mcp_jwt,
@@ -314,11 +315,16 @@ class PerTurnReplayCacheTest(TestCase):
class ResolveTeamJWTTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.owner = User.objects.create_user(username="alice", password="pw")
def setUp(self):
self.key = _make_signing_key()
self.team = Team.objects.create(
id=uuid.uuid4(),
name="Pallas-Harper",
owner=self.owner,
active=True,
active_jti=uuid.uuid4(),
)
@@ -362,10 +368,15 @@ class ResolveTeamJWTTest(TestCase):
class LibrariesForTeamTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.owner = User.objects.create_user(username="alice", password="pw")
def setUp(self):
self.team = Team.objects.create(
id=uuid.uuid4(),
name="T",
owner=self.owner,
active=True,
active_jti=uuid.uuid4(),
)
@@ -415,6 +426,10 @@ class LibrariesForTeamTest(TestCase):
class ResolvedLibrariesForJWTDispatcherTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.owner = User.objects.create_user(username="alice", password="pw")
def test_per_turn_claims_use_libs(self):
claims = {"libs": ["one", "two"]}
self.assertEqual(
@@ -425,6 +440,7 @@ class ResolvedLibrariesForJWTDispatcherTest(TestCase):
team = Team.objects.create(
id=uuid.uuid4(),
name="T",
owner=self.owner,
active=True,
active_jti=uuid.uuid4(),
)
@@ -442,3 +458,51 @@ class ResolvedLibrariesForJWTDispatcherTest(TestCase):
)
out = _resolved_libraries_for_jwt(claims)
self.assertEqual(out, ["lib-team"])
# ---------------------------------------------------------------------------
# _resolve_jwt_actor — team JWTs resolve to the team's owner
# ---------------------------------------------------------------------------
class ResolveJWTActorTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.owner = User.objects.create_user(username="alice", password="pw")
def setUp(self):
self.team = Team.objects.create(
id=uuid.uuid4(),
name="T",
owner=self.owner,
active=True,
active_jti=uuid.uuid4(),
)
def test_team_jwt_resolves_to_team_owner(self):
claims = {"typ": "team", "team_id": self.team.id}
self.assertEqual(_resolve_jwt_actor(claims), self.owner)
def test_inactive_team_rejected(self):
self.team.deactivate()
claims = {"typ": "team", "team_id": self.team.id}
with self.assertRaises(MCPAuthError):
_resolve_jwt_actor(claims)
def test_disabled_owner_rejected(self):
self.owner.is_active = False
self.owner.save(update_fields=["is_active"])
claims = {"typ": "team", "team_id": self.team.id}
with self.assertRaises(MCPAuthError):
_resolve_jwt_actor(claims)
def test_per_turn_jwt_rejected(self):
# No more service-user fallback: per-turn JWTs can't be attributed
# to a Mnemosyne user, so the resolver refuses them.
with self.assertRaises(MCPAuthError):
_resolve_jwt_actor({"libs": ["x"]})
def test_unknown_team_rejected(self):
claims = {"typ": "team", "team_id": uuid.uuid4()}
with self.assertRaises(MCPAuthError):
_resolve_jwt_actor(claims)

28
mnemosyne/mcp_server/tests/test_models.py
View File

@@ -128,15 +128,21 @@ class MCPTokenAllowedLibrariesTest(TestCase):
class TeamTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.owner = get_user_model().objects.create_user(
username="alice", password="pw"
)
def test_create_with_explicit_uuid(self):
tid = uuid.uuid4()
team = Team.objects.create(id=tid, name="Harper")
team = Team.objects.create(id=tid, name="Harper", owner=self.owner)
self.assertEqual(team.id, tid)
self.assertTrue(team.active)
self.assertIsNone(team.active_jti)
def test_rotate_jti_installs_fresh_uuid(self):
team = Team.objects.create(id=uuid.uuid4(), name="t")
team = Team.objects.create(id=uuid.uuid4(), name="t", owner=self.owner)
first = team.rotate_jti()
self.assertIsInstance(first, uuid.UUID)
self.assertEqual(team.active_jti, first)
@@ -146,14 +152,14 @@ class TeamTest(TestCase):
self.assertEqual(team.active_jti, second)
def test_rotate_jti_persists(self):
team = Team.objects.create(id=uuid.uuid4(), name="t")
team = Team.objects.create(id=uuid.uuid4(), name="t", owner=self.owner)
team.rotate_jti()
# Reload from DB and make sure the UUID was committed.
reloaded = Team.objects.get(pk=team.id)
self.assertEqual(reloaded.active_jti, team.active_jti)
def test_deactivate_clears_active_jti(self):
team = Team.objects.create(id=uuid.uuid4(), name="t")
team = Team.objects.create(id=uuid.uuid4(), name="t", owner=self.owner)
team.rotate_jti()
self.assertTrue(team.active)
team.deactivate()
@@ -171,8 +177,16 @@ class TeamTest(TestCase):
class TeamWorkspaceAssignmentTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.owner = get_user_model().objects.create_user(
username="alice", password="pw"
)
def setUp(self):
self.team = Team.objects.create(id=uuid.uuid4(), name="t")
self.team = Team.objects.create(
id=uuid.uuid4(), name="t", owner=self.owner
)
def test_unique_team_workspace_pair(self):
TeamWorkspaceAssignment.objects.create(
@@ -185,7 +199,9 @@ class TeamWorkspaceAssignmentTest(TestCase):
)
def test_same_workspace_different_teams_allowed(self):
other = Team.objects.create(id=uuid.uuid4(), name="t2")
other = Team.objects.create(
id=uuid.uuid4(), name="t2", owner=self.owner
)
TeamWorkspaceAssignment.objects.create(
team=self.team, workspace_id="ws-a"
)

99
mnemosyne/mcp_server/tests/test_teams_api.py
View File

@@ -1,10 +1,11 @@
"""Tests for the ``/mcp_server/api/teams/`` REST control plane.
This is the Daedalus-facing surface described in §7 of
``docs/DAEDALUS_PALLAS_INTEGRATION_v1.md``. We do NOT exercise HTTP
Basic auth here (that's part of DRF / the project's session auth
stack); instead we use :meth:`APIClient.force_authenticate` to focus
on the endpoints' own idempotence and state-transition rules.
``docs/DAEDALUS_PALLAS_INTEGRATION_v1.md``. We do NOT exercise the
DRF Token / Session auth machinery here (that's covered by DRF
itself); instead we use :meth:`APIClient.force_authenticate` to focus
on the endpoints' own idempotence, ownership, and state-transition
rules.
"""
from __future__ import annotations
@@ -36,20 +37,23 @@ def _seed_signing_key() -> MCPSigningKey:
class _AuthenticatedAPITest(TestCase):
"""Shared ``APIClient`` authenticated as the service user.
"""Shared ``APIClient`` authenticated as a regular user.
The real deployment has Daedalus hit these endpoints as
``daedalus-service`` over HTTP Basic, but the view decorator is a
plain ``IsAuthenticated`` — so for unit-test purposes we use
``force_authenticate`` with any active user.
The endpoints scope by ``request.user`` (every team has an
``owner``); ``self.user`` is the team owner and ``self.other_user``
is used for cross-user access tests.
"""
def setUp(self):
self.service_user = User.objects.create_user(
username="daedalus-service", password="pw"
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(username="alice", password="pw")
cls.other_user = User.objects.create_user(
username="bob", password="pw"
)
def setUp(self):
self.client = APIClient()
self.client.force_authenticate(user=self.service_user)
self.client.force_authenticate(user=self.user)
# ---------------------------------------------------------------------------
@@ -106,6 +110,29 @@ class TeamCreateTest(_AuthenticatedAPITest):
self.assertEqual(decoded["sub"], f"team:{tid}")
self.assertEqual(decoded["jti"], str(team.active_jti))
def test_create_sets_request_user_as_owner(self):
tid = uuid.uuid4()
self.client.post(
self.url, {"id": str(tid), "name": "Harper"}, format="json"
)
self.assertEqual(Team.objects.get(pk=tid).owner_id, self.user.id)
def test_same_id_under_other_owner_409s(self):
tid = uuid.uuid4()
# Alice creates the team first.
self.client.post(
self.url, {"id": str(tid), "name": "Harper"}, format="json"
)
# Bob then tries to create with the same id — must be a generic
# conflict, not idempotent and not 200.
self.client.force_authenticate(user=self.other_user)
resp = self.client.post(
self.url, {"id": str(tid), "name": "Bob's"}, format="json"
)
self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT)
# Owner unchanged.
self.assertEqual(Team.objects.get(pk=tid).owner_id, self.user.id)
def test_idempotent_on_same_id_returns_200_without_jwt(self):
tid = uuid.uuid4()
first = self.client.post(
@@ -152,7 +179,11 @@ class TeamDetailTest(_AuthenticatedAPITest):
def setUp(self):
super().setUp()
self.team = Team.objects.create(
id=uuid.uuid4(), name="t", active=True, active_jti=uuid.uuid4()
id=uuid.uuid4(),
name="t",
owner=self.user,
active=True,
active_jti=uuid.uuid4(),
)
TeamWorkspaceAssignment.objects.create(
team=self.team, workspace_id="ws-a"
@@ -197,6 +228,18 @@ class TeamDetailTest(_AuthenticatedAPITest):
resp = self.client.delete(self.url)
self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
def test_get_by_non_owner_returns_404(self):
self.client.force_authenticate(user=self.other_user)
resp = self.client.get(self.url)
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
def test_delete_by_non_owner_returns_404_and_no_op(self):
self.client.force_authenticate(user=self.other_user)
resp = self.client.delete(self.url)
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
# Original team is still active — non-owner couldn't soft-delete it.
self.assertTrue(Team.objects.get(pk=self.team.id).active)
# ---------------------------------------------------------------------------
# PUT /mcp_server/api/teams/{id}/workspaces/
@@ -207,7 +250,11 @@ class TeamWorkspacesTest(_AuthenticatedAPITest):
def setUp(self):
super().setUp()
self.team = Team.objects.create(
id=uuid.uuid4(), name="t", active=True, active_jti=uuid.uuid4()
id=uuid.uuid4(),
name="t",
owner=self.user,
active=True,
active_jti=uuid.uuid4(),
)
self.url = reverse(
"mcp-server-api:team-workspaces",
@@ -308,6 +355,14 @@ class TeamWorkspacesTest(_AuthenticatedAPITest):
)
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
def test_put_by_non_owner_returns_404(self):
self.client.force_authenticate(user=self.other_user)
resp = self.client.put(
self.url, {"workspace_ids": ["ws-x"]}, format="json"
)
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
self.assertEqual(self._ws_ids(), [])
# ---------------------------------------------------------------------------
# POST /mcp_server/api/teams/{id}/rotate/
@@ -319,7 +374,11 @@ class TeamRotateTest(_AuthenticatedAPITest):
super().setUp()
_seed_signing_key()
self.team = Team.objects.create(
id=uuid.uuid4(), name="t", active=True, active_jti=uuid.uuid4()
id=uuid.uuid4(),
name="t",
owner=self.user,
active=True,
active_jti=uuid.uuid4(),
)
self.url = reverse(
"mcp-server-api:team-rotate",
@@ -357,3 +416,11 @@ class TeamRotateTest(_AuthenticatedAPITest):
self.assertEqual(
resp.status_code, status.HTTP_503_SERVICE_UNAVAILABLE
)
def test_rotate_by_non_owner_returns_404(self):
before = self.team.active_jti
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.team.refresh_from_db()
self.assertEqual(self.team.active_jti, before)

24
mnemosyne/mcp_server/tests/test_teams_jwt.py
View File

@@ -12,6 +12,7 @@ import time
import uuid
import jwt as pyjwt
from django.contrib.auth import get_user_model
from django.test import TestCase
from mcp_server.models import MCPSigningKey, Team
@@ -24,10 +25,11 @@ def _make_key(kid: str = "k", is_active: bool = True) -> MCPSigningKey:
)
def _make_team(**overrides) -> Team:
def _make_team(owner, **overrides) -> Team:
data = dict(
id=uuid.uuid4(),
name="t",
owner=owner,
active=True,
active_jti=uuid.uuid4(),
)
@@ -36,9 +38,15 @@ def _make_team(**overrides) -> Team:
class MintTeamJWTHappyPathTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.owner = get_user_model().objects.create_user(
username="alice", password="pw"
)
def setUp(self):
self.key = _make_key("k-1")
self.team = _make_team()
self.team = _make_team(self.owner)
def test_returns_signed_jwt_string(self):
token = mint_team_jwt(self.team)
@@ -88,15 +96,21 @@ class MintTeamJWTHappyPathTest(TestCase):
class MintTeamJWTFailureModesTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.owner = get_user_model().objects.create_user(
username="alice", password="pw"
)
def test_no_signing_key_raises(self):
team = _make_team()
team = _make_team(self.owner)
with self.assertRaises(TeamJWTError) as ctx:
mint_team_jwt(team)
self.assertIn("signing", str(ctx.exception).lower())
def test_missing_active_jti_raises(self):
_make_key()
team = _make_team(active_jti=None)
team = _make_team(self.owner, active_jti=None)
with self.assertRaises(TeamJWTError) as ctx:
mint_team_jwt(team)
# Should name the thing the caller forgot to do.
@@ -106,6 +120,6 @@ class MintTeamJWTFailureModesTest(TestCase):
MCPSigningKey.objects.create(
kid="broken", secret_hex="not-hex!!", is_active=True
)
team = _make_team()
team = _make_team(self.owner)
with self.assertRaises(TeamJWTError):
mint_team_jwt(team)