""" MeshBay Node — WebRTC DataChannel server for browser clients. Browsers cannot use QUIC for NAT traversal (WebTransport doesn't allow choosing the UDP source port — Port-Restricted Cone NAT requires exact port matching). WebRTC DataChannel with ICE/STUN handles this automatically. The MNP protocol (handshake, file_request, file_chunk, chat, etc.) runs identically over WebRTC DataChannel as over QUIC streams. Same E2E encryption, same message types, same msgpack wire format. Wire format on the DataChannel: - Each message is length-prefixed msgpack (4-byte big-endian + msgpack payload) - Same as QUIC streams and TCP+TLS - DataChannel is ordered and reliable (SCTP over DTLS) Signaling flow (handled externally by the hub): Browser → Hub : POST /v1/nodes/{id}/webrtc/offer {sdp, ice_candidates} Hub → Node : WS push {type: "webrtc_offer", sdp, ice_candidates, peer_id} Node → Hub : WS push {type: "webrtc_answer", sdp, ice_candidates, peer_id} Hub → Browser : SSE/response {sdp, ice_candidates} After signaling, DataChannel is P2P — hub is out of the loop. """ import asyncio import base64 import hashlib import hmac import logging import os import re import struct import time from pathlib import Path from typing import Any import jwt import msgpack from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) from meshbay_common import MNP_VERSION from meshbay_common.handshake import ( NONCE_LEN, ROLE_CLIENT, ROLE_NODE, HandshakeError, authorize_token, handshake_transcript, make_proof, verify_proof, webrtc_binding, ) from meshbay_common.adminop import ( ADMIN_CHALLENGE_TTL, OP_DIR_DELETE, OP_FILE_DELETE, OP_INVITE_CREATE, OP_MEMBER_REVOKE, OP_GEK_ROTATE, OP_MEMBER_UNPIN, admin_transcript, ) from meshbay_common.crypto import pk_to_b64, wrap_gek_aes from meshbay_common.join import ( JOIN_TTL, ROLE_MEMBER, ROLE_OPERATOR, join_transcript, ) from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex from meshbay_node import ops from meshbay_node.roots import RootSet, entry_abs_path from meshbay_node.roster import DEFAULT_INVITE_TTL log = logging.getLogger(__name__) CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 # 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). MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024 # 4 GB per file # 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 # ffmpeg is spawned per stream request; without a cap any member can fork-bomb # the node by requesting many streams at once (H6). # # Two was sized when a stream was a burst: the client took segments as fast as # it could append them, so a slot was held for the minute it took to push the # file and then came back. Now that the client only pulls ninety seconds ahead # of the playhead, a slot is held for as long as the film runs — so two slots # means two people can watch anything at all, and the third is refused for the # next hour and a half. The work behind a slot has not changed and is small: # ffmpeg runs `-c copy`, a remux with no encoding in it, and spends most of the # film blocked on a pipe nobody is reading. # # This is the default, not the policy: the right number depends on the machine, # so the operator sets `max_concurrent_streams` under [node] in node.toml. This # value applies when they have said nothing. MAX_CONCURRENT_TRANSCODES = 8 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited # until the native client removes remote keypair bundles entirely. MAX_PRE_PROOF_FETCHES = 4 # Pairing codes carry 40 bits and are single-use, but a connection must not be # allowed to sit there guessing. Failures are audited, so a grind is visible. MAX_JOIN_ATTEMPTS = 5 # Per-connection limits alone would not bind an attacker who can open connections # at will — and the adversary who can mint tokens for any account is the hub. So # failed pairings are also counted node-wide over a window. MAX_JOIN_FAILURES_WINDOW = 20 JOIN_FAILURE_WINDOW = 600 # seconds # Everything a member sends lands here: files from the Files panel and # attachments from the chat alike. One visible directory the operator can look # into, back up or empty — rather than a hidden tree of per-user uuids that # nobody could read, or files scattered wherever someone happened to be looking. UPLOAD_DIR_NAME = "uploads" # Conservative allowlist: also what keeps markup out of filenames, which the node admin # UI used to render unescaped (finding H2). # An allowlist, still — C5a and H2 depend on it — but one that does not assume # the world writes in ASCII. `été.txt` and `rapport (1).pdf` were refused, and # the second of those is a name _free_name generates itself, so the node was # rejecting files it had named. `\w` is Unicode here, which admits letters and # digits of any script while `<`, `>`, `"`, `;`, `/`, `\` and control characters # stay out. The first character must be a letter or digit, so ".." and dotfiles # cannot start one, and a trailing space or dot is refused because it makes two # different files look identical in a list. SAFE_UPLOAD_NAME = re.compile( r"^[^\W_]" # letter or digit — never '.', '-' or space r"[\w .\-()\[\]'\u2019,&+#@]{0,127}" # body: word chars plus mild punctuation r"(? str: """ `filename`, or the first "name (n).ext" that is not taken. Never returns the name of a file that exists, so an upload cannot replace one — the property the per-user quarantine used to provide (C5a). """ if not (directory / filename).exists(): return filename stem, dot, ext = filename.rpartition(".") if not dot: stem, ext = filename, "" for n in range(2, 1000): candidate = f"{stem} ({n}){dot}{ext}" if not (directory / candidate).exists(): return candidate raise FileExistsError(filename) def safe_subdir(roots: RootSet, rel: str) -> Path | None: """ Resolve a client-supplied directory inside one of the group's roots, or refuse. The path arrives from the wire, so every part is checked: the first segment must name a root that is readable right now, each later segment against the same allowlist as filenames, and the resolved result against that root's directory. `..`, absolute paths, symlinks pointing out, and anything with a separator in a segment are all refused here rather than in the caller, so there is one place to get it right. The virtual root itself — `""` — is deliberately **not** resolvable. It is not a directory on anyone's disk: a file cannot be written there and a directory cannot be created there, because it belongs to no volume. Callers that used to receive the shared root for an empty path now receive None, which is the honest answer. The quarantine was the fix for C5a; what actually mattered in it — no overwrite, a name allowlist, and confinement — is kept by this plus the caller's existing checks. """ found = roots.split(rel or "") if found is None: return None root, tail = found if not root.available: return None parts = [seg for seg in tail.split("/") if seg not in ("", ".")] if any(seg == ".." or not SAFE_UPLOAD_NAME.match(seg) for seg in parts): return None try: target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve() base = root.path.resolve() except OSError: return None if target != base and base not in target.parents: return None return target def _extract_dtls_fingerprint(sdp: str) -> bytes: """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes.""" for line in sdp.splitlines(): if line.startswith("a=fingerprint:sha-256 "): hex_str = line.split(" ", 1)[1].replace(":", "") return bytes.fromhex(hex_str) return b"" STREAM_SEGMENT_SIZE = 256 * 1024 # A chunk is a megabyte and the browser keeps eight in flight, so answering them # as they arrive queues 8 MB on the channel with nothing watching. On a LAN that # drains before anyone notices; on a phone that is also uploading, it is minutes # of head-of-line delay for the reader. Above this, wait for room. DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 1024 # What a client may ask for in one go, and how long the node waits for it to ask # again before deciding nobody is watching any more. STREAM_MAX_CREDIT = 256 STREAM_CREDIT_TIMEOUT = 120 # How often that budget is re-examined. A viewer who left stops being # charged for a slot within this, rather than within the timeout. STREAM_CREDIT_POLL = 3 _H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} async def _probe_video(path: str) -> tuple[str | None, float]: """Probe video file with ffprobe, return (MSE codec string, duration).""" import json as _json proc = await asyncio.create_subprocess_exec( "ffprobe", "-v", "error", "-show_entries", "stream=codec_name,profile,level,codec_type", "-show_entries", "format=duration", "-of", "json", path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, _ = await proc.communicate() info = _json.loads(stdout) duration = float(info.get("format", {}).get("duration", 0)) v_codec = a_codec = "" for s in info.get("streams", []): if s.get("codec_type") == "video" and not v_codec: cn = s.get("codec_name", "") if cn == "h264": p = _H264_PROFILES.get(s.get("profile", "High"), "64") lvl = int(s.get("level", 40)) v_codec = f"avc1.{p}00{lvl:02x}" elif cn == "hevc": v_codec = "hev1.1.6.L93.B0" elif cn == "vp9": v_codec = "vp09.00.10.08" elif cn == "av1": v_codec = "av01.0.01M.08" elif s.get("codec_type") == "audio" and not a_codec: cn = s.get("codec_name", "") if cn == "aac": a_codec = "mp4a.40.2" elif cn in ("mp3", "mp2"): a_codec = "mp4a.6b" elif cn == "opus": a_codec = "opus" elif cn == "ac3": a_codec = "ac-3" elif cn == "flac": a_codec = "flac" if not v_codec: return None, duration codec = f"{v_codec},{a_codec}" if a_codec else v_codec return codec, duration def _pack(obj: dict) -> bytes: data = msgpack.packb(obj, use_bin_type=True) return struct.pack(">I", len(data)) + data class _DataChannelBuffer: """ Accumulate DataChannel messages and extract length-prefixed msgpack. Finding H6: the limit was a flat 64 MB applied even before the handshake, so an unauthenticated peer could announce a 64 MB frame and dribble bytes into it, holding that much memory per connection. Until a peer has proved GEK possession it gets a small budget; the large one is for file uploads. """ def __init__(self, max_message: int = MAX_MSG): self._buf = bytearray() self.max_message = max_message def feed(self, data: bytes): self._buf.extend(data) def messages(self): while len(self._buf) >= 4: length = struct.unpack(">I", self._buf[:4])[0] if length > self.max_message: raise ValueError(f"Message too large: {length}") if len(self._buf) < 4 + length: break msg_bytes = bytes(self._buf[4:4 + length]) del self._buf[:4 + length] yield msgpack.unpackb(msg_bytes, raw=False) def _get_remote_ip(pc: RTCPeerConnection) -> str: """Best-effort extraction of the remote peer IP from the ICE transport.""" try: dtls = pc.sctp and pc.sctp.transport ice = dtls and dtls.transport conn = ice and ice._connection if conn and hasattr(conn, '_nominated') and conn._nominated: for pair in conn._nominated.values(): return pair.remote_candidate.host if conn and conn.remote_candidates: return conn.remote_candidates[0].host except Exception: pass return "" class WebRTCPeerSession: """One WebRTC peer connection, handling MNP over a DataChannel.""" def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""): self._pc = pc self._ctx = node_ctx # Every background task this session starts. asyncio keeps only a *weak* # reference to a task, so one that is merely fired and forgotten can be # collected while it is still running — "Task was destroyed but it is # pending!" in the log. For _stream_video that meant its `async with # sem` never reached __aexit__ and the transcode slot was gone for good. # There are two slots: after two abandoned streams the node answered # "Server busy" to everything and no video would start at all. self._tasks: set[asyncio.Task] = set() self._channel: RTCDataChannel | None = None self._buffer = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG) self._pre_proof_fetches = 0 self._user_id: str | None = None self._group_id: str | None = None self._peer_id: str = peer_id self._remote_ip: str = "" self._username: str = "" # Set from the roster: the key this node pinned for this account. Never # from the JWT — the hub picks what goes in there. self._pinned_pk: str = "" # Flow control for video: how many segments the client says it can take. self._stream_credit = 0 self._stream_credit_evt = asyncio.Event() self._stream_stopped = False # When the peer last said anything about this stream. See # _await_stream_credit: silence is what ends a stream, not stinginess. self._stream_heard_at = 0.0 # Diagnostics: how many `stream_more n=0` the peer sent. See # _grant_stream_credit — it tells a paced client from an unpaced one. self._stream_keepalives = 0 # The stream this session currently owns. One viewer plays one film at # a time, so a second request means the first is over — see # _replace_stream for why waiting for it to time out is not an option. self._stream_task: asyncio.Task | None = None # Diagnostics only: when the current stream began and how far it got. self._stream_started_at: float = 0.0 self._stream_segments: int = 0 self._gek_challenge: bytes | None = None # Same value as the GEK challenge, but kept for the life of the connection: # a join_request is signed over it, and it must stay verifiable after the # handshake clears the challenge (an operator pairs while already connected). self._nonce_node: bytes = b"" self._join_attempts = 0 self._nonce_client: bytes = b"" self._admin_ops: dict[str, dict] = {} # op_id → pending admin operation self._uploads: dict[str, dict] = {} # filename → {next_index, bytes} def _setup_channel(self, channel: RTCDataChannel) -> None: self._channel = channel @channel.on("message") def on_message(message): if isinstance(message, str): message = message.encode() self._buffer.feed(message) for msg in self._buffer.messages(): self._handle_message(msg) def _handle_message(self, msg: dict) -> None: mtype = msg.get("type") log.debug("WebRTC recv: %s", mtype) try: if mtype == MNP.HANDSHAKE: self._do_handshake(msg) elif mtype == MNP.HANDSHAKE_RESPONSE: self._do_handshake_response(msg) elif mtype in (MNP.GEK_BUNDLE_FETCH, MNP.KEYPAIR_BUNDLE_FETCH) \ and self._gek_challenge is not None: # Served before the GEK proof by necessity: the client needs its # wrapped bundle in order to compute the proof. That window is a # disclosure surface (C4) — a hub that forges a JWT reaches it — so # it is bounded and audited here, and closed properly when clients # stop storing keypair bundles on other people's nodes. self._pre_proof_fetches += 1 if self._pre_proof_fetches > MAX_PRE_PROOF_FETCHES: self._audit_auth_failed( getattr(self, "_pending_group", ""), "pre-proof fetch flood") self._send({"type": "error", "detail": "Too many requests"}) return self._audit_pre_proof_fetch(mtype) if mtype == MNP.GEK_BUNDLE_FETCH: self._spawn(self._do_gek_bundle_fetch()) else: self._spawn(self._do_keypair_bundle_fetch()) elif mtype == MNP.JOIN_REQUEST and self._nonce_node: # Valid both before the GEK proof (a new member has no GEK to prove # with) and after it (an operator pairing a browser is already # connected). Authority comes from the pairing code and the # signature, never from the session state. self._spawn(self._do_join_request(msg)) elif self._user_id is None: self._send({"type": "error", "detail": "Handshake required"}) elif mtype == MNP.INDEX_SYNC: self._do_index_sync() elif mtype == MNP.FILE_REQUEST: # Spawned rather than answered inline: the reply waits for room # on the channel, and blocking the message loop for that would # stop everything else this peer is doing — including the # uploads whose acks free the very buffer we are waiting on. # Chunks are matched by file and index on the client, so # answering out of order is safe. self._spawn(self._do_file_request(msg)) elif mtype == MNP.STREAM_SEGMENT: self._do_stream_segment(msg) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message(msg) elif mtype == MNP.CHAT_HISTORY: self._do_chat_history(msg) elif mtype == MNP.PING: self._do_ping(msg) elif mtype == MNP.FILE_UPLOAD: self._do_file_upload(msg) elif mtype == MNP.DIR_CREATE: self._do_dir_create(msg) elif mtype == MNP.DIR_DELETE: self._do_dir_delete(msg) elif mtype == MNP.FILE_DELETE: self._do_file_delete(msg) elif mtype == MNP.ADMIN_RESPONSE: self._do_admin_response(msg) elif mtype == MNP.INVITE_CREATE: self._do_invite_create(msg) elif mtype == MNP.MEMBER_REVOKE: self._do_member_revoke(msg) elif mtype == MNP.MEMBER_UNPIN: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: self._do_gek_rotate(msg) elif mtype == MNP.KEYPAIR_BUNDLE_STORE: self._spawn(self._do_keypair_bundle_store(msg)) elif mtype == MNP.KEYPAIR_BUNDLE_DELETE: self._spawn(self._do_keypair_bundle_delete()) elif mtype == MNP.STREAM_REQUEST: sem = self._ctx.get("_transcode_sem") log.info("stream: req file=%s credits=%s slots_free=%s prev=%s", str(msg.get("file_id"))[:12], msg.get("credits"), getattr(sem, "_value", "?"), "alive" if (self._stream_task and not self._stream_task.done()) else "none") self._spawn(self._replace_stream(msg)) elif mtype == MNP.STREAM_MORE: self._grant_stream_credit(msg) elif mtype == "client_diag": # Diagnostics only. The node acts on none of it — it writes it # next to its own view of the same stream, which is the only # place the two halves can be compared when the client is a # phone with no console. # Every field is peer-controlled, so each is stringified and # cut short: this is a log line, not a channel for writing # whatever one likes into the operator's file. def _f(key: str, n: int = 24) -> str: return str(msg.get(key))[:n].replace("\n", " ") if msg.get("event"): # Once per stream or per seek, not once per five seconds — # and a seek nobody asked for looks exactly like a viewer # dragging the scrubber from this side, so it has to be # visible without turning DEBUG on. log.info( "stream: client %s target=%s t=%ss offset=%s ready=%s " "duration=%s ranges=[%s]", _f("event", 16), _f("target"), _f("t"), _f("offset"), _f("ready"), _f("duration"), _f("ranges", 120)) # Debug: one line every five seconds per viewer. Run the daemon # with --log-level debug to see inside a player that is # misbehaving — it is the only view of the browser there is # when the browser is a phone. else: log.debug( "stream: client t=%ss ahead=%ss ready=%s paused=%s " "stalled=%s q=%s inflight=%s appending=%s updating=%s " "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)", _f("t"), _f("ahead"), _f("ready"), _f("paused"), _f("stalled"), _f("q"), _f("inflight"), _f("appending"), _f("updating"), _f("quota"), _f("ms"), _f("err", 80), _f("ranges", 120), self._stream_segments) elif mtype == MNP.STREAM_STOP: age = (time.monotonic() - self._stream_started_at if self._stream_started_at else -1) log.info("stream: stop received %.1fs after start, %d segments sent", age, self._stream_segments) self._stop_stream() else: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: # Log the detail locally; send the peer a generic message. Exception # text here carries filesystem paths and internal state (finding L3). log.error("Error handling %s on DataChannel: %s", mtype, e, exc_info=True) self._send({"type": "error", "detail": "Request failed"}) def _audit(self, event: str, detail: str = "") -> None: audit = self._ctx.get("audit_store") if audit and self._user_id: if not self._remote_ip: self._remote_ip = _get_remote_ip(self._pc) self._spawn(audit.log_event( user_id=self._user_id, event=event, ip=self._remote_ip, username=self._username, group_id=self._group_id or "", detail=detail, )) def _channel_binding(self) -> bytes: """Both DTLS fingerprints, so a proof is valid on this connection only.""" offer_fp = b"" answer_fp = b"" if self._pc.remoteDescription: offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp) if self._pc.localDescription: answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp) if not offer_fp or not answer_fp: return b"" return webrtc_binding(offer_fp, answer_fp) def _do_handshake(self, msg: dict) -> None: group_id = msg.get("group_id", "") try: peer = authorize_token( msg.get("token", ""), self._ctx["hub_pk_pem"], group_id=group_id, hosted_groups=self._ctx.get("groups"), denylist=self._ctx.get("denylist"), ) except HandshakeError as refusal: # HandshakeError messages are authored to be peer-safe, unlike arbitrary # exception text (L3) — the client needs to know *why* it was refused. self._send({"type": "error", "detail": str(refusal), "code": getattr(refusal, "code", "")}) self._audit_auth_failed(group_id, str(refusal)) return try: self._nonce_client = base64.b64decode(msg.get("nonce", "")) except Exception: self._nonce_client = b"" if len(self._nonce_client) < NONCE_LEN: # The client nonce is what makes the NODE's proof fresh (C3). Without # it a recorded ack could be replayed by an impersonating peer. self._send({"type": "error", "detail": "Client nonce required"}) return # Decoded, but NOT authenticated: that happens on the GEK proof. self._pending_sub = peer.user_id self._pending_group = peer.group_id self._pending_username = peer.username gctx = self._ctx["groups"][peer.group_id] if "groups" in self._ctx else self._ctx if not gctx.get("gek"): self._send({ "type": "error", "detail": "Group encryption not initialized — contact node operator", }) return self._gek_challenge = os.urandom(NONCE_LEN) self._nonce_node = self._gek_challenge self._send({ "type": MNP.HANDSHAKE_CHALLENGE, "v": MNP_VERSION, "nonce": base64.b64encode(self._gek_challenge).decode(), # Announced here because a first-time joiner needs it *before* the # ack: join_request signs a transcript naming this node, and someone # who has never held the GEK cannot complete the handshake to learn # it. Unverified at this point — the ack proves it, the client checks # the two match, and a wrong value only makes our own verification # fail. It is never a substitute for the ack's proof and signature. "node_pk": self._node_pk_b64(), }) def _do_handshake_response(self, msg: dict) -> None: if not self._gek_challenge or not hasattr(self, "_pending_sub"): self._send({"type": "error", "detail": "No pending handshake challenge"}) return group_id = self._pending_group gctx = self._ctx["groups"][group_id] if "groups" in self._ctx else self._ctx gek = gctx.get("gek") if not gek: self._send({"type": "error", "detail": "Group encryption not initialized"}) self._gek_challenge = None return try: proof_bytes = base64.b64decode(msg.get("proof", "")) except Exception: self._send({"type": "error", "detail": "Invalid proof encoding"}) return binding = self._channel_binding() if not binding: # Refuse rather than fall back to an unbound proof (L4). self._send({"type": "error", "detail": "Channel binding unavailable"}) self._gek_challenge = None self._audit_auth_failed(group_id, "no channel binding") return if not verify_proof(gek, proof_bytes, ROLE_CLIENT, group_id, self._nonce_client, self._gek_challenge, binding): self._send({"type": "error", "detail": "GEK proof failed"}) self._gek_challenge = None self._audit_auth_failed(group_id, "GEK HMAC mismatch") return self._complete_handshake(gek, binding) self._gek_challenge = None def _complete_handshake(self, gek: bytes, binding: bytes) -> None: # Authenticated peers may send large frames (file uploads); unauthenticated # ones may not (H6). self._buffer.max_message = MAX_MSG self._user_id = self._pending_sub self._group_id = self._pending_group self._username = self._pending_username self._spawn(self._load_pinned_pk()) self._peer_registry()[self._user_id] = self node_user_id = self._ctx.get("node_user_id") log.info("WebRTC handshake OK — user=%s group=%s", self._user_id[:8], self._group_id[:8] if self._group_id else "none") # The node proves itself too (C3): possession of the GEK over the client's # nonce, plus a signature over the same transcript with its long-term key. # Previously the client received an unverifiable node_pk and trusted # is_node_admin from whoever answered — so a peer that had hijacked # signaling could serve a forged index, chat history and permissions. node_transcript = handshake_transcript( ROLE_NODE, self._group_id or "", self._nonce_client, self._gek_challenge or b"", binding) node_proof = make_proof( gek, ROLE_NODE, self._group_id or "", self._nonce_client, self._gek_challenge or b"", binding) ack = { "type": MNP.HANDSHAKE_ACK, "v": MNP_VERSION, "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()), "proof": base64.b64encode(node_proof).decode(), "sig": base64.b64encode( self._ctx["sk_node"].sign(node_transcript)).decode(), "is_node_admin": bool(node_user_id and self._user_id == node_user_id), } if node_user_id: ack["node_user_id"] = node_user_id pk_x_b64 = self._ctx.get("pk_x25519_b64") if pk_x_b64: ack["node_pk_x25519"] = pk_x_b64 self._send(ack) self._audit("handshake") async def _do_gek_bundle_fetch(self) -> None: """Serve the caller's wrapped GEK bundle during the handshake window.""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) return group_id = getattr(self, "_pending_group", "") user_id = getattr(self, "_pending_sub", "") if not group_id or not user_id: self._send({"type": "error", "detail": "No pending handshake"}) return bundle = await bundle_store.fetch(group_id, user_id) if bundle: self._send({ "type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": True, "pk_eph_b64": bundle["pk_eph_b64"], "nonce_b64": bundle["nonce_b64"], "wrapped_b64": bundle["wrapped_b64"], }) else: self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) def _do_invite_create(self, msg: dict) -> None: """ Issue a one-time pairing code for someone the operator wants to admit. Replaces the old invite path, where the inviter fetched the invitee's public key from the hub and wrapped the group key for whatever came back (H3). The node now needs nothing but a name: it will wrap the key itself, later, for a key the invitee proves they hold. """ roster = self._ctx.get("roster") if roster is None: self._send({"type": "error", "detail": "Roster not available"}) return invitee_id = msg.get("user_id", "") group_id = msg.get("group_id") or self._group_id if not invitee_id or not group_id: self._send({"type": "error", "detail": "Missing user_id or group_id"}) return if group_id != self._group_id: self._send({"type": "error", "detail": "Wrong group for this session"}) return if not self._has_admin_authority(): self._send({ "type": "error", "detail": "No operator paired — run `meshbay-node operator pair`", }) return self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, { "group_id": group_id, "user_id": invitee_id, "username": str(msg.get("username", ""))[:64], }) async def _do_keypair_bundle_fetch(self) -> None: """Serve the caller's encrypted keypair bundle during the handshake window.""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) return user_id = getattr(self, "_pending_sub", "") if not user_id: self._send({"type": "error", "detail": "No pending handshake"}) return bundle_enc = await bundle_store.fetch_keypair(user_id) if bundle_enc: self._send({ "type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": True, "bundle_enc": bundle_enc, }) else: self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) async def _do_keypair_bundle_store(self, msg: dict) -> None: """Store an encrypted keypair bundle (user backs up their own keys on node).""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": "error", "detail": "Bundle store not available"}) return bundle_enc = msg.get("bundle_enc", "") if not bundle_enc: self._send({"type": "error", "detail": "Missing bundle_enc"}) return await bundle_store.store_keypair(self._user_id, bundle_enc) log.info("Keypair bundle stored for user=%s", self._user_id[:8]) self._audit("keypair_bundle_store") self._send({ "type": "ack", "v": MNP_VERSION, "detail": "keypair_bundle_stored", }) # ── Pairing and join (H3, M3) ──────────────────────────────────────────── def _join_refuse(self, reason: str, audit_detail: str = "") -> None: self._join_attempts += 1 # Node-wide window, shared across connections: reconnecting must not reset # the budget. now = time.time() failures = [t for t in self._ctx.get("join_failures", []) if now - t < JOIN_FAILURE_WINDOW] failures.append(now) self._ctx["join_failures"] = failures self._audit_join("join_refused", audit_detail or reason) self._send({ "type": MNP.JOIN_RESULT, "v": MNP_VERSION, "ok": False, "reason": reason, }) def _audit_join(self, event: str, detail: str) -> None: audit = self._ctx.get("audit_store") if not audit: return self._remote_ip = self._remote_ip or _get_remote_ip(self._pc) self._spawn(audit.log_event( user_id=self._user_id or getattr(self, "_pending_sub", "unknown"), event=event, ip=self._remote_ip, username=self._username or getattr(self, "_pending_username", ""), group_id=self._group_id or getattr(self, "_pending_group", "") or "", detail=detail, )) async def _do_join_request(self, msg: dict) -> None: """ Pin an identity, or recognise one already pinned. The client signs its own Ed25519 and X25519 keys together with the node's nonce, so the identity key vouches for the encryption key — that is what will make it safe for the node to wrap the GEK for a key that arrived over the wire instead of one fetched from the hub's directory (H3). A first pairing needs a one-time code, which the hub never sees. Afterwards the pin is the credential and a changed key is refused outright, the same rule the client applies to `pk_node` (11.5.8). """ roster = self._ctx.get("roster") if roster is None: self._send({"type": "error", "detail": "Roster not available"}) return if self._join_attempts >= MAX_JOIN_ATTEMPTS: self._send({"type": "error", "detail": "Too many attempts"}) return now = time.time() recent = [t for t in self._ctx.get("join_failures", []) if now - t < JOIN_FAILURE_WINDOW] if len(recent) >= MAX_JOIN_FAILURES_WINDOW: self._audit_join("join_throttled", f"{len(recent)} failures in window") self._send({"type": "error", "detail": "Pairing temporarily locked"}) return user_id = self._user_id or getattr(self, "_pending_sub", "") username = self._username or getattr(self, "_pending_username", "") if not user_id: self._send({"type": "error", "detail": "Handshake required"}) return pk_ed_b64 = msg.get("pk_ed25519", "") pk_x_b64 = msg.get("pk_x25519", "") code = msg.get("code", "") ts = msg.get("ts", 0) try: pk_ed_raw = base64.b64decode(pk_ed_b64) pk_x_raw = base64.b64decode(pk_x_b64) if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32: raise ValueError pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw) except Exception: self._join_refuse("invalid_keys") return if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL: self._join_refuse("stale_request") return # An empty group_id means operator pairing, which is node-wide. Anything # else must be the group this connection authenticated to — a signature # obtained for one group must not name another. group_id = msg.get("group_id", "") or "" session_group = self._group_id or getattr(self, "_pending_group", "") or "" if group_id and group_id != session_group: self._join_refuse("group_mismatch") return transcript = join_transcript( node_pk_b64=self._node_pk_b64(), group_id=group_id, user_id=user_id, pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, nonce_node=self._nonce_node, ts=ts, ) try: sig = base64.b64decode(msg.get("sig", "")) except Exception: self._join_refuse("invalid_signature_encoding") return if not self._verify_sig(pk_ed, transcript, sig): self._join_refuse("signature_invalid") return known = await roster.get_identity(user_id) if known: if known["pk_ed25519"] != pk_ed_b64 or known["pk_x25519"] != pk_x_b64: # The blocking warning, raised where it matters: whoever this is # holds a different key than the person the operator paired. self._join_refuse( "key_changed", f"pinned={known['pk_ed25519'][:16]} presented={pk_ed_b64[:16]}") return # An operator's row is node-wide (empty group), so a lookup for the # group they happen to be opening finds nothing. Fall back to it, or # the client is told it has no role on a node it administers. member = (await roster.get_member(group_id, user_id) or await roster.get_member("", user_id)) await self._join_ok( user_id, pk_x_raw, session_group, role=member["role"] if member else "", recognised=True, ) return if not code: if self._group_join_policy(session_group) == "open": # An open-join group admits anyone the hub calls a member, so a # code would protect nothing — the hub can walk in through the # front door. Pin what turns up and say so in the audit log. await self._pin_and_admit( roster, user_id, username, pk_ed_b64, pk_x_b64, group_id=session_group, role=ROLE_MEMBER, approved_by="open-join", via="tofu") await self._join_ok(user_id, pk_x_raw, session_group, role=ROLE_MEMBER, recognised=False) return self._join_refuse("code_required") return invite = await roster.consume_invite(code, user_id) if not invite: self._join_refuse("code_invalid") return await self._pin_and_admit( # The name comes from the invitation, not from the token: the hub does # not put a username claim in a JWT, so pinning from the session alone # left the roster nameless and `member revoke ` unable to match. roster, user_id, invite["username"] or username, pk_ed_b64, pk_x_b64, group_id=invite["group_id"], role=invite["role"], approved_by=invite["created_by"], via="code") # The roster row comes from the invitation; the key comes from the # connection. An operator pairing is node-wide (empty group), but they # redeemed the code while opening a group and expect to read it — and # is_authorized() already grants an operator every group on this node. await self._join_ok(user_id, pk_x_raw, session_group or invite["group_id"], role=invite["role"], recognised=False) def _group_join_policy(self, group_id: str) -> str: """ Admission policy for a group, read from the node's own configuration. Never from the hub: a hub that could declare a group open would be handed the key to it (§3.4 of docs/invite-pairing-v1.md). """ gctx = (self._ctx.get("groups") or {}).get(group_id) or {} return gctx.get("join_policy", "invite") async def _pin_and_admit( self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str, *, group_id: str, role: str, approved_by: str, via: str, ) -> None: await roster.pin_identity( user_id=user_id, username=username, pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via, ) await roster.set_member( group_id=group_id, user_id=user_id, role=role, status="active", approved_by=approved_by, ) if role == ROLE_OPERATOR: self._ctx["has_admin_authority"] = True log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role) self._audit_join("join_pinned", f"role={role} via={via}") async def _join_ok( self, user_id: str, pk_x_raw: bytes, group_id: str, *, role: str, recognised: bool, ) -> None: """ Answer a join, wrapping the group key for the key the caller just proved. This is the H3 fix. The inviter used to fetch the invitee's public key from the hub and wrap the GEK for whatever came back, so a hub that answered with its own key was handed the group key by an honest member following the protocol exactly. The node now wraps for a key that arrived from its owner over an authenticated channel, bound to a pinned identity. """ reply = { "type": MNP.JOIN_RESULT, "v": MNP_VERSION, "ok": True, "recognised": recognised, "role": role, } roster = self._ctx["roster"] if group_id and not await roster.is_authorized(group_id, user_id): # Pinned on this node, but not admitted to this group. Hub membership # alone must not produce a key. reply["gek"] = False reply["reason"] = "not_authorized_for_group" self._send(reply) self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized") return gctx = (self._ctx.get("groups") or {}).get(group_id) or {} gek = gctx.get("gek") if not gek: reply["gek"] = False reply["reason"] = "no_gek" self._send(reply) return bundle = wrap_gek_aes(gek, pk_x_raw) reply["gek"] = True reply["pk_eph_b64"] = bundle["pk_eph_b64"] reply["nonce_b64"] = bundle["nonce_b64"] reply["wrapped_b64"] = bundle["wrapped_b64"] self._send(reply) self._audit_join("gek_wrapped", f"group={group_id[:8]}") def _do_dir_create(self, msg: dict) -> None: """ Create a directory, for any member of the group. Same confinement as an upload: every segment passes the name allowlist and the result must resolve under the shared root. Making a directory is not a privileged act — a member who can add a file can organise where it goes — but it writes to the operator's disk, so it is audited like one. """ ctx = self._group_ctx() roots: RootSet | None = ctx.get("roots") if not roots: self._send({"type": "error", "detail": "No shared directory"}) return name = str(msg.get("name", "")).strip() if not SAFE_UPLOAD_NAME.match(name): self._send({"type": "error", "detail": "Invalid directory name"}) return # The virtual root is not a directory on anyone's disk, so a member # cannot create one there — that would be adding a root, which is the # operator's configuration and not a file operation. parent_rel = (msg.get("dir") or "").strip("/") if not parent_rel: self._send({"type": "error", "detail": "Choose a folder to create this in"}) return parent = safe_subdir(roots, parent_rel) if parent is None or not parent.is_dir(): self._send({"type": "error", "detail": "Invalid directory"}) return target = safe_subdir(roots, f"{parent_rel}/{name}") if target is None: self._send({"type": "error", "detail": "Invalid directory"}) return if target.exists(): self._send({"type": "error", "detail": "Already exists"}) return target.mkdir(parents=False) virtual = roots.virtual_of(target) or f"{parent_rel}/{name}" log.info("Directory created by %s: %s", self._user_id[:8], virtual) self._audit("dir_create", virtual) self._send({ "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION, "dir": virtual, }) @staticmethod def _names_a_root(roots: RootSet, rel: str) -> bool: """True when `rel` is a bare root name rather than something inside one.""" found = roots.split(rel or "") return found is not None and not found[1] def _do_dir_delete(self, msg: dict) -> None: """ Remove an empty directory, for the node operator. Creating one is not privileged — a member who can add a file may organise where it goes — but removing one is: it acts on a name other members are using, and on the operator's disk. Empty is the whole safety property here. Nothing recursive: refusing a directory with anything in it means this can never destroy content, whatever the caller intended, so the operator deletes the files first and sees what they are losing. """ ctx = self._group_ctx() roots: RootSet | None = ctx.get("roots") if not roots: self._send({"type": "error", "detail": "No shared directory"}) return rel = (msg.get("dir") or "").strip("/") target = safe_subdir(roots, rel) # A root itself is not deletable here: removing one is a configuration # change, and doing it through a file operation would leave the group # config naming a directory nobody can reach. if target is None or self._names_a_root(roots, rel): self._send({"type": "error", "detail": "Invalid directory"}) return if not target.is_dir(): self._send({"type": "error", "detail": "Not a directory"}) return if any(target.iterdir()): self._send({"type": "error", "detail": "Directory is not empty"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for deletion"}) return self._issue_admin_challenge( OP_DIR_DELETE, roots.virtual_of(target) or rel) async def _admin_exec_dir_delete( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: rel = pending["subject"] ctx = self._group_ctx() roots: RootSet | None = ctx.get("roots") target = safe_subdir(roots, rel) if roots else None if (target is None or self._names_a_root(roots, rel) or not target.is_dir()): self._send({"type": "error", "detail": "Not a directory"}) return # Operator only. A file has an uploader who may remove their own; a # directory has none, so there is no second key to accept here. if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"dir_delete:{rel}") return # Checked again after the signature: the emptiness test that let this # through happened before a round trip to the operator's browser, and a # file could have landed in the meantime. if any(target.iterdir()): self._send({"type": "error", "detail": "Directory is not empty"}) return target.rmdir() log.info("Directory removed by %s: %s", self._user_id[:8], rel) self._audit("dir_delete", rel) self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel}) def _do_member_revoke(self, msg: dict) -> None: """ Stop serving the group key to someone, at the operator's request. The same authority as an invite, and the same reason: the roster decides who this node serves, so only a key the node pinned as an operator may change it. Membership on the hub is not consulted — the hub can remove someone from a group, and that stops them reaching the node at all, but it cannot make the node forget them. """ user_id = str(msg.get("user_id", "")).strip() if not user_id: self._send({"type": "error", "detail": "Missing user_id"}) return if user_id == self._user_id: # Removing yourself from your own node is not a member operation; # it would leave the group with nobody able to invite. self._send({"type": "error", "detail": "Cannot revoke yourself"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id) def _do_gek_rotate(self, msg: dict) -> None: """ Ask for a new group key. Operator only, and signed. This is what actually removes a revoked member's access: revocation stops the node serving the *next* key, and they still hold the current one. The node generates the replacement itself — nothing arriving here contributes key material, which is what the C5b rule is about. """ if not self._group_id: self._send({"type": "error", "detail": "No group on this connection"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_GEK_ROTATE, self._group_id) async def _admin_exec_gek_rotate( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"gek_rotate:{pending['subject'][:8]}") return try: result = await self._run_op( ops.set_gek, pending["subject"], rotate=True) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("gek_rotate", pending["subject"]) self._send({ "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION, "group_id": pending["subject"], "authorized_members": result.get("authorized_members", 0), # Said plainly, because rotating is the step people skip: content # already downloaded stays readable to whoever holds it. "note": "members re-receive the key on their next connect; content " "already downloaded is unaffected", }) def _do_member_unpin(self, msg: dict) -> None: """Forget a pinned identity, so someone can pair again with a new key.""" user_id = str(msg.get("user_id", "")).strip() if not user_id: self._send({"type": "error", "detail": "Missing user_id"}) return if user_id == self._user_id: # Unpinning yourself over the connection your pin authorizes would # end that connection's authority mid-operation. self._send({"type": "error", "detail": "Cannot unpin yourself"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_MEMBER_UNPIN, user_id) async def _admin_exec_member_unpin( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: user_id = pending["subject"] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"member_unpin:{user_id[:8]}") return try: await self._run_op(ops.unpin_member, user_id) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("member_unpin", user_id) self._send({"type": MNP.MEMBER_UNPIN_ACK, "v": MNP_VERSION, "user_id": user_id}) async def _run_op(self, fn, *args, **kwargs): """ Call an operation from `meshbay_node.ops` with the daemon's own view. The transport carries its own context and the loopback API carries the daemon state; they overlap but are not the same dict. Handing the MNP path a *second* set of lookups is exactly how two implementations of one operation start disagreeing — C1 and C6 one size down — so the daemon publishes its state here and both adapters call the same function. """ state = self._ctx.get("daemon_state") if state is None: raise ops.OpError("Node state not available", status=503) return await fn(state, *args, **kwargs) async def _admin_exec_member_revoke( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: user_id = pending["subject"] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"member_revoke:{user_id[:8]}") return roster = self._ctx.get("roster") if roster is None: self._send({"type": "error", "detail": "Roster not available"}) return group_id = self._group_id or "" if not await roster.set_status(group_id, user_id, "revoked"): self._send({"type": "error", "detail": "Not a member of this group"}) return # Anyone connected right now keeps the key they already unwrapped; what # they lose is the next one. Rotating it is the operator's call, and the # ack says so rather than implying this undid anything already read. peer = self._peer_registry().get(user_id) if peer is not None: try: await peer.close() except Exception: pass log.info("Member revoked by %s: user=%s group=%s", self._user_id[:8], user_id[:8], group_id[:8] or "-") self._audit("member_revoke", user_id) self._send({ "type": MNP.MEMBER_REVOKE_ACK, "v": MNP_VERSION, "user_id": user_id, "reminder": "they still hold the current group key — rotate it with " "meshbay-node gek-init", }) async def _do_keypair_bundle_delete(self) -> None: """ Withdraw our own key backup from this node. Only ever our own: the user_id comes from the authenticated session, never from the message. Someone who does not want a second browser should not be leaving a PBKDF2-protected blob on every node they have ever joined (C4), and turning the setting off has to remove what is already there — not just stop adding to it. """ bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": "error", "detail": "Bundle store not available"}) return removed = await bundle_store.delete_keypair(self._user_id) if removed: log.info("Keypair bundle withdrawn by user=%s", self._user_id[:8]) self._audit("keypair_bundle_delete") self._send({"type": "ack", "v": MNP_VERSION, "detail": "keypair_bundle_deleted", "removed": removed}) def _audit_pre_proof_fetch(self, mtype: str) -> None: """Record bundle access made before the GEK proof (C4).""" audit = self._ctx.get("audit_store") if not audit: return self._remote_ip = self._remote_ip or _get_remote_ip(self._pc) self._spawn(audit.log_event( user_id=getattr(self, "_pending_sub", "unknown"), event="pre_proof_fetch", ip=self._remote_ip, username=self._username or getattr(self, "_pending_username", ""), group_id=getattr(self, "_pending_group", "") or "", detail=mtype, )) def _audit_auth_failed(self, group_id: str, reason: str) -> None: audit = self._ctx.get("audit_store") if audit: self._remote_ip = _get_remote_ip(self._pc) self._spawn(audit.log_event( user_id="unknown", event="auth_failed", ip=self._remote_ip, group_id=group_id, detail=reason, )) def _spawn(self, coro) -> asyncio.Task: """Run a coroutine in the background and hold on to it. The reference is what keeps the task alive; the done callback is what stops the set growing. Anything that owns a resource for its lifetime — a transcode slot, an ffmpeg process — must go through here rather than `asyncio.ensure_future`. """ task = asyncio.ensure_future(coro) self._tasks.add(task) task.add_done_callback(self._tasks.discard) return task def _group_ctx(self) -> dict: if "groups" in self._ctx and self._group_id: return self._ctx["groups"][self._group_id] return self._ctx def _peer_registry(self) -> dict: """ Connected peers for THIS group only. Finding H1: this used to live on the shared transport context, so a chat message was broadcast to every peer on the node regardless of which group they had authenticated to. """ return self._group_ctx().setdefault("_peers", {}) def _user_names(self) -> dict: """Display-name cache, per group — same leak as _peer_registry (H1).""" return self._group_ctx().setdefault("_user_names", {}) def _do_index_sync(self) -> None: ctx = self._group_ctx() idx = ctx["index"] entries = [ { "id": e.id, "name": e.name, "path": e.path, "size": e.size, "type": e.type, "added_at": e.added_at, "uploader_id": e.uploader_id, } for e in idx.entries ] self._send({ "type": MNP.INDEX_SYNC, "v": MNP_VERSION, "group_id": idx.group_id, "version": idx.version, "entries": entries, # Directories are not index entries, so the client used to infer them # from file paths — which means a folder someone just created, or one # they emptied, simply did not exist as far as the UI was concerned. "dirs": self._list_dirs(ctx.get("roots")), # Which top-level folders are roots, and whether each is readable. # A frozen root's files stay listed, so without this a member cannot # tell "the drive is unplugged" from "it is all still there". "roots": ctx["roots"].describe() if ctx.get("roots") else [], }) @staticmethod def _list_dirs(roots: RootSet | None) -> list[str]: """ Every directory in the group, as members address them, sorted. Each root appears as a directory in its own right, so a root holding no files yet is still somewhere a member can navigate to and upload into. An unavailable root is listed too — its content is frozen, not gone, and hiding it would look exactly like deletion. """ if not roots: return [] out: list[str] = [] for root in roots: out.append(root.name) if not root.available: continue try: for path in sorted(root.path.rglob("*")): if path.is_dir() and not path.name.startswith("."): rel = path.relative_to(root.path) if not any(part.startswith(".") for part in rel.parts): out.append(f"{root.name}/{rel.as_posix()}") except OSError: continue return sorted(out)[:2000] async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg["file_id"] chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) if not entry: log.warning("File not found: %s", file_id[:16]) self._send({"type": "error", "detail": "File not found"}) return file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return log.debug("dl: req file=%s chunk=%s buffered=%s", file_id[:12], chunk_index, getattr(self._channel, "bufferedAmount", "?")) file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( self._ctx["sk_node"], ctx["gek"], file_path, chunk_index, file_hash, entry.id, ) # Backpressure. Without it the node hands the whole window to the # channel at once and the reader sees the first chunk, then nothing for # as long as the link takes to drain the rest. waited = 0.0 while (self._channel is not None and getattr(self._channel, "bufferedAmount", 0) > DOWNLOAD_BUFFER_HIGH and self._channel.readyState == "open" and waited < 60): await asyncio.sleep(0.05) waited += 0.05 if self._channel is None or self._channel.readyState != "open": return self._send(chunk_data) log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s", file_id[:12], chunk_index, len(chunk_data.get("ct") or b""), getattr(self._channel, "bufferedAmount", "?")) if chunk_index == 0: self._audit("file_download", entry.name) def _do_stream_segment(self, msg: dict) -> None: self._spawn(self._do_stream_segment_async(msg)) async def _do_stream_segment_async(self, msg: dict) -> None: """ Legacy HLS segment extraction (superseded by stream_req/MSE). Finding H6: this ran subprocess.run(..., timeout=30) directly inside the event loop, so a single request stalled the whole daemon — every peer, every group — for up to thirty seconds. Now async and under the same transcode semaphore as _stream_video. """ ctx = self._group_ctx() file_id = msg["file_id"] segment_index = msg["segment_index"] segment_duration = msg.get("segment_duration", 4) entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return sem = self._transcode_semaphore() try: async with sem: proc = await asyncio.create_subprocess_exec( "ffmpeg", "-hide_banner", "-loglevel", "error", "-ss", str(segment_index * segment_duration), "-i", str(file_path), "-t", str(segment_duration), "-c:v", "copy", "-c:a", "copy", "-f", "mpegts", "pipe:1", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, ) try: stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30) except asyncio.TimeoutError: proc.kill() await proc.wait() self._send({"type": "error", "detail": "Segment extraction timed out"}) return if proc.returncode != 0 or not stdout: self._send({"type": "error", "detail": "Segment extraction failed"}) return segment_data = stdout except Exception: self._send({"type": "error", "detail": "Segment extraction failed"}) return self._send({ "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION, "file_id": file_id, "segment_index": segment_index, "data_b64": base64.b64encode(segment_data).decode(), "size": len(segment_data), }) def _do_chat_message(self, msg: dict) -> None: # Per-group store — see _peer_registry() and finding H1. Reading chat_store # off the shared transport context sent every group's messages to the first # group's database, and served them back to anyone on the node. chat_store = self._group_ctx().get("chat_store") payload = msg.get("payload", "") sender_name = msg.get("sender_name", "") if sender_name: self._user_names()[self._user_id] = sender_name if chat_store: raw = payload.encode() if isinstance(payload, str) else payload self._spawn(chat_store.save_message( sender_id=self._user_id, iteration=msg.get("iteration", 0), payload=raw, thread_id=msg.get("thread_id"), sender_name=sender_name, )) peers = self._peer_registry() broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "sender_id": self._user_id, "sender_name": sender_name, "payload": payload, "thread_id": msg.get("thread_id"), "timestamp": __import__("time").time(), } for uid, session in list(peers.items()): if uid != self._user_id and session is not self: try: session._send(broadcast) except Exception: pass hub_ws = self._ctx.get("hub_ws") if hub_ws and self._group_id: try: import json as _json self._spawn(hub_ws.send(_json.dumps({ "type": "chat_notify", "group_id": self._group_id, "sender_name": sender_name, # Who actually wrote it, from the authenticated session. The # hub used to fall back to this node's own token subject — # the operator — so everyone was notified of their own # messages and the operator was notified of nobody's. "sender_user_id": self._user_id, }))) except Exception: pass self._send({"type": "ack", "v": MNP_VERSION}) self._audit("chat_message") def _do_ping(self, msg: dict) -> None: """Answer a liveness probe on an open channel, echoing the caller's token. Echoed rather than bare so a client can match the answer to the probe it sent and measure a round trip, instead of being reassured by a reply to some earlier one. """ self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")}) def _do_chat_history(self, msg: dict) -> None: chat_store = self._group_ctx().get("chat_store") if not chat_store: self._send({ "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "messages": [], "has_more": False, }) return # `before` pages backwards from the newest, which is the direction a chat # is actually read. `since` remains for callers that want everything # after a point in time; the browser no longer uses it. before = msg.get("before") limit = max(1, min(int(msg.get("limit", 100)), 200)) self._spawn(self._send_chat_history(chat_store, before, limit)) async def _send_chat_history(self, chat_store, before, limit: int) -> None: if before: msgs = await chat_store.get_before(int(before), limit=limit) else: msgs = await chat_store.get_recent(limit=limit) # Whether the "load older" control has anything left to fetch. Asked # about the oldest row returned, so an empty page correctly says no. has_more = await chat_store.has_before(msgs[0].id) if msgs else False names = self._user_names() self._send({ "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "has_more": has_more, "messages": [ { "id": m.id, "sender_id": m.sender_id, "sender_name": m.sender_name or names.get(m.sender_id, ""), "payload": m.payload.decode("utf-8", errors="replace") if isinstance(m.payload, bytes) else m.payload, "timestamp": m.timestamp, "thread_id": m.thread_id, } for m in msgs ], }) def _do_file_upload(self, msg: dict) -> None: ctx = self._group_ctx() filename = msg.get("filename", "") chunk_index = msg.get("chunk_index", 0) total_chunks = msg.get("total_chunks", 1) data = msg.get("data") if not filename or data is None: self._send({"type": "error", "detail": "Missing filename or data", "filename": filename}) return if not SAFE_UPLOAD_NAME.match(filename): self._send({"type": "error", "detail": "Invalid filename", "filename": filename}) return roots: RootSet | None = ctx.get("roots") upload_root = roots.upload_root if roots else None if upload_root is None: # Refused, never guessed. With several roots, picking one would send # a member's file to a disk the operator did not intend, and that is # discovered weeks later. self._send({"type": "error", "detail": "No upload folder is configured for this group", "filename": filename}) return if not upload_root.available: # The designated root's volume is absent. Falling back to another # root would scatter uploads across disks depending on what happened # to be plugged in. self._send({"type": "error", "detail": f"The upload folder ({upload_root.name}) is " f"currently unavailable", "filename": filename}) return # One destination, chosen by the operator and not by the client: # uploads/ inside the group's designated root. C5a is still honoured — # the name passed the allowlist above, and an existing file is never # replaced, which was the real defect (overwriting a file also made the # attacker its recorded uploader, and therefore able to delete it). rel_dir = f"{upload_root.name}/{UPLOAD_DIR_NAME}" target_dir = upload_root.path / UPLOAD_DIR_NAME try: target_dir.mkdir(parents=True, exist_ok=True) except OSError as e: log.warning("Cannot create upload folder in root %r: %s", upload_root.name, e) self._send({"type": "error", "detail": "Upload folder unavailable", "filename": filename}) return upload_key = f"{rel_dir}/{filename}" state = self._uploads.get(upload_key) # 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 _free_name(target_dir, filename) tmp_path = target_dir / f"{stored_name}.part" final_path = target_dir / stored_name if chunk_index == 0: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. if final_path.exists(): self._send({"type": "error", "detail": "File already exists", "filename": filename}) return state = {"next_index": 0, "bytes": 0, "stored_name": stored_name} self._uploads[upload_key] = state elif state is None: self._send({"type": "error", "detail": "Upload not started", "filename": filename}) 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"]: self._send({"type": "error", "detail": "Unexpected chunk index", "filename": filename}) return if isinstance(data, str): chunk_bytes = base64.b64decode(data) else: chunk_bytes = bytes(data) if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES: self._uploads.pop(upload_key, None) tmp_path.unlink(missing_ok=True) self._send({"type": "error", "detail": "Upload exceeds size limit", "filename": filename}) return with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f: f.write(chunk_bytes) state["next_index"] = chunk_index + 1 state["bytes"] += len(chunk_bytes) self._send({ "type": MNP.FILE_UPLOAD_ACK, "v": MNP_VERSION, "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: self._uploads.pop(upload_key, None) 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, rel_dir, stored_name) def _register_uploader(self, ctx: dict, rel_dir: str, filename: str) -> None: """ Tag the index entry with the uploader's identity after upload completes. The key recorded here 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. """ idx = ctx.get("index") if not idx: return for entry in idx.entries: if entry.name == filename and entry.path == rel_dir: entry.uploader_id = self._user_id entry.uploader_pk = self._pinned_pk return def _do_file_delete(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg.get("file_id", "") if not file_id: self._send({"type": "error", "detail": "Missing file_id"}) return entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return has_uploader_pk = bool(entry.uploader_pk) if not self._has_admin_authority() and not has_uploader_pk: self._send({"type": "error", "detail": "No authorized key for deletion"}) return self._issue_admin_challenge(OP_FILE_DELETE, file_id) # ── Admin operation challenge/response (finding H5) ────────────────────── def _node_pk_b64(self) -> str: return pk_to_b64(self._ctx["sk_node"].public_key()) def _issue_admin_challenge( self, op: str, subject: str, payload: dict | None = None, ) -> None: """ Ask the client to authorize `op` on `subject` with its Ed25519 identity key. The client is sent the transcript *fields*, not opaque bytes, so it can rebuild and inspect what it signs. The node keeps the authoritative copy and rebuilds the transcript itself at verification time — nothing signed is ever taken from the response message. """ nonce = os.urandom(32) ts = int(time.time()) op_id = base64.b64encode(os.urandom(16)).decode() self._admin_ops[op_id] = { "op": op, "subject": subject, "nonce": nonce, "ts": ts, "payload": payload or {}, } self._send({ "type": MNP.ADMIN_CHALLENGE, "v": MNP_VERSION, "op_id": op_id, "op": op, "subject": subject, "nonce": base64.b64encode(nonce).decode(), "ts": ts, "node_pk": self._node_pk_b64(), "group_id": self._group_id or "", }) @staticmethod def _verify_sig(pk: Ed25519PublicKey | None, transcript: bytes, sig: bytes) -> bool: if pk is None: return False try: pk.verify(sig, transcript) return True except Exception: return False async def _load_pinned_pk(self) -> None: """Remember which key this node pinned for the peer we just authenticated.""" roster = self._ctx.get("roster") if roster is None or not self._user_id: return ident = await roster.get_identity(self._user_id) if ident: self._pinned_pk = ident["pk_ed25519"] def _has_admin_authority(self) -> bool: """ Cheap synchronous pre-check: is there anyone who could authorize this? Only decides whether to issue a challenge at all — the gate is `_verify_admin_sig`. The flag is set at startup and refreshed in-process when an operator pairs. """ return bool(self._ctx.get("has_admin_authority")) async def _verify_admin_sig(self, transcript: bytes, sig: bytes) -> bool: """ Check a signature against every key holding node-operator authority. Read from the roster on each call rather than cached: revoking a paired browser must take effect immediately, and admin operations are rare enough that a SQLite read costs nothing. There is one source of operator authority and this is it. `admin_pk_ed25519` in node.toml used to be honoured alongside the roster; it is gone, and a config that still names it is warned about at startup rather than obeyed. """ roster = self._ctx.get("roster") if roster is None: return False for pk_b64 in await roster.operator_pks(): try: pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_b64)) except Exception: continue if self._verify_sig(pk, transcript, sig): return True return False def _do_admin_response(self, msg: dict) -> None: op_id = msg.get("op_id", "") sig_b64 = msg.get("signature", "") pending = self._admin_ops.pop(op_id, None) if not pending: self._send({"type": "error", "detail": "No pending admin operation"}) return if time.time() - pending["ts"] > ADMIN_CHALLENGE_TTL: self._send({"type": "error", "detail": "Admin challenge expired"}) return try: sig_bytes = base64.b64decode(sig_b64) except Exception: self._send({"type": "error", "detail": "Invalid signature encoding"}) return transcript = admin_transcript( op=pending["op"], node_pk_b64=self._node_pk_b64(), group_id=self._group_id or "", subject=pending["subject"], nonce=pending["nonce"], ts=pending["ts"], ) if pending["op"] == OP_FILE_DELETE: self._spawn( self._admin_exec_file_delete(pending, transcript, sig_bytes)) elif pending["op"] == OP_DIR_DELETE: self._spawn( self._admin_exec_dir_delete(pending, transcript, sig_bytes)) elif pending["op"] == OP_MEMBER_REVOKE: self._spawn( self._admin_exec_member_revoke(pending, transcript, sig_bytes)) elif pending["op"] == OP_INVITE_CREATE: self._spawn( self._admin_exec_invite_create(pending, transcript, sig_bytes)) elif pending["op"] == OP_GEK_ROTATE: self._spawn( self._admin_exec_gek_rotate(pending, transcript, sig_bytes)) elif pending["op"] == OP_MEMBER_UNPIN: self._spawn( self._admin_exec_member_unpin(pending, transcript, sig_bytes)) else: self._send({"type": "error", "detail": "Unknown admin operation"}) async def _admin_exec_file_delete( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: file_id = pending["subject"] ctx = self._group_ctx() entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return uploader_pk = None if entry.uploader_pk: try: uploader_pk = Ed25519PublicKey.from_public_bytes( base64.b64decode(entry.uploader_pk)) except Exception: uploader_pk = None # Node operator, or the user who uploaded this file — verified by the key # recorded at upload time, never by a JWT claim (the hub controls those). if not (await self._verify_admin_sig(transcript, sig) or self._verify_sig(uploader_pk, transcript, sig)): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") return self._exec_file_delete(ctx, file_id, entry) async def _admin_exec_invite_create( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: # Node operator only. A group admin who does not run the node has no # authority over who this node admits (deny by default). Delegation is # designed but deferred — see §6.2 of docs/invite-pairing-v1.md. if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}") return roster = self._ctx.get("roster") if roster is None: self._send({"type": "error", "detail": "Roster not available"}) return payload = pending["payload"] code = await roster.create_invite( group_id=payload["group_id"], user_id=payload["user_id"], role=ROLE_MEMBER, created_by=self._user_id or "", ttl=self._ctx.get("invite_ttl", DEFAULT_INVITE_TTL), username=payload.get("username", ""), ) invites = await roster.list_invites() expires = next( (i["expires_at"] for i in invites if i["user_id"] == payload["user_id"] and i["group_id"] == payload["group_id"]), "") log.info("Invite created: group=%s user=%s", payload["group_id"][:8], payload["user_id"][:8]) self._audit("invite_create", f"target={payload['user_id'][:8]}") # The code exists in the clear exactly here and in the operator's hands. self._send({ "type": MNP.INVITE_RESULT, "v": MNP_VERSION, "code": code, "expires_at": expires, "user_id": payload["user_id"], "username": payload.get("username", ""), }) def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: file_path = entry_abs_path(ctx["roots"], entry) if file_path.exists(): file_path.unlink() log.info("File deleted: %s", entry.name) self._audit("file_delete", entry.name) ctx["index"].remove_entry(file_id) self._send({ "type": MNP.FILE_DELETE_ACK, "v": MNP_VERSION, "file_id": file_id, }) def _grant_stream_credit(self, msg: dict) -> None: """ The client has room for more segments. `n` of zero is a keepalive, not a no-op: a viewer whose buffer is already a minute and a half ahead of the playhead deliberately grants nothing, and must still be able to say it is there. Without that, the stall timeout below cannot tell a paused film from a closed tab. """ log.debug("stream credit +%s (had %d, sent %d)", msg.get("n"), self._stream_credit, self._stream_segments) try: n = int(msg.get("n", 1)) except (TypeError, ValueError): n = 1 if n == 0: # The fingerprint of a client that bounds its read-ahead. A client # that never sends one is granting credit per append — which is # what fills the browser's buffer ceiling and wedges the player. self._stream_keepalives += 1 if self._stream_keepalives == 1: log.info("stream: peer is pacing itself (first keepalive at " "%d segments)", self._stream_segments) self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT)) self._stream_heard_at = time.monotonic() self._stream_credit_evt.set() def _stop_stream(self) -> None: """ The viewer was closed. Stop transcoding and let go of the slot. Without this the only thing that ended a stream was the credit timeout, so ffmpeg kept running and held one of the node's two transcode slots for two minutes after nobody was watching — which is how closing a video made the next one answer "server busy". """ self._stream_stopped = True self._stream_credit_evt.set() async def _await_stream_credit(self) -> bool: """ Block until the client has room. False if it stopped asking. Without this the node hands ffmpeg's entire output to the channel as fast as it is produced, and the browser holds a four gigabyte film in a JavaScript array while MediaSource consumes it a segment at a time. """ # Measured from the last thing the peer said, not from the start of the # wait: a viewer that is buffered well ahead sends keepalives and grants # nothing for minutes at a time, and that is a watched film, not a # stalled one. self._stream_heard_at = time.monotonic() waiting_since = 0.0 while self._stream_credit <= 0: if waiting_since == 0.0: waiting_since = time.monotonic() # Debug: a paced viewer runs out of credit between every # window, so this is one line per eight segments — hundreds # per film. It is worth having, but not by default. log.debug("stream: out of credit at %d segments (%.0f MB) — " "waiting for the peer", self._stream_segments, self._stream_segments * STREAM_SEGMENT_SIZE / 1048576) if self._stream_stopped: return False # Checked before the wait as well as after it: a peer that vanishes # sends no credit and fires no event, so waiting the full timeout # on a channel that is already shut is pure dead time on a slot. if self._channel is None or self._channel.readyState != "open": return False self._stream_credit_evt.clear() try: # In slices rather than one long sleep, so a connection that # dies mid-wait is noticed in seconds instead of minutes. The # total budget is unchanged. await asyncio.wait_for(self._stream_credit_evt.wait(), timeout=STREAM_CREDIT_POLL) except asyncio.TimeoutError: silent = time.monotonic() - self._stream_heard_at if silent >= STREAM_CREDIT_TIMEOUT: log.info("Stream stalled: nothing from peer=%s for %.0fs", (self._user_id or "?")[:8], silent) return False continue if self._stream_stopped: return False if self._channel is None or self._channel.readyState != "open": return False if waiting_since: waited_for = time.monotonic() - waiting_since # Only a wait long enough to be a symptom. Normal pacing puts a # gap of a few seconds between windows; a minute means the viewer # is buffered right up and playing, or has stopped watching. level = log.info if waited_for >= 10 else log.debug level("stream: credit arrived after %.1fs", waited_for) self._stream_credit -= 1 return True async def _replace_stream(self, msg: dict) -> None: """Retire this session's previous stream before starting another. A viewer plays one film at a time, so a second request means the first one is finished whatever the client managed to tell us. Relying on `stream_stop` alone was not enough: a browser that is backgrounded, reloaded or simply loses the message never sends it, and the only other thing that ends a stream is STREAM_CREDIT_TIMEOUT — two minutes during which ffmpeg keeps running and holds one of the node's two transcode slots. That is the reported failure exactly: first video fine, second fine, third answered "Server busy" because the first two were still holding both slots. The client shows that as "buffering" forever. Waiting for the old task is what makes the slot available: it is the exit of its `async with sem` that releases it. """ prev = self._stream_task if prev is not None and not prev.done(): t0 = time.monotonic() log.info("stream: retiring previous stream") self._stop_stream() try: await asyncio.wait_for(asyncio.shield(prev), timeout=15) log.info("stream: previous stream ended in %.1fs", time.monotonic() - t0) except asyncio.TimeoutError: log.warning("stream: previous stream STILL RUNNING after 15s") except Exception: pass # it failed on its own; the slot is free either way self._stream_task = asyncio.current_task() await self._stream_video(msg) def _transcode_semaphore(self) -> asyncio.Semaphore: """The node's stream budget, shared across every peer. One ffmpeg per request with no cap lets any member exhaust the node's CPU and process table (H6). The semaphore lives on the transport context rather than the session so that it counts the node's viewers and not one browser's, and it is created once: rebuilding it per call would hand every caller its own budget and cap nothing at all. """ sem = self._ctx.get("_transcode_sem") if sem is None: n = self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES sem = asyncio.Semaphore(n) self._ctx["_transcode_sem"] = sem log.info("stream: %d concurrent viewers allowed", n) return sem async def _stream_video(self, msg: dict) -> None: """Stream a video file as fMP4 segments via MSE-compatible output.""" sem = self._transcode_semaphore() if sem.locked() and sem._value <= 0: self._send({"type": "error", "detail": "Server busy, retry shortly"}) return log.info("stream: waiting for a slot (free=%s)", sem._value) async with sem: log.info("stream: slot acquired (free=%s)", sem._value) try: await self._stream_video_inner(msg) finally: log.info("stream: slot released (free=%s)", sem._value + 1) async def _stream_video_inner(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg.get("file_id", "") entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return gek = ctx.get("gek") file_hash = bytes.fromhex(entry.id) try: codec_str, duration = await _probe_video(str(file_path)) except Exception as e: self._send({"type": "error", "detail": f"Probe failed: {e}"}) return if not codec_str: self._send({"type": "error", "detail": "Unsupported video codec"}) return # Where to begin. Seeking is a stream restarted somewhere else: the # viewer moves the scrubber, this session's previous stream is retired # by _replace_stream, and ffmpeg is spawned again with -ss. try: start = float(msg.get("start", 0) or 0) except (TypeError, ValueError): start = 0.0 # Past the end would produce an empty stream and a player waiting for # segments that are never coming. if duration and start >= duration - 1: start = max(0.0, duration - 5) start = max(0.0, start) # -ss BEFORE -i, which seeks by the container index rather than by # decoding up to the point: milliseconds on a 500 MB film instead of # tens of seconds. It lands on the keyframe at or before `start`, so # the picture can begin a few seconds earlier than asked — which is # what every streaming player does, and why the client is told the # value used rather than left to assume its own. seek_args = ["-ss", f"{start:.3f}"] if start > 0 else [] proc = await asyncio.create_subprocess_exec( "ffmpeg", "-hide_banner", "-loglevel", "error", *seek_args, "-i", str(file_path), "-c", "copy", "-movflags", "frag_keyframe+empty_moov+default_base_moof", "-f", "mp4", "pipe:1", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) self._send({ "type": MNP.STREAM_INIT, "v": MNP_VERSION, "file_id": file_id, "codec": codec_str, "duration": duration, # ffmpeg restarts its timestamps at zero whatever we seek to, so # this is what the client adds back (`SourceBuffer.timestampOffset`) # to put the fragments where they belong on the timeline. "start": start, }) # A client that says nothing gets the old behaviour, which is why this # defaults to unlimited rather than to zero: a stream that waits for # credit from a peer that will never send any is a stream that hangs. try: self._stream_credit = int(msg.get("credits", 0) or 0) except (TypeError, ValueError): self._stream_credit = 0 paced = self._stream_credit > 0 self._stream_stopped = False index = 0 self._stream_started_at = time.monotonic() self._stream_segments = 0 reason = "eof" log.info("stream: stream_init sent file=%s paced=%s credits=%d start=%.1fs", file_id[:12], paced, self._stream_credit, start) try: while True: if paced and not await self._await_stream_credit(): reason = "no-credit-or-gone" break if self._stream_stopped: reason = "stopped-by-peer" log.info("Stream stopped by peer=%s after %d segments", (self._user_id or "?")[:8], index) break data = await proc.stdout.read(STREAM_SEGMENT_SIZE) if not data: break ckey = chunk_key_aes(gek, file_hash, index) nonce, ct = encrypt_chunk_aes(ckey, data) self._send({ "type": MNP.STREAM_DATA, "v": MNP_VERSION, "file_id": file_id, "segment_index": index, "nonce": nonce, "ct": ct, "plaintext_size": len(data), }) index += 1 self._stream_segments = index if index % 100 == 0: # A stream that stops shows up here as a last line, and the # numbers on it say which side stopped it. log.info("stream: %d segments (%.0f MB), credit=%d, " "keepalives=%d, %.0fs in", index, index * STREAM_SEGMENT_SIZE / 1048576, self._stream_credit, self._stream_keepalives, time.monotonic() - self._stream_started_at) await asyncio.sleep(0) except Exception as e: log.error("Stream error: %s", e) finally: try: proc.kill() except ProcessLookupError: pass # `await proc.wait()` on its own is the deadlock the asyncio docs # warn about: ffmpeg fills the stdout pipe we have stopped reading, # and the transport cannot finish closing until that buffer is # drained. Measured on 2026-08-16 with stream: — a viewer closed # the player after 99 segments (25 MB) and the task sat here past # the 15 s handover timeout, holding a transcode slot. The node has # two, so the next video waited and the one after was refused. # # Drain first, then wait with a bound. The slot must come back even # if the process is being stubborn: it has already had SIGKILL, and # the OS will reap it whether or not we are still watching. for pipe in (proc.stdout, proc.stderr): if pipe is None: continue try: await asyncio.wait_for(pipe.read(), timeout=2) except Exception: pass try: await asyncio.wait_for(proc.wait(), timeout=5) except Exception: log.warning("stream: ffmpeg did not reap in 5s — " "releasing the slot regardless") if not self._stream_stopped: self._send({ "type": MNP.STREAM_END, "v": MNP_VERSION, "file_id": file_id, }) log.info("stream: stream ended reason=%s segments=%d after %.1fs", reason, index, time.monotonic() - self._stream_started_at) log.info("Streamed %s: %d segments", entry.name, index) self._audit("stream_video", entry.name) def _send(self, obj: dict) -> None: if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) else: log.warning("WebRTC send skipped: channel=%s", self._channel.readyState if self._channel else "none") async def shutdown_tasks(self) -> None: """Stop everything this session is doing and give back what it holds. Separate from close() because the connection-state handler runs while aiortc is already tearing the peer connection down — calling pc.close() from in there would re-enter it. What matters for the transcode slot is here: cancelling the task runs the exit of its `async with sem`. """ self._stop_stream() for task in list(self._tasks): task.cancel() if self._tasks: await asyncio.gather(*self._tasks, return_exceptions=True) async def close(self) -> None: self._audit("disconnect") if self._user_id: self._peer_registry().pop(self._user_id, None) await self.shutdown_tasks() await self._pc.close() def _read_and_encrypt( sk_node: Ed25519PrivateKey, gek: bytes, file_path: Path, chunk_index: int, file_hash: bytes, file_id: str = "", ) -> dict: with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) ckey = chunk_key_aes(gek, file_hash, chunk_index) nonce, ct = encrypt_chunk_aes(ckey, plaintext) return { "type": MNP.FILE_CHUNK, "v": MNP_VERSION, # Named so a client running several downloads at once can tell whose # reply this is. It used to carry only the index, which made matching a # reply to its request a question of arrival order. "file_id": file_id, "chunk_index": chunk_index, "plaintext_size": len(plaintext), "nonce": nonce, "ct": ct, } class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. Usage: transport = WebRTCTransport(sk_node, hub_pk_pem, gek, roots, index) answer_sdp = await transport.handle_offer(offer_sdp, peer_id) # Return answer_sdp to the browser via hub signaling """ def __init__( self, sk_node: Ed25519PrivateKey, hub_pk_pem: bytes, gek: bytes, roots: RootSet, index: GroupIndex, groups: dict[str, dict] | None = None, denylist: Any | None = None, stun_servers: list[str] | None = None, max_concurrent_streams: int | None = None, ): self._ctx: dict[str, Any] = { "sk_node": sk_node, "hub_pk_pem": hub_pk_pem, "gek": gek, "roots": roots, "index": index, "_peers": {}, # None means "the operator said nothing" — the default applies. It # is read once, when the first stream builds the semaphore. "max_concurrent_streams": max_concurrent_streams, } if groups: self._ctx["groups"] = groups if denylist: self._ctx["denylist"] = denylist self._stun = stun_servers or ["stun:stun.l.google.com:19302"] self._sessions: dict[str, WebRTCPeerSession] = {} async def handle_offer( self, offer_sdp: str, peer_id: str, ) -> tuple[str, list[dict]]: """ Process a WebRTC SDP offer from a browser client. Returns (answer_sdp, ice_candidates) to relay back via hub signaling. ICE candidates are embedded in the SDP (aiortc gathers before returning). """ from aiortc import RTCIceServer, RTCConfiguration config = RTCConfiguration( iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else [] ) pc = RTCPeerConnection(configuration=config) session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id) self._sessions[peer_id] = session @pc.on("datachannel") def on_datachannel(channel: RTCDataChannel): log.info("WebRTC DataChannel opened: %s (peer=%s)", channel.label, peer_id) session._setup_channel(channel) @pc.on("connectionstatechange") async def on_state_change(): state = pc.connectionState log.info("WebRTC connection state: %s (peer=%s)", state, peer_id) if state in ("failed", "closed"): gone = self._sessions.pop(peer_id, None) if gone is not None: # Popping only forgets the session. Its stream went on # transcoding until the credit timeout — measured at 91s # after the connection closed — holding one of the node's # two slots the whole time. Closing the viewer, the tab or # the browser all arrive here, so this is the one place # that covers every way of walking away. await gone.shutdown_tasks() offer = RTCSessionDescription(sdp=offer_sdp, type="offer") await pc.setRemoteDescription(offer) answer = await pc.createAnswer() await pc.setLocalDescription(answer) log.info("WebRTC answer ready for peer=%s", peer_id) return pc.localDescription.sdp, [] async def close_peer(self, peer_id: str) -> None: session = self._sessions.pop(peer_id, None) if session: await session.close() async def close_all(self) -> None: for session in list(self._sessions.values()): await session.close() self._sessions.clear() @property def active_peers(self) -> int: return len(self._sessions)