path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc/blobs.py

"""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:<id>" 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",
        })

    # ── Per-account blobs (playlists, docs/playlists.md §8.2) ────────────────
    #
    # The node stores bytes it cannot read and hands them back. `self._user_id`
    # comes from the authenticated session and never from the message: a
    # `user_id` in the body would let any member of this group read or
    # overwrite any other member's blob, which is finding C5 one size down.

    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})