diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-09-24 02:41:02 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-09-24 16:45:38 +0200 |
| commit | 3544b143b1a1272931377c47bf7e17b94ee5360b (patch) | |
| tree | 3a9b26fa40aa778dde05065da12c3b6103e83a3a /packages/meshbay-node | |
| parent | eaac12bf945cab10f02429aa8d85239f22280554 (diff) | |
| download | meshbay-3544b143b1a1272931377c47bf7e17b94ee5360b.tar.gz | |
refactor(node): move per-account blobs and key bundles out of webrtc_server
BlobsMixin in transport/webrtc/blobs.py, with the blob caps and the kind
pattern.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-node')
3 files changed, 262 insertions, 244 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/blobs.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/blobs.py new file mode 100644 index 0000000..5e67a42 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/blobs.py @@ -0,0 +1,255 @@ +"""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", + }) + + 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}) 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 5024ffd..77f2721 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -143,6 +143,7 @@ from meshbay_node.transport.webrtc.apps.music import MusicMixin from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin from meshbay_node.transport.webrtc.apps.video_meta import VideoMetaMixin +from meshbay_node.transport.webrtc.blobs import BlobsMixin from meshbay_node.transport.webrtc.channel import ( _REPLY_TO, _DataChannelBuffer, @@ -156,32 +157,6 @@ from meshbay_node.transport.wire import index_sync_message log = logging.getLogger(__name__) -# 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})$") - # An invitation link's handle, as `roster.create_link_invite` mints it. _INVITE_ID_RE = re.compile(r"[0-9a-f]{32}") @@ -321,7 +296,10 @@ _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1" _WEBRTC_TRACE_INTERVAL_S = 30.0 -class WebRTCPeerSession(StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin): +class WebRTCPeerSession( + BlobsMixin, + StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, +): """One WebRTC peer connection, handling MNP over a DataChannel.""" def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""): @@ -985,32 +963,6 @@ class WebRTCPeerSession(StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMix if note_activity: note_activity() - 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}) - def _do_invite_create(self, msg: dict) -> None: """ Issue a one-time pairing code for someone the operator wants to admit. @@ -1085,61 +1037,6 @@ class WebRTCPeerSession(StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMix self._issue_admin_challenge( OP_INVITE_CANCEL, invite_id, {"group_id": group_id, "invite_id": invite_id}) - 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` @@ -1147,118 +1044,6 @@ class WebRTCPeerSession(StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMix # `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"}) - # ── Pairing and join (H3, M3) ──────────────────────────────────────────── def _join_refuse(self, reason: str, audit_detail: str = "") -> None: @@ -3197,28 +2982,6 @@ class WebRTCPeerSession(StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMix "reminder": result.get("reminder", ""), }) - 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}) - def _audit_pre_proof_fetch(self, mtype: str) -> None: """Record bundle access made before the GEK proof (C4).""" audit = self._ctx.get("audit_store") diff --git a/packages/meshbay-node/tests/test_user_blob_mnp.py b/packages/meshbay-node/tests/test_user_blob_mnp.py index 85893af..48b9be7 100644 --- a/packages/meshbay-node/tests/test_user_blob_mnp.py +++ b/packages/meshbay-node/tests/test_user_blob_mnp.py @@ -20,12 +20,12 @@ handler methods against a real store, with `_send` captured. import pytest from meshbay_common.protocol import MNP from meshbay_node.bundle_store import BundleStore -from meshbay_node.transport.webrtc_server import ( +from meshbay_node.transport.webrtc.blobs import ( USER_BLOB_ACCOUNT_MAX, USER_BLOB_BODY_MAX, USER_BLOB_MANIFEST_MAX, - WebRTCPeerSession, ) +from meshbay_node.transport.webrtc_server import WebRTCPeerSession @pytest.fixture |