feat(chat): encrypt group chat under per-device epoch keys (MNP 2.0)

Chat messages are sealed with AES-256-GCM under a key derived per group, per
epoch, per *device*, and signed over the ciphertext with the device key the
node pinned. The node relays and archives; it cannot read a message.

There is no switch. MNP goes to 2.0 and MNP_MIN_SUPPORTED moves with it, so a
1.x peer is refused at the handshake with `version_too_old` rather than
admitted and then unable to speak. An opt-in flag was designed and rejected:
every node is a test node, so it would have bought nothing and left a plaintext
branch reachable — C6's lesson one feature later. A test reads the source and
refuses any code that consults a `chat_encrypted` setting.

Not Sender Keys, and `senderkeys.py` is now documented as unused. With
distribution under the group key and a node that serves history to devices
which were not present, the node must retain each chain's earliest key, and a
chain key at iteration i yields every message key from i on by pure HKDF —
forward secrecy is zero either way. What the ratchet was left buying was
stateful client code with silent failure modes, three of them reproduced: any
member could sign as any other, a second device dropped the first's chain, and
the skipped-key cache grew without bound. The reasoning is in
docs/chat-sender-keys.md, which is the specification and the decision record.

Epochs, not rotation: the epoch key is wrapped under the group key at delivery
and never stored under it, so `gek_rotate` is a re-wrap. A group-key-derived
archive key would have made every message ever sent unreadable on the first
`member unpin`, which is the documented step after removing a member. A new
epoch opens on member revoke/unpin, device revoke and `gek_rotate`; old epochs
are kept and still delivered, so history stays readable to everyone who could
already read it, and nothing anywhere deletes one.

Three prerequisites this needed, each a live defect on its own:

* The peer registry was keyed by user_id, so one account's second device
  evicted the first and the broadcast skipped recipients by account — a
  person's phone never saw what they typed on their laptop.
* The handshake authenticated an account, never a device. `device_hello`
  (additive, signed, refused unless the key is a live device of this account in
  the node's own roster) is what lets the node refuse a member claiming
  somebody else's key.
* `_admin_exec_file_delete` authorized against the exact uploading key, so
  device linking had already broken deleting your own file from your other
  device. It now authorizes against any non-revoked device of `uploader_id`.

Found by driving the real panel over the real transport, not by reading source:
`chat_keys_resp` was routed by arrival order and handed to an unanswered
`media_meta_req` — the original frozen-tab defect in a message type that did
not exist when that probe was written. And `_asText` had been deleted with an
unrelated helper beside it; its only caller sits inside a promise the panel
catches, so every conversation rendered empty with nothing in the console.

Existing node data is migrated by QE/migration/migrate_chat_encryption.py
(not versioned, per the QE rule), run with the node stopped.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TZZxYjz8YeWRz13xDi8LJr

1 files changed, 248 insertions, 1 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index a10504e..1bad487 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -24,12 +24,18 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import time as _time
 import re
 from dataclasses import asdict
 from pathlib import Path
 from typing import Any
 
-from meshbay_common.crypto import generate_gek, wrap_gek_aes
+from meshbay_common.chatbox import new_epoch_key
+from meshbay_common.crypto import (
+    generate_gek,
+    unwrap_gek_aes,
+    wrap_gek_aes,
+)
 from meshbay_node.config import DEFAULT_CONFIG_PATH
 from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
 from meshbay_node.roots import RootError, RootSet
@@ -262,6 +268,247 @@ async def unpin_member(state: dict, user_id: str) -> dict:
     return {"status": "unpinned", "user_id": user_id}
 
 
+# ── Chat epoch keys ──────────────────────────────────────────────────────────
+#
+# The key a group's chat archive is encrypted under. Generated here, by the
+# node, and never by a member — the C5b rule is about key material arriving from
+# outside, and this is the same rule that lets `gek_rotate` be a signed
+# instruction rather than a delivery.
+#
+# An *epoch* rather than a rotation, and the distinction is the whole design:
+# opening a new one stops a departing member reading what comes next, while
+# every earlier epoch is kept and still delivered to current members, so the
+# history they could already read stays readable. Rotating instead — replacing
+# the key, as `set_gek` does — would make every message anyone ever sent
+# permanently unreadable to everybody, which is what a plain GEK-derived
+# archive key would have done on the very first `member unpin`
+# (docs/chat-sender-keys.md F4).
+
+
+async def _wrap_for_node(state: dict, key: bytes) -> dict:
+    """
+    Wrap a key to the node's own X25519 key, the way `set_gek` does for the GEK.
+
+    Wrapped, not raw: the claim chat encryption makes is against someone who
+    obtains the node's storage *without the keystore password*, and the node's
+    X25519 private key is what the keystore protects. A raw key in SQLite would
+    leave nothing behind that claim.
+    """
+    pk_x_node_raw = state.get("pk_x25519_raw")
+    if not pk_x_node_raw:
+        raise OpError("Node identity not available", status=503)
+    return wrap_gek_aes(key, pk_x_node_raw)
+
+
+async def chat_epoch_keys(state: dict, group_id: str) -> list[dict]:
+    """
+    Every chat epoch key this group has, oldest first, in the clear *in memory*.
+
+    Cached on the group context: unwrapping is an ECIES operation per epoch and
+    this is on the path of every member connecting to a group with chat on.
+    """
+    ctx = _group_ctx(state, group_id)
+    cached = ctx.get("chat_epoch_keys")
+    if cached is not None:
+        return cached
+
+    bundle_store = state.get("bundle_store")
+    if not bundle_store:
+        raise OpError("Bundle store not available", status=503)
+    sk_x_raw = state.get("sk_x25519_raw")
+    pk_x_raw = state.get("pk_x25519_raw")
+    if not (sk_x_raw and pk_x_raw):
+        raise OpError("Node identity not available", status=503)
+
+    keys: list[dict] = []
+    for row in await bundle_store.fetch_chat_epochs(group_id):
+        try:
+            keys.append({"epoch": row["epoch"],
+                         "key": unwrap_gek_aes(row, sk_x_raw, pk_x_raw)})
+        except Exception as e:
+            # Loud, and not fatal: one unreadable epoch must not take the
+            # readable ones with it. The messages of that epoch are lost, which
+            # is a thing the operator needs told rather than a thing to hide.
+            log.error("chat: epoch %d of group %s will not unwrap (%s) — "
+                      "its messages are unreadable", row["epoch"],
+                      group_id[:8], e)
+    ctx["chat_epoch_keys"] = keys
+    return keys
+
+
+async def open_chat_epoch(state: dict, group_id: str) -> dict:
+    """
+    Open a new chat epoch. Idempotent only in the sense that it always adds one.
+
+    Called when the set of devices that may read *future* messages shrinks: a
+    member removed, a device revoked or unpinned, the group key rotated, or the
+    operator asking directly. Never on a schedule — an epoch nobody needed is an
+    epoch key the node has to keep for ever.
+    """
+    bundle_store = state.get("bundle_store")
+    if not bundle_store:
+        raise OpError("Bundle store not available", status=503)
+
+    epoch = await bundle_store.latest_chat_epoch(group_id) + 1
+    key = new_epoch_key()
+    wrapped = await _wrap_for_node(state, key)
+    await bundle_store.store_chat_epoch(
+        group_id, epoch, wrapped["pk_eph_b64"], wrapped["nonce_b64"],
+        wrapped["wrapped_b64"])
+
+    # Tolerant of a group context that does not exist yet: the daemon opens the
+    # first epoch **while it is building** `groups_ctx`, before publishing it on
+    # the state, because a group with no epoch key is a group nobody can speak
+    # in. Insisting on the context here would make start-up the one moment this
+    # cannot be called.
+    ctx = (state.get("groups_ctx") or {}).get(group_id)
+    if ctx is not None:
+        cached = ctx.get("chat_epoch_keys")
+        if cached is not None:
+            cached.append({"epoch": epoch, "key": key})
+        ctx["chat_epoch"] = epoch
+
+    # The transports hold their own view of the group, exactly as `set_gek`
+    # notes: an epoch that did not reach them would have members sealing under
+    # a key the node no longer thinks is current.
+    for transport_key in ("webrtc", "quic_server"):
+        transport = state.get(transport_key)
+        groups = getattr(transport, "_ctx", {}).get("groups") if transport else None
+        if groups and group_id in groups:
+            groups[group_id]["chat_epoch"] = epoch
+            groups[group_id].pop("chat_epoch_keys", None)
+
+    log.info("Chat epoch %d opened for group %s", epoch, group_id[:8])
+    return {"epoch": epoch}
+
+
+async def ensure_chat_epoch(state: dict, group_id: str) -> int:
+    """The current epoch, opening the first one if the group has none."""
+    bundle_store = state.get("bundle_store")
+    if not bundle_store:
+        raise OpError("Bundle store not available", status=503)
+    epoch = await bundle_store.latest_chat_epoch(group_id)
+    if epoch:
+        return epoch
+    return (await open_chat_epoch(state, group_id))["epoch"]
+
+
+async def chat_status(state: dict, group_id: str) -> dict:
+    """What the operator needs to decide anything about this group's chat."""
+    ctx = _group_ctx(state, group_id)
+    bundle_store = state.get("bundle_store")
+    store = ctx.get("chat_store")
+    plain = sealed = 0
+    if store is not None:
+        plain, sealed = await store.count_by_format()
+    return {
+        "group_id": group_id,
+        "epoch": (await bundle_store.latest_chat_epoch(group_id)
+                  if bundle_store else 0),
+        # Rows written before MNP 2.0. Not a state the node can be *in* — chat
+        # is always encrypted now — but a state its disk can be in until
+        # `chat encrypt-history` has run, and the operator has to be told,
+        # because those messages are the ones still readable off a stolen disk.
+        "plaintext_messages": plain,
+        "encrypted_messages": sealed,
+    }
+
+
+async def encrypt_chat_history(state: dict, group_id: str) -> dict:
+    """
+    Re-encrypt the messages written before this group turned encryption on.
+
+    Deliberately **not** done by the switch. It rewrites the only copy of a
+    conversation, and a toggle that does that is one somebody flips twice; this
+    is an explicit command, it copies the database first, and it runs in one
+    transaction.
+
+    The node can do this at all only because it holds those rows in plaintext —
+    it is the last moment at which anyone can. Afterwards nothing on this
+    machine can read them without an epoch key.
+
+    Messages are sealed under a **synthetic device** belonging to the node, not
+    under the original sender's key: the node does not hold anyone's signing key
+    and must not pretend to. They are marked as such, so a reader is told these
+    carry the node's word for who wrote them — which is all they ever carried,
+    since they were written before signing existed.
+    """
+    import shutil
+
+    from meshbay_common.chatbox import seal
+
+    ctx = _group_ctx(state, group_id)
+    store = ctx.get("chat_store")
+    if store is None:
+        raise OpError("This group has no chat store", status=404)
+
+    epoch = await ensure_chat_epoch(state, group_id)
+    keys = {k["epoch"]: k["key"] for k in await chat_epoch_keys(state, group_id)}
+    key = keys.get(epoch)
+    if not key:
+        raise OpError("No chat key for this group", status=503)
+
+    sk_node = state.get("sk_node")
+    if sk_node is None:
+        raise OpError("Node identity not available", status=503)
+    from cryptography.hazmat.primitives import serialization
+
+    device_raw = sk_node.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    import base64 as _b64
+
+    device_b64 = _b64.b64encode(device_raw).decode()
+
+    backup = store.db_path.with_name(
+        f"{store.db_path.name}.bak-{int(_time.time())}")
+    shutil.copy2(store.db_path, backup)
+
+    converted = 0
+    for row in await store.all_plaintext():
+        text = (row.payload.decode("utf-8", errors="replace")
+                if isinstance(row.payload, bytes) else str(row.payload))
+        env = seal(key, group_id, epoch, device_b64, device_raw, sk_node, {
+            "text": text,
+            "thread_id": row.thread_id,
+            "sender_name": row.sender_name,
+            "sent_at": int(row.timestamp),
+            # The node sealed this after the fact; it did not witness it being
+            # signed. Said in the payload rather than inferred from the device.
+            "migrated": True,
+        })
+        await store.reseal(row.id, epoch=epoch, device=device_raw,
+                           nonce=env["nonce"], ct=env["ct"], sig=env["sig"])
+        converted += 1
+    await store.commit()
+
+    log.info("Chat history re-encrypted for group %s: %d message(s), backup %s",
+             group_id[:8], converted, backup.name)
+    return {"group_id": group_id, "converted": converted,
+            "backup": str(backup), "epoch": epoch}
+
+
+async def prune_chat(state: dict, group_id: str, max_age_days: int) -> dict:
+    """
+    Delete messages older than `max_age_days`. Epoch keys are never touched.
+
+    An epoch whose messages have all aged out costs 32 bytes and keeps the
+    operation reversible in the only direction that matters: nothing that is
+    still stored becomes unreadable because something else was deleted.
+    """
+    ctx = _group_ctx(state, group_id)
+    store = ctx.get("chat_store")
+    if store is None:
+        raise OpError("This group has no chat store", status=404)
+    if max_age_days < 1:
+        raise OpError("max_age_days must be at least 1", status=400)
+    removed = await store.delete_older_than(
+        _time.time() - max_age_days * 86400)
+    log.info("Chat retention for group %s: %d message(s) removed",
+             group_id[:8], removed)
+    return {"group_id": group_id, "removed": removed,
+            "max_age_days": max_age_days}
+
+
 # ── Group keys ───────────────────────────────────────────────────────────────
 
 async def set_gek(state: dict, group_id: str, *, rotate: bool = False) -> dict: