diff options
9 files changed, 424 insertions, 418 deletions
diff --git a/docs/MESHBAY_NODE_PROTOCOL.md b/docs/MESHBAY_NODE_PROTOCOL.md index 3c39835..90b0b41 100644 --- a/docs/MESHBAY_NODE_PROTOCOL.md +++ b/docs/MESHBAY_NODE_PROTOCOL.md @@ -2294,7 +2294,7 @@ LP(x) = uint32be(len(x)) || x every field, no exceptions | `PRE_HANDSHAKE_MAX_MSG` / `MAX_MSG` | 64 KiB / 64 MiB | `webrtc_server.py` / `webrtc/limits.py` | | `CHUNK_SIZE` | 1 MiB | `webrtc/limits.py` | | `DOWNLOAD_BUFFER_HIGH` | 2 MiB | `webrtc/files.py` | -| `MAX_UPLOAD_BYTES` | 8 GiB, default only — `max_upload_gb` overrides it per node | `webrtc_server.py` | +| `MAX_UPLOAD_BYTES` | 8 GiB, default only — `max_upload_gb` overrides it per node | `webrtc/upload_handlers.py` | | Download pipeline / chunk retry (client) | 8 in flight; 6 attempts, 1.5 s apart | `file-utils.js` | | Upload chunk / window / send-buffer high water (client) | 48 KiB / 32 / 1 MiB | `transport.js` | | `UPLOAD_ID_LEN` | 16 bytes, hex on the wire | `protocol.py` | diff --git a/docs/transfers-v1.md b/docs/transfers-v1.md index 711c861..f9c3675 100644 --- a/docs/transfers-v1.md +++ b/docs/transfers-v1.md @@ -58,7 +58,7 @@ and encrypts one 1 MB chunk, waits for room on the channel learns that a download started, and never learns that one ended.** There is nothing to count and nothing to cap. This is the central fact of this work. -**An upload is half-visible.** `_do_file_upload` (`webrtc_server.py:4471`) +**An upload is half-visible.** `_do_file_upload` (`webrtc/upload_handlers.py`) keeps `self._uploads[f"{rel_dir}/{filename}"]` with `next_index` and a `.part` file on disk. That state is **per session object and in memory**: a browser that disconnects mid-upload leaves a `.part` file on the operator's disk that diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index a3bde8f..9faa09c 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -389,7 +389,7 @@ class NodeDaemon: "note_activity": indexer.note_activity, # Bound method, called when an upload finishes. The entry # it belongs to does not exist yet (see - # webrtc_server._register_uploader), so the indexer keeps + # webrtc/upload_handlers.py _register_uploader), so the indexer keeps # the record and stamps the entry when it creates it. "record_upload": indexer.record_upload, # Shown to the operator in Settings, and kept current in 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 `<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 "")) 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 a617d29..9264b60 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -29,7 +29,6 @@ import os import re import time import uuid -from pathlib import Path from typing import Any from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription @@ -104,14 +103,10 @@ from meshbay_common.join import ( ) from meshbay_common.protocol import ( MNP, - UPLOAD_PROBE_INDEX, - file_upload_ack_wire, - file_upload_payload, ) from meshbay_node import ops from meshbay_node import transfers as transfers_mod -from meshbay_node import uploads as uploads_mod from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer @@ -120,10 +115,7 @@ 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 ( - SAFE_UPLOAD_NAME, RootSet, - _free_name, - off_disk, ) from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK from meshbay_node.transport.webrtc.apps.music import MusicMixin @@ -140,8 +132,9 @@ from meshbay_node.transport.webrtc.channel import ( ) from meshbay_node.transport.webrtc.chat import ChatMixin from meshbay_node.transport.webrtc.files import FilesMixin -from meshbay_node.transport.webrtc.limits import LEASE_NONE, LEASE_QUEUED, MAX_MSG +from meshbay_node.transport.webrtc.limits import MAX_MSG from meshbay_node.transport.webrtc.transfer_handlers import TransferMixin +from meshbay_node.transport.webrtc.upload_handlers import UploadMixin log = logging.getLogger(__name__) @@ -149,18 +142,6 @@ log = logging.getLogger(__name__) _INVITE_ID_RE = re.compile(r"[0-9a-f]{32}") -# 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 - # 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 @@ -209,7 +190,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0 class WebRTCPeerSession( - BlobsMixin, ChatMixin, FilesMixin, TransferMixin, + BlobsMixin, ChatMixin, FilesMixin, TransferMixin, UploadMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, ): """One WebRTC peer connection, handling MNP over a DataChannel.""" @@ -2671,20 +2652,6 @@ class WebRTCPeerSession( # ── Upload ceiling and transfer slots ──────────────────────────────────── - 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 _register_peer(self) -> None: """Add this connection to its group's peer set. @@ -2731,365 +2698,6 @@ class WebRTCPeerSession( """ self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")}) - 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 "")) - # ── Admin operation challenge/response (finding H5) ────────────────────── def _node_pk_b64(self) -> str: @@ -3482,12 +3090,6 @@ class WebRTCPeerSession( await self._pc.close() -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) - - class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py index e404b47..a94789a 100644 --- a/packages/meshbay-node/tests/test_disk_io_off_loop.py +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -34,8 +34,7 @@ from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node.roots import Root, RootSet from meshbay_node.transfers import LeaselessReads -from meshbay_node.transport import webrtc_server -from meshbay_node.transport.webrtc import files, media_tools +from meshbay_node.transport.webrtc import files, media_tools, upload_handlers from meshbay_node.transport.webrtc_server import WebRTCPeerSession from node_source import webrtc_files @@ -316,8 +315,8 @@ def _upload_session(tmp_path): async def test_a_slow_upload_write_does_not_stop_the_loop(tmp_path, monkeypatch): session, shared = _upload_session(tmp_path) - monkeypatch.setattr(webrtc_server, "_append_chunk", - _slow(webrtc_server._append_chunk)) + monkeypatch.setattr(upload_handlers, "_append_chunk", + _slow(upload_handlers._append_chunk)) with _Ticker() as ticker: await session._do_file_upload(sealed_upload( @@ -344,8 +343,8 @@ async def test_chunks_of_one_upload_keep_their_order_under_a_slow_disk(tmp_path, session, shared = _upload_session(tmp_path) pieces = [b"first-", b"second-", b"third-", b"fourth"] - monkeypatch.setattr(webrtc_server, "_append_chunk", - _slow(webrtc_server._append_chunk)) + monkeypatch.setattr(upload_handlers, "_append_chunk", + _slow(upload_handlers._append_chunk)) # Fired together and in order, which is what the dispatcher does: it creates # one task per message as it arrives. diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 7ace21b..dd859b3 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -29,7 +29,7 @@ def _safe_name_re(): rather than aborting collection of the whole module and hiding every other finding's result. """ - from meshbay_node.transport.webrtc_server import SAFE_UPLOAD_NAME + from meshbay_node.roots import SAFE_UPLOAD_NAME return SAFE_UPLOAD_NAME @@ -122,7 +122,7 @@ def test_upload_rejects_names_that_lie_about_themselves(name): def test_the_node_never_generates_a_name_it_would_refuse(tmp_path): """_free_name resolves a collision by appending " (n)"; that has to be legal.""" - from meshbay_node.transport.webrtc_server import _free_name + from meshbay_node.roots import _free_name (tmp_path / "clip.mp4").touch() (tmp_path / "clip (1).mp4").touch() chosen = _free_name(tmp_path, "clip.mp4") diff --git a/packages/meshbay-node/tests/test_upload_size_cap.py b/packages/meshbay-node/tests/test_upload_size_cap.py index dbb14d9..7675f32 100644 --- a/packages/meshbay-node/tests/test_upload_size_cap.py +++ b/packages/meshbay-node/tests/test_upload_size_cap.py @@ -19,12 +19,8 @@ from pathlib import Path import pytest from meshbay_node.config import load_config from meshbay_node.roots import RootSet -from meshbay_node.transport.webrtc_server import ( - GB_BYTES, - MAX_UPLOAD_BYTES, - WebRTCPeerSession, - WebRTCTransport, -) +from meshbay_node.transport.webrtc.upload_handlers import GB_BYTES, MAX_UPLOAD_BYTES +from meshbay_node.transport.webrtc_server import WebRTCPeerSession, WebRTCTransport from node_source import webrtc_source |