path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py

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