diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-09-07 17:50:28 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-09-07 17:50:28 +0200 |
| commit | 36cebf25d0e0f24cf63be4380ccb5d03da726a74 (patch) | |
| tree | 8509ec4cf68a058f7383299e11bdea97ab06cadf /packages/meshbay-node | |
| parent | 8883d60d0afa2ed9dd1ef68bc21fe1b9a65a59ff (diff) | |
| download | meshbay-36cebf25d0e0f24cf63be4380ccb5d03da726a74.tar.gz | |
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
Diffstat (limited to 'packages/meshbay-node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/bundle_store.py | 73 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/chat/__init__.py | 20 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/chat/store.py | 186 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 103 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ops.py | 249 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/quic_server.py | 14 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 572 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ui/app.py | 18 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_chat_encryption.py | 517 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_chat_history_binary.py | 181 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_chat_multidevice.py | 160 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_cli_dispatch.py | 11 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_device_on_connection.py | 287 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_webrtc_transport.py | 64 |
14 files changed, 2346 insertions, 109 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/bundle_store.py b/packages/meshbay-node/src/meshbay_node/bundle_store.py index 7f03caa..86b8de4 100644 --- a/packages/meshbay-node/src/meshbay_node/bundle_store.py +++ b/packages/meshbay-node/src/meshbay_node/bundle_store.py @@ -30,6 +30,30 @@ CREATE TABLE IF NOT EXISTS gek_bundles ( ); """ +# Chat epoch keys. Wrapped to the node's own X25519 key, exactly as the node's +# copy of the group key is — never stored raw. +# +# That is the whole basis of the claim chat encryption makes: "unreadable to +# someone who obtains the node's storage without the keystore password". A +# plaintext table beside chat.db would collapse it to nothing, silently, and it +# is the obvious thing to write. `test_chat_key_storage.py` reads the file back +# and refuses to find the live key in it. +# +# Rows are kept, never replaced: opening a new epoch must not make the history +# of the old one unreadable to the members who could already read it, which is +# the difference between an epoch and a rotation. +_SCHEMA_CHAT_EPOCHS = """\ +CREATE TABLE IF NOT EXISTS chat_epochs ( + group_id TEXT NOT NULL, + epoch INTEGER NOT NULL, + pk_eph_b64 TEXT NOT NULL, + nonce_b64 TEXT NOT NULL, + wrapped_b64 TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + PRIMARY KEY (group_id, epoch) +); +""" + _SCHEMA_KEYPAIR = """\ CREATE TABLE IF NOT EXISTS keypair_bundles ( user_id TEXT PRIMARY KEY, @@ -50,6 +74,7 @@ class BundleStore: self._db = await aiosqlite.connect(str(self._db_path)) await self._db.execute(_SCHEMA_GEK) await self._db.execute(_SCHEMA_KEYPAIR) + await self._db.execute(_SCHEMA_CHAT_EPOCHS) await self._migrate_keypair_recovery() await self._db.commit() @@ -155,6 +180,54 @@ class BundleStore: await self._db.commit() return cur.rowcount > 0 + # ── Chat epoch keys ────────────────────────────────────────────────── + + async def store_chat_epoch( + self, + group_id: str, + epoch: int, + pk_eph_b64: str, + nonce_b64: str, + wrapped_b64: str, + ) -> None: + """ + Record one epoch key, wrapped to the node's own key. + + `INSERT OR IGNORE`, not `REPLACE`: an epoch's key is written once and is + then the only way to read the messages sent under it. Overwriting one — + which a retry, or two callers racing to open the same epoch, would do — + would destroy that history with no error anywhere. + """ + assert self._db + await self._db.execute( + "INSERT OR IGNORE INTO chat_epochs " + "(group_id, epoch, pk_eph_b64, nonce_b64, wrapped_b64) " + "VALUES (?, ?, ?, ?, ?)", + (group_id, epoch, pk_eph_b64, nonce_b64, wrapped_b64), + ) + await self._db.commit() + + async def fetch_chat_epochs(self, group_id: str) -> list[dict]: + """Every epoch this group has had, oldest first.""" + assert self._db + async with self._db.execute( + "SELECT epoch, pk_eph_b64, nonce_b64, wrapped_b64 FROM chat_epochs " + "WHERE group_id = ? ORDER BY epoch", (group_id,) + ) as cur: + rows = await cur.fetchall() + return [{"epoch": r[0], "pk_eph_b64": r[1], "nonce_b64": r[2], + "wrapped_b64": r[3]} for r in rows] + + async def latest_chat_epoch(self, group_id: str) -> int: + """The highest epoch number, or 0 when the group has none yet.""" + assert self._db + async with self._db.execute( + "SELECT MAX(epoch) FROM chat_epochs WHERE group_id = ?", + (group_id,) + ) as cur: + row = await cur.fetchone() + return int(row[0] or 0) + async def close(self) -> None: if self._db: await self._db.close() diff --git a/packages/meshbay-node/src/meshbay_node/chat/__init__.py b/packages/meshbay-node/src/meshbay_node/chat/__init__.py index f647e19..cb2c868 100644 --- a/packages/meshbay-node/src/meshbay_node/chat/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/chat/__init__.py @@ -1,4 +1,18 @@ -"""MeshBay Node — chat module (Sender Keys encrypted group messaging).""" -from .store import ChatStore +"""MeshBay Node — chat storage and relay. -__all__ = ["ChatStore"] +Encryption is the client's: the node holds an epoch key it delivers to members +and never a plaintext message once a group has the switch on. See +`docs/chat-sender-keys.md`. It is not the Sender Keys ratchet this module's +docstring used to name — `senderkeys.py` is unused by production and is kept for +a possible future 1:1 DM, alongside `ratchet.py`. +""" +from .store import ( + FORMAT_PLAIN, + FORMAT_SEALED_V1, + ChatStore, + ReplayedMessage, + StoredMessage, +) + +__all__ = ["ChatStore", "StoredMessage", "ReplayedMessage", + "FORMAT_PLAIN", "FORMAT_SEALED_V1"] 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") diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index f9e992c..e270b6c 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -415,6 +415,11 @@ class NodeDaemon: # Whether the node unfurls links members post here. "chat_link_preview": await self._roster.chat_link_preview( group_cfg.id) if self._roster else True, + # Which chat epoch key is current. Opened here if the + # group has none, because chat is always encrypted (MNP + # 2.0) and a group with no epoch is a group nobody can + # speak in — the node cannot wait for an operator to notice. + "chat_epoch": await self._ensure_chat_epoch(group_cfg.id), # Whether TMDB lookups run for this group at all — # per-group (2026-08-24, used to be node-wide), same # "read once, kept current in place by the signed op" @@ -633,6 +638,10 @@ class NodeDaemon: self._state["bundle_store"] = self._bundle_store self._state["roster"] = self._roster self._state["node_user_id"] = session.user_id + # The node's own Ed25519 key. Needed by `encrypt_chat_history`, + # which seals migrated messages under a synthetic device of the + # node's rather than pretending to hold a member's signing key. + self._state["sk_node"] = keys.sk_ed25519 self._state["webrtc"] = self._webrtc self._state["quic_server"] = self._quic_server self._state["hub"] = hub @@ -869,6 +878,7 @@ class NodeDaemon: "chat_link_preview": ( await self._roster.chat_link_preview(group_cfg.id) if self._roster else True), + "chat_epoch": await self._ensure_chat_epoch(group_cfg.id), "tmdb_enabled": ( await self._roster.tmdb_enabled(group_cfg.id) if self._roster else True), @@ -1007,6 +1017,35 @@ class NodeDaemon: log.warning("No unwrappable GEK bundle found for group %s", group_id[:8]) return None + async def _ensure_chat_epoch(self, group_id: str) -> int: + """ + The group's current chat epoch, opening the first one if it has none. + + Chat is always encrypted, so a group with no epoch key is a group in + which nobody can say anything. Attaching one is the node's job and + happens here rather than on the first message: a failure at start-up is + in the log the operator is already reading, and a failure on someone's + first message is a chat that mysteriously refuses them. + + Never fatal. A group whose epoch cannot be opened keeps every other + function — files, video, the index — and only its chat is unusable, + which is strictly better than refusing to host the group at all. + """ + if not self._bundle_store: + return 0 + try: + epoch = await self._bundle_store.latest_chat_epoch(group_id) + if epoch: + return epoch + from meshbay_node import ops + + return (await ops.open_chat_epoch(self._state, group_id))["epoch"] + except Exception as e: + log.error("chat: no epoch key for group %s (%s) — chat is " + "unusable in this group until this is fixed", + group_id[:8], e) + return 0 + async def _progress_pusher(self, indexer: DirectoryIndexer, interval: float = 2.0) -> None: """ @@ -1759,7 +1798,8 @@ def main() -> None: parser.add_argument("command", nargs="?", choices=["init", "reset", "status", "gek-init", "gek", "operator", "member", "group", "root", - "file", "video", "denylist", "stun", "reload", + "file", "video", "chat", "denylist", "stun", + "reload", "restart-daemon", "autostart", "service", "calibrate-argon2"], help="init: provision config + keystore | reset: erase all " @@ -1770,7 +1810,9 @@ def main() -> None: "| root list|add|remove|set|eject|plug " "| gek init|rotate | file list|rm " "| video rematch: re-resolve TMDB matches for a group's " - "videos | denylist show|clear " + "videos " + "| chat status|rotate|encrypt-history|prune " + "| denylist show|clear " "| stun list|add|remove|reset " "| reload: re-read node.toml (hot; systemd or the " "loopback API) | restart-daemon: restart the node " @@ -1827,7 +1869,7 @@ def main() -> None: # Query commands print a report; library logging would interleave with it. quiet = args.command in ("status", "gek-init", "gek", "operator", - "member", "group", "root", "file", "video", + "member", "group", "root", "file", "video", "chat", "denylist", "stun", "reload", "restart-daemon", "reset") logging.basicConfig( @@ -2370,6 +2412,61 @@ def main() -> None: sys.exit(1) return + if args.command == "chat": + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + sub = args.subcommand or "status" + group_id = _resolve_group(cfg, args.group) + + if sub == "status": + out = _daemon_api(cfg, f"/api/groups/{group_id}/chat") + print(f"encryption always on (MNP {MNP_VERSION})") + print(f"epoch {out.get('epoch', 0)}") + print(f"messages {out.get('encrypted_messages', 0)} encrypted, " + f"{out.get('plaintext_messages', 0)} in the clear") + if out.get("plaintext_messages"): + print("\nThose messages were written before this node spoke MNP " + "2.0 and are\nstill readable off this disk. " + "`chat encrypt-history` converts them.") + return + + if sub == "rotate": + out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/epoch", + method="POST") + print(f"chat epoch {out['epoch']} opened") + print("Everyone still in the group keeps reading the history; " + "whoever left\ncannot read what is written from now on.") + return + + if sub == "encrypt-history": + if not args.yes: + print("This rewrites the only copy of this group's older " + "messages.") + print("A backup of chat.db is taken first, beside it.") + if input("re-encrypt now? [y/N] ").strip().lower() not in ("y", "yes"): + print("cancelled") + return + out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/encrypt-history", + method="POST", timeout=300) + print(f"re-encrypted {out['converted']} message(s) under epoch " + f"{out['epoch']}") + print(f"backup {out['backup']}") + return + + if sub == "prune": + days = int(args.target or 0) + if days < 1: + print("usage: meshbay-node chat prune <days> [--group G]") + sys.exit(1) + out = _daemon_api( + cfg, f"/api/groups/{group_id}/chat/prune?max_age_days={days}", + method="POST") + print(f"removed {out['removed']} message(s) older than {days} day(s)") + return + + print("usage: meshbay-node chat " + "status|rotate|encrypt-history|prune [--group G]") + sys.exit(1) + if args.command == "denylist": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "show" 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: diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 360b9ac..284b488 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -23,6 +23,7 @@ import logging import os import struct import subprocess +import uuid from pathlib import Path from typing import Any, Callable @@ -207,6 +208,9 @@ class _MNPServerProtocol(QuicConnectionProtocol): def __init__(self, *args, node_ctx: dict, **kwargs): super().__init__(*args, **kwargs) self._ctx = node_ctx # shared server context (keys, index, etc.) + # Per connection, never per account — one person may hold several + # devices. See webrtc_server.WebRTCPeerSession._registry_key. + self._registry_key: str = uuid.uuid4().hex self._user_id: str | None = None self._group_id: str | None = None self._buffers: dict[int, _StreamBuffer] = {} @@ -362,7 +366,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._user_id = peer.user_id self._group_id = peer.group_id - self._peer_registry()[self._user_id] = self + self._peer_registry()[self._registry_key] = self transcript = handshake_transcript( ROLE_NODE, peer.group_id, self._nonce_client, self._gek_challenge, binding) @@ -507,8 +511,10 @@ class _MNPServerProtocol(QuicConnectionProtocol): "thread_id": msg.get("thread_id"), "group_id": self._group_id or "", } - for uid, proto in list(self._peer_registry().items()): - if uid != self._user_id and proto is not self: + # Per connection, not per account — see the WebRTC path and + # docs/chat-sender-keys.md F7. A person's other devices are recipients. + for proto in list(self._peer_registry().values()): + if proto is not self: try: proto._send(0, broadcast) except Exception: @@ -518,7 +524,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): def connection_lost(self, exc) -> None: if self._user_id: - self._peer_registry().pop(self._user_id, None) + self._peer_registry().pop(self._registry_key, None) for task in list(self._tasks): task.cancel() super().connection_lost(exc) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 8e357c9..9774831 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -31,6 +31,7 @@ import os import struct import tempfile import time +import uuid from pathlib import Path from typing import Any @@ -78,6 +79,7 @@ from meshbay_common.adminop import ( OP_PHOTO_ROOTS, OP_APP_DIRECTORIES, OP_CHAT_DIRECTORY, + OP_CHAT_EPOCH, OP_CHAT_LINK_PREVIEW, OP_ROOT_ADD, OP_ROOT_REMOVE, @@ -93,9 +95,10 @@ from meshbay_common.device import ( DEVICE_TTL, device_add_transcript, device_code_hash, + device_hello_transcript, device_request_transcript, ) -from meshbay_common.groupbox import PURPOSE_ACK, seal +from meshbay_common.groupbox import PURPOSE_ACK, PURPOSE_CHAT_KEYS, seal from meshbay_common.join import ( JOIN_TTL, ROLE_MEMBER, @@ -103,6 +106,8 @@ from meshbay_common.join import ( join_transcript, ) from meshbay_common.protocol import MNP, chunk_ciphertext, file_chunk_wire +from meshbay_common.chatbox import NONCE_LEN as CHAT_NONCE_LEN, SIG_LEN as CHAT_SIG_LEN +from meshbay_node.chat import FORMAT_PLAIN, FORMAT_SEALED_V1, ReplayedMessage from meshbay_node.transport.wire import index_sync_message from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer @@ -326,9 +331,27 @@ class WebRTCPeerSession: self._peer_id: str = peer_id self._remote_ip: str = "" self._username: str = "" + # This connection's key in the group's peer registry. **Per connection, + # never per account**: one person may hold several devices here, and + # keying the registry by user_id made the second evict the first — the + # same "keyed by account where it should be keyed by device" mistake as + # `pin_identity`'s old INSERT OR REPLACE and as GroupSenderKeyStore's + # silent overwrite. Symptom was invisible: two devices of one account + # could not both be connected, and whichever disconnected took the + # other's chat delivery with it. See docs/chat-sender-keys.md F7. + self._registry_key: str = uuid.uuid4().hex # Set from the roster: the key this node pinned for this account. Never # from the JWT — the hub picks what goes in there. + # + # This is the account's *oldest* live device unless `device_hello` has + # told us better — see _do_device_hello. Treat it as "a device of this + # account", not "the device on this connection", anywhere that has not + # checked `_device_confirmed`. self._pinned_pk: str = "" + # True once this connection proved which device it is. Until then the + # node knows the account and not the key, which is all it ever knew + # before device linking existed. + self._device_confirmed: bool = False # Flow control for video: how many segments the client says it can take. self._stream_credit = 0 self._stream_credit_evt = asyncio.Event() @@ -470,6 +493,8 @@ class WebRTCPeerSession: self._spawn(self._do_device_list(msg)) elif mtype == MNP.DEVICE_REVOKE: self._spawn(self._do_device_revoke(msg)) + elif mtype == MNP.DEVICE_HELLO and self._nonce_node: + self._spawn(self._do_device_hello(msg)) elif mtype == MNP.MEMBER_UPLOAD: self._do_member_upload(msg) elif mtype == MNP.APPS_ENABLED: @@ -492,6 +517,10 @@ class WebRTCPeerSession: self._do_chat_directory(msg) elif mtype == MNP.CHAT_LINK_PREVIEW: self._do_chat_link_preview(msg) + elif mtype == MNP.CHAT_EPOCH: + self._do_chat_epoch(msg) + elif mtype == MNP.CHAT_KEYS_REQ: + self._spawn(self._do_chat_keys_req(msg)) elif mtype == MNP.MEDIA_META_REQ: self._spawn(self._do_media_meta_request(msg)) elif mtype == MNP.SEASON_META_REQ: @@ -743,7 +772,7 @@ class WebRTCPeerSession: self._username = self._pending_username self._spawn(self._load_pinned_pk()) - self._peer_registry()[self._user_id] = self + self._register_peer() node_user_id = self._ctx.get("node_user_id") log.info("WebRTC handshake OK — user=%s group=%s", @@ -827,6 +856,15 @@ class WebRTCPeerSession: # on, which is what it did before this existed. "chat_link_preview": bool( self._group_ctx().get("chat_link_preview", True)), + # Which chat epoch key a client should be sealing under. Inside + # the sealed part of the ack like every other configuration field, + # so it carries an authentication tag from a key the hub does not + # hold — a forged epoch would have a client sealing under a key the + # group has retired. + # + # No `chat_encrypted` beside it: there is no switch. A peer that + # reached this point speaks MNP 2.0, and 2.0 has no plaintext chat. + "chat_epoch": int(self._group_ctx().get("chat_epoch", 0) or 0), # So a client that connects mid-scan shows the indexing state # immediately, instead of waiting for the next periodic # INDEX_PROGRESS push. Never a path or filename — see @@ -1378,6 +1416,79 @@ class WebRTCPeerSession: self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION, "pk_ed25519": pk_ed_b64}) + async def _do_device_hello(self, msg: dict) -> None: + """ + Learn which of this account's devices is on this connection. + + The handshake authenticates a *group membership* (the GEK-HMAC) and an + *account* (the hub's token). It has never authenticated a device, and + while one account meant one key that was the same statement. It stopped + being so on 2026-08-18, and `_load_pinned_pk` — which resolves the + account's oldest live device — has been standing in for the real answer + ever since, including as the recorded uploader of every file. + + What is checked, in order: the key is a live device *of this account* in + the node's own roster (never a token claim — that is + `per-node-identity-v1.md`'s rule), the timestamp is fresh, and the + signature verifies over a transcript naming this node, this group and + this connection's nonce. A key that is merely well-formed proves nothing. + + Idempotent for the same key, refused for a different one: a connection + does not get to change device half way through, which would let one + session's uploads be attributed to two. + """ + roster = self._ctx.get("roster") + if roster is None or not self._user_id: + self._send({"type": "error", "detail": "Roster not available"}) + return + if not self._spend_device_attempt(): + return + + pk_ed_b64 = str(msg.get("pk_ed25519", "")) + ts = int(msg.get("ts", 0) or 0) + if not pk_ed_b64: + self._send({"type": "error", "detail": "Missing device key"}) + return + if self._device_confirmed and pk_ed_b64 != self._pinned_pk: + self._send({"type": "error", + "detail": "This connection is already another device"}) + return + if abs(time.time() - ts) > DEVICE_TTL: + self._send({"type": "error", "detail": "Stale device_hello"}) + return + + device = await roster.find_device(self._user_id, pk_ed_b64) + if device is None: + self._audit("device_hello_refused", pk_ed_b64[:16]) + self._send({"type": "error", + "detail": "Not a device paired here"}) + return + + transcript = device_hello_transcript( + node_pk_b64=self._node_pk_b64(), group_id=self._group_id or "", + user_id=self._user_id, pk_ed25519_b64=pk_ed_b64, + nonce_node=self._nonce_node, ts=ts) + try: + pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64)) + except Exception: + self._send({"type": "error", "detail": "Unreadable device key"}) + return + try: + sig = base64.b64decode(msg.get("sig", "")) + except Exception: + sig = b"" + if not self._verify_sig(pk, transcript, sig): + self._audit("device_hello_refused", pk_ed_b64[:16]) + self._send({"type": "error", "detail": "Signature verification failed"}) + return + + self._pinned_pk = pk_ed_b64 + self._device_confirmed = True + log.info("Device identified on connection: user=%s device=%s", + self._user_id[:8], pk_ed_b64[:16]) + self._send({"type": MNP.DEVICE_HELLO_ACK, "v": MNP_VERSION, + "pk_ed25519": pk_ed_b64}) + async def _do_device_list(self, msg: dict) -> None: """This account's devices. Anyone may read their own, nobody else's.""" roster = self._ctx.get("roster") @@ -1398,6 +1509,43 @@ class WebRTCPeerSession: ], }) + async def _new_chat_epoch(self, group_id: str, reason: str) -> None: + """ + Open a chat epoch because the set of devices that may read future + messages just shrank. + + Called on every removal — a member, a device, an unpin — and on group + key rotation, because the operator rotates precisely when someone has + left. It is the exact counterpart of "still rotate the GEK, the + ex-member holds the current one": revocation stops the node handing + over the *next* key, and nothing else takes the current one away. + + Best effort by design: a failure here must never turn a successful + revocation into a refused one — the revocation is the control, and this + is the follow-through. It is logged loudly instead, because an operator + who removed someone needs to know if the chat key did not move. + """ + if not group_id: + return + try: + result = await self._run_op(ops.open_chat_epoch, group_id) + except Exception as e: + log.error("chat: could not open a new epoch for group %s after " + "%s (%s) — the removed party still holds the current " + "chat key", group_id[:8], reason, e) + self._audit("chat_epoch_failed", reason) + return + self._audit("chat_epoch", f"{reason}:{result['epoch']}") + # Everyone still connected picks the new key up without reconnecting. + for session in list( + (self._ctx.get("groups") or {}).get(group_id, {}) + .get("_peers", {}).values()): + try: + session._send({"type": MNP.CHAT_EPOCH_ACK, "v": MNP_VERSION, + "epoch": result["epoch"]}) + except Exception: + pass + async def _do_device_revoke(self, msg: dict) -> None: """ Retire one of this account's devices — a lost laptop. @@ -1446,6 +1594,9 @@ class WebRTCPeerSession: return await roster.revoke_device(self._user_id, target) + # A revoked device holds every chat key it ever received — a lost laptop + # reads the group's chat until the epoch moves. + await self._new_chat_epoch(self._group_id or "", "device_revoke") self._audit("device_revoked", f"{target[:16]} by {signer[:16]}") log.info("Device revoked for %s: %s", self._user_id[:8], target[:16]) self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION, @@ -1769,6 +1920,11 @@ class WebRTCPeerSession: except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return + # The operator is rotating because somebody left, and the chat archive + # key is not derived from the group key — so rotating that one does not + # move this one. Doing both here is what makes "rotate after a removal" + # mean the same thing for chat as it does for files. + await self._new_chat_epoch(pending["subject"], "gek_rotate") self._audit("gek_rotate", pending["subject"]) self._send({ "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION, @@ -1806,6 +1962,7 @@ class WebRTCPeerSession: return try: await self._run_op(ops.unpin_member, user_id) + await self._new_chat_epoch(self._group_id or "", "member_unpin") except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return @@ -2280,6 +2437,76 @@ class WebRTCPeerSession: self._broadcast_to_group({"type": MNP.CHAT_LINK_PREVIEW_ACK, "v": MNP_VERSION, "enabled": enabled}) + def _do_chat_epoch(self, msg: dict) -> None: + """ + Open a new chat epoch by hand. Operator only, and signed. + + There is no switch to turn chat encryption on: MNP 2.0 has no plaintext + chat to fall back to. What an operator may want to do deliberately is + move the key on — the same instruction as `gek_rotate`, and signed for + the same reason. The removals that matter (member revoke, member unpin, + device revoke, `gek_rotate`) already open one by themselves. + """ + group_id = str(msg.get("group_id", "")).strip() or self._group_id + if not group_id: + self._send({"type": "error", "detail": "No group on this connection"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_CHAT_EPOCH, group_id, group_id=group_id) + + async def _admin_exec_chat_epoch( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"chat_epoch:{pending['subject'][:8]}") + return + try: + result = await self._run_op(ops.open_chat_epoch, pending["subject"]) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("chat_epoch", f"manual:{result['epoch']}") + self._broadcast_to_group({"type": MNP.CHAT_EPOCH_ACK, + "v": MNP_VERSION, "epoch": result["epoch"]}) + + async def _do_chat_keys_req(self, msg: dict) -> None: + """ + Hand this member every chat epoch key the group has, sealed. + + Sealed under a group-derived subkey rather than sent in clear: the same + reasoning as the index and the handshake ack, and one step stronger + here, because the payload *is* key material. A peer that has completed + the handshake holds the group key and can open it; anything short of + that gets a ciphertext. + + **Every** live epoch, not just the current one, which is what keeps the + history readable to a member who joined after it was written and to a + device linked this morning. Whether a new member should receive the back + catalogue at all is a policy question with a per-group answer; the shape + is here so that answer can be given without a wire change. + """ + gctx = self._group_ctx() + gek = gctx.get("gek") + if not gek: + self._send({"type": "error", "detail": "Group encryption not initialized"}) + return + try: + keys = await self._run_op(ops.chat_epoch_keys, self._group_id or "") + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + + payload = {"epochs": [{"epoch": k["epoch"], "key": k["key"]} + for k in keys], + "current": keys[-1]["epoch"] if keys else 0} + sealed = seal(gek, PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP, + self._group_id or "", payload) + self._send({"type": MNP.CHAT_KEYS_RESP, "v": MNP_VERSION, + "group_id": self._group_id or "", **sealed}) + def _broadcast_to_group(self, notice: dict) -> None: """ Tell everyone connected to this group about a setting that changed. @@ -2852,6 +3079,7 @@ class WebRTCPeerSession: try: result = await self._run_op( ops.revoke_member, user_id, self._group_id or "") + await self._new_chat_epoch(self._group_id or "", "member_revoke") except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return @@ -2859,8 +3087,11 @@ class WebRTCPeerSession: # Anyone connected right now keeps the key they already unwrapped; what # they lose is the next one. Rotating it is the operator's call, and the # ack says so rather than implying this undid anything already read. - peer = self._peer_registry().get(user_id) - if peer is not None: + # Every connection that account holds, not "the" one: with device + # linking a person may be connected from several at once, and the + # registry is keyed per connection precisely because it cannot hold + # only one of them. + for peer in self._sessions_of(user_id): try: await peer.close() except Exception: @@ -2961,6 +3192,29 @@ class WebRTCPeerSession: "total_bytes": progress.total_bytes, } + def _register_peer(self) -> None: + """Add this connection to its group's peer set. + + One place decides the key, and it is `_registry_key` — per connection, + never per account. Written as a method so a test drives the real + registration rather than a second copy of this line that agrees with it + by construction. + """ + self._peer_registry()[self._registry_key] = self + + def _unregister_peer(self) -> None: + self._peer_registry().pop(self._registry_key, None) + + def _sessions_of(self, user_id: str) -> list["WebRTCPeerSession"]: + """Every live connection this account holds in this group. + + Never "the" connection: with device linking a person may be connected + from a laptop and a phone at once, and an operation that acts on one of + them at random is a revocation that leaves a session running. + """ + return [s for s in list(self._peer_registry().values()) + if s._user_id == user_id] + def _peer_registry(self) -> dict: """ Connected peers for THIS group only. @@ -3817,22 +4071,65 @@ class WebRTCPeerSession: }) def _do_chat_message(self, msg: dict) -> None: - # Per-group store — see _peer_registry() and finding H1. Reading chat_store - # off the shared transport context sent every group's messages to the first - # group's database, and served them back to anyone on the node. - chat_store = self._group_ctx().get("chat_store") - payload = msg.get("payload", "") + """ + Store one message and hand it to everyone else in this group. + + The node is a relay and an archive here, not a reader: once a group has + chat encryption on, `payload` is a ciphertext it cannot open, and every + decision below is made from fields that stay in clear — which is why + those fields are the ones that must be *authenticated* rather than + merely present. + + `sender_id` comes from the authenticated session and never from the wire + (NS6). What the wire may now assert is the sending *device*, and that is + checked against this connection rather than believed: a member who could + name any device could sign as anyone once receivers verify signatures. + """ + # Per-group store — see _peer_registry() and finding H1. Reading + # chat_store off the shared transport context sent every group's + # messages to the first group's database, and served them back to + # anyone on the node. + gctx = self._group_ctx() + chat_store = gctx.get("chat_store") sender_name = msg.get("sender_name", "") + + # Two shapes, and keeping them apart is what makes this deployable. + # + # A plaintext message is exactly what it has always been: a string in + # `payload`. A sealed one carries its ciphertext in `ct`, beside the + # `nonce`/`device`/`sig` that authenticate it. Putting the ciphertext in + # `payload` instead would have been tidier and wrong: `payload` reaches + # older clients — the UI ships inside the desktop package now, so it can + # be months behind the node — and they would render bytes where they + # expect text. A field they have never heard of is ignored instead. + fmt = int(msg.get("format", 0) or 0) + epoch = int(msg.get("epoch", 0) or 0) + device = msg.get("device") + nonce = msg.get("nonce") + sig = msg.get("sig") + + if fmt == FORMAT_SEALED_V1: + payload = "" + raw = bytes(msg.get("ct") or b"") + else: + payload = msg.get("payload", "") + raw = (payload.encode() if isinstance(payload, str) + else bytes(payload or b"")) + + refusal = self._check_chat_envelope(gctx, fmt, raw, device, nonce, sig) + if refusal: + self._send({"type": "error", "detail": refusal}) + self._audit("chat_refused", refusal) + return + if sender_name: self._user_names()[self._user_id] = sender_name if chat_store: - raw = payload.encode() if isinstance(payload, str) else payload - self._spawn(chat_store.save_message( - sender_id=self._user_id, - iteration=msg.get("iteration", 0), - payload=raw, - thread_id=msg.get("thread_id"), - sender_name=sender_name, + self._spawn(self._store_chat_message( + chat_store, + iteration=msg.get("iteration", 0), payload=raw, + thread_id=msg.get("thread_id"), sender_name=sender_name, + format=fmt, epoch=epoch, device=device, nonce=nonce, sig=sig, )) peers = self._peer_registry() @@ -3843,10 +4140,21 @@ class WebRTCPeerSession: "sender_name": sender_name, "payload": payload, "thread_id": msg.get("thread_id"), - "timestamp": __import__("time").time(), + "timestamp": time.time(), + "format": fmt, + "epoch": epoch, + "device": device, + "nonce": nonce, + "sig": sig, } - for uid, session in list(peers.items()): - if uid != self._user_id and session is not self: + if fmt == FORMAT_SEALED_V1: + broadcast["ct"] = raw + # Excludes this connection, not this account. The sender's other + # devices are ordinary recipients: they did not compose the message and + # have no local echo of it, so skipping them by user_id left a person's + # second device silently missing everything they said from the first. + for session in list(peers.values()): + if session is not self: try: session._send(broadcast) except Exception: @@ -3859,11 +4167,16 @@ class WebRTCPeerSession: self._spawn(hub_ws.send(_json.dumps({ "type": "chat_notify", "group_id": self._group_id, - "sender_name": sender_name, - # Who actually wrote it, from the authenticated session. The - # hub used to fall back to this node's own token subject — - # the operator — so everyone was notified of their own - # messages and the operator was notified of nobody's. + # No sender_name. The body is unreadable to the hub the + # moment a group turns encryption on, and shipping the + # author's display name beside it would leave the hub a + # per-message record of who spoke where — the metadata the + # feature is otherwise about not producing. The hub renders + # "New message in <group>". + # + # `sender_user_id` stays: the hub needs it to not notify + # the author of their own message, and it already knows the + # group's membership. "sender_user_id": self._user_id, }))) except Exception: @@ -3872,6 +4185,66 @@ class WebRTCPeerSession: self._send({"type": "ack", "v": MNP_VERSION}) self._audit("chat_message") + def _check_chat_envelope(self, gctx: dict, fmt: int, ct: bytes, device, + nonce, sig) -> str: + """ + Why this message is refused, or "" to accept it. + + Two rules, and the first is the one that matters: + + **A device may only send as itself.** `device` is what receivers verify + a signature against, so a member free to name another member's key could + be that member to everyone — worse than the node-asserted attribution it + replaces (NS6), not better. The connection has proved which device it is + (`device_hello`), and this must match it. + + **Plaintext is refused, always.** Not "accepts and marks", and not + "unless a switch says otherwise": a member who can post in clear into a + group whose members believe their chat is encrypted is a downgrade, and + C6 is the standing lesson that the bypass left open is the one that gets + used. There is no switch to leave open — MNP 2.0 refuses a 1.x peer at + the handshake, so nothing that reaches here is unable to seal. + + `FORMAT_PLAIN` still exists, because rows written before 2.0 are still + in `chat.db` and still served. It is a *storage* state, never something + this accepts from the wire. + """ + if fmt != FORMAT_SEALED_V1: + return "Chat messages must be encrypted" + + if not (isinstance(device, (bytes, bytearray)) + and isinstance(nonce, (bytes, bytearray)) + and isinstance(sig, (bytes, bytearray))): + return "Sealed chat message is missing its envelope" + if len(nonce) != CHAT_NONCE_LEN or len(sig) != CHAT_SIG_LEN: + return "Sealed chat message has a malformed envelope" + if not ct: + return "Sealed chat message has no ciphertext" + + claimed = base64.b64encode(bytes(device)).decode() + if not self._device_confirmed: + return ("Identify this device before sending chat (device_hello)") + if claimed != self._pinned_pk: + return "That is not the device on this connection" + + return "" + + async def _store_chat_message(self, chat_store, **kwargs) -> None: + """ + Persist one message, treating a replay as already-done. + + A replayed message is a *validly signed* copy of a real one, so nothing + about the signature refuses it; the unique `(device, nonce)` does. It is + logged and dropped rather than raised at the sender: the message it + duplicates is already stored, so there is nothing for anyone to retry. + """ + try: + await chat_store.save_message(sender_id=self._user_id, **kwargs) + except ReplayedMessage: + log.warning("Replayed chat message from %s dropped", + (self._user_id or "?")[:8]) + self._audit("chat_replay_dropped") + def _do_ping(self, msg: dict) -> None: """Answer a liveness probe on an open channel, echoing the caller's token. @@ -3912,20 +4285,49 @@ class WebRTCPeerSession: "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "has_more": has_more, - "messages": [ - { - "id": m.id, - "sender_id": m.sender_id, - "sender_name": m.sender_name or names.get(m.sender_id, ""), - "payload": m.payload.decode("utf-8", errors="replace") - if isinstance(m.payload, bytes) else m.payload, - "timestamp": m.timestamp, - "thread_id": m.thread_id, - } - for m in msgs - ], + # `payload` goes out as **bytes**, never decoded here. It used to be + # `.decode("utf-8", errors="replace")`, which substitutes U+FFFD for + # every byte that is not valid UTF-8 — fine while chat was text, and + # silent destruction of a ciphertext. Live messages would have kept + # working (they are relayed, not re-read), so the symptom would have + # been "history won't decrypt", which is the hardest possible place + # to look. msgpack carries `bin` on both sides; the client decides + # how to read it from `format`. + "messages": [self._history_row(m, names) for m in msgs], }) + @staticmethod + def _history_row(m, names: dict) -> dict: + """One stored message on the wire. + + A plaintext row goes out under `payload` as a string, exactly as it + always has — an older client reads this response and must keep working. + A sealed row's ciphertext goes out under `ct` as bytes and `payload` + stays empty: decoding a ciphertext as UTF-8 (which is what this did, + with `errors="replace"`) substitutes U+FFFD for most of it, and the + symptom would have been history that will not decrypt while live + messages worked — the hardest possible place to look. + """ + row = { + "id": m.id, + "sender_id": m.sender_id, + "sender_name": m.sender_name or names.get(m.sender_id, ""), + "timestamp": m.timestamp, + "thread_id": m.thread_id, + "format": m.format, + "epoch": m.epoch, + "device": m.device, + "nonce": m.nonce, + "sig": m.sig, + } + if m.format == FORMAT_SEALED_V1: + row["payload"] = "" + row["ct"] = m.payload + else: + row["payload"] = (m.payload.decode("utf-8", errors="replace") + if isinstance(m.payload, bytes) else m.payload) + return row + def _link_preview_rate_ok(self) -> bool: """ True when this preview fetch is within both the per-connection and the @@ -4226,8 +4628,11 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - has_uploader_pk = bool(entry.uploader_pk) - if not self._has_admin_authority() and not has_uploader_pk: + # An owner is an *account* now, so an entry that records one is + # challengeable even if the device that uploaded it is gone. + has_uploader = bool(entry.uploader_pk + or getattr(entry, "uploader_id", "")) + if not self._has_admin_authority() and not has_uploader: self._send({"type": "error", "detail": "No authorized key for deletion"}) return @@ -4283,13 +4688,66 @@ class WebRTCPeerSession: except Exception: return False + async def _verify_uploader_sig(self, entry, transcript: bytes, + sig: bytes) -> bool: + """ + Whether this signature comes from a live device of the file's uploader. + + Every non-revoked device of `entry.uploader_id` is tried, the same way + `_verify_device_signer` tries every device that may approve a new one. + Two properties worth keeping straight: + + - **Ownership survives device revocation.** A retired laptop's uploads + keep their owner, because the account is what owns them; the revoked + key simply is not among the ones that may act. + - **Ownership survives the account losing every device**, where nothing + verifies here and the operator remains able to delete — which is the + behaviour a group needs when someone leaves. + + Falls back to the recorded `uploader_pk` only when the roster cannot + answer at all (no roster wired, or no `uploader_id` on an entry written + before that field existed). That is the pre-device-linking behaviour, so + an old index does not become undeletable. + """ + roster = self._ctx.get("roster") + uploader_id = getattr(entry, "uploader_id", "") or "" + if roster is not None and uploader_id: + for device in await roster.list_devices(uploader_id): + try: + pk = Ed25519PublicKey.from_public_bytes( + base64.b64decode(device["pk_ed25519"])) + except Exception: + continue + if self._verify_sig(pk, transcript, sig): + return True + return False + + if not entry.uploader_pk: + return False + try: + pk = Ed25519PublicKey.from_public_bytes( + base64.b64decode(entry.uploader_pk)) + except Exception: + return False + return self._verify_sig(pk, transcript, sig) + async def _load_pinned_pk(self) -> None: - """Remember which key this node pinned for the peer we just authenticated.""" + """ + A key this node pinned for the account we just authenticated. + + `get_identity` returns the account's **oldest** live device, which is a + stand-in, not an answer: the handshake never said which device is on + this connection. `device_hello` is the answer, and it arrives later — + so this must never overwrite a confirmed one. It is spawned from + `_complete_handshake` and can therefore finish *after* a fast client has + already identified itself, which is exactly the ordering that would put + the wrong key back. + """ roster = self._ctx.get("roster") - if roster is None or not self._user_id: + if roster is None or not self._user_id or self._device_confirmed: return ident = await roster.get_identity(self._user_id) - if ident: + if ident and not self._device_confirmed: self._pinned_pk = ident["pk_ed25519"] def _is_node_admin(self) -> bool: @@ -4433,6 +4891,9 @@ class WebRTCPeerSession: elif pending["op"] == OP_CHAT_LINK_PREVIEW: self._spawn( self._admin_exec_chat_link_preview(pending, transcript, sig_bytes)) + elif pending["op"] == OP_CHAT_EPOCH: + self._spawn( + self._admin_exec_chat_epoch(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_UPDATE: self._spawn( self._admin_exec_root_update(pending, transcript, sig_bytes)) @@ -4461,18 +4922,23 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - uploader_pk = None - if entry.uploader_pk: - try: - uploader_pk = Ed25519PublicKey.from_public_bytes( - base64.b64decode(entry.uploader_pk)) - except Exception: - uploader_pk = None - - # Node operator, or the user who uploaded this file — verified by the key - # recorded at upload time, never by a JWT claim (the hub controls those). + # Node operator, or the account that uploaded this file — **any of its + # non-revoked devices**, resolved through the node's own roster. + # + # This used to verify against `entry.uploader_pk` alone, the exact key + # that uploaded. Device linking broke that on 2026-08-18 without + # anything failing loudly: a file uploaded from a phone could not be + # deleted from the same person's laptop, and the only symptom was + # "Signature verification failed" on their own file + # (docs/desktop-client-v1.md §4.8 A). + # + # `uploader_pk` is kept, and stops being the authorization key: it is + # now the audit record of *which device* did it. Authorization is by + # account, through the roster — never through a token claim, which is + # the protection `per-node-identity-v1.md` added and which a lookup by + # `uploader_id` in the hub's world would give straight back. if not (await self._verify_admin_sig(transcript, sig) - or self._verify_sig(uploader_pk, transcript, sig)): + or await self._verify_uploader_sig(entry, transcript, sig)): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") return @@ -4941,7 +5407,7 @@ class WebRTCPeerSession: async def close(self) -> None: self._audit("disconnect") if self._user_id: - self._peer_registry().pop(self._user_id, None) + self._unregister_peer() await self.shutdown_tasks() await self._pc.close() diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index fc6c04a..2b99f20 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -334,6 +334,24 @@ def create_ui_app(state: dict) -> FastAPI: async def unpin_member(user_id: str): return await _op(lambda: ops.unpin_member(state, user_id)) + # ── Chat encryption (operator only, localhost) ───────────────────────── + + @app.get("/api/groups/{group_id}/chat") + async def chat_status(group_id: str): + return await _op(lambda: ops.chat_status(state, group_id)) + + @app.post("/api/groups/{group_id}/chat/epoch") + async def rotate_chat_epoch(group_id: str): + return await _op(lambda: ops.open_chat_epoch(state, group_id)) + + @app.post("/api/groups/{group_id}/chat/encrypt-history") + async def encrypt_chat_history(group_id: str): + return await _op(lambda: ops.encrypt_chat_history(state, group_id)) + + @app.post("/api/groups/{group_id}/chat/prune") + async def prune_chat(group_id: str, max_age_days: int): + return await _op(lambda: ops.prune_chat(state, group_id, max_age_days)) + # ── GEK initialization (operator only, localhost) ────────────────────── @app.post("/api/groups/{group_id}/gek") diff --git a/packages/meshbay-node/tests/test_chat_encryption.py b/packages/meshbay-node/tests/test_chat_encryption.py new file mode 100644 index 0000000..ea4de2f --- /dev/null +++ b/packages/meshbay-node/tests/test_chat_encryption.py @@ -0,0 +1,517 @@ +""" +Chat encryption: what the node stores, what it refuses, and what survives. + +Design A of `docs/chat-sender-keys.md`. Every test here is written as "this +does not work" or "this still works after X" — the regressions the plan's +register names, in the order they would bite. + +The load-bearing ones are the last three. Rotation is the failure the design +exists to avoid: a chat key derived from the group key would have made every +message ever sent unreadable on the first `member unpin`, for everybody, +including the operator, and that is the *documented* procedure after removing +someone. Key storage is the failure that would make the whole feature a +decoration. Downgrade is C6's lesson, one feature later. +""" + +import base64 +import time +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.chatbox import open_message, seal +from meshbay_common.crypto import generate_gek +from meshbay_common.groupbox import PURPOSE_CHAT_KEYS, unseal +from meshbay_common.protocol import MNP +from meshbay_node import ops +from meshbay_node.bundle_store import BundleStore +from meshbay_node.chat import FORMAT_SEALED_V1, ChatStore +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import open_roster +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root + +GROUP = "g" * 32 + + +def _device(): + sk = Ed25519PrivateKey.generate() + raw = sk.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + return sk, raw, base64.b64encode(raw).decode() + + +@pytest.fixture +async def node(tmp_path): + """A daemon state with the pieces the chat path actually touches.""" + roster = await open_roster(tmp_path) + bundles = BundleStore(tmp_path / "bundles.db") + await bundles.open() + chat = ChatStore(tmp_path / "chat.db") + await chat.open() + + sk_x = Ed25519PrivateKey.generate() # stand-in shape; X25519 below + from cryptography.hazmat.primitives.asymmetric.x25519 import ( + X25519PrivateKey, + ) + sk_x = X25519PrivateKey.generate() + sk_x_raw = sk_x.private_bytes( + serialization.Encoding.Raw, serialization.PrivateFormat.Raw, + serialization.NoEncryption()) + pk_x_raw = sk_x.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + gek = generate_gek() + + group_ctx = { + "gek": gek, "index": index, "roots": one_root(shared), + "chat_store": chat, "chat_epoch": 0, + "_peers": {}, + } + state = { + "roster": roster, "bundle_store": bundles, + "sk_x25519_raw": sk_x_raw, "pk_x25519_raw": pk_x_raw, + "groups_ctx": {GROUP: group_ctx}, "node_user_id": "operator", + } + yield {"state": state, "group_ctx": group_ctx, "gek": gek, + "chat": chat, "roster": roster, "bundles": bundles, + "index": index, "tmp_path": tmp_path} + await chat.close() + await bundles.close() + await roster.close() + + +def _session(node, user_id="alice", device_b64=""): + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = {"groups": {GROUP: node["group_ctx"]}, + "daemon_state": node["state"]} + session._group_id = GROUP + session._user_id = user_id + session._username = user_id + session._pinned_pk = device_b64 + session._device_confirmed = bool(device_b64) + session._registry_key = f"conn-{user_id}-{len(node['group_ctx']['_peers'])}" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +async def _drain(session, coro_holder): + """`_spawn` stubbed to await inline, so a test sees the store written.""" + pass + + +def _spawn_inline(session): + import asyncio + + pending = [] + session._spawn = lambda coro: pending.append( + asyncio.get_event_loop().create_task(coro)) + return pending + + +async def _send_sealed(node, session, sk, device_raw, device_b64, text, + epoch=None): + keys = await ops.chat_epoch_keys(node["state"], GROUP) + epoch = epoch or keys[-1]["epoch"] + key = next(k["key"] for k in keys if k["epoch"] == epoch) + env = seal(key, GROUP, epoch, device_b64, device_raw, sk, + {"text": text, "sender_name": session._user_id}) + pending = _spawn_inline(session) + session._do_chat_message({ + "format": FORMAT_SEALED_V1, "epoch": epoch, "device": device_raw, + "ct": env["ct"], "nonce": env["nonce"], "sig": env["sig"], + }) + for task in pending: + await task + return env + + +# ── the archive survives what would destroy it ─────────────────────────────── + +async def test_history_survives_a_group_key_rotation(node): + """ + R1, and the reason Design A exists. + + A chat key derived from the group key would be gone the moment the operator + rotates — which is the documented step after removing a member. Every + message ever sent would become unreadable, for everybody. The epoch key is + wrapped under the group key *at delivery* and never stored under it, so a + rotation is a re-wrap and costs nothing. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + sk, raw, b64 = _device() + session = _session(node, device_b64=b64) + await _send_sealed(node, session, sk, raw, b64, "before the rotation") + + # Rotate the group key, exactly as the operator does after a removal. + node["group_ctx"]["gek"] = generate_gek() + + keys = await ops.chat_epoch_keys(node["state"], GROUP) + stored = (await node["chat"].get_recent(10))[0] + opened = open_message( + next(k["key"] for k in keys if k["epoch"] == stored.epoch), + GROUP, stored.epoch, b64, stored.nonce, stored.payload) + assert opened["text"] == "before the rotation", ( + "rotating the group key must not make the chat archive unreadable — " + "F4, and the whole reason the epoch key is not derived from it") + + +async def test_a_new_epoch_does_not_orphan_the_old_ones(node): + """ + R2. Opening an epoch stops a removed member reading what comes *next*; it + must leave what they could already read readable to everybody else. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + sk, raw, b64 = _device() + session = _session(node, device_b64=b64) + await _send_sealed(node, session, sk, raw, b64, "epoch one") + + await ops.open_chat_epoch(node["state"], GROUP) + await _send_sealed(node, session, sk, raw, b64, "epoch two") + + keys = {k["epoch"]: k["key"] + for k in await ops.chat_epoch_keys(node["state"], GROUP)} + assert len(keys) == 2 + texts = [] + for m in await node["chat"].get_recent(10): + texts.append(open_message(keys[m.epoch], GROUP, m.epoch, b64, + m.nonce, m.payload)["text"]) + assert texts == ["epoch one", "epoch two"] + + +async def test_an_epoch_key_is_never_written_in_the_clear(node): + """ + R15. The claim chat encryption makes is against someone who takes the + node's storage *without the keystore password*. An epoch key sitting in a + plaintext SQLite beside chat.db would collapse that to nothing, silently, + and it is the obvious thing to write. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + keys = await ops.chat_epoch_keys(node["state"], GROUP) + assert keys + + live = keys[-1]["key"] + for path in sorted(node["tmp_path"].rglob("*")): + if not path.is_file(): + continue + assert live not in path.read_bytes(), ( + f"the live chat epoch key appears verbatim in {path.name} — " + "it must be wrapped to the node's own key, as the GEK is") + + +async def test_the_stored_message_contains_neither_text_nor_display_name(node): + """ + What "encrypted at rest" has to mean. The display name is inside the + envelope too: on the wire it is a field any peer can set to anything, and + the node caches it to render history, so leaving it outside would both + leak it and leave spoofing free. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + sk, raw, b64 = _device() + session = _session(node, device_b64=b64) + await _send_sealed(node, session, sk, raw, b64, "a secret message") + + blob = (node["tmp_path"] / "chat.db").read_bytes() + assert b"a secret message" not in blob + stored = (await node["chat"].get_recent(10))[0] + assert stored.format == FORMAT_SEALED_V1 + assert b"a secret message" not in stored.payload + + +# ── refusals ───────────────────────────────────────────────────────────────── + +async def test_plaintext_is_refused_always(node): + """ + R5 / C6's lesson one feature later, and now unconditional: there is no + switch to leave in the wrong position. A member who can post in clear into + a group whose members believe their chat is encrypted is a downgrade anyone + could ask for. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + session = _session(node) + _spawn_inline(session) + session._do_chat_message({"payload": "in the clear", "sender_name": "alice"}) + + assert session.sent[-1]["type"] == "error" + assert await node["chat"].message_count() == 0 + + +async def test_there_is_no_setting_that_re_enables_plaintext(node): + """ + The switch is gone, not defaulted. A `chat_encrypted` in the group context + — left by an older node's roster row, or invented by anything reading one — + must not be consulted, or the bypass is back with a name. + """ + node["group_ctx"]["chat_encrypted"] = False + await ops.ensure_chat_epoch(node["state"], GROUP) + session = _session(node) + _spawn_inline(session) + session._do_chat_message({"payload": "in the clear", "sender_name": "alice"}) + + assert session.sent[-1]["type"] == "error" + assert await node["chat"].message_count() == 0 + + source = (Path(__file__).parent.parent / "src" / "meshbay_node" + / "transport" / "webrtc_server.py").read_text(encoding="utf-8") + assert 'get("chat_encrypted"' not in source, ( + "nothing may read a chat_encrypted setting — there is no switch") + + +async def test_a_member_cannot_send_as_another_members_device(node): + """ + The hole that would have made encrypted chat *worse* than plaintext chat. + + Receivers verify a signature against the `device` field, so a member free + to name somebody else's key could be that member to everyone — which is + exactly what `GroupSenderKeyStore.add_sender` allowed, one design earlier + (`docs/chat-sender-keys.md` F1). The connection has proved which device it + is, and the claim must match it. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + _sk_alice, raw_alice, b64_alice = _device() + sk_mallory, raw_mallory, b64_mallory = _device() + + session = _session(node, user_id="mallory", device_b64=b64_mallory) + keys = await ops.chat_epoch_keys(node["state"], GROUP) + epoch, key = keys[-1]["epoch"], keys[-1]["key"] + # Correctly sealed and correctly signed — by Mallory, claiming to be Alice. + env = seal(key, GROUP, epoch, b64_alice, raw_alice, sk_mallory, + {"text": "not from alice"}) + _spawn_inline(session) + session._do_chat_message({ + "format": FORMAT_SEALED_V1, "epoch": epoch, "device": raw_alice, + "ct": env["ct"], "nonce": env["nonce"], "sig": env["sig"], + }) + + assert session.sent[-1]["type"] == "error" + assert await node["chat"].message_count() == 0 + + +async def test_a_signed_message_cannot_be_replayed(node): + """ + A replay is a *validly signed* copy of a real message, so nothing about + the signature refuses it. The unique (device, nonce) does — and the nonce + is already required to be unique for AES-GCM to be safe, so it costs + nothing to make it a key. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + sk, raw, b64 = _device() + session = _session(node, device_b64=b64) + env = await _send_sealed(node, session, sk, raw, b64, "said once") + assert await node["chat"].message_count() == 1 + + keys = await ops.chat_epoch_keys(node["state"], GROUP) + pending = _spawn_inline(session) + session._do_chat_message({ + "format": FORMAT_SEALED_V1, "epoch": keys[-1]["epoch"], "device": raw, + "ct": env["ct"], "nonce": env["nonce"], "sig": env["sig"], + }) + for task in pending: + await task + assert await node["chat"].message_count() == 1, ( + "a replayed message must not be stored twice") + + +async def test_an_unidentified_connection_cannot_send_a_signed_message(node): + """ + `device_hello` is what makes "that is not the device on this connection" + checkable at all. Without it the node knows the account and not the key, + and a `device` field would be an assertion nobody verified. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + sk, raw, b64 = _device() + session = _session(node) # no device_hello + keys = await ops.chat_epoch_keys(node["state"], GROUP) + epoch, key = keys[-1]["epoch"], keys[-1]["key"] + env = seal(key, GROUP, epoch, b64, raw, sk, {"text": "x"}) + _spawn_inline(session) + session._do_chat_message({ + "format": FORMAT_SEALED_V1, "epoch": epoch, "device": raw, + "ct": env["ct"], "nonce": env["nonce"], "sig": env["sig"], + }) + assert session.sent[-1]["type"] == "error" + + +# ── key delivery ───────────────────────────────────────────────────────────── + +async def test_the_keys_are_delivered_sealed_under_the_group_key(node): + """ + Sealed for the same reason the index and the ack are, one step stronger: + the payload *is* key material. A peer that has completed the handshake + holds the group key and can open it; anything short of that gets a + ciphertext. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + await ops.open_chat_epoch(node["state"], GROUP) + session = _session(node) + await session._do_chat_keys_req({}) + + resp = session.sent[-1] + assert resp["type"] == MNP.CHAT_KEYS_RESP + assert "epochs" not in resp, "the keys must not travel in clear" + payload = unseal(node["gek"], PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP, + GROUP, resp) + assert [e["epoch"] for e in payload["epochs"]] == [1, 2] + assert payload["current"] == 2 + for e in payload["epochs"]: + assert len(e["key"]) == 32 + + +async def test_every_epoch_is_delivered_not_just_the_current_one(node): + """ + R2 again, from the delivery side: this is what lets a device linked this + morning read a conversation from last year. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + for _ in range(3): + await ops.open_chat_epoch(node["state"], GROUP) + session = _session(node) + await session._do_chat_keys_req({}) + payload = unseal(node["gek"], PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP, + GROUP, session.sent[-1]) + assert [e["epoch"] for e in payload["epochs"]] == [1, 2, 3, 4] + + +# ── epochs move when access shrinks ───────────────────────────────────────── + +async def test_revoking_a_device_opens_a_new_epoch(node): + """ + A revoked device holds every chat key it ever received. Revocation stops + the node handing over the *next* one; nothing else takes the current one + away — the exact counterpart of "still rotate the GEK". + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + before = await node["bundles"].latest_chat_epoch(GROUP) + session = _session(node) + await session._new_chat_epoch(GROUP, "device_revoke") + assert await node["bundles"].latest_chat_epoch(GROUP) == before + 1 + + +async def test_a_group_always_gets_an_epoch(node): + """ + Chat is always encrypted, so a group with no epoch key is a group nobody + can speak in. `ensure_chat_epoch` is what the daemon calls at group load — + at start-up, where a failure lands in the log the operator is already + reading rather than on somebody's first message. + """ + assert await node["bundles"].latest_chat_epoch(GROUP) == 0 + epoch = await ops.ensure_chat_epoch(node["state"], GROUP) + assert epoch == 1 + # Idempotent: called at every group load, and a second epoch per restart + # would be a key nobody needed and the node keeps for ever. + assert await ops.ensure_chat_epoch(node["state"], GROUP) == 1 + + +async def test_an_epoch_key_is_never_deleted(node): + """ + Nothing in the system removes an epoch key, and nothing may: the messages + sealed under it become unreadable the moment it goes, for everybody. The + only operation that touches the table adds a row. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + sk, raw, b64 = _device() + session = _session(node, device_b64=b64) + await _send_sealed(node, session, sk, raw, b64, "still readable") + await ops.open_chat_epoch(node["state"], GROUP) + await ops.prune_chat(node["state"], GROUP, 3650) + + keys = await ops.chat_epoch_keys(node["state"], GROUP) + assert [k["epoch"] for k in keys] == [1, 2] + + source = (Path(__file__).parent.parent / "src" / "meshbay_node" + / "bundle_store.py").read_text(encoding="utf-8") + assert "DELETE FROM chat_epochs" not in source + assert "INSERT OR REPLACE INTO chat_epochs" not in source, ( + "an epoch key is written once — REPLACE would destroy the history " + "sealed under it, with no error anywhere") + + +# ── the explicit history migration, and retention ─────────────────────────── + +async def test_encrypt_history_converts_the_old_plaintext(node): + """ + The migration for a node that ran before MNP 2.0. + + The plaintext row is written straight into the store, because that is the + only way one can exist now: `_do_chat_message` refuses plaintext outright. + Such rows are the ones still readable off a stolen disk, and the node can + convert them only because it holds them in the clear — it is the last + moment at which anyone can. + """ + node["state"]["sk_node"] = Ed25519PrivateKey.generate() + await node["chat"].save_message( + sender_id="alice", iteration=0, payload=b"written in the clear", + sender_name="alice") + await ops.ensure_chat_epoch(node["state"], GROUP) + + result = await ops.encrypt_chat_history(node["state"], GROUP) + + assert result["converted"] == 1 + stored = (await node["chat"].get_recent(10))[0] + assert stored.format == FORMAT_SEALED_V1 + assert b"written in the clear" not in stored.payload + assert stored.sender_name == "", ( + "the display name moves inside the envelope — leaving it would keep in " + "the clear the one field the sealing was for") + + keys = {k["epoch"]: k["key"] + for k in await ops.chat_epoch_keys(node["state"], GROUP)} + device_b64 = base64.b64encode(stored.device).decode() + opened = open_message(keys[stored.epoch], GROUP, stored.epoch, device_b64, + stored.nonce, stored.payload) + assert opened["text"] == "written in the clear" + assert opened["sender_name"] == "alice" + assert opened["migrated"] is True, ( + "a migrated message carries the node's word for who wrote it, which is " + "all it ever carried — that has to be visible, not inferred") + + +async def test_encrypt_history_backs_the_database_up_first(node): + node["state"]["sk_node"] = Ed25519PrivateKey.generate() + await node["chat"].save_message( + sender_id="alice", iteration=0, payload=b"one", sender_name="alice") + await ops.ensure_chat_epoch(node["state"], GROUP) + + result = await ops.encrypt_chat_history(node["state"], GROUP) + + from pathlib import Path + backup = Path(result["backup"]) + assert backup.exists() and backup.stat().st_size > 0 + assert b"one" in backup.read_bytes(), ( + "the backup is taken before the rewrite, or it is not a backup") + + +async def test_retention_deletes_messages_and_never_epoch_keys(node): + """ + R16. An epoch whose messages have all aged out costs 32 bytes; deleting it + would make anything still stored under it unreadable. + """ + await ops.ensure_chat_epoch(node["state"], GROUP) + sk, raw, b64 = _device() + session = _session(node, device_b64=b64) + await _send_sealed(node, session, sk, raw, b64, "old news") + + # Age it past the cutoff. + await node["chat"]._db.execute( + "UPDATE messages SET timestamp = ?", (time.time() - 40 * 86400,)) + await node["chat"].commit() + + result = await ops.prune_chat(node["state"], GROUP, 30) + + assert result["removed"] == 1 + assert await node["chat"].message_count() == 0 + assert await ops.chat_epoch_keys(node["state"], GROUP), ( + "retention deletes messages, never keys") + + +async def test_retention_refuses_a_zero_day_window(node): + """`prune 0` would delete the whole conversation and read as a typo.""" + with pytest.raises(ops.OpError): + await ops.prune_chat(node["state"], GROUP, 0) diff --git a/packages/meshbay-node/tests/test_chat_history_binary.py b/packages/meshbay-node/tests/test_chat_history_binary.py new file mode 100644 index 0000000..18efbf5 --- /dev/null +++ b/packages/meshbay-node/tests/test_chat_history_binary.py @@ -0,0 +1,181 @@ +""" +A ciphertext must survive the history path. + +`_send_chat_history` used to put every stored payload through +`.decode("utf-8", errors="replace")`, which substitutes U+FFFD for every byte +that is not valid UTF-8 — i.e. for most of a ciphertext. Live messages are +relayed rather than re-read, so they would have kept working: the symptom would +have been "history will not decrypt" and nothing else, which is the hardest +possible place to look for a wire-format error. + +The fix keeps plaintext exactly where it has always been (a string in +`payload`, which older clients read) and gives ciphertext its own `ct` field. +That way this is not a compatibility break either — `docs/chat-sender-keys.md` +R3. +""" + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.protocol import MNP +from meshbay_node.chat import FORMAT_PLAIN, FORMAT_SEALED_V1, ChatStore +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root + +GROUP = "g" * 32 + +# Deliberately not valid UTF-8: a lone continuation byte, an over-long form and +# a bare 0xff, which is what a random AES-GCM ciphertext is full of. +CIPHERTEXT = bytes([0x80, 0xff, 0xc0, 0x80, 0xfe, 0x00, 0x41, 0xed, 0xa0, 0x80]) + + +@pytest.fixture +async def store(tmp_path): + s = ChatStore(tmp_path / "chat.db") + await s.open() + yield s + await s.close() + + +def _session(store, tmp_path): + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = {"groups": {GROUP: { + "index": index, "roots": one_root(shared), "chat_store": store}}} + session._group_id = GROUP + session._user_id = "alice" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +async def test_a_ciphertext_survives_the_history_path(store, tmp_path): + await store.save_message( + sender_id="alice", iteration=0, payload=CIPHERTEXT, + format=FORMAT_SEALED_V1, epoch=1, device=b"\x01" * 32, + nonce=b"\x02" * 12, sig=b"\x03" * 64) + + session = _session(store, tmp_path) + await session._send_chat_history(store, None, 50) + + resp = session.sent[-1] + assert resp["type"] == MNP.CHAT_HISTORY_RESPONSE + row = resp["messages"][0] + assert row["ct"] == CIPHERTEXT, ( + "the ciphertext must come back byte for byte — decoded as UTF-8 with " + "errors='replace' it comes back as U+FFFD and nothing decrypts") + assert row["format"] == FORMAT_SEALED_V1 + assert row["epoch"] == 1 + assert row["nonce"] == b"\x02" * 12 + assert row["sig"] == b"\x03" * 64 + + +async def test_plaintext_history_keeps_the_shape_older_clients_read( + store, tmp_path): + """ + The compatibility half. The UI ships inside the desktop package now, so a + client can be months behind the node; a plaintext message must still arrive + as a string under `payload`, exactly as it always has. + """ + await store.save_message(sender_id="alice", iteration=0, + payload="bonjour ç'est moi".encode()) + + session = _session(store, tmp_path) + await session._send_chat_history(store, None, 50) + + row = session.sent[-1]["messages"][0] + assert row["payload"] == "bonjour ç'est moi" + assert isinstance(row["payload"], str) + assert row["format"] == FORMAT_PLAIN + assert "ct" not in row + + +async def test_a_mixed_history_reads_both_ways(store, tmp_path): + """ + R4: rows written before a group turned encryption on keep rendering. The + switch never rewrites anything, so every group that turns it on has a + history of both kinds for ever. + """ + await store.save_message(sender_id="alice", iteration=0, payload=b"before") + await store.save_message( + sender_id="alice", iteration=0, payload=CIPHERTEXT, + format=FORMAT_SEALED_V1, epoch=1, device=b"\x01" * 32, + nonce=b"\x02" * 12, sig=b"\x03" * 64) + + session = _session(store, tmp_path) + await session._send_chat_history(store, None, 50) + + rows = session.sent[-1]["messages"] + assert [r["format"] for r in rows] == [FORMAT_PLAIN, FORMAT_SEALED_V1] + assert rows[0]["payload"] == "before" + assert rows[1]["ct"] == CIPHERTEXT + + +async def test_an_existing_database_opens_and_keeps_its_rows(tmp_path): + """ + The migration, from the only angle that matters: a chat.db written before + the new columns existed must open, keep every row, and read back as + plaintext. `CREATE TABLE IF NOT EXISTS` adds no column to a table that is + already there — the same trap `create_all()` is recorded for on the hub. + """ + import aiosqlite + + path = tmp_path / "old_chat.db" + async with aiosqlite.connect(str(path)) as db: + await db.execute( + "CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT, " + "sender_id TEXT NOT NULL, iteration INTEGER NOT NULL, " + "payload BLOB NOT NULL, timestamp REAL NOT NULL, " + "thread_id TEXT DEFAULT NULL, sender_name TEXT DEFAULT '')") + await db.execute( + "INSERT INTO messages (sender_id, iteration, payload, timestamp) " + "VALUES ('alice', 0, ?, 1700000000.0)", (b"an old message",)) + await db.commit() + + store = ChatStore(path) + await store.open() + try: + rows = await store.get_recent(10) + assert len(rows) == 1 + assert rows[0].payload == b"an old message" + assert rows[0].format == FORMAT_PLAIN + assert rows[0].epoch == 0 + assert rows[0].device is None + # And it is still writable, including with the new columns. + await store.save_message( + sender_id="bob", iteration=0, payload=CIPHERTEXT, + format=FORMAT_SEALED_V1, epoch=1, device=b"\x09" * 32, + nonce=b"\x08" * 12, sig=b"\x07" * 64) + assert await store.message_count() == 2 + finally: + await store.close() + + +async def test_opening_twice_keeps_every_column(tmp_path): + """ + The migrations are swallowed per statement, not per batch: one 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. + """ + path = tmp_path / "twice.db" + for _ in range(2): + store = ChatStore(path) + await store.open() + await store.close() + + store = ChatStore(path) + await store.open() + try: + await store.save_message( + sender_id="alice", iteration=0, payload=CIPHERTEXT, + format=FORMAT_SEALED_V1, epoch=3, device=b"\x01" * 32, + nonce=b"\x02" * 12, sig=b"\x03" * 64) + row = (await store.get_recent(1))[0] + assert row.epoch == 3 and row.sig == b"\x03" * 64 + finally: + await store.close() diff --git a/packages/meshbay-node/tests/test_chat_multidevice.py b/packages/meshbay-node/tests/test_chat_multidevice.py new file mode 100644 index 0000000..d718b2a --- /dev/null +++ b/packages/meshbay-node/tests/test_chat_multidevice.py @@ -0,0 +1,160 @@ +""" +One account, several devices, on one node. + +Device linking (2026-08-18) made `identities` a table keyed by +`(user_id, pk_ed25519)`, so a person legitimately holds several keys here. The +chat path never followed: the peer registry was keyed by `user_id`, so the +second connection of one account **evicted the first**, and the broadcast loop +skipped recipients by account, so a person's own other devices never received +what they said. + +Neither shows up as an error anywhere. The first is a message that silently +reaches nobody after a second device connects and disconnects; the second is a +phone that never shows what was typed on the laptop. Both are +`docs/chat-sender-keys.md` F7, and both are the same "keyed by account where it +should be keyed by connection" mistake as `pin_identity`'s old INSERT OR REPLACE. +""" + +from pathlib import Path + +import base64 +import hashlib + +from aiortc import RTCPeerConnection +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root + + +def _sealed(device_raw: bytes, text: bytes = b"ciphertext") -> dict: + """A well-formed sealed envelope. + + The bytes are not a real ciphertext and do not need to be: the node never + opens one. What it *does* check is the envelope's shape and that the device + is the connection's own, and going through the real `_do_chat_message` + rather than around it is the point — these tests are about delivery, and + delivery now runs after that check. + """ + return {"format": 1, "epoch": 1, "device": device_raw, "ct": text, + "nonce": b"\x02" * 12, "sig": b"\x03" * 64} + + +def _session(ctx: dict, user_id: str, group_id: str) -> WebRTCPeerSession: + """A peer session with only what the chat path touches wired up. + + Built through the real `__init__` — with a bare RTCPeerConnection, which + costs under a millisecond and opens no socket — so `_registry_key` is the + one production assigns. Constructing it in the test instead would make + these tests agree with the fix by construction, which is precisely the + trap the repo's own notes record. + """ + session = WebRTCPeerSession(pc=RTCPeerConnection(), node_ctx=ctx) + session._group_id = group_id + session._user_id = user_id + session._username = user_id + # Each connection is a distinct device of that account — which is the whole + # subject here, and what `_check_chat_envelope` compares a message against. + session._pinned_pk = base64.b64encode(_device_raw(session)).decode() + session._device_confirmed = True + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + session._spawn = lambda coro: coro.close() + return session + + +def _ctx(tmp_path: Path, group_id: str) -> dict: + index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate()) + root = tmp_path / "shared" + root.mkdir(exist_ok=True) + return {"groups": {group_id: { + "index": index, "roots": one_root(root), "chat_store": None, + }}} + + +GROUP = "g" * 32 + + +def _device_raw(session) -> bytes: + """A stable 32-byte stand-in for this connection's device key. + + Derived from the registry key, so two sessions of one account get two + devices — which is exactly the situation being tested, and a shared one + would make the envelope check pass for the wrong reason. + """ + return hashlib.sha256(session._registry_key.encode()).digest() + + +def test_two_devices_of_one_account_both_stay_registered(tmp_path): + """ + F7: keyed by `user_id`, the second device overwrote the first, and closing + either then removed the other's entry — so one person's two devices could + never both be reachable. + """ + ctx = _ctx(tmp_path, GROUP) + laptop = _session(ctx, "alice", GROUP) + phone = _session(ctx, "alice", GROUP) + + laptop._register_peer() + phone._register_peer() + + registry = laptop._peer_registry() + assert len(registry) == 2, ( + "one account's two devices must both be in the registry; keyed by " + "user_id the second silently replaced the first") + assert set(registry.values()) == {laptop, phone} + + +def test_a_message_reaches_the_senders_other_device(tmp_path): + """ + F7, the visible half: the broadcast excluded recipients whose `user_id` + matched the sender's, so everything typed on the laptop was missing from + the phone — with no error and no way to notice but to hold both. + """ + ctx = _ctx(tmp_path, GROUP) + laptop = _session(ctx, "alice", GROUP) + phone = _session(ctx, "alice", GROUP) + bob = _session(ctx, "bob", GROUP) + for s in (laptop, phone, bob): + s._register_peer() + laptop._user_names = lambda: ctx["groups"][GROUP].setdefault("_names", {}) + + laptop._do_chat_message(_sealed(_device_raw(laptop), b"hello-ciphertext")) + + def chats(session): + return [m for m in session.sent if m.get("type") == "chat_msg"] + + assert len(chats(phone)) == 1, ( + "the sender's other device is an ordinary recipient — it composed " + "nothing and has no local echo to fall back on") + assert chats(phone)[0]["ct"] == b"hello-ciphertext" + assert len(chats(bob)) == 1 + assert chats(laptop) == [], "the composing connection must not echo to itself" + + +def test_closing_one_device_leaves_the_other_connected(tmp_path): + """ + The teardown half. `close()` popped `self._user_id`, so the phone + disconnecting unregistered the laptop, which then received nothing for the + rest of its session while still reading as connected. + """ + ctx = _ctx(tmp_path, GROUP) + laptop = _session(ctx, "alice", GROUP) + phone = _session(ctx, "alice", GROUP) + for s in (laptop, phone): + s._register_peer() + + phone._unregister_peer() + + assert laptop._registry_key in laptop._peer_registry() + assert len(laptop._peer_registry()) == 1 + + +def test_registry_key_is_per_connection_not_per_account(tmp_path): + """The property the two tests above depend on, asserted directly.""" + ctx = _ctx(tmp_path, GROUP) + a = _session(ctx, "alice", GROUP) + b = _session(ctx, "alice", GROUP) + assert a._registry_key != b._registry_key diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py index cf91564..6f43772 100644 --- a/packages/meshbay-node/tests/test_cli_dispatch.py +++ b/packages/meshbay-node/tests/test_cli_dispatch.py @@ -49,6 +49,11 @@ VERBS = [ ["file", "list"], ["file", "rm", "abc", "--yes"], ["video", "rematch", "--yes"], + ["chat", "status"], + ["chat", "rotate"], + ["chat", "encrypt-history", "--yes"], + ["chat", "prune", "30"], + ["chat", "prune"], # missing days: usage, then exit ["denylist", "show"], ["denylist", "clear", "--yes"], ["stun", "list"], @@ -78,6 +83,12 @@ def stub_daemon(monkeypatch, tmp_path): "user_id": "u", "authorized_members": 0, "errors": [], "name": "g", "group_id": "g", "shared_dir": str(tmp_path), "config": str(tmp_path / "node.toml"), + # Chat encryption: the switch's answer, the epoch a rotation + # opened, and what a history re-encryption reports. + "enabled": False, "epoch": 1, "converted": 0, + "backup": str(tmp_path / "chat.db.bak"), + "encrypted": False, "plaintext_messages": 0, + "encrypted_messages": 0, "max_age_days": 30, } monkeypatch.setattr(daemon_mod, "_daemon_api", fake_api) diff --git a/packages/meshbay-node/tests/test_device_on_connection.py b/packages/meshbay-node/tests/test_device_on_connection.py new file mode 100644 index 0000000..3da8a8c --- /dev/null +++ b/packages/meshbay-node/tests/test_device_on_connection.py @@ -0,0 +1,287 @@ +""" +Which device is on this connection, and what depends on knowing. + +The MNP handshake authenticates a *group membership* (the GEK-HMAC) and an +*account* (the hub's token). It has never authenticated a device. While one +person meant one key on a node those were the same statement; device linking +(2026-08-18) ended that, and two things were left resolving "the account's +oldest live device" and calling it the answer: + + * `_load_pinned_pk`, whose result is recorded as `entry.uploader_pk` on every + upload — so a phone's uploads were attributed to a laptop; + * `_admin_exec_file_delete`, which authorized deletion against **that exact + key** — so a person could not delete their own file from their other device, + and the only symptom was "Signature verification failed" on their own upload + (`docs/desktop-client-v1.md` §4.8 A). + +`device_hello` closes the first: additive, signed, refused unless the key is a +live device *of this account in the node's own roster*. The second is closed by +authorizing against the account rather than the key. +""" + +import base64 +import time + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import pk_to_b64 +from meshbay_common.device import device_hello_transcript +from meshbay_common.protocol import MNP +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import open_roster +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root + +GROUP = "g" * 32 +NONCE = b"\x11" * 32 + + +@pytest.fixture +async def roster(tmp_path): + r = await open_roster(tmp_path) + yield r + await r.close() + + +def _keys(): + sk_ed = Ed25519PrivateKey.generate() + sk_x = Ed25519PrivateKey.generate() # stand-in; only its b64 is used + return sk_ed, pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key()) + + +def _session(tmp_path, roster, user_id="alice"): + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = { + "roots": one_root(shared), "index": index, "sk_node": index.sk_node, + "roster": roster, + "groups": {GROUP: {"gek": b"\x01" * 32, "index": index, + "roots": one_root(shared)}}, + } + session._group_id = GROUP + session._user_id = user_id + session._username = user_id + session._pinned_pk = "" + session._device_confirmed = False + session._nonce_node = NONCE + session._remote_ip = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +async def _hello(session, sk, pk_ed, *, ts=None, user_id=None): + ts = int(time.time()) if ts is None else ts + transcript = device_hello_transcript( + node_pk_b64=session._node_pk_b64(), group_id=session._group_id, + user_id=user_id or session._user_id, pk_ed25519_b64=pk_ed, + nonce_node=NONCE, ts=ts) + await session._do_device_hello({ + "pk_ed25519": pk_ed, "ts": ts, + "sig": base64.b64encode(sk.sign(transcript)).decode(), + }) + + +def _last(session): + return session.sent[-1] if session.sent else {} + + +# ── device_hello ───────────────────────────────────────────────────────────── + +async def test_a_pinned_device_identifies_itself(tmp_path, roster): + sk_a, pk_ed_a, pk_x_a = _keys() + sk_b, pk_ed_b, pk_x_b = _keys() + await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code") + await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device") + + session = _session(tmp_path, roster) + await _hello(session, sk_b, pk_ed_b) + + assert _last(session)["type"] == MNP.DEVICE_HELLO_ACK + assert session._pinned_pk == pk_ed_b, ( + "the connection must be the device that signed, not the account's " + "oldest key") + assert session._device_confirmed is True + + +async def test_a_key_this_node_never_pinned_is_refused(tmp_path, roster): + sk_a, pk_ed_a, pk_x_a = _keys() + await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code") + sk_x, pk_ed_x, _ = _keys() + + session = _session(tmp_path, roster) + await _hello(session, sk_x, pk_ed_x) + + assert _last(session)["type"] == "error" + assert session._device_confirmed is False + + +async def test_a_revoked_device_cannot_identify_itself(tmp_path, roster): + sk_a, pk_ed_a, pk_x_a = _keys() + sk_b, pk_ed_b, pk_x_b = _keys() + await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code") + await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device") + await roster.revoke_device("alice", pk_ed_b) + + session = _session(tmp_path, roster) + await _hello(session, sk_b, pk_ed_b) + + assert _last(session)["type"] == "error", ( + "a revoked key must stay refused — that is why revocation marks the " + "row instead of deleting it") + + +async def test_another_accounts_device_cannot_identify_here(tmp_path, roster): + """The roster lookup is scoped to *this* account, never to the key alone.""" + sk_a, pk_ed_a, pk_x_a = _keys() + sk_m, pk_ed_m, pk_x_m = _keys() + await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code") + await roster.pin_identity("mallory", "mallory", pk_ed_m, pk_x_m, via="code") + + session = _session(tmp_path, roster, user_id="alice") + await _hello(session, sk_m, pk_ed_m) + + assert _last(session)["type"] == "error" + + +async def test_a_signature_for_another_connection_does_not_transfer( + tmp_path, roster): + """`nonce_node` binds the statement to one connection (L4).""" + sk_a, pk_ed_a, pk_x_a = _keys() + await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code") + + session = _session(tmp_path, roster) + ts = int(time.time()) + transcript = device_hello_transcript( + node_pk_b64=session._node_pk_b64(), group_id=GROUP, + user_id="alice", pk_ed25519_b64=pk_ed_a, + nonce_node=b"\x99" * 32, ts=ts) # another connection's nonce + await session._do_device_hello({ + "pk_ed25519": pk_ed_a, "ts": ts, + "sig": base64.b64encode(sk_a.sign(transcript)).decode(), + }) + + assert _last(session)["type"] == "error" + + +async def test_a_stale_hello_is_refused(tmp_path, roster): + sk_a, pk_ed_a, pk_x_a = _keys() + await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code") + session = _session(tmp_path, roster) + await _hello(session, sk_a, pk_ed_a, ts=int(time.time()) - 3600) + assert _last(session)["type"] == "error" + + +async def test_a_connection_cannot_become_a_second_device(tmp_path, roster): + """ + Both keys are legitimately this account's, and it is still refused: one + connection's uploads must be attributable to one device. + """ + sk_a, pk_ed_a, pk_x_a = _keys() + sk_b, pk_ed_b, pk_x_b = _keys() + await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code") + await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device") + + session = _session(tmp_path, roster) + await _hello(session, sk_a, pk_ed_a) + assert _last(session)["type"] == MNP.DEVICE_HELLO_ACK + await _hello(session, sk_b, pk_ed_b) + assert _last(session)["type"] == "error" + assert session._pinned_pk == pk_ed_a + + +async def test_the_late_roster_load_does_not_undo_a_confirmed_device( + tmp_path, roster): + """ + `_load_pinned_pk` is spawned at handshake and can finish *after* a fast + client has identified itself. It must not put the account's oldest key back + — a race that would have been intermittent and attributed to nothing. + """ + sk_a, pk_ed_a, pk_x_a = _keys() + sk_b, pk_ed_b, pk_x_b = _keys() + await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code") + await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device") + + session = _session(tmp_path, roster) + await _hello(session, sk_b, pk_ed_b) + await session._load_pinned_pk() # arrives late + + assert session._pinned_pk == pk_ed_b + + +# ── deletion is authorized by account, not by the exact device ─────────────── + +class _Entry: + def __init__(self, uploader_id="", uploader_pk=""): + self.uploader_id = uploader_id + self.uploader_pk = uploader_pk + + +async def test_a_second_device_can_delete_the_first_devices_upload( + tmp_path, roster): + """ + §4.8 A. Alice uploads from her phone and deletes from her desktop. Before + the fix this failed with "Signature verification failed" on her own file. + """ + sk_phone, pk_phone, pk_x_phone = _keys() + sk_desk, pk_desk, pk_x_desk = _keys() + await roster.pin_identity("alice", "alice", pk_phone, pk_x_phone, via="code") + await roster.pin_identity("alice", "alice", pk_desk, pk_x_desk, via="device") + + session = _session(tmp_path, roster) + entry = _Entry(uploader_id="alice", uploader_pk=pk_phone) + transcript = b"delete-this-file" + + assert await session._verify_uploader_sig( + entry, transcript, sk_desk.sign(transcript)) is True + + +async def test_a_stranger_still_cannot_delete_someone_elses_upload( + tmp_path, roster): + sk_alice, pk_alice, pk_x_alice = _keys() + sk_mallory, pk_mallory, pk_x_mallory = _keys() + await roster.pin_identity("alice", "alice", pk_alice, pk_x_alice, via="code") + await roster.pin_identity("mallory", "mallory", pk_mallory, pk_x_mallory, + via="code") + + session = _session(tmp_path, roster) + entry = _Entry(uploader_id="alice", uploader_pk=pk_alice) + transcript = b"delete-this-file" + + assert await session._verify_uploader_sig( + entry, transcript, sk_mallory.sign(transcript)) is False + + +async def test_a_revoked_device_can_no_longer_delete(tmp_path, roster): + """Ownership survives revocation; the revoked *key* stops being able to act.""" + sk_old, pk_old, pk_x_old = _keys() + sk_new, pk_new, pk_x_new = _keys() + await roster.pin_identity("alice", "alice", pk_old, pk_x_old, via="code") + await roster.pin_identity("alice", "alice", pk_new, pk_x_new, via="device") + await roster.revoke_device("alice", pk_old) + + session = _session(tmp_path, roster) + entry = _Entry(uploader_id="alice", uploader_pk=pk_old) + transcript = b"delete-this-file" + + assert await session._verify_uploader_sig( + entry, transcript, sk_old.sign(transcript)) is False + assert await session._verify_uploader_sig( + entry, transcript, sk_new.sign(transcript)) is True, ( + "the file is still Alice's — a retired laptop does not orphan its uploads") + + +async def test_an_entry_with_no_uploader_id_falls_back_to_the_recorded_key( + tmp_path, roster): + """An index written before `uploader_id` existed must not become undeletable.""" + sk_a, pk_a, pk_x_a = _keys() + session = _session(tmp_path, roster) + entry = _Entry(uploader_id="", uploader_pk=pk_a) + transcript = b"delete-this-file" + + assert await session._verify_uploader_sig( + entry, transcript, sk_a.sign(transcript)) is True diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index dc74752..ea13d96 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -244,6 +244,35 @@ async def _open_channel(transport, peer_id): return pc, ch, q +def _sealed_chat(session, text: bytes = b"ciphertext") -> dict: + """ + A chat message in the shape MNP 2.0 requires, on a live session. + + There is no plaintext chat any more, so a test that wants to exercise + delivery has to send a real envelope. The bytes need not be a real + ciphertext — the node never opens one — but the envelope's shape and the + device claim are checked, and the device must be the one this connection + identified itself as. Identifying it here is what `device_hello` does over + the wire; doing it directly keeps this test about chat rather than about + device linking, which `test_device_on_connection.py` covers. + """ + device = hashlib.sha256(session._registry_key.encode()).digest() + session._pinned_pk = base64.b64encode(device).decode() + session._device_confirmed = True + return { + "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, + "format": 1, "epoch": 1, "device": device, "ct": text, + "nonce": b"\x02" * 12, "sig": b"\x03" * 64, + } + + +def _only_session(transport): + """The one live peer session on a transport, for tests that made one.""" + sessions = list(transport._sessions.values()) + assert len(sessions) == 1, f"expected one session, got {len(sessions)}" + return sessions[0] + + async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None, group_id=TEST_GROUP): """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue).""" @@ -574,11 +603,8 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm browser_pc, channel, received = await _setup_peer( transport, sk_hub, gek, "peer-chat") - channel.send(_pack({ - "type": MNP.CHAT_MESSAGE, - "v": MNP_VERSION, - "payload": "hello from browser", - })) + channel.send(_pack(_sealed_chat(_only_session(transport), + b"hello from browser"))) chat_ack = await asyncio.wait_for(received.get(), timeout=5.0) assert chat_ack["type"] == "ack" @@ -593,7 +619,12 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm hist = await asyncio.wait_for(received.get(), timeout=5.0) assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE assert len(hist["messages"]) == 1 - assert hist["messages"][0]["payload"] == "hello from browser" + # The ciphertext comes back under `ct`, byte for byte — `payload` is the + # plaintext field and stays empty for a sealed row. Decoding a ciphertext + # as UTF-8, which the history path used to do, would mangle it. + assert hist["messages"][0]["ct"] == b"hello from browser" + assert hist["messages"][0]["payload"] == "" + assert hist["messages"][0]["format"] == 1 assert hist["messages"][0]["sender_id"] == "user-001" await chat_store.close() @@ -650,17 +681,22 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path) pc_a, ch_a, q_a = await _setup_peer(transport, sk_hub, gek, "peer-A", "user-A") pc_b, ch_b, q_b = await _setup_peer(transport, sk_hub, gek, "peer-B", "user-B") - ch_a.send(_pack({ - "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "hi from A", - })) + session_a = next(s for s in transport._sessions.values() + if s._user_id == "user-A") + ch_a.send(_pack(_sealed_chat(session_a, b"hi from A"))) ack_a = await asyncio.wait_for(q_a.get(), timeout=5.0) assert ack_a["type"] == "ack" broadcast = await asyncio.wait_for(q_b.get(), timeout=5.0) assert broadcast["type"] == MNP.CHAT_MESSAGE + # `sender_id` is still the node's, from the authenticated session (NS6). + # What it now carries beside it is the sending device and a signature over + # the ciphertext, which is what makes the claim checkable by the receiver + # rather than taken on the node's word. assert broadcast["sender_id"] == "user-A" - assert broadcast["payload"] == "hi from A" + assert broadcast["ct"] == b"hi from A" + assert broadcast["device"] == base64.b64decode(session_a._pinned_pk) await chat_store.close() await pc_a.close() @@ -731,12 +767,16 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir): browser_pc, channel, received = await _setup_peer( transport, sk_hub, gek, "peer-cleanup") - assert "user-001" in transport._ctx["_peers"] + # Keyed per connection, not per account (docs/chat-sender-keys.md F7), so + # membership is asserted by the session object rather than by user_id — + # one account may hold several entries here. + peers = transport._ctx["_peers"] + assert [s._user_id for s in peers.values()] == ["user-001"] assert transport.active_peers == 1 await transport.close_peer("peer-cleanup") - assert "user-001" not in transport._ctx["_peers"] + assert transport._ctx["_peers"] == {} assert transport.active_peers == 0 await browser_pc.close() |