feat(mcp): store MCP tokens as SHA-256 hashes instead of plaintext

Replace plaintext token storage with SHA-256 hashes so leaked database
contents cannot be used to authenticate. Plaintext is generated, shown
once at creation time, and never persisted.

- Add `hash_token()` helper and `MCPTokenManager.create_token()` that
  returns `(instance, plaintext)`.
- Replace `token` field with indexed `token_hash`; look up bearers by
  hashing the incoming value.
- Update dashboard, management command, and admin to surface plaintext
  only at creation. Disable admin "add" since it cannot reveal plaintext.
- Migration drops the old `token` column and adds `token_hash`;
  pre-existing tokens are invalidated and must be reissued.
This commit is contained in:
2026-04-27 09:01:36 -04:00
parent 2df22941d2
commit 81426327bf
24 changed files with 950 additions and 50 deletions

View File

@@ -17,16 +17,18 @@ class ResolveMCPUserTest(TestCase):
self.user = User.objects.create_user(
username="bob", email="bob@example.com", password="pw"
)
self.token = MCPToken.objects.create(user=self.user, name="t")
self.token, self.plaintext = MCPToken.objects.create_token(
user=self.user, name="t"
)
def test_resolves_valid_token(self):
user, token = resolve_mcp_user(self.token.token)
user, token = resolve_mcp_user(self.plaintext)
self.assertEqual(user.pk, self.user.pk)
self.assertEqual(token.pk, self.token.pk)
def test_records_usage(self):
self.assertIsNone(self.token.last_used_at)
resolve_mcp_user(self.token.token)
resolve_mcp_user(self.plaintext)
self.token.refresh_from_db()
self.assertIsNotNone(self.token.last_used_at)
@@ -38,16 +40,31 @@ class ResolveMCPUserTest(TestCase):
self.token.is_active = False
self.token.save()
with self.assertRaises(MCPAuthError):
resolve_mcp_user(self.token.token)
resolve_mcp_user(self.plaintext)
def test_expired_token_raises(self):
self.token.expires_at = timezone.now() - timedelta(hours=1)
self.token.save()
with self.assertRaises(MCPAuthError):
resolve_mcp_user(self.token.token)
resolve_mcp_user(self.plaintext)
def test_disabled_user_raises(self):
self.user.is_active = False
self.user.save()
with self.assertRaises(MCPAuthError):
resolve_mcp_user(self.token.token)
resolve_mcp_user(self.plaintext)
def test_plaintext_not_in_db(self):
# Defense in depth: scan every column for the plaintext value.
from django.db import connection
plaintext = self.plaintext
with connection.cursor() as cur:
cur.execute("SELECT * FROM mcp_server_mcptoken")
rows = cur.fetchall()
for row in rows:
for value in row:
self.assertNotEqual(
value, plaintext,
f"Plaintext token leaked into the database: {value!r}",
)

View File

@@ -0,0 +1,58 @@
"""Form tests for the MCP token dashboard."""
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
User = get_user_model()
class CreateFormTest(TestCase):
def test_required_fields(self):
form = MCPTokenCreateForm(data={})
self.assertFalse(form.is_valid())
self.assertIn("name", form.errors)
def test_name_only_is_valid(self):
form = MCPTokenCreateForm(data={"name": "Test"})
self.assertTrue(form.is_valid(), form.errors)
def test_tool_choices_match_registered_tools(self):
form = MCPTokenCreateForm()
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"}:
self.assertIn(expected, choices)
class EditFormTest(TestCase):
def setUp(self):
self.user = User.objects.create_user(username="alice", password="pw")
self.token, _ = MCPToken.objects.create_token(
user=self.user, name="t", allowed_tools=["search"]
)
def test_initial_allowed_tools_populated(self):
form = MCPTokenEditForm(instance=self.token)
self.assertEqual(form.fields["allowed_tools"].initial, ["search"])
def test_save_updates_metadata(self):
form = MCPTokenEditForm(
data={
"name": "Renamed",
"is_active": False,
"expires_at": "",
"allowed_tools": ["search", "get_chunk"],
},
instance=self.token,
)
self.assertTrue(form.is_valid(), form.errors)
instance = form.save(commit=False)
instance.allowed_tools = form.cleaned_data["allowed_tools"]
instance.save()
self.token.refresh_from_db()
self.assertEqual(self.token.name, "Renamed")
self.assertFalse(self.token.is_active)
self.assertEqual(self.token.allowed_tools, ["search", "get_chunk"])

View File

@@ -6,7 +6,7 @@ from django.contrib.auth import get_user_model
from django.test import TestCase
from django.utils import timezone
from mcp_server.models import MCPToken
from mcp_server.models import MCPToken, hash_token
User = get_user_model()
@@ -17,21 +17,33 @@ class MCPTokenModelTest(TestCase):
username="alice", email="alice@example.com", password="pw"
)
def test_token_auto_generated(self):
token = MCPToken.objects.create(user=self.user, name="t")
self.assertTrue(token.token)
self.assertGreater(len(token.token), 20)
def test_create_token_returns_plaintext_and_stores_hash(self):
token, plaintext = MCPToken.objects.create_token(user=self.user, name="t")
self.assertTrue(plaintext)
self.assertGreater(len(plaintext), 20)
# Database stores hash, not plaintext
self.assertEqual(len(token.token_hash), 64)
self.assertNotEqual(token.token_hash, plaintext)
self.assertEqual(token.token_hash, hash_token(plaintext))
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")
self.assertNotIn(plaintext, token.token_hash)
def test_active_token_is_valid(self):
token = MCPToken.objects.create(user=self.user, name="t")
token, _ = MCPToken.objects.create_token(user=self.user, name="t")
self.assertTrue(token.is_valid)
def test_inactive_token_not_valid(self):
token = MCPToken.objects.create(user=self.user, name="t", is_active=False)
token, _ = MCPToken.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, _ = MCPToken.objects.create_token(
user=self.user,
name="t",
expires_at=timezone.now() - timedelta(hours=1),
@@ -39,25 +51,27 @@ class MCPTokenModelTest(TestCase):
self.assertFalse(token.is_valid)
def test_unrestricted_permits_all(self):
token = MCPToken.objects.create(user=self.user, name="t")
token, _ = MCPToken.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, _ = MCPToken.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(user=self.user, name="t")
token, _ = MCPToken.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(self):
token = MCPToken.objects.create(user=self.user, name="t")
def test_masked_token_is_hash_prefix(self):
token, plaintext = MCPToken.objects.create_token(user=self.user, name="t")
masked = token.get_masked_token()
self.assertTrue(masked.endswith(token.token[-8:]))
self.assertIn("*", masked)
self.assertTrue(masked.startswith("mcp_…"))
self.assertIn(token.token_hash[:8], masked)
# Plaintext must never leak through the masked display
self.assertNotIn(plaintext, masked)

View File

@@ -0,0 +1,177 @@
"""View tests for the MCP 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
User = get_user_model()
class TokenListViewTest(TestCase):
def setUp(self):
self.user = User.objects.create_user(
username="alice", email="alice@example.com", password="pw"
)
self.url = reverse("mcp_server:mcp-token-list")
def test_login_required(self):
resp = self.client.get(self.url)
self.assertEqual(resp.status_code, 302)
self.assertIn("/login/", resp.url)
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")
self.client.force_login(self.user)
resp = self.client.get(self.url)
self.assertContains(resp, "mine")
self.assertNotContains(resp, "theirs")
def test_empty_state(self):
self.client.force_login(self.user)
resp = self.client.get(self.url)
self.assertContains(resp, "No MCP 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")
def test_get_renders_form(self):
resp = self.client.get(self.url)
self.assertEqual(resp.status_code, 200)
self.assertContains(resp, "Generate MCP Token")
def test_post_creates_token_and_shows_plaintext_once(self):
resp = self.client.post(self.url, {"name": "Claude Desktop"})
self.assertEqual(resp.status_code, 200)
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")
self.assertNotContains(resp, token.token_hash) # hash is not what we display
# And the detail page never renders the plaintext.
body = resp.content.decode()
# Find the plaintext from the response: the only long alphanumeric
# block inside the #mcp-plaintext div.
import re
m = re.search(r'id="mcp-plaintext">([A-Za-z0-9_\-]+)<', body)
self.assertIsNotNone(m, "plaintext block not found in response")
plaintext = m.group(1)
# Sanity: round-tripping the plaintext through hash_token reproduces
# what's stored.
from mcp_server.models import hash_token
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])
)
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)
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")
def test_renders_token(self):
resp = self.client.get(
reverse("mcp_server:mcp-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")
resp = self.client.get(
reverse("mcp_server:mcp-token-detail", args=[other_token.pk])
)
self.assertEqual(resp.status_code, 404)
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")
def test_post_updates_metadata(self):
resp = self.client.post(
reverse("mcp_server:mcp-token-edit", args=[self.token.pk]),
{
"name": "Renamed",
"is_active": "on",
"expires_at": "",
"allowed_tools": ["search"],
},
)
self.assertEqual(resp.status_code, 302)
self.token.refresh_from_db()
self.assertEqual(self.token.name, "Renamed")
self.assertEqual(self.token.allowed_tools, ["search"])
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")
resp = self.client.post(
reverse("mcp_server:mcp-token-edit", args=[other_token.pk]),
{"name": "hacked"},
)
self.assertEqual(resp.status_code, 404)
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")
def test_revoke_sets_inactive_keeps_row(self):
url = reverse("mcp_server:mcp-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())
def test_get_not_allowed(self):
url = reverse("mcp_server:mcp-token-revoke", args=[self.token.pk])
resp = self.client.get(url)
self.assertEqual(resp.status_code, 405)
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")
def test_delete_removes_row(self):
url = reverse("mcp_server:mcp-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())
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])
resp = self.client.post(url)
self.assertEqual(resp.status_code, 404)
self.assertTrue(MCPToken.objects.filter(pk=other_token.pk).exists())