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

"""Uploads: a member's sealed chunks written to a partial file, checked, and
renamed into place under the name the node chose."""

import asyncio
import logging
from pathlib import Path

from meshbay_common.protocol import UPLOAD_PROBE_INDEX, file_upload_ack_wire, file_upload_payload

from meshbay_node import uploads as uploads_mod
from meshbay_node.roots import SAFE_UPLOAD_NAME, RootSet, _free_name, off_disk
from meshbay_node.transport.webrtc.disk import _append_chunk
from meshbay_node.transport.webrtc.limits import LEASE_NONE, LEASE_QUEUED

log = logging.getLogger("meshbay_node.transport.webrtc_server")


# Upload limits (finding C5a). Uploads used to land directly in the shared root under
# a name the client chose, overwriting whatever was already there — which both violated
# node sovereignty and defeated the delete authorization (overwrite a file, become its
# recorded uploader, then delete it legitimately).
# The ceiling is the operator's to set (`max_upload_gb` in node.toml, the Node
# page and `meshbay-node transfers max-size`) because it is their disk that
# fills: this is only the default a node starts from when they have said
# nothing. It is read from the transport context on every chunk, so a change
# applies to an upload already in flight.
MAX_UPLOAD_BYTES = 8 * 1024 * 1024 * 1024   # 8 GB per file
GB_BYTES = 1024 * 1024 * 1024


class UploadMixin:
    def _max_upload_bytes(self) -> int:
        """The per-file upload ceiling this node is running with, in bytes.

        Read from the transport context rather than captured once, for the same
        reason the transfer pools are refreshed there: the operator can change
        it from the Node page or the CLI while an upload is running, and a
        ceiling that only applies after a restart is not the one they were
        shown. `None` means they have said nothing and the default stands.
        """
        gb = self._ctx.get("max_upload_gb")
        if not gb:
            return MAX_UPLOAD_BYTES
        return max(1, int(float(gb) * GB_BYTES))

    def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads:
        """This group's uploads in progress, created on first use.

        In the group context rather than on the session, so a client that
        reconnects finds its own upload where it left it — and so the reaper has
        something to ask "is anyone still writing this?".
        """
        store = ctx.get("partial_uploads")
        if store is None:
            store = uploads_mod.PartialUploads()
            ctx["partial_uploads"] = store
        return store

    @staticmethod
    def _upload_lock(ctx: dict) -> asyncio.Lock:
        """
        One lock per group, beside the state it protects.

        Not per session: `partial_uploads` lives in the group context so a
        client that reconnects finds its upload where it left it, which means
        two sessions of the same member share the position of one `.part` file.
        A lock on the session would let them interleave — and the loop no longer
        serializes them for free now that a chunk write is awaited.
        """
        lock = ctx.get("upload_lock")
        if lock is None:
            lock = asyncio.Lock()
            ctx["upload_lock"] = lock
        return lock

    async def _do_file_upload(self, msg: dict) -> None:
        """
        One chunk of an upload, in the order it arrived.

        The chunk ordering rule — `chunk_index != state.next_index` is refused —
        used to hold for free: the handler was synchronous, so nothing could run
        between the check and the `advance` that answers it. Awaiting the write
        opens that gap, and two chunks of one upload racing through it is a
        `.part` file with a hole in it or a chunk refused for arriving on time.
        So the check, the write and the advance are one critical section again.

        The order is the arrival order: `_dispatch_message` runs per message as
        it arrives and creates these tasks in that order, tasks start in
        creation order, and this lock is the first thing each one waits on, so
        its queue of waiters is in arrival order too.
        """
        async with self._upload_lock(self._group_ctx()):
            await self._upload_chunk(msg)

    async def _upload_chunk(self, msg: dict) -> None:
        """
        One chunk of an upload, sealed under the group key (MNP 2.0).

        Sealing this direction is not symmetry for its own sake. Downloads have
        been under a GEK-derived key since the beginning; uploads carried the
        filename and the raw bytes in plain msgpack, so the same file was
        ciphertext leaving a node and plaintext arriving at one. The node holds
        the GEK for its own group, so it opens the payload here — before it
        decides a destination, before it touches the disk — and refuses a chunk
        that does not open.

        `upload_id` is the correlation key and stays in clear; `filename`, `dir`
        and `root` moved inside the seal, which is why every refusal below names
        the upload rather than the file. A `code` says which refusal it is, and
        the client already knows what it sent.
        """
        ctx = self._group_ctx()
        upload_id = str(msg.get("upload_id") or "")[:64]

        # Say the slot is being used, chunk by chunk, exactly as `_do_file_req`
        # does for a download.
        #
        # A grant nobody takes up is reclaimed after GRANT_DEADLINE_SECS and, on
        # the third miss, abandoned. Uploads were not gated by the lease, so the
        # file still arrived — but the widget follows the lease, so a 3.5 GB
        # upload showed "waiting, 0 ahead" for a minute and a half while it was
        # in fact transferring, and the node logged three reclaims against a
        # transfer that never stopped. Measured, from the journal:
        #
        #   11:52:49 open upload 919ebf54 -> granted
        #   11:53:19 reclaimed 919ebf54 (not_taken_up)
        #   11:54:19 reclaimed 919ebf54 (abandoned)
        #   11:55:48 Upload complete: ... (3 522 297 517 bytes)
        #
        # `_do_file_request` marks a lease alive for exactly the same reason;
        # both sides of a transfer have to say they are still moving, or the
        # sweeper reclaims whichever one forgot.
        #
        # And a queued lease is refused here rather than written to disk. There
        # is no leaseless fallback on this side — a write is never "browsing" —
        # so the two cases part company: a lease this node has not granted is a
        # member taking a slot they were told to wait for, and an unknown `tr`
        # is the reconnect case, where the client is re-opening leases that died
        # with the old connection and the four upload protections (§6.4) are
        # what bound it meanwhile.
        tr = str(msg.get("tr") or "")[:64]
        if tr:
            state = self._lease_of(tr)
            if state == LEASE_QUEUED:
                self._send({
                    "type": "error",
                    "detail": "This upload is waiting for a slot.",
                    "code": "lease_not_granted",
                    "upload_id": upload_id,
                    "tr": tr,
                })
                return
            if state == LEASE_NONE:
                self._note_unleased(tr)

        gek = ctx.get("gek")
        if not gek:
            self._send({"type": "error", "detail": "Group encryption not initialized",
                        "code": "no_group_key", "upload_id": upload_id})
            return

        try:
            payload = file_upload_payload(gek, self._group_id or "", msg)
        except Exception:
            # Deliberately one answer for "not sealed at all" and "sealed wrong":
            # distinguishing them tells a peer which of the two it got right.
            # An MNP 1.x client lands here, which is the whole of the upgrade
            # story — everything else it does still works.
            self._audit("upload_refused", "unsealed")
            self._send({
                "type": "error",
                "detail": "This upload did not open under the group key — the "
                          "client may be running an older version",
                "code": "upload_not_sealed",
                "upload_id": upload_id,
            })
            return

        filename = payload.get("filename") or ""
        data = payload.get("data")
        # From the clear part of the message, so peer-controlled and unchecked
        # by the AEAD. Everything below compares and adds to them.
        try:
            chunk_index = int(msg.get("chunk_index", 0))
            total_chunks = int(msg.get("total_chunks", 1))
        except (TypeError, ValueError):
            self._send({"type": "error", "detail": "Invalid chunk index",
                        "code": "bad_chunk_index", "upload_id": upload_id})
            return

        def _refuse(detail: str, code: str = "") -> None:
            """A refusal names the upload, never the file: the name is sealed."""
            out = {"type": "error", "detail": detail, "upload_id": upload_id}
            if code:
                out["code"] = code
            self._send(out)

        # Types first, and before any state is created. What comes out of a
        # sealed payload is authenticated, not validated: it is msgpack a
        # member wrote, and `SAFE_UPLOAD_NAME.match(123)` raises where a
        # refusal was meant.
        if not isinstance(filename, str) or not filename:
            _refuse("Missing filename or data", "upload_incomplete")
            return
        # Bytes, always: base64 was the shape of the old plaintext `data` field
        # and there is no sealed message that can carry a string here.
        if not isinstance(data, (bytes, bytearray)):
            _refuse("Invalid chunk encoding", "bad_chunk_encoding")
            return
        chunk_bytes = bytes(data)

        if not SAFE_UPLOAD_NAME.match(filename):
            _refuse("Invalid filename", "invalid_filename")
            return

        roots: RootSet | None = ctx.get("roots")
        if not roots:
            _refuse("No directories configured for this group", "no_roots")
            return

        # The client names the root it is uploading into — it is browsing one,
        # and with several writable roots any other choice is a guess. It names
        # a root, never a path: the destination inside it is decided below and
        # is not negotiable, which is what keeps C5a closed.
        #
        # An unknown name is refused rather than falling back to a writable
        # root, because "the file went somewhere else" is discovered weeks
        # later — the same reason the old single upload root was never guessed.
        # `dir` is the folder being browsed, as a virtual path
        # (`Media/Films/1999`); `root` is the older, coarser form and is what
        # its first segment means on its own. Both are sealed now, so a refusal
        # below can no longer quote them back.
        target_rel = str(payload.get("dir") or "").strip().strip("/")
        target_root_name = (target_rel.split("/")[0] if target_rel
                            else str(payload.get("root") or "").strip())
        upload_root = None
        if target_root_name:
            upload_root = roots.by_name(target_root_name)
            if upload_root is None:
                _refuse("No such directory in this group", "no_such_root")
                return
        else:
            writable = roots.writable_roots
            upload_root = writable[0] if writable else None

        if upload_root is None:
            _refuse("No writable directory in this group", "no_writable_root")
            return
        if not upload_root.writable:
            _refuse("That directory is read-only", "root_read_only")
            self._audit("upload_refused", filename[:64])
            return
        if not upload_root.available:
            _refuse("That directory is currently unavailable", "root_unavailable")
            return

        # The folder the sender is looking at, and no subdirectory of the node's
        # invention.
        #
        # Uploads used to be confined to `<root>/uploads/`, created on demand.
        # That was the last of v5's quarantine (the per-user layer went on
        # 2026-08-14, for the same reason): a shared directory nobody can
        # organise is not a shared directory, and a folder appearing beside the
        # operator's library because somebody sent a file is the node deciding
        # how their disk is arranged.
        #
        # What made the quarantine worth having is not the subdirectory — it is
        # the filename allowlist, the size cap, the chunk ordering, and the
        # no-overwrite rule below. All four are unchanged.
        #
        # `resolve()` and not a join: it refuses `..`, absolute segments and
        # anything whose resolved form escapes its root, symlinks included. The
        # client names *where among the group's own folders*, never a path on
        # the operator's filesystem.
        if target_rel:
            target_dir = await off_disk(roots, roots.resolve, target_rel)
            if target_dir is None or not await off_disk(roots, target_dir.is_dir):
                _refuse("Not a directory in this group", "no_such_directory")
                return
            rel_dir = target_rel
        else:
            # A client that names nothing: the first writable root is where its
            # one destination is.
            target_dir = upload_root.path
            rel_dir = upload_root.name
            if not await off_disk(roots, target_dir.is_dir):
                _refuse("That directory is currently unavailable",
                        "root_unavailable")
                return

        # Held by the group, not by this connection.
        #
        # This used to be `self._uploads`, on the session. A dropped link threw
        # the position away and the next chunk was refused with `not_started`:
        # an upload interrupted at 99% could only be started again from zero, on
        # a connection flaky enough to have interrupted it once. And the state
        # it lost was the only thing that knew about the `.part` file left
        # behind — see `uploads.orphaned_parts`, which is the other half of this.
        #
        # Keyed by member as well as by name, because a shared directory means
        # two people can be sending IMG_1234.jpg at the same moment and neither
        # may inherit the other's position.
        uploads = self._partial_uploads(ctx)
        user_id = self._user_id or ""
        state = uploads.get(user_id, rel_dir, filename)
        # A shared directory means two people can send the same name. Refusing the
        # second is safe but silly — everyone's camera produces IMG_1234.jpg — so
        # a free name is found instead. Never a replacement.
        stored_name = (state.stored_name if state
                       else await off_disk(roots, _free_name, target_dir, filename))
        tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}"
        final_path = target_dir / stored_name

        if chunk_index == UPLOAD_PROBE_INDEX:
            # "Where am I?", asked inside the seal rather than on a clear
            # message, because the answer is about a file whose name is exactly
            # what sealing this path was for.
            #
            # It writes nothing, creates no state and reserves no name: a client
            # that asks and then goes away has cost this node one reply. Every
            # check above has already run, so it cannot be used to ask questions
            # about a directory the caller may not write to.
            self._send(file_upload_ack_wire(
                gek, self._group_id or "",
                upload_id=upload_id,
                chunk_index=UPLOAD_PROBE_INDEX,
                filename=filename,
                # Only what is really on disk. Without state, `_free_name` above
                # picked a name nothing has claimed yet, and reporting it would
                # promise a destination the real chunk 0 may not choose.
                stored_as=state.stored_name if state else "",
                dir=rel_dir,
                resume_from=state.next_index if state else 0,
            ))
            return

        if chunk_index == 0:
            # Backstop: _free_name already guarantees this, and it stays because
            # it asserts the invariant where the write happens.
            if await off_disk(roots, final_path.exists):
                _refuse("File already exists", "already_exists")
                return
            state = uploads.start(user_id, rel_dir, filename, stored_name,
                                  part_path=tmp_path)
        elif state is None:
            _refuse("Upload not started", "not_started")
            return

        # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends
        # blindly to whatever .part file is already on disk.
        if chunk_index != state.next_index:
            _refuse("Unexpected chunk index", "bad_chunk_index")
            return

        if state.bytes + len(chunk_bytes) > self._max_upload_bytes():
            uploads.drop(user_id, rel_dir, filename)
            await off_disk(roots, tmp_path.unlink, True)
            _refuse("Upload exceeds size limit", "too_large")
            return

        await off_disk(roots, _append_chunk, tmp_path, chunk_bytes, chunk_index == 0)
        uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes))

        self._send(file_upload_ack_wire(
            gek, self._group_id or "",
            upload_id=upload_id,
            chunk_index=chunk_index,
            filename=filename,
            # What it is actually called on disk, which a chat attachment has to
            # reference and the uploader deserves to be told.
            stored_as=stored_name,
            dir=rel_dir,
        ))

        if chunk_index + 1 >= total_chunks:
            uploads.drop(user_id, rel_dir, filename)
            await off_disk(roots, tmp_path.rename, final_path)
            log.info("Upload complete: %s (%d chunks, %d bytes)",
                     stored_name, total_chunks, state.bytes)
            self._audit("file_upload", f"{rel_dir}/{stored_name}")
            self._register_uploader(ctx, final_path)

    def _register_uploader(self, ctx: dict, file_path: Path) -> None:
        """
        Record who sent this file, for the index entry that does not exist yet.

        The key recorded is the one this node pinned, not the one the token
        carried. `pk_user` was a hub-chosen claim, and it decided who could later
        delete the file: a hub issuing a token naming its own key could delete
        anyone's uploads on any node. Deletion is supposed to be authorized by the
        node, and this closes the last place where it was not.

        **The entry is not here to be tagged.** This used to walk `ctx["index"]`
        for the name just written and set the fields on it; at this point the
        watchdog has not fired (it debounces for two seconds and then hashes)
        and the file was a `.part` until the line above, which is not indexable
        — so the walk matched nothing, every time, and said nothing about it.
        The indexer stamps the entry from this record when it creates it.
        """
        record = ctx.get("record_upload")
        if record is None:
            return
        self._spawn(record(file_path, self._user_id or "", self._pinned_pk or ""))