diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 404 |
1 files changed, 3 insertions, 401 deletions
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. |