Extract Layer 1 to cores/mizan-python (mizan-core)

Pull cache/keys.py (HMAC cache key derivation) and mwt.py (Mizan Web Token)
out of backends/mizan-django and into a new cores/mizan-python package.
mizan-django re-imports them via the new mizan_core module.

Naming: directory cores/mizan-python/, distribution mizan-core, importable
module mizan_core. mizan-django keeps its existing 'mizan' distribution slot
on PyPI; the two coexist as distinct packages.

Wiring:
- backends/mizan-django/pyproject.toml gains a 'mizan-core' dep with a
  [tool.uv.sources] path entry (editable install from ../../cores/mizan-python).
- Makefile install target prepends 'cd cores/mizan-python && uv pip install -e .'
- 3 import sites in mizan-django updated: cache/__init__.py, jwt/functions.py,
  client/executor.py — all now import from mizan_core.

Test split:
- 3 unit-test classes (CacheKeyDerivationTests, MWTCreationTests,
  PermissionKeyTests) move to cores/mizan-python/tests/, rewritten against
  unittest.TestCase (no Django dep). The cross-language pin test (pinned
  HMAC hex digests against mizan-ts) moves with CacheKeyDerivationTests.
- Integration tests stay in mizan-django (CacheBackendTests, CachePurgeTests,
  CacheIntegrationTests, RevParameterTests, MWTAuthIntegrationTests) — they
  need the Django request flow.

Verified:
- mizan-core: 15/15 pass (incl. cross-language pin)
- mizan-django: 348 pass, 21 skip, 0 fail
- mizan-ts: edge-compat 34/34 pass — protocol invariant holds, the moved
  Python derive_cache_key still produces the exact hex digests TS pins against.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-06 01:32:54 -04:00
parent 9d2781b52c
commit 37e61c646b
14 changed files with 202 additions and 192 deletions

View File

View File

@@ -0,0 +1,68 @@
"""Unit tests for cache key derivation. Includes the cross-language pin against mizan-ts."""
from unittest import TestCase
from mizan_core.cache.keys import derive_cache_key
class CacheKeyDerivationTests(TestCase):
"""Tests that HMAC cache key derivation is deterministic and correct."""
SECRET = "test-cache-secret"
def test_deterministic_output(self):
"""Same inputs always produce the same key."""
key1 = derive_cache_key(self.SECRET, "user", {"user_id": "5"})
key2 = derive_cache_key(self.SECRET, "user", {"user_id": "5"})
self.assertEqual(key1, key2)
self.assertTrue(key1.startswith("ctx:user:"))
self.assertEqual(len(key1), len("ctx:user:") + 64) # prefix + SHA-256 hex
def test_param_order_irrelevant(self):
"""Parameter ordering does not affect the key."""
key1 = derive_cache_key(self.SECRET, "ctx", {"a": "1", "b": "2"})
key2 = derive_cache_key(self.SECRET, "ctx", {"b": "2", "a": "1"})
self.assertEqual(key1, key2)
def test_different_user_ids_different_keys(self):
"""Different user_ids produce different cache keys."""
key1 = derive_cache_key(self.SECRET, "user", {"user_id": "5"}, user_id="5")
key2 = derive_cache_key(self.SECRET, "user", {"user_id": "5"}, user_id="6")
self.assertNotEqual(key1, key2)
def test_rev_changes_key(self):
"""Different rev values produce different cache keys."""
key1 = derive_cache_key(self.SECRET, "user", {"user_id": "5"}, rev=0)
key2 = derive_cache_key(self.SECRET, "user", {"user_id": "5"}, rev=1)
self.assertNotEqual(key1, key2)
def test_no_delimiter_collision(self):
"""JSON-canonical form prevents delimiter-free concatenation collisions."""
# "user" + user_id="12" + params="3" vs "user1" + user_id="2" + params="3"
key1 = derive_cache_key(self.SECRET, "user", {"id": "3"}, user_id="12")
key2 = derive_cache_key(self.SECRET, "user1", {"id": "3"}, user_id="2")
self.assertNotEqual(key1, key2)
def test_public_vs_user_scoped(self):
"""Public (no user_id) and user-scoped produce different keys."""
public = derive_cache_key(self.SECRET, "products", {"id": "1"})
scoped = derive_cache_key(self.SECRET, "products", {"id": "1"}, user_id="5")
self.assertNotEqual(public, scoped)
def test_cross_language_pin(self):
"""Pinned HMAC values — must match TypeScript adapter exactly."""
pin_secret = "test-pin-secret-that-is-32bytes!"
public_key = derive_cache_key(pin_secret, "user", {"user_id": "5"}, rev=0)
self.assertEqual(
public_key,
"ctx:user:605a1ca5ad5994e9b765c8d1b330474c2a0d51a7b8fbbdc402f992da7ba902f6",
)
user_scoped_key = derive_cache_key(
pin_secret, "user", {"user_id": "5"}, user_id="5", rev=0,
)
self.assertEqual(
user_scoped_key,
"ctx:user:30fc08eb46ee4ff2cf7d317e97dca90fd616511e0587304416f71dc863338dc2",
)

View File

@@ -0,0 +1,97 @@
"""Unit tests for MWT creation, decoding, and permission key derivation."""
from unittest import TestCase
from unittest.mock import MagicMock
from mizan_core.mwt import (
MWTUser,
compute_permission_key,
create_mwt,
decode_mwt,
)
def _make_user(**kwargs):
user = MagicMock()
user.pk = kwargs.get("pk", 1)
user.is_staff = kwargs.get("is_staff", False)
user.is_superuser = kwargs.get("is_superuser", False)
user.get_all_permissions = MagicMock(return_value=kwargs.get("perms", set()))
return user
class MWTCreationTests(TestCase):
"""Tests for MWT creation and decoding."""
SECRET = "test-mwt-secret-that-is-32bytes!"
def test_create_and_decode(self):
"""Create an MWT and decode it successfully."""
user = _make_user(pk=42, is_staff=True)
token = create_mwt(user, self.SECRET, ttl=300)
payload = decode_mwt(token, self.SECRET)
self.assertIsNotNone(payload)
self.assertEqual(payload.sub, "42")
self.assertTrue(payload.staff)
self.assertFalse(payload.super)
self.assertEqual(payload.kid, "v1")
self.assertEqual(len(payload.pkey), 64)
def test_decode_expired(self):
"""Expired MWT returns None."""
user = _make_user()
token = create_mwt(user, self.SECRET, ttl=-1)
payload = decode_mwt(token, self.SECRET)
self.assertIsNone(payload)
def test_decode_wrong_secret(self):
"""MWT signed with wrong secret returns None."""
user = _make_user()
token = create_mwt(user, self.SECRET)
payload = decode_mwt(token, "wrong-secret")
self.assertIsNone(payload)
def test_decode_wrong_audience(self):
"""MWT with wrong audience returns None."""
user = _make_user()
token = create_mwt(user, self.SECRET, audience="app1")
payload = decode_mwt(token, self.SECRET, audience="app2")
self.assertIsNone(payload)
def test_mwt_user_has_pkey(self):
"""MWTUser carries the permission key."""
user = _make_user(pk=5, perms={"app.view_thing"})
token = create_mwt(user, self.SECRET)
payload = decode_mwt(token, self.SECRET)
mwt_user = MWTUser(payload)
self.assertEqual(mwt_user.pk, 5)
self.assertTrue(mwt_user.is_authenticated)
self.assertEqual(len(mwt_user.pkey), 64)
class PermissionKeyTests(TestCase):
"""Tests for pkey determinism and sensitivity."""
def test_deterministic(self):
"""Same permissions produce same pkey."""
user = _make_user(perms={"app.view_thing", "app.add_thing"})
pkey1 = compute_permission_key(user)
pkey2 = compute_permission_key(user)
self.assertEqual(pkey1, pkey2)
def test_changes_on_permission_change(self):
"""Different permissions produce different pkey."""
user1 = _make_user(perms={"app.view_thing"})
user2 = _make_user(perms={"app.view_thing", "app.add_thing"})
self.assertNotEqual(compute_permission_key(user1), compute_permission_key(user2))
def test_changes_on_staff_change(self):
"""Staff status change produces different pkey."""
user_normal = _make_user(is_staff=False)
user_staff = _make_user(is_staff=True)
self.assertNotEqual(
compute_permission_key(user_normal),
compute_permission_key(user_staff),
)