"""Django ORM models for the MCP server app. This module defines every Postgres-backed row the Mnemosyne MCP surface relies on: * :class:`UserToken` — opaque bearer tokens (SHA-256 hashed at rest); used on both the MCP surface and the Daedalus DRF REST surface. * :class:`MCPSigningKey` — HMAC signing keys (``HS256``) for JWTs, keyed by ``kid``. Used by the legacy per-turn path *and* by team JWTs minted in §7 of ``DAEDALUS_PALLAS_INTEGRATION_v1.md``. * :class:`LibraryMembership` — Postgres-side role membership for Neo4j-resident Libraries. Referenced by Library ``uid`` string because Library is a neomodel ``StructuredNode``, not a Django ORM model. * :class:`Team` — Pallas deployment identity inside Mnemosyne. Stable UUID = ``PallasInstance.id`` on the Daedalus side. * :class:`TeamWorkspaceAssignment` — which Daedalus workspaces a given team is allowed to see. Queried live on every request so revocation via ``DELETE`` / ``PUT /workspaces/`` is instantaneous. See ``mnemosyne/docs/DAEDALUS_PALLAS_INTEGRATION_v1.md`` for the complete credential / authorization model. """ from __future__ import annotations import hashlib import secrets import uuid from django.conf import settings from django.db import models from django.utils import timezone def hash_token(plaintext: str) -> str: """SHA-256 hex digest of an MCP bearer token. 64 chars.""" return hashlib.sha256(plaintext.encode("utf-8")).hexdigest() # --------------------------------------------------------------------------- # Library memberships # --------------------------------------------------------------------------- class LibraryMembership(models.Model): """Role of a user on a Neo4j-resident Library. Library lives in Neo4j (``library.models.Library``, a neomodel ``StructuredNode``), so this table joins by the Library's ``uid`` string rather than a foreign key. Consumers that want the Library's live state (name, description, workspace_id, …) must look it up separately via ``Library.nodes.get(uid=…)``. Roles are ordered (owner > manager > reader) but not hierarchical in storage: a user with owner rights is represented by a single row with ``role="owner"``, not multiple rows. Callers deciding whether a user may *grant* a Library into a ``UserToken`` should check for ``role__in=("owner", "manager")``. """ class Role(models.TextChoices): OWNER = "owner", "Owner" MANAGER = "manager", "Manager" READER = "reader", "Reader" user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="library_memberships", ) library_uid = models.CharField(max_length=64, db_index=True) role = models.CharField(max_length=10, choices=Role.choices) created_at = models.DateTimeField(auto_now_add=True) class Meta: # One (user, library_uid, role) triple per row. A user may not # hold both ``manager`` and ``reader`` on the same library — # callers must consolidate to the higher role. constraints = [ models.UniqueConstraint( fields=("user", "library_uid"), name="unique_library_membership_per_user", ) ] indexes = [ models.Index(fields=("library_uid", "role")), ] ordering = ["library_uid", "role"] def __str__(self): # pragma: no cover - trivial return f"{self.user} → {self.library_uid} ({self.role})" # --------------------------------------------------------------------------- # Opaque bearer tokens # --------------------------------------------------------------------------- class UserTokenManager(models.Manager): def create_token( self, *, user, name, allowed_tools=None, allowed_libraries=None, expires_at=None, ): """Generate a new bearer token, store its hash, and return (instance, plaintext). The plaintext is returned exactly once and is never persisted. Callers must surface it to the human and rely on the user to copy it; after this method returns, the plaintext is unrecoverable from the database. """ plaintext = secrets.token_urlsafe(48) instance = self.create( user=user, name=name, token_hash=hash_token(plaintext), allowed_tools=list(allowed_tools or []), allowed_libraries=list(allowed_libraries or []), expires_at=expires_at, ) return instance, plaintext class UserToken(models.Model): """Per-user opaque bearer token for authenticating to Mnemosyne. A single generic credential model used on both surfaces: * **MCP (``/mcp/``)** — third-party tool clients (Claude Desktop, …). ``allowed_tools`` / ``allowed_libraries`` scope the token; an empty ``allowed_libraries`` is fail-closed (the token sees nothing), not an implicit "all". * **DRF REST (``/library/api/*``, ``/mcp_server/api/teams/*``)** — Daedalus calls authenticated as the owning user. The scope fields are ignored on this surface; ``Team.owner`` and ``Library.owner_username`` enforce access. Tokens are hashed at rest (SHA-256, 64-char hex). Plaintext exists only in memory at creation time, on the wire to the client, and in the user's own storage. A leaked database backup discloses no usable credentials. """ user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="api_tokens", ) token_hash = models.CharField(max_length=64, unique=True, db_index=True) name = models.CharField(max_length=100) is_active = models.BooleanField(default=True) expires_at = models.DateTimeField(null=True, blank=True) last_used_at = models.DateTimeField(null=True, blank=True) allowed_tools = models.JSONField(default=list, blank=True) # JSON list of Library.uid strings. Fail-closed: empty → zero libraries # *for MCP callers*; ignored on the REST surface (see class docstring). # We cannot use a ``ManyToManyField(Library)`` because Library is a # neomodel ``StructuredNode`` in Neo4j, not a Django ORM model. allowed_libraries = models.JSONField(default=list, blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) objects = UserTokenManager() class Meta: ordering = ["-created_at"] verbose_name = "API Token" verbose_name_plural = "API Tokens" def __str__(self): return f"{self.name} ({self.user})" @property def is_valid(self) -> bool: if not self.is_active: return False if self.expires_at and self.expires_at < timezone.now(): return False return True def can_use_tool(self, tool_name: str) -> bool: if not self.allowed_tools: return True return tool_name in self.allowed_tools def record_usage(self): self.last_used_at = timezone.now() self.save(update_fields=["last_used_at"]) def get_masked_token(self) -> str: """Token-id-style display for admin and dashboard. Plaintext is unrecoverable, so we display the first 8 chars of the hash prefixed with ``tok_…``. Stable per token, never reveals plaintext. """ return f"tok_…{self.token_hash[:8]}" # --------------------------------------------------------------------------- # Signing keys # --------------------------------------------------------------------------- class MCPSigningKeyManager(models.Manager): def active(self): """Active keys, newest first. Multiple may overlap during rotation.""" return self.filter(is_active=True).order_by("-created_at") def by_kid(self, kid: str): return self.filter(kid=kid).first() def current(self): """Most recently seeded active key — used when minting new tokens.""" return self.filter(is_active=True).order_by("-created_at").first() class MCPSigningKey(models.Model): """HMAC signing key used for every Mnemosyne JWT (``HS256``). Two populations of tokens share this keyring: * **Per-turn JWTs** (legacy, category 2) minted by Daedalus with ``exp`` ≤ 10 minutes. Retired in Phase 4. * **Team JWTs** (category 3) minted by Mnemosyne itself with ``exp`` = 10 years. Signed with whichever ``MCPSigningKey`` was ``objects.current()`` at mint time. Rotation: seed a new active key, distribute the secret to Daedalus (for the per-turn path) and re-issue every team token via ``POST /mcp_server/api/teams/{id}/rotate/`` (for the team path), then flip the old one ``is_active=False``. """ kid = models.CharField(max_length=64, unique=True, db_index=True) secret_hex = models.CharField(max_length=128) # 256-bit secret = 64 hex is_active = models.BooleanField(default=True) created_at = models.DateTimeField(auto_now_add=True) retired_at = models.DateTimeField(null=True, blank=True) note = models.TextField(blank=True) objects = MCPSigningKeyManager() class Meta: ordering = ["-created_at"] def __str__(self): suffix = "active" if self.is_active else "retired" return f"{self.kid} ({suffix})" def retire(self): self.is_active = False self.retired_at = timezone.now() self.save(update_fields=["is_active", "retired_at"]) # --------------------------------------------------------------------------- # Pallas teams # --------------------------------------------------------------------------- class Team(models.Model): """A Pallas deployment as seen by Mnemosyne. ``id`` is the Daedalus ``PallasInstance.id`` UUID and is stable across re-deployments / host moves of the same Pallas instance. ``active_jti`` identifies the single currently-valid team JWT for this team. On ``POST /rotate/`` we generate a new UUID here and re-mint; the previous JWT is invalidated immediately because the auth middleware compares the incoming ``jti`` against this value. ``active=False`` soft-deletes the team — every validation will reject tokens for an inactive team, so revocation survives restart without needing a cache purge. ``owner`` is the Mnemosyne user that created the team. Team management endpoints scope by ``owner`` so that one user cannot manage another user's teams. """ id = models.UUIDField(primary_key=True, editable=False) name = models.CharField(max_length=200) owner = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="teams", ) active = models.BooleanField(default=True) active_jti = models.UUIDField(null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["name"] def __str__(self): suffix = "" if self.active else " [inactive]" return f"{self.name}{suffix}" def rotate_jti(self) -> uuid.UUID: """Install a fresh ``active_jti``. Returns the new UUID.""" self.active_jti = uuid.uuid4() self.save(update_fields=["active_jti", "updated_at"]) return self.active_jti def deactivate(self): """Soft-delete the team. All its JWTs stop validating on next request.""" self.active = False self.active_jti = None self.save(update_fields=["active", "active_jti", "updated_at"]) class TeamWorkspaceAssignment(models.Model): """Grant a team read access to a Daedalus workspace's libraries. Queried live on every request via:: MATCH (l:Library) WHERE l.workspace_id IN $workspace_ids RETURN l.uid so attach/detach is visible to subsequent requests without any cache invalidation. ``workspace_id`` is a plain string (Daedalus owns the namespace) rather than a foreign key, mirroring how ``library.models.IngestJob.library_uid`` references Neo4j state. """ team = models.ForeignKey( Team, on_delete=models.CASCADE, related_name="workspace_assignments", ) workspace_id = models.CharField(max_length=64, db_index=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: constraints = [ models.UniqueConstraint( fields=("team", "workspace_id"), name="unique_team_workspace", ) ] ordering = ["team", "workspace_id"] def __str__(self): # pragma: no cover - trivial return f"{self.team} ↔ {self.workspace_id}"