diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/bundle_store.py | 115 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 142 |
2 files changed, 257 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/bundle_store.py b/packages/meshbay-node/src/meshbay_node/bundle_store.py index 86b8de4..c65b94d 100644 --- a/packages/meshbay-node/src/meshbay_node/bundle_store.py +++ b/packages/meshbay-node/src/meshbay_node/bundle_store.py @@ -63,6 +63,38 @@ CREATE TABLE IF NOT EXISTS keypair_bundles ( ); """ +# Per-account blobs the node holds and cannot read — playlists today +# (docs/playlists.md §3.3). The same shape as a keypair bundle, with a +# different payload, so this adds no trust boundary the node was not already +# on the wrong side of for this same account. +# +# Two deliberate differences from keypair_bundles, each a mistake avoided +# rather than a preference: +# +# `blob_enc` is a BLOB, not base64 TEXT. A keypair bundle is a few hundred +# bytes and nobody will ever notice the 33% tax; a playlist is hundreds of +# kilobytes, where a third is not a rounding error. +# +# `kind` is a namespace, not an enum: "playlists" is the manifest and +# "playlist:<uuid>" is one playlist's tracks. That is what lets one playlist +# be rewritten without re-uploading the whole collection, and it costs no +# second table. webrtc_server.py validates the shape. +# +# blob_enc_recovery is declared now and left NULL — the same column, on the +# same kind of table, that keypair_bundles needed a PRAGMA migration for one +# release late. CREATE TABLE IF NOT EXISTS never adds a column. +_SCHEMA_USER_BLOBS = """\ +CREATE TABLE IF NOT EXISTS user_blobs ( + user_id TEXT NOT NULL, + kind TEXT NOT NULL, + rev INTEGER NOT NULL, + blob_enc BLOB NOT NULL, + blob_enc_recovery BLOB, + stored_at TEXT NOT NULL DEFAULT (datetime('now')), + PRIMARY KEY (user_id, kind) +); +""" + class BundleStore: def __init__(self, db_path: Path): @@ -75,6 +107,7 @@ class BundleStore: await self._db.execute(_SCHEMA_GEK) await self._db.execute(_SCHEMA_KEYPAIR) await self._db.execute(_SCHEMA_CHAT_EPOCHS) + await self._db.execute(_SCHEMA_USER_BLOBS) await self._migrate_keypair_recovery() await self._db.commit() @@ -180,6 +213,88 @@ class BundleStore: await self._db.commit() return cur.rowcount > 0 + # ── Per-account blobs (playlists) ──────────────────────────────────── + # + # Opaque throughout: the node stores bytes it cannot read, returns them, and + # never looks inside. Every method takes `user_id` from the caller, which + # takes it from the authenticated session and never from a message body — + # a user_id off the wire would let any member read or overwrite any other + # member's blob. + + async def store_user_blob( + self, + user_id: str, + kind: str, + rev: int, + blob_enc: bytes, + ) -> None: + """ + Write one blob, replacing whatever was there. + + Older revisions are not kept. The client is the authority on what the + merged state is (docs/playlists.md §6.4) and holds its own copy, so a + node keeping history would buy nothing and would mean the node deciding + which revision is current — which is exactly what it must not do. + """ + assert self._db + await self._db.execute( + "INSERT INTO user_blobs (user_id, kind, rev, blob_enc, stored_at) " + "VALUES (?, ?, ?, ?, datetime('now')) " + "ON CONFLICT(user_id, kind) DO UPDATE SET " + " rev = excluded.rev, " + " blob_enc = excluded.blob_enc, " + " stored_at = excluded.stored_at", + (user_id, kind, rev, blob_enc), + ) + await self._db.commit() + + async def fetch_user_blob(self, user_id: str, kind: str) -> dict | None: + assert self._db + async with self._db.execute( + "SELECT rev, blob_enc FROM user_blobs WHERE user_id = ? AND kind = ?", + (user_id, kind), + ) as cursor: + row = await cursor.fetchone() + if not row: + return None + return {"rev": row[0], "blob_enc": row[1]} + + async def list_user_blobs(self, user_id: str) -> list[dict]: + """ + Which blobs exist and at what revision — never a payload. + + A client that has lost its local state (a cache clear, a new device) + cannot otherwise discover which playlists exist: their kinds carry + client-generated UUIDs, and guessing is not a plan. + """ + assert self._db + async with self._db.execute( + "SELECT kind, rev FROM user_blobs WHERE user_id = ? ORDER BY kind", + (user_id,), + ) as cursor: + return [{"kind": r[0], "rev": r[1]} for r in await cursor.fetchall()] + + async def delete_user_blob(self, user_id: str, kind: str) -> bool: + assert self._db + cur = await self._db.execute( + "DELETE FROM user_blobs WHERE user_id = ? AND kind = ?", + (user_id, kind)) + await self._db.commit() + return cur.rowcount > 0 + + async def user_blob_total_bytes(self, user_id: str) -> int: + """ + What this account is already using, so a store can refuse to take it + over the per-account cap. An unbounded write primitive pointed at + somebody else's disk needs a number that is actually checked. + """ + assert self._db + async with self._db.execute( + "SELECT COALESCE(SUM(LENGTH(blob_enc)), 0) FROM user_blobs " + "WHERE user_id = ?", (user_id,)) as cursor: + row = await cursor.fetchone() + return int(row[0]) if row else 0 + # ── Chat epoch keys ────────────────────────────────────────────────── async def store_chat_epoch( 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 af41061..298b3cd 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -29,6 +29,7 @@ import hashlib import hmac import logging import os +import re import struct import tempfile import time @@ -146,6 +147,32 @@ log = logging.getLogger(__name__) CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 +# 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})$") + # Chat link-preview results, kept in memory only (draft-v6 §2.7: 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 @@ -682,6 +709,14 @@ class WebRTCPeerSession: self._spawn(self._do_keypair_bundle_store(msg)) elif mtype == MNP.KEYPAIR_BUNDLE_DELETE: self._spawn(self._do_keypair_bundle_delete()) + elif mtype == MNP.USER_BLOB_STORE: + self._spawn(self._do_user_blob_store(msg)) + elif mtype == MNP.USER_BLOB_FETCH: + self._spawn(self._do_user_blob_fetch(msg)) + elif mtype == MNP.USER_BLOB_LIST: + self._spawn(self._do_user_blob_list()) + elif mtype == MNP.USER_BLOB_DELETE: + self._spawn(self._do_user_blob_delete(msg)) elif mtype == MNP.STREAM_REQUEST: sem = self._ctx.get("_transcode_sem") log.info("stream: req file=%s credits=%s slots_free=%s prev=%s", @@ -1136,6 +1171,113 @@ class WebRTCPeerSession: "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_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._send({"type": "error", "detail": "Unknown blob kind"}) + 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._send({"type": "error", "detail": "Missing blob_enc"}) + return + blob = bytes(blob) + + rev = msg.get("rev") + if not isinstance(rev, int) or rev < 0: + self._send({"type": "error", "detail": "Missing rev"}) + 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._send({"type": "error", + "detail": f"Blob too large ({len(blob)} > {limit})"}) + 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._send({"type": "error", + "detail": f"Account blob quota exceeded " + f"({used + len(blob)} > {USER_BLOB_ACCOUNT_MAX})"}) + 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._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"}) + # ── Pairing and join (H3, M3) ──────────────────────────────────────────── def _join_refuse(self, reason: str, audit_detail: str = "") -> None: |