diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-09-24 09:03:06 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-09-24 16:45:38 +0200 |
| commit | ffdb70784f11de0df8162cea5e9f5abfac91ea96 (patch) | |
| tree | 70d7577aaef5e4b5b493444bb8d1d81d8e416ce0 /packages/meshbay-node/src/meshbay_node/transport | |
| parent | fd620dc2b61b5ef6fe72581321f488ebf8faf746 (diff) | |
| download | meshbay-ffdb70784f11de0df8162cea5e9f5abfac91ea96.tar.gz | |
refactor(node): move file browsing and folder/file ops out of webrtc_server
FilesMixin in transport/webrtc/files.py; the blocking disk helpers join
_locate in webrtc/disk.py, and the lease states go to webrtc/limits.py.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
4 files changed, 489 insertions, 461 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py index fd3184c..bc8d33f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py @@ -2,7 +2,10 @@ from pathlib import Path +from meshbay_common.protocol import file_chunk_wire + from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path +from meshbay_node.transport.webrtc.limits import CHUNK_SIZE def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: @@ -21,3 +24,53 @@ def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: if not path.exists(): return None, "File not on disk" return path, None + + +def _mkdir_if_absent(target: Path) -> str | None: + """ + Create a directory unless it is already there, or say why not. + + Both in one call, not a check awaited and then an act: the disk thread is + one worker, so nothing can slip between them. Split across two awaits, two + members creating the same name would both find nothing there and the second + `mkdir` would raise where a refusal was meant. Blocking; called through + `off_disk`. + """ + if target.exists(): + return "Already exists" + target.mkdir(parents=False) + return None + + +def _is_empty_dir(target: Path) -> bool: + """Blocking; called through `off_disk`.""" + return not any(target.iterdir()) + + +def _rmdir_if_empty(target: Path) -> bool: + """ + Remove a directory if nothing is in it. False if something is. + + The emptiness test and the removal are one call for the reason the caller + re-tests at all: the first test happened before a round trip to the + operator's browser, and a file can land in between. Two awaits here would + reopen the same window one size smaller. Blocking; called through `off_disk`. + """ + if any(target.iterdir()): + return False + target.rmdir() + return True + + +def _read_and_encrypt( + gek: bytes, + file_path: Path, + chunk_index: int, + file_hash: bytes, + file_id: str = "", +) -> dict: + """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`.""" + with open(file_path, "rb") as f: + f.seek(chunk_index * CHUNK_SIZE) + plaintext = f.read(CHUNK_SIZE) + return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py new file mode 100644 index 0000000..e76e23e --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py @@ -0,0 +1,427 @@ +"""Browsing and fetching a group's files: the index, file chunks and +thumbnails, and the signed folder and file operations.""" + +import asyncio +import base64 +import logging + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey +from meshbay_common import MNP_VERSION +from meshbay_common.adminop import OP_DIR_DELETE, OP_FILE_DELETE +from meshbay_common.protocol import MNP, file_chunk_wire + +from meshbay_node.roots import ROOT_NOT_SERVED, SAFE_UPLOAD_NAME, RootSet, off_disk, safe_subdir +from meshbay_node.transport.webrtc.disk import ( + _is_empty_dir, + _locate, + _mkdir_if_absent, + _read_and_encrypt, + _rmdir_if_empty, +) +from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, LEASE_GRANTED, LEASE_QUEUED +from meshbay_node.transport.wire import index_sync_message + +log = logging.getLogger("meshbay_node.transport.webrtc_server") + + +# A chunk is a megabyte and the browser keeps eight in flight, so answering them +# as they arrive queues 8 MB on the channel with nothing watching. On a LAN that +# drains before anyone notices; on a phone that is also uploading, it is minutes +# of head-of-line delay for the reader. Above this, wait for room. +DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 1024 + + +class FilesMixin: + async def _do_dir_create(self, msg: dict) -> None: + """ + Create a directory, for any member of the group. + + Same confinement as an upload: every segment passes the name allowlist and + the result must resolve under the shared root. Making a directory is not a + privileged act — a member who can add a file can organise where it goes — + but it writes to the operator's disk, so it is audited like one. + """ + ctx = self._group_ctx() + roots: RootSet | None = ctx.get("roots") + if not roots: + self._send({"type": "error", "detail": "No shared directory"}) + return + + name = str(msg.get("name", "")).strip() + if not SAFE_UPLOAD_NAME.match(name): + self._send({"type": "error", "detail": "Invalid directory name"}) + return + + # The virtual root is not a directory on anyone's disk, so a member + # cannot create one there — that would be adding a root, which is the + # operator's configuration and not a file operation. + parent_rel = (msg.get("dir") or "").strip("/") + if not parent_rel: + self._send({"type": "error", + "detail": "Choose a folder to create this in"}) + return + + # Read-only means read-only, and creating a folder writes to the + # operator's disk. `_do_file_upload` gained this check with the RO/RW + # model and this one did not — so a member could not add a file to a + # published library but could still leave empty directories in it. + owner = roots.split(parent_rel) + if owner is None: + self._send({"type": "error", "detail": "Invalid directory"}) + return + parent_root, _tail = owner + if not parent_root.writable: + self._send({"type": "error", + "detail": f"Directory '{parent_root.name}' is read-only", + "code": "root_read_only"}) + self._audit("dir_create_refused", parent_rel[:64]) + return + if not parent_root.available: + self._send({"type": "error", + "detail": f"Directory '{parent_root.name}' is " + f"currently unavailable", + "code": "root_unavailable"}) + return + + parent = await off_disk(roots, safe_subdir, roots, parent_rel) + if parent is None or not await off_disk(roots, parent.is_dir): + self._send({"type": "error", "detail": "Invalid directory"}) + return + + target = await off_disk(roots, safe_subdir, roots, f"{parent_rel}/{name}") + if target is None: + self._send({"type": "error", "detail": "Invalid directory"}) + return + refusal = await off_disk(roots, _mkdir_if_absent, target) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) + return + virtual = roots.virtual_of(target) or f"{parent_rel}/{name}" + log.info("Directory created by %s: %s", self._user_id[:8], virtual) + self._audit("dir_create", virtual) + self._send({ + "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION, + "dir": virtual, + }) + + @staticmethod + def _names_a_root(roots: RootSet, rel: str) -> bool: + """True when `rel` is a bare root name rather than something inside one.""" + found = roots.split(rel or "") + return found is not None and not found[1] + + async def _do_dir_delete(self, msg: dict) -> None: + """ + Remove an empty directory, for the node operator. + + Creating one is not privileged — a member who can add a file may organise + where it goes — but removing one is: it acts on a name other members are + using, and on the operator's disk. Empty is the whole safety property + here. Nothing recursive: refusing a directory with anything in it means + this can never destroy content, whatever the caller intended, so the + operator deletes the files first and sees what they are losing. + """ + ctx = self._group_ctx() + roots: RootSet | None = ctx.get("roots") + if not roots: + self._send({"type": "error", "detail": "No shared directory"}) + return + + rel = (msg.get("dir") or "").strip("/") + target = await off_disk(roots, safe_subdir, roots, rel) + # A root itself is not deletable here: removing one is a configuration + # change, and doing it through a file operation would leave the group + # config naming a directory nobody can reach. + if target is None or self._names_a_root(roots, rel): + self._send({"type": "error", "detail": "Invalid directory"}) + return + if not await off_disk(roots, target.is_dir): + self._send({"type": "error", "detail": "Not a directory"}) + return + if not await off_disk(roots, _is_empty_dir, target): + self._send({"type": "error", "detail": "Directory is not empty"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for deletion"}) + return + + self._issue_admin_challenge( + OP_DIR_DELETE, roots.virtual_of(target) or rel) + + async def _admin_exec_dir_delete( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + rel = pending["subject"] + ctx = self._group_ctx() + roots: RootSet | None = ctx.get("roots") + target = await off_disk(roots, safe_subdir, roots, rel) if roots else None + if (target is None or self._names_a_root(roots, rel) + or not await off_disk(roots, target.is_dir)): + self._send({"type": "error", "detail": "Not a directory"}) + return + + # Operator only. A file has an uploader who may remove their own; a + # directory has none, so there is no second key to accept here. + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"dir_delete:{rel}") + return + + # Checked again after the signature: the emptiness test that let this + # through happened before a round trip to the operator's browser, and a + # file could have landed in the meantime. + if not await off_disk(roots, _rmdir_if_empty, target): + self._send({"type": "error", "detail": "Directory is not empty"}) + return + log.info("Directory removed by %s: %s", self._user_id[:8], rel) + self._audit("dir_delete", rel) + self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel}) + + def _do_index_sync(self) -> None: + ctx = self._group_ctx() + self._send(index_sync_message(ctx["index"], ctx.get("roots"))) + + async def _try_serve_thumbnail( + self, thumb_hash: str, chunk_index: int, gek: bytes | None, + ) -> dict | None: + """ + docs/MESHBAY_DESIGN.md §6.5: a thumbnail is served through the same + chunked file_req path as a real file, resolved against the media + cache instead of the index when the id doesn't match a file. + Sliced by `chunk_index` like a real file's chunks, not just handed + back whole: a thumbnail/poster/cover never approached CHUNK_SIZE so + this used to be equivalent to "only chunk 0 exists", but an audio + transcode result (docs/MESHBAY_DESIGN.md §9.8, the WMA/Musepack exception) is + cached in the same media_cache blob store and can be several MB — + genuinely multi-chunk, same as a file read straight off disk. + """ + media_cache = self._ctx.get("media_cache") + if media_cache is None: + return None + blob = await media_cache.get_thumb(thumb_hash) + if blob is None: + return None + start = chunk_index * CHUNK_SIZE + if start > len(blob) or (start == len(blob) and chunk_index != 0): + return None + piece = blob[start:start + CHUNK_SIZE] + return file_chunk_wire( + gek, piece, chunk_index, bytes.fromhex(thumb_hash), thumb_hash) + + async def _do_file_request(self, msg: dict) -> None: + ctx = self._group_ctx() + # A chunk request is what "this transfer is alive" looks like. Nothing + # marked a lease used, so `used` stayed False for the whole download and + # the sweeper revoked the grant every 30 s as never-taken-up — while the + # file was transferring at 20 MB/s. Found in the node's own log, which + # repeated the same two reclaims every 30 s for as long as the daemon + # ran. + # + # And it is also what makes the caps real: `tr` names a lease or it + # does not, and `_lease_of` is the only thing that decides which. + tr = str(msg.get("tr") or "")[:64] + leased = False + if tr: + state = self._lease_of(tr) + if state == LEASE_QUEUED: + self._send({ + "type": "error", + "detail": "This transfer is waiting for a slot.", + "code": "lease_not_granted", + "tr": tr, + }) + return + leased = state == LEASE_GRANTED + if not leased: + self._note_unleased(tr) + file_id = msg["file_id"] + chunk_index = msg["chunk_index"] + entry = ctx["index"].get_entry(file_id) + if not entry: + 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", + file_id[:16], chunk_index) + self._send(thumb) + return + log.warning("File not found: %s", file_id[:16]) + self._send({"type": "error", "detail": "File not found"}) + return + + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) + return + + # A real index entry, asked for without a lease: browsing, or a client + # helping itself to the whole library outside every cap. + # + # Both look identical here — which is why the bound is a small count of + # files rather than a judgement about what the read is for. Thumbnails, + # posters and cover art never reach this line: they resolve through + # `_try_serve_thumbnail` above, out of a cache the node built itself, + # and are never leased, never counted, never queued. + # + # `not leased`, not `not tr`: a `tr` the node cannot match to a granted + # lease of this session is not a transfer, whatever the client calls it, + # and reading the field as a boolean is what let any string at all + # bypass this ceiling and the member cap behind it. + if not leased: + if not self._leaseless.admit(str(file_id)): + self._send({ + "type": "error", + "detail": "Too many files open at once without a transfer. " + "Download this one instead of previewing it.", + "code": "transfer_required", + "file_id": file_id, + }) + return + + log.debug("dl: req file=%s chunk=%s buffered=%s", + file_id[:12], chunk_index, + getattr(self._channel, "bufferedAmount", "?")) + file_hash = bytes.fromhex(entry.id) + chunk_data = await off_disk( + ctx["roots"], _read_and_encrypt, + ctx["gek"], file_path, chunk_index, file_hash, entry.id) + # Backpressure. Without it the node hands the whole window to the + # channel at once and the reader sees the first chunk, then nothing for + # as long as the link takes to drain the rest. + waited = 0.0 + while (self._channel is not None + and getattr(self._channel, "bufferedAmount", 0) > DOWNLOAD_BUFFER_HIGH + and self._channel.readyState == "open" + and waited < 60): + await asyncio.sleep(0.05) + waited += 0.05 + if self._channel is None or self._channel.readyState != "open": + return + self._send(chunk_data) + log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s", + file_id[:12], chunk_index, len(chunk_data.get("ct") or b""), + getattr(self._channel, "bufferedAmount", "?")) + if chunk_index == 0: + self._audit("file_download", entry.name) + # The last chunk is the only "close" a leaseless read has. Without this + # the session carries the entry until it goes idle, and the person who + # just looked at two photos cannot look at a third for a minute. + if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size: + self._leaseless.finish(str(file_id)) + + def _do_file_delete(self, msg: dict) -> None: + ctx = self._group_ctx() + file_id = msg.get("file_id", "") + if not file_id: + self._send({"type": "error", "detail": "Missing file_id"}) + return + + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + # An owner is an *account* now, so an entry that records one is + # challengeable even if the device that uploaded it is gone. + has_uploader = bool(entry.uploader_pk + or getattr(entry, "uploader_id", "")) + if not self._has_admin_authority() and not has_uploader: + self._send({"type": "error", "detail": "No authorized key for deletion"}) + return + + self._issue_admin_challenge(OP_FILE_DELETE, file_id) + + async def _verify_uploader_sig(self, entry, transcript: bytes, + sig: bytes) -> bool: + """ + Whether this signature comes from a live device of the file's uploader. + + Every non-revoked device of `entry.uploader_id` is tried, the same way + `_verify_device_signer` tries every device that may approve a new one. + Two properties worth keeping straight: + + - **Ownership survives device revocation.** A retired laptop's uploads + keep their owner, because the account is what owns them; the revoked + key simply is not among the ones that may act. + - **Ownership survives the account losing every device**, where nothing + verifies here and the operator remains able to delete — which is the + behaviour a group needs when someone leaves. + + Falls back to the recorded `uploader_pk` only when the roster cannot + answer at all (no roster wired, or no `uploader_id` on an entry written + before that field existed). That is the pre-device-linking behaviour, so + an old index does not become undeletable. + """ + roster = self._ctx.get("roster") + uploader_id = getattr(entry, "uploader_id", "") or "" + if roster is not None and uploader_id: + for device in await roster.list_devices(uploader_id): + try: + pk = Ed25519PublicKey.from_public_bytes( + base64.b64decode(device["pk_ed25519"])) + except Exception: + continue + if self._verify_sig(pk, transcript, sig): + return True + return False + + if not entry.uploader_pk: + return False + try: + pk = Ed25519PublicKey.from_public_bytes( + base64.b64decode(entry.uploader_pk)) + except Exception: + return False + return self._verify_sig(pk, transcript, sig) + + async def _admin_exec_file_delete( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + file_id = pending["subject"] + ctx = self._group_ctx() + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + # Node operator, or the account that uploaded this file — **any of its + # non-revoked devices**, resolved through the node's own roster. + # + # This used to verify against `entry.uploader_pk` alone, the exact key + # that uploaded. Device linking broke that on 2026-08-18 without + # anything failing loudly: a file uploaded from a phone could not be + # deleted from the same person's laptop, and the only symptom was + # "Signature verification failed" on their own file + # (docs/MESHBAY_DESIGN.md §3.3). + # + # `uploader_pk` is kept, and stops being the authorization key: it is + # now the audit record of *which device* did it. Authorization is by + # account, through the roster — never through a token claim, which is + # the protection per-node identity keys give (docs/MESHBAY_DESIGN.md + # §3.2) and which a lookup by `uploader_id` in the hub's world would + # give straight back. + if not (await self._verify_admin_sig(transcript, sig) + or await self._verify_uploader_sig(entry, transcript, sig)): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") + return + + await self._exec_file_delete(ctx, file_id, entry) + + async def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal == ROOT_NOT_SERVED: + # Frozen, not gone: removing the entry would lose a file that is + # still on a drive the node cannot read right now. + self._send({"type": "error", "detail": ROOT_NOT_SERVED}) + return + if file_path is not None: + await off_disk(ctx["roots"], file_path.unlink) + log.info("File deleted: %s", entry.name) + self._audit("file_delete", entry.name) + + ctx["index"].remove_entry(file_id) + self._send({ + "type": MNP.FILE_DELETE_ACK, + "v": MNP_VERSION, + "file_id": file_id, + }) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py index 89fe86b..7d458f4 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py @@ -3,3 +3,9 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 + + +# What the `tr` on a chunk request turned out to be (see `_lease_of`). +LEASE_GRANTED = "granted" +LEASE_QUEUED = "queued" +LEASE_NONE = "none" 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 889c411..cb95822 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -105,7 +105,6 @@ from meshbay_common.join import ( from meshbay_common.protocol import ( MNP, UPLOAD_PROBE_INDEX, - file_chunk_wire, file_upload_ack_wire, file_upload_payload, ) @@ -121,12 +120,10 @@ from meshbay_node.indexer.indexer import DirectoryIndexer # in media_probe.py so the indexer package (imported just above) can call it # too, for index-time enrichment, without a circular import. from meshbay_node.roots import ( - ROOT_NOT_SERVED, SAFE_UPLOAD_NAME, RootSet, _free_name, off_disk, - safe_subdir, ) from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK from meshbay_node.transfers import TransferSlots @@ -143,9 +140,8 @@ from meshbay_node.transport.webrtc.channel import ( _pack, ) from meshbay_node.transport.webrtc.chat import ChatMixin -from meshbay_node.transport.webrtc.disk import _locate -from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, MAX_MSG -from meshbay_node.transport.wire import index_sync_message +from meshbay_node.transport.webrtc.files import FilesMixin +from meshbay_node.transport.webrtc.limits import LEASE_GRANTED, LEASE_NONE, LEASE_QUEUED, MAX_MSG log = logging.getLogger(__name__) @@ -165,11 +161,6 @@ _INVITE_ID_RE = re.compile(r"[0-9a-f]{32}") MAX_UPLOAD_BYTES = 8 * 1024 * 1024 * 1024 # 8 GB per file GB_BYTES = 1024 * 1024 * 1024 -# What the `tr` on a chunk request turned out to be (see `_lease_of`). -LEASE_GRANTED = "granted" -LEASE_QUEUED = "queued" -LEASE_NONE = "none" - # Budget for an unauthenticated peer: enough for a handshake and a bundle fetch, # nowhere near enough to be a memory-exhaustion primitive (H6). PRE_HANDSHAKE_MAX_MSG = 64 * 1024 @@ -202,11 +193,6 @@ JOIN_FAILURE_WINDOW = 600 # seconds # nobody could read, or files scattered wherever someone happened to be looking. -# A chunk is a megabyte and the browser keeps eight in flight, so answering them -# as they arrive queues 8 MB on the channel with nothing watching. On a LAN that -# drains before anyone notices; on a phone that is also uploading, it is minutes -# of head-of-line delay for the reader. Above this, wait for room. -DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 1024 # How often transfer leases are swept. Nothing depends on it being # prompt -- the session teardown is the reclaim that matters and is # immediate; this catches peers that vanished without the connection @@ -229,7 +215,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0 class WebRTCPeerSession( - BlobsMixin, ChatMixin, + BlobsMixin, ChatMixin, FilesMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, ): """One WebRTC peer connection, handling MNP over a DataChannel.""" @@ -1656,151 +1642,6 @@ class WebRTCPeerSession( self._send(reply) self._audit_join("gek_wrapped", f"group={group_id[:8]}") - async def _do_dir_create(self, msg: dict) -> None: - """ - Create a directory, for any member of the group. - - Same confinement as an upload: every segment passes the name allowlist and - the result must resolve under the shared root. Making a directory is not a - privileged act — a member who can add a file can organise where it goes — - but it writes to the operator's disk, so it is audited like one. - """ - ctx = self._group_ctx() - roots: RootSet | None = ctx.get("roots") - if not roots: - self._send({"type": "error", "detail": "No shared directory"}) - return - - name = str(msg.get("name", "")).strip() - if not SAFE_UPLOAD_NAME.match(name): - self._send({"type": "error", "detail": "Invalid directory name"}) - return - - # The virtual root is not a directory on anyone's disk, so a member - # cannot create one there — that would be adding a root, which is the - # operator's configuration and not a file operation. - parent_rel = (msg.get("dir") or "").strip("/") - if not parent_rel: - self._send({"type": "error", - "detail": "Choose a folder to create this in"}) - return - - # Read-only means read-only, and creating a folder writes to the - # operator's disk. `_do_file_upload` gained this check with the RO/RW - # model and this one did not — so a member could not add a file to a - # published library but could still leave empty directories in it. - owner = roots.split(parent_rel) - if owner is None: - self._send({"type": "error", "detail": "Invalid directory"}) - return - parent_root, _tail = owner - if not parent_root.writable: - self._send({"type": "error", - "detail": f"Directory '{parent_root.name}' is read-only", - "code": "root_read_only"}) - self._audit("dir_create_refused", parent_rel[:64]) - return - if not parent_root.available: - self._send({"type": "error", - "detail": f"Directory '{parent_root.name}' is " - f"currently unavailable", - "code": "root_unavailable"}) - return - - parent = await off_disk(roots, safe_subdir, roots, parent_rel) - if parent is None or not await off_disk(roots, parent.is_dir): - self._send({"type": "error", "detail": "Invalid directory"}) - return - - target = await off_disk(roots, safe_subdir, roots, f"{parent_rel}/{name}") - if target is None: - self._send({"type": "error", "detail": "Invalid directory"}) - return - refusal = await off_disk(roots, _mkdir_if_absent, target) - if refusal is not None: - self._send({"type": "error", "detail": refusal}) - return - virtual = roots.virtual_of(target) or f"{parent_rel}/{name}" - log.info("Directory created by %s: %s", self._user_id[:8], virtual) - self._audit("dir_create", virtual) - self._send({ - "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION, - "dir": virtual, - }) - - @staticmethod - def _names_a_root(roots: RootSet, rel: str) -> bool: - """True when `rel` is a bare root name rather than something inside one.""" - found = roots.split(rel or "") - return found is not None and not found[1] - - async def _do_dir_delete(self, msg: dict) -> None: - """ - Remove an empty directory, for the node operator. - - Creating one is not privileged — a member who can add a file may organise - where it goes — but removing one is: it acts on a name other members are - using, and on the operator's disk. Empty is the whole safety property - here. Nothing recursive: refusing a directory with anything in it means - this can never destroy content, whatever the caller intended, so the - operator deletes the files first and sees what they are losing. - """ - ctx = self._group_ctx() - roots: RootSet | None = ctx.get("roots") - if not roots: - self._send({"type": "error", "detail": "No shared directory"}) - return - - rel = (msg.get("dir") or "").strip("/") - target = await off_disk(roots, safe_subdir, roots, rel) - # A root itself is not deletable here: removing one is a configuration - # change, and doing it through a file operation would leave the group - # config naming a directory nobody can reach. - if target is None or self._names_a_root(roots, rel): - self._send({"type": "error", "detail": "Invalid directory"}) - return - if not await off_disk(roots, target.is_dir): - self._send({"type": "error", "detail": "Not a directory"}) - return - if not await off_disk(roots, _is_empty_dir, target): - self._send({"type": "error", "detail": "Directory is not empty"}) - return - if not self._has_admin_authority(): - self._send({"type": "error", "detail": "No authorized key for deletion"}) - return - - self._issue_admin_challenge( - OP_DIR_DELETE, roots.virtual_of(target) or rel) - - async def _admin_exec_dir_delete( - self, pending: dict, transcript: bytes, sig: bytes, - ) -> None: - rel = pending["subject"] - ctx = self._group_ctx() - roots: RootSet | None = ctx.get("roots") - target = await off_disk(roots, safe_subdir, roots, rel) if roots else None - if (target is None or self._names_a_root(roots, rel) - or not await off_disk(roots, target.is_dir)): - self._send({"type": "error", "detail": "Not a directory"}) - return - - # Operator only. A file has an uploader who may remove their own; a - # directory has none, so there is no second key to accept here. - if not await self._verify_admin_sig(transcript, sig): - self._send({"type": "error", "detail": "Signature verification failed"}) - self._audit("admin_auth_failed", f"dir_delete:{rel}") - return - - # Checked again after the signature: the emptiness test that let this - # through happened before a round trip to the operator's browser, and a - # file could have landed in the meantime. - if not await off_disk(roots, _rmdir_if_empty, target): - self._send({"type": "error", "detail": "Directory is not empty"}) - return - log.info("Directory removed by %s: %s", self._user_id[:8], rel) - self._audit("dir_delete", rel) - self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel}) - def _do_member_revoke(self, msg: dict) -> None: """ Stop serving the group key to someone, at the operator's request. @@ -3137,137 +2978,6 @@ class WebRTCPeerSession( """Display-name cache, per group — same leak as _peer_registry (H1).""" return self._group_ctx().setdefault("_user_names", {}) - def _do_index_sync(self) -> None: - ctx = self._group_ctx() - self._send(index_sync_message(ctx["index"], ctx.get("roots"))) - - async def _try_serve_thumbnail( - self, thumb_hash: str, chunk_index: int, gek: bytes | None, - ) -> dict | None: - """ - docs/MESHBAY_DESIGN.md §6.5: a thumbnail is served through the same - chunked file_req path as a real file, resolved against the media - cache instead of the index when the id doesn't match a file. - Sliced by `chunk_index` like a real file's chunks, not just handed - back whole: a thumbnail/poster/cover never approached CHUNK_SIZE so - this used to be equivalent to "only chunk 0 exists", but an audio - transcode result (docs/MESHBAY_DESIGN.md §9.8, the WMA/Musepack exception) is - cached in the same media_cache blob store and can be several MB — - genuinely multi-chunk, same as a file read straight off disk. - """ - media_cache = self._ctx.get("media_cache") - if media_cache is None: - return None - blob = await media_cache.get_thumb(thumb_hash) - if blob is None: - return None - start = chunk_index * CHUNK_SIZE - if start > len(blob) or (start == len(blob) and chunk_index != 0): - return None - piece = blob[start:start + CHUNK_SIZE] - return file_chunk_wire( - gek, piece, chunk_index, bytes.fromhex(thumb_hash), thumb_hash) - - async def _do_file_request(self, msg: dict) -> None: - ctx = self._group_ctx() - # A chunk request is what "this transfer is alive" looks like. Nothing - # marked a lease used, so `used` stayed False for the whole download and - # the sweeper revoked the grant every 30 s as never-taken-up — while the - # file was transferring at 20 MB/s. Found in the node's own log, which - # repeated the same two reclaims every 30 s for as long as the daemon - # ran. - # - # And it is also what makes the caps real: `tr` names a lease or it - # does not, and `_lease_of` is the only thing that decides which. - tr = str(msg.get("tr") or "")[:64] - leased = False - if tr: - state = self._lease_of(tr) - if state == LEASE_QUEUED: - self._send({ - "type": "error", - "detail": "This transfer is waiting for a slot.", - "code": "lease_not_granted", - "tr": tr, - }) - return - leased = state == LEASE_GRANTED - if not leased: - self._note_unleased(tr) - file_id = msg["file_id"] - chunk_index = msg["chunk_index"] - entry = ctx["index"].get_entry(file_id) - if not entry: - 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", - file_id[:16], chunk_index) - self._send(thumb) - return - log.warning("File not found: %s", file_id[:16]) - self._send({"type": "error", "detail": "File not found"}) - return - - file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) - if refusal is not None: - self._send({"type": "error", "detail": refusal}) - return - - # A real index entry, asked for without a lease: browsing, or a client - # helping itself to the whole library outside every cap. - # - # Both look identical here — which is why the bound is a small count of - # files rather than a judgement about what the read is for. Thumbnails, - # posters and cover art never reach this line: they resolve through - # `_try_serve_thumbnail` above, out of a cache the node built itself, - # and are never leased, never counted, never queued. - # - # `not leased`, not `not tr`: a `tr` the node cannot match to a granted - # lease of this session is not a transfer, whatever the client calls it, - # and reading the field as a boolean is what let any string at all - # bypass this ceiling and the member cap behind it. - if not leased: - if not self._leaseless.admit(str(file_id)): - self._send({ - "type": "error", - "detail": "Too many files open at once without a transfer. " - "Download this one instead of previewing it.", - "code": "transfer_required", - "file_id": file_id, - }) - return - - log.debug("dl: req file=%s chunk=%s buffered=%s", - file_id[:12], chunk_index, - getattr(self._channel, "bufferedAmount", "?")) - file_hash = bytes.fromhex(entry.id) - chunk_data = await off_disk( - ctx["roots"], _read_and_encrypt, - ctx["gek"], file_path, chunk_index, file_hash, entry.id) - # Backpressure. Without it the node hands the whole window to the - # channel at once and the reader sees the first chunk, then nothing for - # as long as the link takes to drain the rest. - waited = 0.0 - while (self._channel is not None - and getattr(self._channel, "bufferedAmount", 0) > DOWNLOAD_BUFFER_HIGH - and self._channel.readyState == "open" - and waited < 60): - await asyncio.sleep(0.05) - waited += 0.05 - if self._channel is None or self._channel.readyState != "open": - return - self._send(chunk_data) - log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s", - file_id[:12], chunk_index, len(chunk_data.get("ct") or b""), - getattr(self._channel, "bufferedAmount", "?")) - if chunk_index == 0: - self._audit("file_download", entry.name) - # The last chunk is the only "close" a leaseless read has. Without this - # the session carries the entry until it goes idle, and the person who - # just looked at two photos cannot look at a third for a minute. - if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size: - self._leaseless.finish(str(file_id)) - def _do_ping(self, msg: dict) -> None: """Answer a liveness probe on an open channel, echoing the caller's token. @@ -3636,28 +3346,6 @@ class WebRTCPeerSession( return self._spawn(record(file_path, self._user_id or "", self._pinned_pk or "")) - def _do_file_delete(self, msg: dict) -> None: - ctx = self._group_ctx() - file_id = msg.get("file_id", "") - if not file_id: - self._send({"type": "error", "detail": "Missing file_id"}) - return - - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send({"type": "error", "detail": "File not found"}) - return - - # An owner is an *account* now, so an entry that records one is - # challengeable even if the device that uploaded it is gone. - has_uploader = bool(entry.uploader_pk - or getattr(entry, "uploader_id", "")) - if not self._has_admin_authority() and not has_uploader: - self._send({"type": "error", "detail": "No authorized key for deletion"}) - return - - self._issue_admin_challenge(OP_FILE_DELETE, file_id) - # ── Admin operation challenge/response (finding H5) ────────────────────── def _node_pk_b64(self) -> str: @@ -3708,49 +3396,6 @@ class WebRTCPeerSession( except Exception: return False - async def _verify_uploader_sig(self, entry, transcript: bytes, - sig: bytes) -> bool: - """ - Whether this signature comes from a live device of the file's uploader. - - Every non-revoked device of `entry.uploader_id` is tried, the same way - `_verify_device_signer` tries every device that may approve a new one. - Two properties worth keeping straight: - - - **Ownership survives device revocation.** A retired laptop's uploads - keep their owner, because the account is what owns them; the revoked - key simply is not among the ones that may act. - - **Ownership survives the account losing every device**, where nothing - verifies here and the operator remains able to delete — which is the - behaviour a group needs when someone leaves. - - Falls back to the recorded `uploader_pk` only when the roster cannot - answer at all (no roster wired, or no `uploader_id` on an entry written - before that field existed). That is the pre-device-linking behaviour, so - an old index does not become undeletable. - """ - roster = self._ctx.get("roster") - uploader_id = getattr(entry, "uploader_id", "") or "" - if roster is not None and uploader_id: - for device in await roster.list_devices(uploader_id): - try: - pk = Ed25519PublicKey.from_public_bytes( - base64.b64decode(device["pk_ed25519"])) - except Exception: - continue - if self._verify_sig(pk, transcript, sig): - return True - return False - - if not entry.uploader_pk: - return False - try: - pk = Ed25519PublicKey.from_public_bytes( - base64.b64decode(entry.uploader_pk)) - except Exception: - return False - return self._verify_sig(pk, transcript, sig) - async def _load_pinned_pk(self) -> None: """ A key this node pinned for the account we just authenticated. @@ -3972,40 +3617,6 @@ class WebRTCPeerSession( else: self._send({"type": "error", "detail": "Unknown admin operation"}) - async def _admin_exec_file_delete( - self, pending: dict, transcript: bytes, sig: bytes, - ) -> None: - file_id = pending["subject"] - ctx = self._group_ctx() - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send({"type": "error", "detail": "File not found"}) - return - - # Node operator, or the account that uploaded this file — **any of its - # non-revoked devices**, resolved through the node's own roster. - # - # This used to verify against `entry.uploader_pk` alone, the exact key - # that uploaded. Device linking broke that on 2026-08-18 without - # anything failing loudly: a file uploaded from a phone could not be - # deleted from the same person's laptop, and the only symptom was - # "Signature verification failed" on their own file - # (docs/MESHBAY_DESIGN.md §3.3). - # - # `uploader_pk` is kept, and stops being the authorization key: it is - # now the audit record of *which device* did it. Authorization is by - # account, through the roster — never through a token claim, which is - # the protection per-node identity keys give (docs/MESHBAY_DESIGN.md - # §3.2) and which a lookup by `uploader_id` in the hub's world would - # give straight back. - if not (await self._verify_admin_sig(transcript, sig) - or await self._verify_uploader_sig(entry, transcript, sig)): - self._send({"type": "error", "detail": "Signature verification failed"}) - self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") - return - - await self._exec_file_delete(ctx, file_id, entry) - async def _admin_exec_invite_create( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: @@ -4081,25 +3692,6 @@ class WebRTCPeerSession( self._send({"type": "ack", "v": MNP_VERSION, "detail": "invite_cancelled", "invite_id": payload["invite_id"]}) - async def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: - file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) - if refusal == ROOT_NOT_SERVED: - # Frozen, not gone: removing the entry would lose a file that is - # still on a drive the node cannot read right now. - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if file_path is not None: - await off_disk(ctx["roots"], file_path.unlink) - log.info("File deleted: %s", entry.name) - self._audit("file_delete", entry.name) - - ctx["index"].remove_entry(file_id) - self._send({ - "type": MNP.FILE_DELETE_ACK, - "v": MNP_VERSION, - "file_id": file_id, - }) - def _send(self, obj: dict) -> None: # Stamp the reply with the id of the request being answered, so the # caller never has to guess. Only for this session's own replies: a @@ -4146,62 +3738,12 @@ class WebRTCPeerSession( await self._pc.close() -def _mkdir_if_absent(target: Path) -> str | None: - """ - Create a directory unless it is already there, or say why not. - - Both in one call, not a check awaited and then an act: the disk thread is - one worker, so nothing can slip between them. Split across two awaits, two - members creating the same name would both find nothing there and the second - `mkdir` would raise where a refusal was meant. Blocking; called through - `off_disk`. - """ - if target.exists(): - return "Already exists" - target.mkdir(parents=False) - return None - - -def _is_empty_dir(target: Path) -> bool: - """Blocking; called through `off_disk`.""" - return not any(target.iterdir()) - - -def _rmdir_if_empty(target: Path) -> bool: - """ - Remove a directory if nothing is in it. False if something is. - - The emptiness test and the removal are one call for the reason the caller - re-tests at all: the first test happened before a round trip to the - operator's browser, and a file can land in between. Two awaits here would - reopen the same window one size smaller. Blocking; called through `off_disk`. - """ - if any(target.iterdir()): - return False - target.rmdir() - return True - - def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None: """Add one chunk to a partial upload. Blocking; called through `off_disk`.""" with open(tmp_path, "wb" if first else "ab") as f: f.write(chunk_bytes) -def _read_and_encrypt( - gek: bytes, - file_path: Path, - chunk_index: int, - file_hash: bytes, - file_id: str = "", -) -> dict: - """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`.""" - with open(file_path, "rb") as f: - f.seek(chunk_index * CHUNK_SIZE) - plaintext = f.read(CHUNK_SIZE) - return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) - - class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. |