diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -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 |
2 files changed, 529 insertions, 57 deletions
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() |