Verify Python CultMesh authority leases

packages/cultcache-py/PARITY.md

@@ -89,8 +89,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
   store beside the cache file by default.
 - Python CultMesh authority leases:
   Lease objects own validity and peer/role/shard coverage checks; the catalog
-  owns sorted listing, lookup, upsert, and authorization delegation, matching
-  the C# reference authority boundary.
+  owns sorted listing, lookup, upsert, signature verification policy, and
+  authorization delegation, matching the C# reference authority boundary.
+  Callers can require verified signatures and use the packaged HMAC verifier to
+  issue and verify canonical lease payloads keyed by issuer runtime.
 - Python CultMesh local server:
   Serves a `CultMeshNode` plus Verse and peer catalogs over the same framed
   MessagePack request/response lane used by `CultNetRawClient` and

packages/cultcache-py/README.md

@@ -217,11 +217,13 @@ python -m cultnet_py.interop_peer probe --runtime-id python-prober --discovery-p
 ```
 
 `cultmesh_py` includes a local cache-backed node, schema-v0 helpers for the
-CultMesh Verse catalog and peer exchange wire messages, local authority lease
-checks, stream transport negotiation, committed simulation fact documents, and
-local prediction reconciliation:
+CultMesh Verse catalog and peer exchange wire messages, local and verified
+authority lease checks, stream transport negotiation, committed simulation fact
+documents, and local prediction reconciliation:
 
 ```python
+from datetime import UTC, datetime, timedelta
+
 from cultcache_py import define_database_entry_type
 from cultnet_py import (
     CultNetClientAuthorityScope,
@@ -233,7 +235,14 @@ from cultnet_py import (
     CultNetShardReplicator,
     CultNetShardReplicatorOptions,
 )
-from cultmesh_py import CultMesh, CultMeshGameSessionOptions, CultMeshNodeOptions, peer_exchange_request
+from cultmesh_py import (
+    CultMesh,
+    CultMeshAuthorityLease,
+    CultMeshGameSessionOptions,
+    CultMeshHmacAuthorityLeaseVerifier,
+    CultMeshNodeOptions,
+    peer_exchange_request,
+)
 
 note_doc = define_database_entry_type(
     "mesh.note",
@@ -267,6 +276,23 @@ shard_log_wire = node.database.create_shard_log_response(shard_id="interop", sha
 
 peers = CultMesh.create_peer_catalog()
 response = peers.create_response(peer_exchange_request("pex-1", verse_id="local"))
+lease_verifier = CultMeshHmacAuthorityLeaseVerifier({"odin": b"shared-lease-key"})
+lease_catalog = CultMesh.create_authority_lease_catalog(
+    signature_verifier=lease_verifier.verify,
+    require_verified_signatures=True,
+)
+lease_valid_from = datetime.now(UTC)
+lease = lease_verifier.issue(CultMeshAuthorityLease(
+    lease_id="lease:python-runtime",
+    verse_id="local",
+    peer_id="python-runtime",
+    roles=("shard-primary",),
+    valid_from=lease_valid_from,
+    expires_at=lease_valid_from + timedelta(minutes=5),
+    shard_ids=("interop",),
+    issuer_runtime_id="odin",
+))
+lease_catalog.upsert(lease)
 session = CultMesh.create_game_session(
     node,
     CultMeshGameSessionOptions(

packages/cultcache-py/src/cultmesh_py/__init__.py

@@ -21,10 +21,12 @@ from .wire import (
     PEER_EXCHANGE_RESPONSE,
     VERSE_CATALOG_REQUEST,
     VERSE_CATALOG_RESPONSE,
+    AuthorityLeaseVerifier,
     CultMeshPeerCard,
     CultMeshPeerCatalog,
     CultMeshAuthorityLease,
     CultMeshAuthorityLeaseCatalog,
+    CultMeshHmacAuthorityLeaseVerifier,
     CultMeshStreamCatalog,
     CultMeshStreamConsumerProfile,
     CultMeshStreamDescriptor,
@@ -57,6 +59,8 @@ __all__ = [
     "CultMeshSimulationFactCommitter",
     "CultMeshAuthorityLease",
     "CultMeshAuthorityLeaseCatalog",
+    "CultMeshHmacAuthorityLeaseVerifier",
+    "AuthorityLeaseVerifier",
     "CultMeshDiscoveryClient",
     "CultMeshPeerExchangeClient",
     "CultMeshVerseDiscoveryClient",

packages/cultcache-py/src/cultmesh_py/facade.py

@@ -10,6 +10,7 @@ from .server import CultMeshLocalServer
 from .session import CultMeshGameSession, CultMeshGameSessionOptions
 from .simulation import CultMeshSimulationFactCommitter
 from .wire import (
+    AuthorityLeaseVerifier,
     CultMeshAuthorityLeaseCatalog,
     CultMeshPeerCatalog,
     CultMeshStreamCatalog,
@@ -61,8 +62,15 @@ class CultMesh:
         return CultMeshPeerCatalog()
 
     @staticmethod
-    def create_authority_lease_catalog() -> CultMeshAuthorityLeaseCatalog:
-        return CultMeshAuthorityLeaseCatalog()
+    def create_authority_lease_catalog(
+        *,
+        signature_verifier: AuthorityLeaseVerifier | None = None,
+        require_verified_signatures: bool = False,
+    ) -> CultMeshAuthorityLeaseCatalog:
+        return CultMeshAuthorityLeaseCatalog(
+            signature_verifier=signature_verifier,
+            require_verified_signatures=require_verified_signatures,
+        )
 
     @staticmethod
     def create_stream_catalog() -> CultMeshStreamCatalog:

packages/cultcache-py/src/cultmesh_py/wire.py

@@ -1,8 +1,13 @@
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+import base64
+import hmac
+from dataclasses import dataclass, field, replace
 from datetime import datetime
-from typing import Any, Callable
+from hashlib import sha256
+from typing import Any, Callable, Mapping
+
+import msgpack
 
 
 VERSE_CATALOG_REQUEST = "cultmesh.verse_catalog_request.v0"
@@ -242,6 +247,23 @@ class CultMeshAuthorityLease:
         object.__setattr__(self, "roles", _clean_tuple(self.roles))
         object.__setattr__(self, "shard_ids", _clean_tuple(self.shard_ids))
 
+    def signature_payload(self) -> bytes:
+        return msgpack.packb([
+            "cultmesh.authority_lease.v0",
+            self.lease_id,
+            self.verse_id,
+            self.peer_id,
+            list(self.roles),
+            self.valid_from.isoformat(),
+            self.expires_at.isoformat(),
+            list(self.shard_ids),
+            self.issuer_runtime_id,
+        ], use_bin_type=True)
+
+    def with_signature(self, signature: str) -> "CultMeshAuthorityLease":
+        require_non_empty(signature, "signature")
+        return replace(self, signature=signature)
+
     def is_valid_at(self, at: datetime) -> bool:
         return self.valid_from <= at < self.expires_at
 
@@ -265,9 +287,42 @@ class CultMeshAuthorityLease:
         )
 
 
+AuthorityLeaseVerifier = Callable[[CultMeshAuthorityLease], bool]
+
+
+@dataclass(frozen=True)
+class CultMeshHmacAuthorityLeaseVerifier:
+    keys_by_issuer: Mapping[str, bytes | str]
+
+    def sign(self, lease: CultMeshAuthorityLease) -> str:
+        key = self._key_for(lease.issuer_runtime_id)
+        digest = hmac.new(key, lease.signature_payload(), sha256).digest()
+        encoded = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
+        return f"hmac-sha256:{encoded}"
+
+    def issue(self, lease: CultMeshAuthorityLease) -> CultMeshAuthorityLease:
+        return lease.with_signature(self.sign(lease))
+
+    def verify(self, lease: CultMeshAuthorityLease) -> bool:
+        signature = lease.signature or ""
+        if not signature:
+            return False
+        try:
+            expected = self.sign(lease)
+        except KeyError:
+            return False
+        return hmac.compare_digest(signature, expected)
+
+    def _key_for(self, issuer_runtime_id: str) -> bytes:
+        key = self.keys_by_issuer[issuer_runtime_id]
+        return key.encode("utf-8") if isinstance(key, str) else key
+
+
 @dataclass
 class CultMeshAuthorityLeaseCatalog:
     _leases: dict[str, CultMeshAuthorityLease] = field(default_factory=dict)
+    signature_verifier: AuthorityLeaseVerifier | None = None
+    require_verified_signatures: bool = False
 
     @property
     def leases(self) -> tuple[CultMeshAuthorityLease, ...]:
@@ -299,8 +354,19 @@ class CultMeshAuthorityLeaseCatalog:
         lease = self._leases.get(peer.authority_lease_id)
         if lease is None:
             return False
+        if not self._is_verified(lease):
+            return False
         return lease.covers(peer, role, shard_id, at)
 
+    def is_verified(self, lease_id: str) -> bool:
+        lease = self.get(lease_id)
+        return False if lease is None else self._is_verified(lease)
+
+    def _is_verified(self, lease: CultMeshAuthorityLease) -> bool:
+        if self.signature_verifier is None:
+            return not self.require_verified_signatures
+        return self.signature_verifier(lease)
+
 
 ZERO_COPY_TRANSPORTS = {
     "shared-memory",

packages/cultcache-py/tests/test_cultcache.py

@@ -87,6 +87,7 @@ from cultmesh_py import (
     CultMeshSimulationFact,
     CultMeshAuthorityLease,
     CultMeshAuthorityLeaseCatalog,
+    CultMeshHmacAuthorityLeaseVerifier,
     CultMeshDiscoveryClient,
     CultMeshPeerExchangeClient,
     CultMeshStreamCatalog,
@@ -2636,6 +2637,63 @@ class CultCacheTests(unittest.TestCase):
         self.assertFalse(leases.is_authorized(peer, "read-replica", at=now))
         self.assertIsNone(leases.get("missing"))
 
+    def test_cultmesh_authority_lease_catalog_can_require_verified_signatures(self) -> None:
+        peer = CultMeshPeerCard(
+            peer_id="voidbot-local",
+            verse_id="local",
+            endpoints=("cultmesh://localhost",),
+            roles=("shard-primary",),
+            shard_ids=("shard-a",),
+            authority_lease_id="lease:voidbot-local",
+        )
+        now = datetime.now(UTC)
+        unsigned = CultMeshAuthorityLease(
+            lease_id="lease:voidbot-local",
+            verse_id="local",
+            peer_id="voidbot-local",
+            roles=("shard-primary",),
+            valid_from=now - timedelta(seconds=1),
+            expires_at=now + timedelta(seconds=30),
+            shard_ids=("shard-a",),
+            issuer_runtime_id="odin",
+        )
+        verifier = CultMeshHmacAuthorityLeaseVerifier({"odin": b"lease-secret"})
+        signed = verifier.issue(unsigned)
+        strict_leases = CultMesh.create_authority_lease_catalog(
+            signature_verifier=verifier.verify,
+            require_verified_signatures=True,
+        )
+        strict_leases.upsert(unsigned)
+        self.assertFalse(strict_leases.is_verified(unsigned.lease_id))
+        self.assertFalse(strict_leases.is_authorized(peer, "shard-primary", shard_id="shard-a", at=now))
+
+        strict_leases.upsert(signed)
+        self.assertTrue(strict_leases.is_verified(signed.lease_id))
+        self.assertTrue(strict_leases.is_authorized(peer, "shard-primary", shard_id="shard-a", at=now))
+
+        wrong_key = CultMesh.create_authority_lease_catalog(
+            signature_verifier=CultMeshHmacAuthorityLeaseVerifier({"odin": b"wrong-secret"}).verify,
+            require_verified_signatures=True,
+        )
+        wrong_key.upsert(signed)
+        self.assertFalse(wrong_key.is_verified(signed.lease_id))
+        self.assertFalse(wrong_key.is_authorized(peer, "shard-primary", shard_id="shard-a", at=now))
+
+        tampered = CultMeshAuthorityLease(
+            lease_id=signed.lease_id,
+            verse_id=signed.verse_id,
+            peer_id=signed.peer_id,
+            roles=("shard-primary", "read-replica"),
+            valid_from=signed.valid_from,
+            expires_at=signed.expires_at,
+            shard_ids=signed.shard_ids,
+            issuer_runtime_id=signed.issuer_runtime_id,
+            signature=signed.signature,
+        )
+        strict_leases.upsert(tampered)
+        self.assertFalse(strict_leases.is_verified(tampered.lease_id))
+        self.assertFalse(strict_leases.is_authorized(peer, "shard-primary", shard_id="shard-a", at=now))
+
     def test_cultmesh_stream_catalog_negotiates_transport_and_latest_frame(self) -> None:
         streams = CultMeshStreamCatalog()
         streams.declare(

src/GameCult.Networking/Contracts/cultnet-distributed-database.md

@@ -144,8 +144,8 @@ Not implemented yet:
 Add discovery fanout and operational polish:
 
 - add periodic gossip fanout over known peer cards
-- add real signature verification for authority leases
-- add snapshot throttling and size limits for public mesh edges
+- standardize cross-runtime authority lease signature algorithms beyond the
+  package-local Python HMAC verifier
 - add simulation-frame rollback/resimulation helpers around predicted input
   streams
 - add peer-to-peer fanout for observation gossip and candidate propagation