diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
12 files changed, 129 insertions, 60 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py index 8f3fb74..d548103 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py @@ -212,16 +212,19 @@ class QuicChunkClient: "session — refusing to handshake without channel binding") binding = quic_binding(self._peer_cert_der) - # MNP 3.4: a signed challenge proves the node key before we send anything - # else. A wrong signature is refused; an absent one is an older node. - if reply.get("sig"): - try: - Ed25519PublicKey.from_public_bytes( - base64.b64decode(reply.get("node_pk", "")) - ).verify(base64.b64decode(reply["sig"]), challenge_transcript( - self._group_id, nonce_c, nonce_s, binding)) - except Exception as exc: - raise ConnectionError(f"Node challenge signature invalid: {exc}") from exc + # A signed challenge proves the node key before we send anything else. + # Every node we can reach signs (the floor is 4.0, signing is 3.4), and + # one without a certificate to bind could not complete the proof anyway, + # so a missing signature is refused like a wrong one. + if not reply.get("sig"): + raise ConnectionError("Node challenge is not signed") + try: + Ed25519PublicKey.from_public_bytes( + base64.b64decode(reply.get("node_pk", "")) + ).verify(base64.b64decode(reply["sig"]), challenge_transcript( + self._group_id, nonce_c, nonce_s, binding)) + except Exception as exc: + raise ConnectionError(f"Node challenge signature invalid: {exc}") from exc self._proto._send(self._ctrl_stream, { "type": MNP.HANDSHAKE_RESPONSE, diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py index f42d8e8..daaee62 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py @@ -5,6 +5,7 @@ import base64 import os import time +import msgpack from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from meshbay_common import MNP_VERSION from meshbay_common.adminop import ( @@ -42,6 +43,16 @@ from meshbay_common.adminop import ( from meshbay_common.crypto import pk_to_b64 from meshbay_common.protocol import MNP +# What one connection may have waiting for a signature. Anyone authenticated can +# ask for a challenge — the signature is what is checked, and it comes later — so +# without a bound a member who never answers makes the node keep every request, +# payload and all, for the life of the connection (§13.5b). A person signs one +# operation at a time; a handful covers a settings page saving several at once. +MAX_PENDING_ADMIN_OPS = 8 +# The subject and payload of one pending operation, packed. A path is at most a +# few KiB, and the largest field a legitimate request carries is a directory list. +MAX_ADMIN_OP_BYTES = 64 * 1024 + # Which executor runs each signed operation once its signature has been # checked. Every one runs as a task of the session. _ADMIN_EXECUTORS = { @@ -98,8 +109,22 @@ class AdminMixin: (e.g. root management from a NodePage connection). """ gid = group_id if group_id is not None else (self._group_id or "") + now = time.time() + for op_id, pending in list(self._admin_ops.items()): + if now - pending["ts"] > ADMIN_CHALLENGE_TTL: + del self._admin_ops[op_id] + if len(self._admin_ops) >= MAX_PENDING_ADMIN_OPS: + self._send({"type": "error", "detail": "Too many operations waiting for a " + "signature", "code": "too_many_pending"}) + self._audit("admin_pending_flood", op) + return + if len(msgpack.packb([subject, payload or {}], use_bin_type=True)) \ + > MAX_ADMIN_OP_BYTES: + self._send({"type": "error", "detail": "Request too large", + "code": "too_large"}) + return nonce = os.urandom(32) - ts = int(time.time()) + ts = int(now) op_id = base64.b64encode(os.urandom(16)).decode() self._admin_ops[op_id] = { "op": op, "subject": subject, "nonce": nonce, "ts": ts, diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admission.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admission.py index b02e59a..e678a13 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admission.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admission.py @@ -8,7 +8,12 @@ import time from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from meshbay_common import MNP_VERSION -from meshbay_common.adminop import OP_INVITE_CANCEL, OP_INVITE_CREATE, OP_INVITE_LINK_CREATE +from meshbay_common.adminop import ( + OP_INVITE_CANCEL, + OP_INVITE_CREATE, + OP_INVITE_LINK_CREATE, + invite_create_subject, +) from meshbay_common.crypto import wrap_gek_aes from meshbay_common.device import ( DEVICE_TTL, @@ -71,11 +76,13 @@ class AdmissionMixin: }) return - self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, { - "group_id": group_id, - "user_id": invitee_id, - "username": str(msg.get("username", ""))[:64], - }) + username = str(msg.get("username", ""))[:64] + self._issue_admin_challenge( + OP_INVITE_CREATE, invite_create_subject(invitee_id, username), { + "group_id": group_id, + "user_id": invitee_id, + "username": username, + }) def _do_invite_link_create(self, msg: dict) -> None: """ diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py index f2b6fa5..857db33 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py @@ -64,6 +64,8 @@ class MusicMixin: """ ctx = self._group_ctx() file_id = msg.get("file_id", "") + if self._refuse_blocked(file_id): + return entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py index a425c65..4337e24 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py @@ -228,6 +228,8 @@ class StreamingMixin: async def _stream_video_inner(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg.get("file_id", "") + if self._refuse_blocked(file_id): + return entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py index bd2ab52..70781eb 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py @@ -39,6 +39,8 @@ class SubtitlesMixin: """ ctx = self._group_ctx() file_id = msg.get("file_id", "") + if self._refuse_blocked(file_id): + return entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/video_meta.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/video_meta.py index 834bac4..a9b35dc 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/video_meta.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/video_meta.py @@ -11,6 +11,7 @@ from meshbay_common.adminop import ( OP_TMDB_ENABLED, OP_TMDB_OVERRIDE, OP_TMDB_REMATCH, + tmdb_config_subject, ) from meshbay_common.protocol import MNP @@ -60,12 +61,12 @@ class VideoMetaMixin: if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return - # The subject is the signed, audited, human-shown string — it must - # never contain the token itself (it would end up in the audit log - # in plaintext). The actual token travels only in `payload`, which - # is node-side context, never re-sent or re-verified from the wire. - # The language is not a secret, so it travels in the subject itself. - subject = f"custom_token={'yes' if token else 'no'},language={language or 'default'}" + # The subject is the signed, audited string, so it must never contain + # the token itself (it would end up in the audit log in plaintext); it + # carries the token's SHA-256 instead, which binds the signature to this + # token without writing it down. `None` (unchanged) and `""` (clear) + # stay distinct, for the token and the language alike. + subject = tmdb_config_subject(token, language) self._issue_admin_challenge( OP_TMDB_CONFIG, subject, payload={"token": token, "language": language}, diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py index c3fb623..882ddbc 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py @@ -246,6 +246,25 @@ class SessionCore: return self._ctx["groups"].get(self._group_id) or {} return self._ctx + def _hidden_ids(self): + """The content blocklist, in a public group; nothing anywhere else. + + A private group's content never reaches the hub, so nothing in it can have + been blocked there (docs/MESHBAY_DESIGN.md §7.5, `meshbay_node.blocklist`). + """ + if self._group_ctx().get("visibility") != "public": + return () + return self._ctx.get("blocklist") or () + + def _refuse_blocked(self, file_id) -> bool: + """Refuse a file the blocklist names, in a public group. True if refused.""" + if not isinstance(file_id, str) or file_id not in self._hidden_ids(): + return False + self._send({"type": "error", "detail": "This file is not available here.", + "code": "content_blocked", "file_id": file_id}) + self._audit("content_blocked", file_id[:16]) + return True + def _register_peer(self) -> None: """Add this connection to its group's peer set. diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py index e76e23e..f629563 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py @@ -179,7 +179,7 @@ class FilesMixin: def _do_index_sync(self) -> None: ctx = self._group_ctx() - self._send(index_sync_message(ctx["index"], ctx.get("roots"))) + self._send(index_sync_message(ctx["index"], ctx.get("roots"), self._hidden_ids())) async def _try_serve_thumbnail( self, thumb_hash: str, chunk_index: int, gek: bytes | None, @@ -236,8 +236,18 @@ class FilesMixin: self._note_unleased(tr) file_id = msg["file_id"] chunk_index = msg["chunk_index"] + if self._refuse_blocked(file_id): + return entry = ctx["index"].get_entry(file_id) if not entry: + # A blocked file's own thumbnail is a preview of it. + hidden = self._hidden_ids() + if hidden and file_id in { + e.thumb_hash for e in map(ctx["index"].get_entry, hidden) + if e is not None and e.thumb_hash}: + self._send({"type": "error", "detail": "This file is not available here.", + "code": "content_blocked", "file_id": file_id}) + return thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek")) if thumb is not None: log.debug("file_req file_id=%s chunk=%s: served as thumbnail", diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/handshake.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/handshake.py index eceb2b4..f701072 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/handshake.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/handshake.py @@ -24,7 +24,6 @@ from meshbay_common.handshake import ( ) from meshbay_common.protocol import MNP -from meshbay_node import transfers as transfers_mod from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node.transport.webrtc.channel import _extract_dtls_fingerprint, _get_remote_ip from meshbay_node.transport.webrtc.limits import MAX_MSG @@ -266,19 +265,6 @@ class HandshakeMixin: # No `chat_encrypted` beside it: there is no switch. A peer that # reached this point speaks MNP 2.0, and 2.0 has no plaintext chat. "chat_epoch": int(self._group_ctx().get("chat_epoch", 0) or 0), - # This member's own transfer caps in this group, so the interface - # can say "2 of 2 of your slots are busy" rather than draw a bare - # spinner. Absent reads as "no limit known" and the hint is simply - # not drawn — never as "unlimited", which would have the interface - # contradicting the node. - "transfer_limits": { - "download": self._slots().member_cap( - transfers_mod.DOWNLOAD, - (self._group_id or "", self._user_id or "")), - "upload": self._slots().member_cap( - transfers_mod.UPLOAD, - (self._group_id or "", self._user_id or "")), - }, # So a client that connects mid-scan shows the indexing state # immediately, instead of waiting for the next periodic # INDEX_PROGRESS push. Never a path or filename — see diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py index 9d58d8c..f7bbbfa 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py @@ -14,6 +14,8 @@ from meshbay_common.adminop import ( OP_ROOT_UPDATE, OP_SET_SCAN_SETTINGS, OP_TRANSFER_LIMITS, + group_attach_subject, + root_add_subject, ) from meshbay_common.protocol import MNP @@ -246,10 +248,11 @@ class NodeOpsMixin: # is ignored rather than obeyed: on load it forces every other root # read-only, which is the model the RO/RW one replaced. A second # writable directory is `root_add` with `writable`. + writable = bool(msg.get("writable", True)) + # The directory being exposed is signed, not only the group's name. self._issue_admin_challenge( - OP_GROUP_ATTACH, name, - payload={"name": name, "shared_dir": shared_dir, - "writable": bool(msg.get("writable", True))}, + OP_GROUP_ATTACH, group_attach_subject(name, shared_dir, writable), + payload={"name": name, "shared_dir": shared_dir, "writable": writable}, group_id="") async def _admin_exec_group_attach( @@ -342,16 +345,20 @@ class NodeOpsMixin: if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return + payload = { + "group_id": target_group, "path": path, + "name": str(msg.get("name", ""))[:128], + "kind": str(msg.get("kind", "generic"))[:16], + "writable": bool(msg.get("writable", msg.get("upload", False))), + "removable": bool(msg.get("removable", False)), + } + # Everything the executor acts on is signed — `writable` decides whether + # every member may write there. The group is in the transcript itself. self._issue_admin_challenge( - OP_ROOT_ADD, path, - payload={ - "group_id": target_group, "path": path, - "name": str(msg.get("name", ""))[:128], - "kind": str(msg.get("kind", "generic"))[:16], - "writable": bool(msg.get("writable", msg.get("upload", False))), - "removable": bool(msg.get("removable", False)), - }, - group_id=target_group) + OP_ROOT_ADD, + root_add_subject(path, payload["name"], payload["kind"], + payload["writable"], payload["removable"]), + payload=payload, group_id=target_group) async def _admin_exec_root_add( self, pending: dict, transcript: bytes, sig: bytes, diff --git a/packages/meshbay-node/src/meshbay_node/transport/wire.py b/packages/meshbay-node/src/meshbay_node/transport/wire.py index 6986b01..258edb5 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/wire.py +++ b/packages/meshbay-node/src/meshbay_node/transport/wire.py @@ -12,11 +12,10 @@ envelope produced by `GroupIndex.serialize()`. Same message type, two encodings, consumer each and nothing asserting they matched. Same failure mode as the two `file_chunk` encoders, and the same fix: one builder, used by both. -`GroupIndex.serialize()`/`deserialize()` are unchanged and still tested — they remain -a correct signed index envelope — but they no longer describe any MNP message. Read -them as an at-rest/interchange format, not as a wire contract. It is also not a -candidate for reuse below: it compresses with zstd, which no browser can decompress -(`DecompressionStream` offers gzip and deflate only). +That envelope, and `GroupIndex.serialize()`/`deserialize()` which produced it, are +gone: once both transports built `index_sync` here, nothing stored or exchanged it. +It was never a candidate for reuse below either — it compressed with zstd, which no +browser can decompress (`DecompressionStream` offers gzip and deflate only). Since MNP 1.0 both messages carry their payload **sealed under a GEK-derived subkey** (`meshbay_common.groupbox`). Only the routing fields — `type`, `v`, `group_id` — stay @@ -66,7 +65,7 @@ def list_dirs(roots: RootSet | None) -> list[str]: return sorted(out)[:MAX_DIRS] -def index_sync_message(index, roots: RootSet | None) -> dict: +def index_sync_message(index, roots: RootSet | None, hidden=()) -> dict: """ The full `index_sync` message for one group. @@ -74,10 +73,13 @@ def index_sync_message(index, roots: RootSet | None) -> dict: without them a folder someone just created, or one they emptied, does not exist as far as a client is concerned, and a member cannot tell "the drive is unplugged" from "it is all still there". + + `hidden` is the content blocklist in a public group (`meshbay_node.blocklist`): + those entries are not sent at all. """ payload = { "version": index.version, - "entries": [index_entry_wire(e) for e in index.entries], + "entries": [index_entry_wire(e) for e in index.entries if e.id not in hidden], "dirs": list_dirs(roots), "roots": roots.describe() if roots else [], } @@ -89,7 +91,7 @@ def index_sync_message(index, roots: RootSet | None) -> dict: } -def index_delta_message(index, delta, roots=None) -> dict: +def index_delta_message(index, delta, roots=None, hidden=()) -> dict: """ One `index_delta` — what changed since the last thing this node broadcast. @@ -104,13 +106,16 @@ def index_delta_message(index, delta, roots=None) -> dict: page: the delta that told them something had changed was the one message that could not say what. It is a handful of dicts, bounded by the number of directories a group has, and it is sealed with the rest. + + `hidden`, as for `index_sync_message`: a blocked entry is never added or + updated; its deletion still goes out. """ payload = { "base_version": delta.base_version, "version": delta.version, - "additions": [index_entry_wire(e) for e in delta.additions], + "additions": [index_entry_wire(e) for e in delta.additions if e.id not in hidden], "deletions": list(delta.deletions), - "updates": [index_entry_wire(e) for e in delta.updates], + "updates": [index_entry_wire(e) for e in delta.updates if e.id not in hidden], } if roots is not None: payload["roots"] = roots.describe() |