From 02aeb0a751dce80fac1bfca31b6f736558953b77 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Thu, 24 Sep 2026 10:12:37 +0200 Subject: refactor(node): move uploads out of webrtc_server UploadMixin in transport/webrtc/upload_handlers.py, with the upload cap; _upload_chunk moved unchanged, _append_chunk joins webrtc/disk.py. Co-Authored-By: Claude Opus 5.5 --- .../src/meshbay_node/transport/webrtc/disk.py | 6 + .../transport/webrtc/upload_handlers.py | 403 +++++++++++++++++++++ 2 files changed, 409 insertions(+) create mode 100644 packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc') 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 bc8d33f..2e484de 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py @@ -74,3 +74,9 @@ def _read_and_encrypt( f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) + + +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) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py new file mode 100644 index 0000000..1c6d1ce --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py @@ -0,0 +1,403 @@ +"""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 `/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 "")) -- cgit v1.2.3