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