"""Group chat: sealed messages, history, the epoch keys, link previews, and the operator's chat settings.""" import base64 import logging import time import blake3 from meshbay_common import MNP_VERSION from meshbay_common.adminop import OP_CHAT_DIRECTORY, OP_CHAT_EPOCH, OP_CHAT_LINK_PREVIEW from meshbay_common.chatbox import NONCE_LEN as CHAT_NONCE_LEN from meshbay_common.chatbox import SIG_LEN as CHAT_SIG_LEN from meshbay_common.groupbox import PURPOSE_CHAT_KEYS, seal from meshbay_common.protocol import MNP from meshbay_node import linkpreview, ops from meshbay_node.chat import FORMAT_SEALED_V1, ReplayedMessage log = logging.getLogger("meshbay_node.transport.webrtc_server") # Chat link-preview results, kept in memory only (docs/MESHBAY_DESIGN.md §6.5: # the node # produces enrichment on demand and keeps nothing durable — the asking device # caches). Bounded and time-limited so a busy group cannot grow it without end # and a page that changed its card is picked up within the hour. _LINK_PREVIEW_TTL = 3600 _LINK_PREVIEW_MAX = 256 _link_preview_cache: dict[str, tuple[float, dict]] = {} def _link_preview_cache_get(url: str) -> dict | None: hit = _link_preview_cache.get(url) if hit is None: return None ts, value = hit if time.time() - ts > _LINK_PREVIEW_TTL: _link_preview_cache.pop(url, None) return None return value def _link_preview_cache_put(url: str, value: dict) -> None: if not url: return if len(_link_preview_cache) >= _LINK_PREVIEW_MAX: oldest = min(_link_preview_cache, key=lambda k: _link_preview_cache[k][0]) _link_preview_cache.pop(oldest, None) _link_preview_cache[url] = (time.time(), value) # A member pasting a link is normal; a member — or a hub minting tokens for many # accounts — firing hundreds is amplification/DoS and a way to make the node # reach arbitrary hosts on demand (finding M3). Only a real outbound fetch is # counted (a cache hit costs nothing), and the ceilings are generous enough that # ordinary chat never meets them. _LINK_PREVIEW_RATE_WINDOW = 60.0 _LINK_PREVIEW_RATE_PER_CONN = 15 _LINK_PREVIEW_RATE_NODE = 60 # Chat limits. A message is a member-supplied write onto the operator's disk # (`chat.db`, where retention is a manual CLI command — §6.6), relayed from there # to every other connected member and turned into a notification for every member # of the group. Nothing bounded any of it: the only ceiling was the frame size, # 64 MB once the handshake is done, so one member in a loop could fill the # operator's disk and saturate everyone else's connection. Uploads — the other # member-supplied write — have carried four protections and a size cap since # C5a; this is the same question asked of the path nobody had asked it of. # # 64 KB of ciphertext is about sixty thousand characters. The sealed payload is # the text, a thread id, a display name and a timestamp: an attachment is a file # on a root and travels as a reference (§4.5), so nothing legitimate comes close. MAX_CHAT_CIPHERTEXT = 64 * 1024 # Per account per group, not per connection: a second tab does not make a person # type faster, and keying on the session would hand a script one budget per # socket. Sixty a minute is far above a human and far below a flood. _CHAT_RATE_WINDOW = 60.0 _CHAT_RATE_PER_ACCOUNT = 60 # When the map of senders grows past this, the stale entries are dropped. A node # with more live chatters than this in one window is not the case being bounded. _CHAT_RATE_MAX_TRACKED = 1000 class ChatMixin: 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 def _do_chat_directory(self, msg: dict) -> None: """ Where chat attachments are written. Unlike every other app directory this one is a destination, so it has to be on a read-write root — checked by `ops.set_chat_directory` after the signature, which is where the refusal actually lives. """ path = msg.get("path") if not isinstance(path, str): self._send({"type": "error", "detail": "Missing or invalid 'path'"}) return path = path.strip("/") if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_CHAT_DIRECTORY, path) async def _admin_exec_chat_directory( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: path = pending["subject"] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"chat_directory:{path}") return try: await self._run_op( ops.set_chat_directory, self._group_id or "", path) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("chat_directory", path) self._broadcast_to_group( {"type": MNP.CHAT_DIRECTORY_ACK, "v": MNP_VERSION, "path": path}) def _do_chat_link_preview(self, msg: dict) -> None: """ Whether the node fetches a page's title and image when a member posts a link — outbound traffic on the operator's connection, from a message they did not write, so it is signed like everything else that decides what leaves this machine. """ enabled = msg.get("enabled") if not isinstance(enabled, bool): self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_CHAT_LINK_PREVIEW, "on" if enabled else "off") async def _admin_exec_chat_link_preview( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: enabled = pending["subject"] == "on" if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"chat_link_preview:{pending['subject']}") return try: await self._run_op( ops.set_chat_link_preview, self._group_id or "", enabled) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("chat_link_preview", pending["subject"]) 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 _do_chat_message(self, msg: dict) -> None: """ 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 # Separate from the envelope check above, and deliberately: that one asks # whether the message is well formed and authentic, this one asks what it # costs everyone else. Before either is written to disk or relayed. detail, code = self._chat_bounds_refusal(raw) if detail: self._send({"type": "error", "detail": detail, "code": code}) self._audit("chat_refused", code) return if sender_name: self._user_names()[self._user_id] = sender_name if chat_store: 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() broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "sender_id": self._user_id, "sender_name": sender_name, "payload": payload, "thread_id": msg.get("thread_id"), "timestamp": time.time(), "format": fmt, "epoch": epoch, "device": device, "nonce": nonce, "sig": sig, } 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: pass hub_ws = self._ctx.get("hub_ws") if hub_ws and self._group_id: try: import json as _json self._spawn(hub_ws.send(_json.dumps({ "type": "chat_notify", "group_id": self._group_id, # 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 ". # # `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: pass 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 "" def _chat_bounds_refusal(self, ct: bytes) -> tuple[str, str]: """What this message would cost the others, or ("", "") to accept it. Two bounds, and each answers a different half of "who pays". A message is written to `chat.db` on the operator's disk and kept — retention is a manual command (§6.6) — then relayed to every other connected member and turned into a notification for every member of the group. So **size** bounds what one message costs, and **rate** bounds how often one member may impose it. There is deliberately no node-wide ceiling to go with the per-account one. The link-preview limiter has both because a preview spends the *node's* egress and its third-party quota, which is one shared thing; a chat message spends the sender's own group. A node-wide chat ceiling would let a busy group silence a quiet one, which is the same class of defect this bound exists to close, one level up. """ if len(ct) > MAX_CHAT_CIPHERTEXT: return ("This message is too large to send in chat — " "send a large file as an attachment instead.", "chat_too_large") if not self._chat_rate_ok(): return ("Too many messages just now — wait a moment.", "chat_rate_limited") return ("", "") def _chat_rate_ok(self) -> bool: """True when this sender is within their window; records it when so. Keyed by (group, account) on the transport context rather than on the session: the sender is authenticated, so this is the one identifier a second tab — or fifty of them — cannot multiply. The window is trimmed on every call, and the map of senders is swept when it grows, so neither can be the memory leak the bound was added to prevent. """ now = time.monotonic() hits: dict = self._ctx.setdefault("chat_hits", {}) if len(hits) > _CHAT_RATE_MAX_TRACKED: for key, times in list(hits.items()): if not times or now - times[-1] >= _CHAT_RATE_WINDOW: hits.pop(key, None) key = (self._group_id or "", self._user_id or "") mine = [t for t in hits.get(key, ()) if now - t < _CHAT_RATE_WINDOW] if len(mine) >= _CHAT_RATE_PER_ACCOUNT: hits[key] = mine return False mine.append(now) hits[key] = mine return True 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_chat_history(self, msg: dict) -> None: chat_store = self._group_ctx().get("chat_store") if not chat_store: self._send({ "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "messages": [], "has_more": False, }) return # `before` pages backwards from the newest, which is the direction a chat # is actually read. `since` remains for callers that want everything # after a point in time; the browser no longer uses it. before = msg.get("before") limit = max(1, min(int(msg.get("limit", 100)), 200)) self._spawn(self._send_chat_history(chat_store, before, limit)) async def _send_chat_history(self, chat_store, before, limit: int) -> None: if before: msgs = await chat_store.get_before(int(before), limit=limit) else: msgs = await chat_store.get_recent(limit=limit) # Whether the "load older" control has anything left to fetch. Asked # about the oldest row returned, so an empty page correctly says no. has_more = await chat_store.has_before(msgs[0].id) if msgs else False names = self._user_names() self._send({ "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "has_more": has_more, # `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 node-wide window; records it when so, and both counts are trimmed to the window on every call so the lists cannot grow without bound. """ now = time.monotonic() w = _LINK_PREVIEW_RATE_WINDOW mine = [t for t in getattr(self, "_link_preview_hits", []) if now - t < w] node = [t for t in self._ctx.get("link_preview_hits", []) if now - t < w] if (len(mine) >= _LINK_PREVIEW_RATE_PER_CONN or len(node) >= _LINK_PREVIEW_RATE_NODE): self._link_preview_hits = mine self._ctx["link_preview_hits"] = node return False mine.append(now) node.append(now) self._link_preview_hits = mine self._ctx["link_preview_hits"] = node return True async def _do_link_preview_request(self, msg: dict) -> None: """ Unfurl a URL a member pasted into chat (docs/MESHBAY_DESIGN.md §6.5's enrichment rule: the client asks, the node produces on demand, the asking device caches — nothing durable here). `linkpreview.safe_url` is the SSRF gate: the URL a *member* chose decides an outbound request from the operator's machine, so http(s) only and the resolved address must be globally routable. Failure of any kind — blocked, unreachable, not HTML, nothing worth showing — comes back as `ok: false`, the way a TMDB miss does; the client then just shows the bare link. """ url = msg.get("url") key = url if isinstance(url, str) else "" # Checked before the cache, not after: the operator turning previews # off has to stop serving the ones already fetched too, or the setting # takes effect only for links nobody has posted yet. Refused as an # ordinary miss — the client shows the bare link, which is exactly what # "no preview" looks like for a page that has none. if not self._group_ctx().get("chat_link_preview", True): self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION, "url": key, "ok": False}) return cached = _link_preview_cache_get(key) if cached is not None: self._send({**cached, "type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION}) return if not self._link_preview_rate_ok(): # Same shape as any other miss — the client shows the bare link. A # rate-limited result is not cached, so it is retried once the # window clears rather than pinned as "no preview". log.debug("link_preview_req: rate-limited (peer=%s)", self._peer_id) self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION, "url": key, "ok": False}) return resp: dict = {"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION, "url": key, "ok": False} try: meta = await linkpreview.fetch_preview(url) if meta is not None: resp.update(ok=True, title=meta["title"], description=meta["description"], site_name=meta["site_name"]) image_url = meta.get("image_url") media_cache = self._ctx.get("media_cache") if image_url and media_cache is not None: synthetic_id = f"linkpreview:{image_url}" thumb_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id) if thumb_hash is None: jpeg = await linkpreview.fetch_image(image_url) if jpeg: thumb_hash = blake3.blake3(jpeg).hexdigest() await media_cache.put_thumb(thumb_hash, synthetic_id, jpeg) if thumb_hash: resp["image_thumb_hash"] = thumb_hash except Exception as e: log.debug("link_preview_req %s: %s", key[:80], e) _link_preview_cache_put(key, {k: v for k, v in resp.items() if k not in ("type", "v")}) self._send(resp)