"""What a member stores on the node for themselves: per-account blobs (playlists), and the key bundles that carry their keys between devices.""" import logging import re from meshbay_common import MNP_VERSION from meshbay_common.protocol import MNP log = logging.getLogger("meshbay_node.transport.webrtc_server") # Per-account blobs (docs/playlists.md §4.3). These are an unbounded write # primitive pointed at somebody else's disk, so every one of them is checked — # and every check **refuses**, never truncates. A truncating cap silently loses # tracks, which is the one failure the whole playlist design exists to prevent. # # The numbers are sized against the measured shape: ~300 bytes per track before # compression, deflate worth about three on a payload this repetitive. A 1 MB # body is therefore roughly ten thousand tracks in one playlist, and the # manifest holds names and revisions only. USER_BLOB_MANIFEST_MAX = 64 * 1024 USER_BLOB_BODY_MAX = 1024 * 1024 USER_BLOB_ACCOUNT_MAX = 8 * 1024 * 1024 # "playlists" is the manifest; "playlist:" is one playlist's tracks. A # pattern rather than a set, because the ids are client-generated — but a # pattern, not anything at all, or the table becomes a key/value store for # whatever a client feels like writing. # # The character class is deliberately wider than a UUID: the reserved id is the # word "favorites" (docs/playlists.md §5.1), so a hex-only pattern refuses the # one playlist every account has. It stays narrow enough to carry no structure # of its own — no "/", no ".", no second ":" — so a kind can never be read as a # path or as anything but one name in one namespace. _USER_BLOB_KIND_RE = re.compile( r"^(playlists|playlist:[A-Za-z0-9_-]{1,64})$") class BlobsMixin: async def _do_gek_bundle_fetch(self) -> None: """Serve the caller's wrapped GEK bundle during the handshake window.""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) return group_id = getattr(self, "_pending_group", "") user_id = getattr(self, "_pending_sub", "") if not group_id or not user_id: self._send({"type": "error", "detail": "No pending handshake"}) return bundle = await bundle_store.fetch(group_id, user_id) if bundle: self._send({ "type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": True, "pk_eph_b64": bundle["pk_eph_b64"], "nonce_b64": bundle["nonce_b64"], "wrapped_b64": bundle["wrapped_b64"], }) else: self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) async def _do_keypair_bundle_fetch(self) -> None: """Serve the caller's encrypted keypair bundle during the handshake window.""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) return user_id = getattr(self, "_pending_sub", "") if not user_id: self._send({"type": "error", "detail": "No pending handshake"}) return kp = await bundle_store.fetch_keypair(user_id) if kp and kp.get("bundle_enc"): resp = { "type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": True, "bundle_enc": kp["bundle_enc"], } # The recovery-wrapped copy (MNP 0.14) rides along when present, so a # client holding the recovery key can re-wrap it under a new # passphrase — docs/MESHBAY_DESIGN.md §3.6. if kp.get("bundle_enc_recovery"): resp["bundle_enc_recovery"] = kp["bundle_enc_recovery"] self._send(resp) else: self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) async def _do_keypair_bundle_store(self, msg: dict) -> None: """Store an encrypted keypair bundle (user backs up their own keys on node).""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": "error", "detail": "Bundle store not available"}) return bundle_enc = msg.get("bundle_enc", "") if not bundle_enc: self._send({"type": "error", "detail": "Missing bundle_enc"}) return # Optional second copy wrapped under the recovery key (MNP 0.14). Omitted # by an older client and by a plain re-backup; the store keeps any # existing recovery copy when this is absent. recovery = msg.get("bundle_enc_recovery") or None await bundle_store.store_keypair(self._user_id, bundle_enc, recovery) log.info("Keypair bundle stored for user=%s (recovery=%s)", self._user_id[:8], bool(recovery)) self._audit("keypair_bundle_store") self._send({ "type": "ack", "v": MNP_VERSION, "detail": "keypair_bundle_stored", }) def _user_blob_refuse(self, detail: str, kind: str = "") -> None: """ Refuse, and say so in the audit log. A refusal used to be invisible here: the audit line was written only after a store *succeeded*, so a client whose writes were all being turned away looked exactly like a client that never wrote — which is how a wedged sync went unnoticed for two hours. """ self._audit("user_blob_refused", f"{kind} {detail}".strip()) self._send({"type": "error", "detail": detail}) def _user_blob_kind(self, msg: dict) -> str | None: """The validated `kind`, or None having already refused.""" kind = msg.get("kind") if not isinstance(kind, str) or not _USER_BLOB_KIND_RE.match(kind): self._user_blob_refuse("Unknown blob kind", str(kind)[:40]) return None return kind def _user_blob_store_ok(self): store = self._ctx.get("bundle_store") if not store: self._send({"type": "error", "detail": "Bundle store not available"}) return None if not self._user_id: self._send({"type": "error", "detail": "Not authenticated"}) return None return store async def _do_user_blob_store(self, msg: dict) -> None: store = self._user_blob_store_ok() if not store: return kind = self._user_blob_kind(msg) if not kind: return blob = msg.get("blob_enc") if not isinstance(blob, (bytes, bytearray)) or not blob: self._user_blob_refuse("Missing blob_enc", kind) return blob = bytes(blob) rev = msg.get("rev") if not isinstance(rev, int) or rev < 0: self._user_blob_refuse("Missing rev", kind) return limit = (USER_BLOB_MANIFEST_MAX if kind == "playlists" else USER_BLOB_BODY_MAX) if len(blob) > limit: # A stated reason, not a bare error: the client turns this into a # sentence the reader can act on ("this playlist is too large"), # and a refusal nobody can read is a support case. self._user_blob_refuse(f"Blob too large ({len(blob)} > {limit})", kind) return # What the account already uses, minus whatever this call replaces. used = await store.user_blob_total_bytes(self._user_id) existing = await store.fetch_user_blob(self._user_id, kind) if existing: used -= len(existing["blob_enc"]) if used + len(blob) > USER_BLOB_ACCOUNT_MAX: self._user_blob_refuse( f"Account blob quota exceeded " f"({used + len(blob)} > {USER_BLOB_ACCOUNT_MAX})", kind) return await store.store_user_blob(self._user_id, kind, rev, blob) self._audit("user_blob_store", f"{kind} rev={rev} bytes={len(blob)}") self._send({"type": "ack", "v": MNP_VERSION, "detail": "user_blob_stored"}) async def _do_user_blob_fetch(self, msg: dict) -> None: store = self._user_blob_store_ok() if not store: return kind = self._user_blob_kind(msg) if not kind: return row = await store.fetch_user_blob(self._user_id, kind) self._audit("user_blob_fetch", kind) self._send({ "type": MNP.USER_BLOB_RESP, "v": MNP_VERSION, "kind": kind, # A kind this account has never written is `null`, not an error: # "no playlist here yet" is the ordinary state of a fresh node and # the client must not read it as a failure. "rev": row["rev"] if row else None, "blob_enc": row["blob_enc"] if row else None, }) async def _do_user_blob_list(self) -> None: store = self._user_blob_store_ok() if not store: return blobs = await store.list_user_blobs(self._user_id) self._audit("user_blob_list", f"{len(blobs)} blobs") self._send({"type": MNP.USER_BLOB_LIST_RESP, "v": MNP_VERSION, "blobs": blobs}) async def _do_user_blob_delete(self, msg: dict) -> None: store = self._user_blob_store_ok() if not store: return kind = self._user_blob_kind(msg) if not kind: return removed = await store.delete_user_blob(self._user_id, kind) self._audit("user_blob_delete", kind) self._send({"type": "ack", "v": MNP_VERSION, "detail": "user_blob_deleted" if removed else "user_blob_absent"}) async def _do_keypair_bundle_delete(self) -> None: """ Withdraw our own key backup from this node. Only ever our own: the user_id comes from the authenticated session, never from the message. Someone who does not want a second browser should not be leaving a PBKDF2-protected blob on every node they have ever joined (C4), and turning the setting off has to remove what is already there — not just stop adding to it. """ bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": "error", "detail": "Bundle store not available"}) return removed = await bundle_store.delete_keypair(self._user_id) if removed: log.info("Keypair bundle withdrawn by user=%s", self._user_id[:8]) self._audit("keypair_bundle_delete") self._send({"type": "ack", "v": MNP_VERSION, "detail": "keypair_bundle_deleted", "removed": removed})