summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-07 17:50:28 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-07 17:50:28 +0200
commit36cebf25d0e0f24cf63be4380ccb5d03da726a74 (patch)
tree8509ec4cf68a058f7383299e11bdea97ab06cadf /packages/meshbay-node/src
parent8883d60d0afa2ed9dd1ef68bc21fe1b9a65a59ff (diff)
downloadmeshbay-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/src')
-rw-r--r--packages/meshbay-node/src/meshbay_node/bundle_store.py73
-rw-r--r--packages/meshbay-node/src/meshbay_node/chat/__init__.py20
-rw-r--r--packages/meshbay-node/src/meshbay_node/chat/store.py186
-rw-r--r--packages/meshbay-node/src/meshbay_node/daemon.py103
-rw-r--r--packages/meshbay-node/src/meshbay_node/ops.py249
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_server.py14
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py572
-rw-r--r--packages/meshbay-node/src/meshbay_node/ui/app.py18
8 files changed, 1138 insertions, 97 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")