diff options
Diffstat (limited to 'packages')
4 files changed, 670 insertions, 654 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/chat.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/chat.py new file mode 100644 index 0000000..26ec27c --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/chat.py @@ -0,0 +1,660 @@ +"""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 <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: + 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) 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 77f2721..889c411 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -32,7 +32,6 @@ import uuid from pathlib import Path from typing import Any -import blake3 from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, @@ -71,12 +70,6 @@ from meshbay_common.adminop import ( OP_TRANSFER_LIMITS, admin_transcript, ) -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.crypto import pk_to_b64, wrap_gek_aes from meshbay_common.device import ( DEVICE_TTL, @@ -86,7 +79,6 @@ from meshbay_common.device import ( ) from meshbay_common.groupbox import ( PURPOSE_ACK, - PURPOSE_CHAT_KEYS, PURPOSE_ROSTER, seal, ) @@ -118,10 +110,9 @@ from meshbay_common.protocol import ( file_upload_payload, ) -from meshbay_node import linkpreview, ops +from meshbay_node import ops from meshbay_node import transfers as transfers_mod from meshbay_node import uploads as uploads_mod -from meshbay_node.chat import FORMAT_SEALED_V1, ReplayedMessage from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer @@ -151,6 +142,7 @@ from meshbay_node.transport.webrtc.channel import ( _get_remote_ip, _pack, ) +from meshbay_node.transport.webrtc.chat import ChatMixin from meshbay_node.transport.webrtc.disk import _locate from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, MAX_MSG from meshbay_node.transport.wire import index_sync_message @@ -160,66 +152,6 @@ log = logging.getLogger(__name__) # An invitation link's handle, as `roster.create_link_invite` mints it. _INVITE_ID_RE = re.compile(r"[0-9a-f]{32}") -# 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 # Upload limits (finding C5a). Uploads used to land directly in the shared root under # a name the client chose, overwriting whatever was already there — which both violated @@ -297,7 +229,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0 class WebRTCPeerSession( - BlobsMixin, + BlobsMixin, ChatMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, ): """One WebRTC peer connection, handling MNP over a DataChannel.""" @@ -1553,43 +1485,6 @@ 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. @@ -2140,78 +2035,6 @@ class WebRTCPeerSession( # ── Chat ───────────────────────────────────────────────────────────── - 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_search_listed(self, msg: dict) -> None: """ Whether this group's files appear in members' cross-group Search. @@ -2244,41 +2067,6 @@ class WebRTCPeerSession( self._broadcast_to_group({"type": MNP.SEARCH_LISTED_ACK, "v": MNP_VERSION, "listed": listed}) - 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_group_roster_req(self, msg: dict) -> None: """ Who is in this group, and which device keys they hold. @@ -2319,41 +2107,6 @@ class WebRTCPeerSession( self._send({"type": MNP.GROUP_ROSTER_RESP, "v": MNP_VERSION, "group_id": self._group_id or "", **sealed}) - 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. @@ -3515,240 +3268,6 @@ class WebRTCPeerSession( if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size: self._leaseless.finish(str(file_id)) - 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 <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: - 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_ping(self, msg: dict) -> None: """Answer a liveness probe on an open channel, echoing the caller's token. @@ -3758,169 +3277,6 @@ class WebRTCPeerSession( """ self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")}) - 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) - def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads: """This group's uploads in progress, created on first use. diff --git a/packages/meshbay-node/tests/test_chat_is_bounded.py b/packages/meshbay-node/tests/test_chat_is_bounded.py index 5dc1128..3332601 100644 --- a/packages/meshbay-node/tests/test_chat_is_bounded.py +++ b/packages/meshbay-node/tests/test_chat_is_bounded.py @@ -36,7 +36,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.chatbox import NONCE_LEN, SIG_LEN from meshbay_common.crypto import pk_to_b64 from meshbay_node.chat import FORMAT_SEALED_V1, ChatStore -from meshbay_node.transport import webrtc_server as ws +from meshbay_node.transport.webrtc import chat as ws from meshbay_node.transport.webrtc_server import WebRTCPeerSession # Read with a default rather than imported. Against the source these were diff --git a/packages/meshbay-node/tests/test_link_preview_request.py b/packages/meshbay-node/tests/test_link_preview_request.py index fe7dec6..0e1bd06 100644 --- a/packages/meshbay-node/tests/test_link_preview_request.py +++ b/packages/meshbay-node/tests/test_link_preview_request.py @@ -13,7 +13,7 @@ import pytest from meshbay_common.protocol import MNP from meshbay_node import linkpreview from meshbay_node.media_cache import MediaCache -from meshbay_node.transport import webrtc_server +from meshbay_node.transport.webrtc import chat from meshbay_node.transport.webrtc_server import WebRTCPeerSession pytestmark = pytest.mark.asyncio @@ -29,9 +29,9 @@ async def media_cache(tmp_path): @pytest.fixture(autouse=True) def _clear_cache(): - webrtc_server._link_preview_cache.clear() + chat._link_preview_cache.clear() yield - webrtc_server._link_preview_cache.clear() + chat._link_preview_cache.clear() def _session(media_cache, ctx=None): @@ -83,7 +83,7 @@ async def test_unfurlable_failure_is_ok_false(media_cache, monkeypatch): async def test_rate_limit_per_connection(media_cache, monkeypatch): """A member firing many previews is bounded; over the ceiling the reply is a plain `ok: false` (bare link) and no outbound fetch is made.""" - monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_PER_CONN", 3) + monkeypatch.setattr(chat, "_LINK_PREVIEW_RATE_PER_CONN", 3) calls = {"n": 0} async def counting_preview(url, **k): @@ -105,8 +105,8 @@ async def test_rate_limit_per_connection(media_cache, monkeypatch): async def test_rate_limit_is_node_wide(media_cache, monkeypatch): """Two connections share the node-wide ceiling.""" - monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_PER_CONN", 100) - monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_NODE", 2) + monkeypatch.setattr(chat, "_LINK_PREVIEW_RATE_PER_CONN", 100) + monkeypatch.setattr(chat, "_LINK_PREVIEW_RATE_NODE", 2) calls = {"n": 0} async def counting_preview(url, **k): |