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, 153 insertions, 33 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/chat/store.py b/packages/meshbay-node/src/meshbay_node/chat/store.py
index 6f23905..9e5ee90 100644
--- a/packages/meshbay-node/src/meshbay_node/chat/store.py
+++ b/packages/meshbay-node/src/meshbay_node/chat/store.py
@@ -1,9 +1,26 @@
 """
 MeshBay Node — SQLite-backed chat message store.
 
-One database per group. Stores encrypted Sender Keys messages for offline
-retrieval and history. Messages are stored as received (ciphertext) —
-decryption happens on the client side.
+One database per group. The node is a relay and an archive: it stores what it
+was handed, serves it back, and — once a group has chat encryption switched on —
+cannot read any of it. Decryption happens in the client, which is the only place
+that holds the epoch key (`docs/chat-sender-keys.md` §5).
+
+Three things about the schema are load-bearing rather than incidental:
+
+* **`payload` is bytes, always.** It used to be a UTF-8 string in practice, and
+  the history path decoded it with `errors="replace"` — which substitutes
+  U+FFFD for every byte that is not valid UTF-8, i.e. for most of a ciphertext.
+  That would have corrupted history while live messages worked, which reads as
+  an intermittent decryption bug rather than as a wire-format error.
+* **`format` says how to read a row**, so messages written before a group turned
+  encryption on keep rendering. Nothing is ever rewritten in place by the
+  switch; see `chat encrypt-history` for the explicit, backed-up alternative.
+* **`(device, nonce)` is unique.** The nonce is 96 random bits chosen per
+  message by the sending device, so it is already required to be unique for
+  AES-GCM to be safe — making it a key costs nothing and turns a replayed
+  message (which is validly signed, being a copy of a real one) into an
+  integrity error instead of a duplicate.
 """
 
 import logging
@@ -15,6 +32,17 @@ import aiosqlite
 
 log = logging.getLogger(__name__)
 
+
+class ReplayedMessage(Exception):
+    """This device has already sent a message under this nonce."""
+
+# Plaintext, as every message was before chat encryption existed. Rows keep it
+# for ever; nothing rewrites them.
+FORMAT_PLAIN = 0
+# Sealed under a chat epoch key: `payload` is the AES-256-GCM ciphertext,
+# `nonce` its 96-bit nonce, `sig` the sender device's Ed25519 signature.
+FORMAT_SEALED_V1 = 1
+
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS messages (
     id          INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -29,8 +57,27 @@ CREATE INDEX IF NOT EXISTS idx_messages_ts ON messages(timestamp);
 CREATE INDEX IF NOT EXISTS idx_messages_thread ON messages(thread_id);
 """
 
-_MIGRATE_SENDER_NAME = (
-    "ALTER TABLE messages ADD COLUMN sender_name TEXT DEFAULT ''"
+# Additive, every one with a default, so an existing chat.db opens unchanged and
+# a node that is downgraded still reads its own rows. `CREATE TABLE IF NOT
+# EXISTS` adds no column to a table that already exists — the same trap
+# `create_all()` is recorded for on the hub — so each of these runs on its own
+# and a duplicate-column error is the expected outcome on the second start.
+_MIGRATIONS = (
+    "ALTER TABLE messages ADD COLUMN sender_name TEXT DEFAULT ''",
+    "ALTER TABLE messages ADD COLUMN format INTEGER NOT NULL DEFAULT 0",
+    "ALTER TABLE messages ADD COLUMN epoch INTEGER NOT NULL DEFAULT 0",
+    "ALTER TABLE messages ADD COLUMN device BLOB DEFAULT NULL",
+    "ALTER TABLE messages ADD COLUMN nonce BLOB DEFAULT NULL",
+    "ALTER TABLE messages ADD COLUMN sig BLOB DEFAULT NULL",
+)
+
+# A replay is a validly signed copy of a real message, so nothing about the
+# signature refuses it. The nonce does: it is per message, per device, and a
+# repeat is either an attack or a bug. Partial, because plaintext rows carry no
+# nonce at all and NULLs are distinct in SQLite anyway.
+_REPLAY_INDEX = (
+    "CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_replay "
+    "ON messages(device, nonce) WHERE nonce IS NOT NULL"
 )
 
 
@@ -43,11 +90,24 @@ class StoredMessage:
     timestamp: float
     thread_id: str | None
     sender_name: str = ""
+    format: int = FORMAT_PLAIN
+    epoch: int = 0
+    device: bytes | None = None
+    nonce: bytes | None = None
+    sig: bytes | None = None
+
+
+_COLUMNS = ("id, sender_id, iteration, payload, timestamp, thread_id, "
+            "sender_name, format, epoch, device, nonce, sig")
 
 
 def _row(r) -> StoredMessage:
-    return StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], payload=r[3],
-                         timestamp=r[4], thread_id=r[5], sender_name=r[6] or "")
+    return StoredMessage(
+        id=r[0], sender_id=r[1], iteration=r[2], payload=r[3],
+        timestamp=r[4], thread_id=r[5], sender_name=r[6] or "",
+        format=r[7] or FORMAT_PLAIN, epoch=r[8] or 0,
+        device=r[9], nonce=r[10], sig=r[11],
+    )
 
 
 class ChatStore:
@@ -61,10 +121,17 @@ class ChatStore:
         self._db_path.parent.mkdir(parents=True, exist_ok=True)
         self._db = await aiosqlite.connect(str(self._db_path))
         await self._db.executescript(_SCHEMA)
-        try:
-            await self._db.execute(_MIGRATE_SENDER_NAME)
-        except Exception:
-            pass
+        for statement in _MIGRATIONS:
+            try:
+                await self._db.execute(statement)
+            except Exception:
+                # Already applied. Swallowed per column rather than per batch:
+                # one loop with a shared `try` would stop at the first
+                # already-present column and silently skip every later one, so a
+                # node upgraded twice would be missing the newest fields with
+                # nothing to show for it.
+                pass
+        await self._db.execute(_REPLAY_INDEX)
         await self._db.commit()
 
     async def close(self) -> None:
@@ -86,14 +153,32 @@ class ChatStore:
         payload: bytes,
         thread_id: str | None = None,
         sender_name: str = "",
+        *,
+        format: int = FORMAT_PLAIN,
+        epoch: int = 0,
+        device: bytes | None = None,
+        nonce: bytes | None = None,
+        sig: bytes | None = None,
     ) -> int:
-        """Store a message. Returns the row id."""
+        """
+        Store a message. Returns the row id.
+
+        Raises `ReplayedMessage` if this device has already used this nonce —
+        see `_REPLAY_INDEX`. The caller must not turn that into a generic
+        failure the sender retries: it means the message is already stored.
+        """
         ts = time.time()
-        cursor = await self._db.execute(
-            "INSERT INTO messages (sender_id, iteration, payload, timestamp, thread_id, sender_name) "
-            "VALUES (?, ?, ?, ?, ?, ?)",
-            (sender_id, iteration, payload, ts, thread_id, sender_name),
-        )
+        try:
+            cursor = await self._db.execute(
+                "INSERT INTO messages (sender_id, iteration, payload, timestamp, "
+                " thread_id, sender_name, format, epoch, device, nonce, sig) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (sender_id, iteration, payload, ts, thread_id, sender_name,
+                 format, epoch, device, nonce, sig),
+            )
+        except aiosqlite.IntegrityError as e:
+            await self._db.rollback()
+            raise ReplayedMessage(str(e)) from e
         await self._db.commit()
         return cursor.lastrowid
 
@@ -104,17 +189,12 @@ class ChatStore:
     ) -> list[StoredMessage]:
         """Get messages after a timestamp, most recent last."""
         cursor = await self._db.execute(
-            "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+            f"SELECT {_COLUMNS} "
             "FROM messages WHERE timestamp > ? ORDER BY timestamp ASC LIMIT ?",
             (since, limit),
         )
         rows = await cursor.fetchall()
-        return [
-            StoredMessage(id=r[0], sender_id=r[1], iteration=r[2],
-                          payload=r[3], timestamp=r[4], thread_id=r[5],
-                          sender_name=r[6] or "")
-            for r in rows
-        ]
+        return [_row(r) for r in rows]
 
     async def get_recent(self, limit: int = 100) -> list[StoredMessage]:
         """The newest `limit` messages, oldest first so they render in order.
@@ -125,7 +205,7 @@ class ChatStore:
         rows means ordering DESC in SQL and reversing here.
         """
         cursor = await self._db.execute(
-            "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+            f"SELECT {_COLUMNS} "
             "FROM messages ORDER BY id DESC LIMIT ?",
             (limit,),
         )
@@ -141,7 +221,7 @@ class ChatStore:
         `id` is AUTOINCREMENT: unique, and ordered by insertion.
         """
         cursor = await self._db.execute(
-            "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+            f"SELECT {_COLUMNS} "
             "FROM messages WHERE id < ? ORDER BY id DESC LIMIT ?",
             (before_id, limit),
         )
@@ -157,17 +237,57 @@ class ChatStore:
     async def get_thread(self, thread_id: str, limit: int = 100) -> list[StoredMessage]:
         """Get messages in a thread."""
         cursor = await self._db.execute(
-            "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+            f"SELECT {_COLUMNS} "
             "FROM messages WHERE thread_id = ? ORDER BY timestamp ASC LIMIT ?",
             (thread_id, limit),
         )
         rows = await cursor.fetchall()
-        return [
-            StoredMessage(id=r[0], sender_id=r[1], iteration=r[2],
-                          payload=r[3], timestamp=r[4], thread_id=r[5],
-                          sender_name=r[6] or "")
-            for r in rows
-        ]
+        return [_row(r) for r in rows]
+
+    @property
+    def db_path(self) -> Path:
+        """Where this store lives — the re-encryption command backs it up."""
+        return self._db_path
+
+    async def count_by_format(self) -> tuple[int, int]:
+        """(plaintext, sealed). What the operator is deciding from."""
+        cursor = await self._db.execute(
+            "SELECT format, COUNT(*) FROM messages GROUP BY format")
+        counts = {row[0]: row[1] for row in await cursor.fetchall()}
+        return (counts.get(FORMAT_PLAIN, 0), counts.get(FORMAT_SEALED_V1, 0))
+
+    async def all_plaintext(self) -> list[StoredMessage]:
+        """Every message still stored in the clear, oldest first."""
+        cursor = await self._db.execute(
+            f"SELECT {_COLUMNS} FROM messages WHERE format = ? ORDER BY id",
+            (FORMAT_PLAIN,))
+        return [_row(r) for r in await cursor.fetchall()]
+
+    async def reseal(self, message_id: int, *, epoch: int, device: bytes,
+                     nonce: bytes, ct: bytes, sig: bytes) -> None:
+        """
+        Replace one plaintext row with its sealed form. **Does not commit** —
+        the caller commits once, so a re-encryption that fails half way leaves
+        the database as it was rather than half readable.
+
+        `sender_name` is cleared because it moves inside the envelope; leaving
+        it would keep in the clear the one field the sealing was for.
+        """
+        await self._db.execute(
+            "UPDATE messages SET format = ?, epoch = ?, device = ?, "
+            " nonce = ?, payload = ?, sig = ?, sender_name = '' "
+            "WHERE id = ?",
+            (FORMAT_SEALED_V1, epoch, device, nonce, ct, sig, message_id))
+
+    async def commit(self) -> None:
+        await self._db.commit()
+
+    async def delete_older_than(self, cutoff: float) -> int:
+        """Retention. Returns how many rows went."""
+        cursor = await self._db.execute(
+            "DELETE FROM messages WHERE timestamp < ?", (cutoff,))
+        await self._db.commit()
+        return cursor.rowcount
 
     async def message_count(self) -> int:
         cursor = await self._db.execute("SELECT COUNT(*) FROM messages")