""" 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 logging import os import time import uuid from typing import Any from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) from meshbay_common import MNP_VERSION from meshbay_common.adminop import ( ADMIN_CHALLENGE_TTL, OP_APP_DIRECTORIES, OP_APPS_ENABLED, OP_CHAT_DIRECTORY, OP_CHAT_EPOCH, OP_CHAT_LINK_PREVIEW, OP_DIR_DELETE, OP_FILE_DELETE, OP_GEK_ROTATE, OP_GROUP_ATTACH, OP_GROUP_DETACH, OP_INVITE_CANCEL, OP_INVITE_CREATE, OP_INVITE_LINK_CREATE, OP_MEMBER_REVOKE, OP_MEMBER_UNPIN, OP_MUSICBRAINZ_ENABLED, OP_ROOT_ADD, OP_ROOT_EJECT, OP_ROOT_PLUG, OP_ROOT_REMOVE, OP_ROOT_UPDATE, OP_SEARCH_LISTED, OP_SET_SCAN_SETTINGS, OP_TMDB_CONFIG, OP_TMDB_ENABLED, OP_TMDB_OVERRIDE, OP_TMDB_REMATCH, OP_TRANSFER_LIMITS, admin_transcript, ) from meshbay_common.crypto import pk_to_b64 from meshbay_common.groupbox import ( PURPOSE_ACK, PURPOSE_ROSTER, seal, ) from meshbay_common.handshake import ( MNP_MIN_SUPPORTED, NONCE_LEN, ROLE_CLIENT, ROLE_NODE, HandshakeError, authorize_token, challenge_transcript, check_version, handshake_transcript, make_proof, verify_proof, webrtc_binding, ) from meshbay_common.protocol import ( MNP, ) from meshbay_node import ops from meshbay_node import transfers as transfers_mod from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer # Re-imported under its original name: every call site and existing test in # this module still refers to it as `_probe_video`. The implementation lives # 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 ( RootSet, ) from meshbay_node.transport.webrtc.admission import AdmissionMixin from meshbay_node.transport.webrtc.apps.music import MusicMixin from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin from meshbay_node.transport.webrtc.apps.video_meta import VideoMetaMixin from meshbay_node.transport.webrtc.blobs import BlobsMixin from meshbay_node.transport.webrtc.channel import ( _REPLY_TO, _DataChannelBuffer, _extract_dtls_fingerprint, _get_remote_ip, _pack, ) from meshbay_node.transport.webrtc.chat import ChatMixin from meshbay_node.transport.webrtc.files import FilesMixin 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__) # 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 # How many peer connections this node holds at once, and how long one may stay # without completing the MNP handshake. The budget above bounds what *one* # unauthenticated peer costs; these bound how many there may be and how long # each lasts, which is the other half and was missing. The hub caps three # offers in flight per account — a limit on each caller, not on this machine — # so the cost to an operator grew with the number of people in their groups. # Sized to be unreachable in ordinary use: a browser holds one connection per # open group, and a handshake unfinished after a minute is not going to finish. MAX_PEER_SESSIONS = 64 UNAUTHENTICATED_SESSION_TIMEOUT = 60 # seconds # 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 # 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. # Opt-in, off by default: a per-session heartbeat log (message count, time # since the last message, ICE state) and ICE-state-change logging, on top of # the connectionstatechange logging that already runs unconditionally. Added # while chasing a report of the browser side going unresponsive after a # mobile screen lock; --log-level DEBUG was not the right knob for this, # since it is already used for the per-message request/response tracing # every group index lookup produces, and turning that on for days of normal # operation just to catch one intermittent session is not viable. Set # MESHBAY_WEBRTC_TRACE=1 in the node's environment for the duration of a # debugging session. _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1" _WEBRTC_TRACE_INTERVAL_S = 30.0 class WebRTCPeerSession( AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, TransferMixin, UploadMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, ): """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 = "" # This connection's key in the group's peer registry. **Per connection, # never per account**: one person may hold several devices here, and # keying the registry by user_id makes the second evict the first, and # the symptom is invisible: two devices of one account cannot both be # connected, and whichever disconnects takes the other's chat delivery # with it. self._registry_key: str = uuid.uuid4().hex # Set from the roster: the key this node pinned for this account. Never # from the JWT — the hub picks what goes in there. # # This is the account's *oldest* live device unless `device_hello` has # told us better — see _do_device_hello. Treat it as "a device of this # account", not "the device on this connection", anywhere that has not # checked `_device_confirmed`. self._pinned_pk: str = "" # True once this connection proved which device it is. Until then the # node knows the account and not the key, which is all it ever knew # before device linking existed. self._device_confirmed: bool = False # 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 # Uploads in progress live in the group context, not here: see # `_partial_uploads` and `uploads.py`. # # Leaseless reads, though, *are* this connection's: the bound is on what # one session may do while claiming to be browsing, not a pool shared # between them. Three tabs open is browsing in three tabs. self._leaseless = transfers_mod.LeaselessReads() # Whether this session has already been noted as transferring under a # lease the node does not have (see `_note_unleased`). One line per # connection, not per chunk. self._unleased_noted = False # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message # arrived, so the heartbeat can report silence duration. self._last_msg_at: float = 0.0 def _setup_channel(self, channel: RTCDataChannel) -> None: self._channel = channel self._msg_count = 0 @channel.on("message") def on_message(message): if isinstance(message, str): message = message.encode() self._msg_count += 1 self._last_msg_at = time.monotonic() if self._msg_count <= 3: log.info("WebRTC data received: %d bytes, msg #%d (peer=%s)", len(message), self._msg_count, self._peer_id) self._buffer.feed(message) for msg in self._buffer.messages(): self._handle_message(msg) if _WEBRTC_TRACE: self._spawn(self._trace_heartbeat()) async def _trace_heartbeat(self) -> None: """Diagnostics only (_WEBRTC_TRACE): periodic proof-of-life for this session, so a gap in these lines pinpoints when the node stopped hearing from a peer that (from its own side) may still look connected.""" while True: await asyncio.sleep(_WEBRTC_TRACE_INTERVAL_S) silence = time.monotonic() - self._last_msg_at if self._last_msg_at else -1 log.info( "WebRTC heartbeat peer=%s msgs=%d silence=%.0fs pc=%s ice=%s", self._peer_id, self._msg_count, silence, self._pc.connectionState, self._pc.iceConnectionState, ) def _handle_message(self, msg: dict) -> None: """Answer one MNP message, under the correlation id it carries. The id is published for the whole handler — see _REPLY_TO — so that every reply _send puts on the wire, including the ones a spawned task sends much later and the generic refusal below, names the request it answers. Resetting on the way out only clears it for *this* call: a task spawned in between captured its own copy of the context when it was created and keeps answering under the right id. """ token = _REPLY_TO.set((self, msg.get("req_id"))) try: self._dispatch_message(msg) finally: _REPLY_TO.reset(token) def _dispatch_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.CHAT_MESSAGE: self._do_chat_message(msg) elif mtype == MNP.CHAT_HISTORY: self._do_chat_history(msg) elif mtype == MNP.LINK_PREVIEW_REQ: self._spawn(self._do_link_preview_request(msg)) elif mtype == MNP.PING: self._do_ping(msg) elif mtype == MNP.TRANSFER_OPEN: self._do_transfer_open(msg) elif mtype == MNP.TRANSFER_CLOSE: self._do_transfer_close(msg) elif mtype == MNP.FILE_UPLOAD: self._spawn(self._do_file_upload(msg)) elif mtype == MNP.DIR_CREATE: self._spawn(self._do_dir_create(msg)) elif mtype == MNP.DIR_DELETE: self._spawn(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.INVITE_LINK_CREATE: self._do_invite_link_create(msg) elif mtype == MNP.INVITE_CANCEL: self._do_invite_cancel(msg) elif mtype == MNP.MEMBER_REVOKE: self._do_member_revoke(msg) elif mtype == MNP.DEVICE_REQUEST: self._spawn(self._do_device_request(msg)) elif mtype == MNP.DEVICE_LOOKUP: self._spawn(self._do_device_lookup(msg)) elif mtype == MNP.DEVICE_ADD: self._spawn(self._do_device_add(msg)) elif mtype == MNP.DEVICE_LIST: self._spawn(self._do_device_list(msg)) elif mtype == MNP.DEVICE_REVOKE: self._spawn(self._do_device_revoke(msg)) elif mtype == MNP.DEVICE_HELLO: self._spawn(self._do_device_hello(msg)) elif mtype == MNP.APPS_ENABLED: self._do_apps_enabled(msg) elif mtype == MNP.TRANSFER_LIMITS: self._do_transfer_limits(msg) elif mtype == MNP.SET_SCAN_SETTINGS: self._do_set_scan_settings(msg) elif mtype == MNP.TMDB_CONFIG: self._do_tmdb_config(msg) elif mtype == MNP.TMDB_ENABLED: self._do_tmdb_enabled(msg) elif mtype == MNP.APP_DIRECTORIES: self._do_app_directories(msg) elif mtype == MNP.CHAT_DIRECTORY: self._do_chat_directory(msg) elif mtype == MNP.CHAT_LINK_PREVIEW: self._do_chat_link_preview(msg) elif mtype == MNP.SEARCH_LISTED: self._do_search_listed(msg) elif mtype == MNP.CHAT_EPOCH: self._do_chat_epoch(msg) elif mtype == MNP.CHAT_KEYS_REQ: self._spawn(self._do_chat_keys_req(msg)) elif mtype == MNP.GROUP_ROSTER_REQ: self._spawn(self._do_group_roster_req(msg)) elif mtype == MNP.MEDIA_META_REQ: self._spawn(self._do_media_meta_request(msg)) elif mtype == MNP.SEASON_META_REQ: self._spawn(self._do_season_meta_request(msg)) elif mtype == MNP.TMDB_SEARCH_REQ: self._spawn(self._do_tmdb_search_request(msg)) elif mtype == MNP.TMDB_OVERRIDE: self._do_tmdb_override(msg) elif mtype == MNP.TMDB_REMATCH: self._do_tmdb_rematch(msg) elif mtype == MNP.MUSICBRAINZ_ENABLED: self._do_musicbrainz_enabled(msg) elif mtype == MNP.MUSIC_META_REQ: self._spawn(self._do_music_meta_request(msg)) elif mtype == MNP.AUDIO_TRANSCODE_REQ: self._spawn(self._do_audio_transcode_request(msg)) elif mtype == MNP.SUBTITLE_REQ: self._spawn(self._do_subtitle_request(msg)) elif mtype == MNP.MEMBER_UNPIN: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: self._do_gek_rotate(msg) elif mtype == MNP.NODE_STATUS: self._spawn(self._do_node_status(msg)) elif mtype == MNP.ROOT_ADD: self._do_root_add(msg) elif mtype == MNP.ROOT_REMOVE: self._do_root_remove(msg) elif mtype == MNP.ROOT_UPDATE: self._do_root_update(msg) elif mtype == MNP.ROOT_EJECT: self._do_root_eject(msg) elif mtype == MNP.ROOT_PLUG: self._do_root_plug(msg) elif mtype == MNP.ROSTER_READ: self._spawn(self._do_roster_read(msg)) elif mtype == MNP.DENYLIST_READ: self._spawn(self._do_denylist_read(msg)) elif mtype == MNP.DENYLIST_CLEAR: self._spawn(self._do_denylist_clear(msg)) elif mtype == MNP.GROUP_ATTACH: self._do_group_attach(msg) elif mtype == MNP.GROUP_DETACH: self._do_group_detach(msg) elif mtype == MNP.NODE_SETTINGS_SET: self._spawn(self._do_node_settings_set(msg)) elif mtype == MNP.NODE_RELOAD: self._spawn(self._do_node_reload(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.USER_BLOB_STORE: self._spawn(self._do_user_blob_store(msg)) elif mtype == MNP.USER_BLOB_FETCH: self._spawn(self._do_user_blob_fetch(msg)) elif mtype == MNP.USER_BLOB_LIST: self._spawn(self._do_user_blob_list()) elif mtype == MNP.USER_BLOB_DELETE: self._spawn(self._do_user_blob_delete(msg)) 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: # `ahead` on its own cannot say whether a short buffer is # the player's own gate holding or the network failing to # keep up, and those two want opposite answers. `limit` is # what the gate is set to for this film and `budget` the # byte budget it was derived from, so the three read as one # sentence. log.debug( "stream: client t=%ss ahead=%ss/%ss budget=%sMB " "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("limit"), _f("budgetMB"), _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", "") log.info("WebRTC handshake request: group=%s (peer=%s)", group_id[:8] if group_id else "none", self._peer_id) # Before the token, and before anything is decided from it: a peer we # cannot speak to is refused with a code it can act on, rather than # served messages it will misread as missing fields (L2). try: check_version(msg.get("v", ""), msg.get("v_min", "")) except HandshakeError as refusal: self._send({"type": "error", "detail": str(refusal), "code": refusal.code}) return 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"): log.warning("Handshake refused — no GEK for group=%s", peer.group_id[:8]) 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 log.info("WebRTC handshake challenge sent (peer=%s)", self._peer_id) self._send({ "type": MNP.HANDSHAKE_CHALLENGE, "v": MNP_VERSION, # Our half of the range. The client refuses us on this rather than # discovering the mismatch when a field it expected is not there. "v_min": MNP_MIN_SUPPORTED, "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(), # ...except that since 3.4 it is signed, so a client that already # knows which key to expect can check it before it sends a code. **self._challenge_sig(peer.group_id, self._channel_binding()), }) def _challenge_sig(self, group_id: str, binding: bytes) -> dict: """ `{"sig": ...}` over the challenge transcript, or nothing (MNP 3.4). What makes `node_pk` above more than an announcement: a client about to send an invitation code can check this node holds the key it was told to expect, before the code leaves. No binding means no signature rather than an unbound one — a signature that is not tied to the channel is one somebody can relay, and the handshake proof refuses that case anyway. """ if not binding: return {} transcript = challenge_transcript( group_id, self._nonce_client, self._gek_challenge, binding) return {"sig": base64.b64encode(self._ctx["sk_node"].sign(transcript)).decode()} 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._register_peer() 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) # Everything the client needs in order to *authenticate* us stays in clear — # node_pk, proof and sig are what it checks before it would trust a # decryption, so they cannot themselves be behind one. The configuration # below is sealed under a GEK-derived subkey, which gives it an # authentication tag from a key the hub does not hold. Until MNP 1.0 the # signed transcript named no ack field at all, so is_node_admin, # enabled_apps, the app directories and the rest were authenticated by DTLS # channel and nothing else. config = { "is_node_admin": self._is_node_admin(), # Which group "applications" to show. Absent/empty falls back to # every registered one client-side, so a node that predates this # setting (or one whose context has not loaded it yet) hides # nothing. "enabled_apps": list(self._group_ctx().get("enabled_apps") or []), # Read once and kept current in place by the signed op, and # surfaced here rather than only via tmdb_enabled_ack, so a client # that connects after the operator configured it does not have to # wait for a live change to find out. "tmdb_enabled": bool(self._group_ctx().get("tmdb_enabled", True)), # Token/language stay node-wide (one shared credential/cache) — # via daemon_state, kept current by tmdb_config_ack. "tmdb_token_customized": bool( self._ctx.get("daemon_state", {}).get("tmdb_token_customized", False)), "tmdb_language": str( self._ctx.get("daemon_state", {}).get("tmdb_language") or ""), # Music app (docs/MESHBAY_DESIGN.md §9.8) — same shape as the TMDB # fields above. No language field: MusicBrainz search doesn't # take one the way TMDB does. "musicbrainz_enabled": bool(self._group_ctx().get("musicbrainz_enabled", True)), # Every app's configured folders — `_directories`, keyed by # the app's own registry name, always a list. Built by daemon.py's # `_app_directories_ctx`, and the only form on the wire: the # `video_root` / `audio_root` / `photo_roots` scalars that used to # ride here are gone. One folder was never the general case, and two # spellings of one answer meant whichever the reader consulted first # decided it. # # `chat_directory` below is the one surviving second name, and it is # safe for the reason those were not: it is *derived* from this list # on every build rather than stored beside it, so the two cannot # drift apart. **self._app_directories_ack(), # Where chat attachments are written — the singular form, because # Chat genuinely has one destination. "" means the operator has not # chosen, and the paperclip says so. "chat_directory": self._group_ctx().get("chat_directory") or "", # Whether the node unfurls links members post here. Absent means # on, which is what it did before this existed. "chat_link_preview": bool( self._group_ctx().get("chat_link_preview", True)), # Whether the reader's cross-group Search should list this group. # Presentation only: the index below is served to Search and to the # group page alike, and this cannot tell them apart. Sealed like the # rest, so the hub cannot flip it. Absent means listed. "search_listed": bool(self._group_ctx().get("search_listed", True)), # Which chat epoch key a client should be sealing under. Inside # the sealed part of the ack like every other configuration field, # so it carries an authentication tag from a key the hub does not # hold — a forged epoch would have a client sealing under a key the # group has retired. # # No `chat_encrypted` beside it: there is no switch. A peer that # reached this point speaks MNP 2.0, and 2.0 has no plaintext chat. "chat_epoch": int(self._group_ctx().get("chat_epoch", 0) or 0), # This member's own transfer caps in this group, so the interface # can say "2 of 2 of your slots are busy" rather than draw a bare # spinner. Absent reads as "no limit known" and the hint is simply # not drawn — never as "unlimited", which would have the interface # contradicting the node. "transfer_limits": { "download": self._slots().member_cap( transfers_mod.DOWNLOAD, (self._group_id or "", self._user_id or "")), "upload": self._slots().member_cap( transfers_mod.UPLOAD, (self._group_id or "", self._user_id or "")), }, # So a client that connects mid-scan shows the indexing state # immediately, instead of waiting for the next periodic # INDEX_PROGRESS push. Never a path or filename — see # IndexProgress in indexer.py. "indexing": self._indexing_status(), # Current values only — not enforced from here, just shown to # the operator in Settings so the number on screen matches what # the indexer is actually doing (set_scan_settings, ops.py). "scan_settings": { "reconcile_interval_secs": self._group_ctx().get( "reconcile_interval_secs", DirectoryIndexer.DEFAULT_RECONCILE_SECS), "debounce_secs": self._group_ctx().get( "debounce_secs", DirectoryIndexer.DEFAULT_DEBOUNCE_SECS), }, } if node_user_id: config["node_user_id"] = node_user_id pk_x_b64 = self._ctx.get("pk_x25519_b64") if pk_x_b64: config["node_pk_x25519"] = pk_x_b64 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(), **seal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, self._group_id or "", config), } self._send(ack) self._audit("handshake") # Someone is here now — reconcile's backstop should be prompt again # rather than however far its backoff had stretched while nobody # was connected (indexer.py DirectoryIndexer.note_activity). note_activity = self._group_ctx().get("note_activity") if note_activity: note_activity() # ── Per-account blobs (playlists, docs/playlists.md §8.2) ──────────────── # # The node stores bytes it cannot read and hands them back. `self._user_id` # comes from the authenticated session and never from the message: a # `user_id` in the body would let any member of this group read or # overwrite any other member's blob, which is finding C5 one size down. # ── Pairing and join (H3, M3) ──────────────────────────────────────────── # ── Device linking ─────────────────────────────────────────────────────── # # A person may hold several devices on one node. The authority admitting a # new one is a key the node already pinned — never the hub, which has stored # no user keys since 2026-08-14 and therefore cannot countersign anything. # See docs/MESHBAY_DESIGN.md §3.3. 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. """ group_id = str(msg.get("group_id", "")).strip() or self._group_id if not 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, group_id, group_id=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 # The operator is rotating because somebody left, and the chat archive # key is not derived from the group key — so rotating that one does not # move this one. Doing both here is what makes "rotate after a removal" # mean the same thing for chat as it does for files. await self._new_chat_epoch(pending["subject"], "gek_rotate") 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) await self._new_chat_epoch(self._group_id or "", "member_unpin") 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}) # Every "application" a group can show. Photos joins this set (and # apps.js's registry, client-side) when it lands; nothing else about # this handler changes. DEFAULT_APPS (roster.py) deliberately does not # include "video" or "music" — both can make outbound third-party # network calls (TMDB, MusicBrainz) once enabled, so an operator opts a # group in explicitly rather than getting it for free # (docs/MESHBAY_DESIGN.md §9.7, §9.8). # `helloworld` is the reference implementation (docs/MESHBAY_DESIGN.md # §9.4), hidden client-side behind `?dev=1`. It is here because the # allow-list is server-side enforcement — a client that names an app this # node does not know is refused — and an app the node refused could not # demonstrate anything. This entry and the client's registry line are the # whole of what adding an application costs. ALLOWED_APPS = frozenset({"chat", "files", "video", "music", "photo", "helloworld"}) def _do_apps_enabled(self, msg: dict) -> None: """ Turn a group "application" on or off for everyone, for this group. Signed like the root ops: this decides what a member sees, and an unsigned message would let any member turn a disabled one back on. """ apps = msg.get("apps") if not isinstance(apps, list) or not apps: self._send({"type": "error", "detail": "Missing or empty apps"}) return unknown = set(apps) - self.ALLOWED_APPS if unknown: self._send({"type": "error", "detail": f"Unknown app(s): {', '.join(sorted(unknown))}"}) return # Files is not a toggle: MNP permits root exploration regardless of # what this list says, so hiding the tab only ever misled. Added at the # front, the same order ops.set_enabled_apps writes, so the landing-tab # preference sees one list and not two. if "files" not in apps: apps.insert(0, "files") if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return # The subject is what the operator is shown before signing, and what # the client compares its own request against (transport.js) — a # canonical form so both sides build the same transcript. self._issue_admin_challenge(OP_APPS_ENABLED, ",".join(sorted(apps))) async def _admin_exec_apps_enabled( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: apps = pending["subject"].split(",") if pending["subject"] else [] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"apps_enabled:{pending['subject']}") return try: await self._run_op( ops.set_enabled_apps, self._group_id or "", apps) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("apps_enabled", pending["subject"]) # Everyone already connected is told, so a disabled tab disappears # without waiting for a reconnection. notice = {"type": MNP.APPS_ENABLED_ACK, "v": MNP_VERSION, "apps": apps} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass # ── App directories (generic) ──────────────────────────────────────── def _do_app_directories(self, msg: dict) -> None: """ Which folder(s) an application works over, for any application. One handler for every application, keyed by the app's own name: adding an application adds no message type, and there is no per-app handler differing only in the key it writes and whether it carries a string or a list. `app` must be one this node knows (`ALLOWED_APPS`) — a client-supplied key is otherwise a way to write arbitrary rows into `group_settings`. The paths are checked by `ops._validate_app_dirs`, which runs after the signature: this is a settings change, not a capability, so refusing early here would be a courtesy rather than the control. """ app = str(msg.get("app", "")).strip() dirs = msg.get("directories") if app not in self.ALLOWED_APPS: self._send({"type": "error", "detail": f"Unknown app {app!r}"}) return if not isinstance(dirs, list) or not all(isinstance(d, str) for d in dirs): self._send({"type": "error", "detail": "Missing or invalid 'directories'"}) return clean = sorted({d.strip("/") for d in dirs if d.strip("/")}) if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return # The app is in the subject, not only the paths: an operator shown # "Media/Films" alone cannot tell which application is about to be # pointed at it, and two apps' challenges would be indistinguishable. self._issue_admin_challenge( OP_APP_DIRECTORIES, f"{app}:{','.join(clean)}") async def _admin_exec_app_directories( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: app, _, joined = pending["subject"].partition(":") dirs = joined.split(",") if joined else [] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"app_directories:{pending['subject']}") return try: result = await self._run_op( ops.set_app_directories, self._group_id or "", app, dirs) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("app_directories", pending["subject"]) self._broadcast_to_group({"type": MNP.APP_DIRECTORIES_ACK, "v": MNP_VERSION, "app": app, "directories": result["directories"]}) # ── Chat ───────────────────────────────────────────────────────────── def _do_search_listed(self, msg: dict) -> None: """ Whether this group's files appear in members' cross-group Search. Signed because it changes what every member's Search shows, not because it protects anything — see ops.set_search_listed. """ listed = msg.get("listed") if not isinstance(listed, bool): self._send({"type": "error", "detail": "Missing or invalid 'listed'"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_SEARCH_LISTED, "on" if listed else "off") async def _admin_exec_search_listed( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: listed = pending["subject"] == "on" if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"search_listed:{pending['subject']}") return try: await self._run_op(ops.set_search_listed, self._group_id or "", listed) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("search_listed", pending["subject"]) self._broadcast_to_group({"type": MNP.SEARCH_LISTED_ACK, "v": MNP_VERSION, "listed": listed}) async def _do_group_roster_req(self, msg: dict) -> None: """ Who is in this group, and which device keys they hold. Answers **any member**, not only the operator — that is the whole point. A member verifies for themselves that a message came from a device belonging to the account it claims, instead of taking the node's `sender_id` on trust. What makes that possible is relayed here: each device's key, which already-pinned key countersigned it, and the signature plus the nonce and timestamp needed to rebuild what was signed. Sealed under a GEK-derived subkey, for the same reason the index is: it is the group's membership, and a peer that has not completed the handshake has no business reading it. What this deliberately does not do is *decide* anything. The node hands over evidence; the client checks the chain and keeps its own pins. A node that lies here is caught by a client that has seen the account before, which is the property Tier 2 buys and the reason the node is not asked to assert trust. """ gctx = self._group_ctx() gek = gctx.get("gek") roster = self._ctx.get("roster") if not gek: self._send({"type": "error", "detail": "Group encryption not initialized"}) return if roster is None: self._send({"type": "error", "detail": "Roster not available"}) return devices = await roster.group_devices(self._group_id or "") payload = {"devices": devices, "node_pk": self._node_pk_b64()} sealed = seal(gek, PURPOSE_ROSTER, MNP.GROUP_ROSTER_RESP, self._group_id or "", payload) self._send({"type": MNP.GROUP_ROSTER_RESP, "v": MNP_VERSION, "group_id": self._group_id or "", **sealed}) def _broadcast_to_group(self, notice: dict) -> None: """ Tell everyone connected to this group about a setting that changed. Enforcement never depends on this reaching them — the node is what refuses — but a control that stays on screen until the next reconnection is a control people use. """ for _uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass # Reconcile's backstop and the watchdog debounce (indexer.py # DirectoryIndexer) — how hard the node works on the operator's own # disk, not a member-facing permission. Signed for the same reason as # apps_enabled: consistency of the authorization model, not because a # wrong value here is itself dangerous. MIN_RECONCILE_SECS = 10.0 MAX_RECONCILE_SECS = 24 * 3600.0 MIN_DEBOUNCE_SECS = 0.0 MAX_DEBOUNCE_SECS = 300.0 def _do_set_scan_settings(self, msg: dict) -> None: try: reconcile = float(msg.get("reconcile_interval_secs")) debounce = float(msg.get("debounce_secs")) except (TypeError, ValueError): self._send({"type": "error", "detail": "Invalid scan settings"}) return if not (self.MIN_RECONCILE_SECS <= reconcile <= self.MAX_RECONCILE_SECS): self._send({"type": "error", "detail": f"reconcile_interval_secs must be between " f"{self.MIN_RECONCILE_SECS:.0f} and " f"{self.MAX_RECONCILE_SECS:.0f}"}) return if not (self.MIN_DEBOUNCE_SECS <= debounce <= self.MAX_DEBOUNCE_SECS): self._send({"type": "error", "detail": f"debounce_secs must be between " f"{self.MIN_DEBOUNCE_SECS:.0f} and " f"{self.MAX_DEBOUNCE_SECS:.0f}"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}") MIN_TRANSFER_LIMIT = 1 MAX_TRANSFER_LIMIT = 32 def _do_transfer_limits(self, msg: dict) -> None: """How many transfers one member may run at once in this group. Zero is not "unlimited" and is refused: a member who may not transfer at all is a member the operator revokes, and reading 0 as no-limit would make the most dangerous value the easiest to type by accident. """ try: downloads = int(msg.get("downloads")) uploads = int(msg.get("uploads")) except (TypeError, ValueError): self._send({"type": "error", "detail": "Invalid transfer limits"}) return for value in (downloads, uploads): if not (self.MIN_TRANSFER_LIMIT <= value <= self.MAX_TRANSFER_LIMIT): self._send({"type": "error", "detail": f"transfer limits must be between " f"{self.MIN_TRANSFER_LIMIT} and " f"{self.MAX_TRANSFER_LIMIT}"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_TRANSFER_LIMITS, f"d={downloads},u={uploads}") async def _admin_exec_transfer_limits( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: try: parts = dict(p.split("=") for p in pending["subject"].split(",")) downloads, uploads = int(parts["d"]), int(parts["u"]) except (ValueError, KeyError): self._send({"type": "error", "detail": "Invalid transfer limits"}) return if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"transfer_limits:{pending['subject']}") return try: result = await self._run_op( ops.set_transfer_limits, self._group_id or "", downloads, uploads) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("transfer_limits", pending["subject"]) notice = {"type": MNP.TRANSFER_LIMITS_ACK, "v": MNP_VERSION, "limits": result["limits"]} for session in list(self._peer_registry().values()): try: session._send(notice) except Exception: pass async def _admin_exec_set_scan_settings( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: try: reconcile_s, debounce_s = pending["subject"].split(",") reconcile, debounce = float(reconcile_s), float(debounce_s) except (ValueError, KeyError): self._send({"type": "error", "detail": "Invalid scan settings"}) return if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"set_scan_settings:{pending['subject']}") return try: result = await self._run_op( ops.set_scan_settings, self._group_id or "", reconcile, debounce) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("set_scan_settings", pending["subject"]) notice = {"type": MNP.SET_SCAN_SETTINGS_ACK, "v": MNP_VERSION, **result} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass # ── Node management (D5) ───────────────────────────────────────────────── async def _do_node_status(self, msg: dict) -> None: """All groups, roots, peers — the operator's overview. Including every root's absolute path, which is why this is gated on a proved operator device and not on an account the hub named. """ node_uid = self._ctx.get("node_user_id") log.info("node_status: user=%s node_user=%s owner=%s device=%s", self._user_id, node_uid, self._is_node_admin(), "confirmed" if self._device_confirmed else "unidentified") if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return try: result = await self._run_op(ops.list_groups) self._send({"type": MNP.NODE_STATUS_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("node_status failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) async def _do_node_settings_set(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return settings = msg.get("settings", {}) if not settings: self._send({"type": "error", "detail": "No settings provided"}) return try: result = await self._run_op(ops.set_node_settings, settings) self._send({"type": MNP.NODE_SETTINGS_SET_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("node_settings_set failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) async def _do_roster_read(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return group_id = str(msg.get("group_id", "")).strip() try: result = await self._run_op(ops.read_roster, group_id) self._send({"type": MNP.ROSTER_READ_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("roster_read failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) async def _do_denylist_read(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return try: result = await self._run_op(ops.read_denylist) self._send({"type": MNP.DENYLIST_READ_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("denylist_read failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) async def _do_denylist_clear(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return subject = str(msg.get("subject", "")).strip() try: result = await self._run_op(ops.clear_denylist, subject=subject) self._audit("denylist_clear", subject or "all") self._send({"type": MNP.DENYLIST_CLEAR_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("denylist_clear failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) def _do_group_attach(self, msg: dict) -> None: name = str(msg.get("name", "")).strip() shared_dir = str(msg.get("shared_dir", "")).strip() if not name or not shared_dir: self._send({"type": "error", "detail": "Missing name or shared_dir"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return # `upload_dir` is not read here any more, and a client still sending it # is ignored rather than obeyed: on load it forces every other root # read-only, which is the model the RO/RW one replaced. A second # writable directory is `root_add` with `writable`. self._issue_admin_challenge( OP_GROUP_ATTACH, name, payload={"name": name, "shared_dir": shared_dir, "writable": bool(msg.get("writable", True))}, group_id="") async def _admin_exec_group_attach( 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"group_attach:{pending['subject'][:16]}") return p = pending.get("payload") or {} try: result = await self._run_op( ops.attach_group, p["name"], p["shared_dir"], writable=bool(p.get("writable", True))) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("group_attach", pending["subject"]) self._send({"type": MNP.GROUP_ATTACH_ACK, "v": MNP_VERSION, **result}) state = self._ctx.get("daemon_state") reload_fn = state.get("reload_fn") if state else None if reload_fn: try: await reload_fn() except Exception as e: log.error("Reload after group_attach failed: %s", e) def _do_group_detach(self, msg: dict) -> None: name = str(msg.get("name", "")).strip() if not name: self._send({"type": "error", "detail": "Missing group name or id"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_GROUP_DETACH, name, payload={"name": name}, group_id="") async def _admin_exec_group_detach( 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"group_detach:{pending['subject'][:16]}") return p = pending.get("payload") or {} try: result = await self._run_op(ops.detach_group, p["name"]) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("group_detach", pending["subject"]) self._send({"type": MNP.GROUP_DETACH_ACK, "v": MNP_VERSION, **result}) state = self._ctx.get("daemon_state") reload_fn = state.get("reload_fn") if state else None if reload_fn: try: await reload_fn() except Exception as e: log.error("Reload after group_detach failed: %s", e) async def _do_node_reload(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return state = self._ctx.get("daemon_state") reload_fn = state.get("reload_fn") if state else None if not reload_fn: self._send({"type": "error", "detail": "Reload not available"}) return try: await reload_fn() self._send({"type": MNP.NODE_RELOAD_ACK, "v": MNP_VERSION, "status": "reloaded"}) except Exception as e: log.error("node_reload failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Reload failed"}) def _do_root_add(self, msg: dict) -> None: target_group = str(msg.get("group_id", "")).strip() path = str(msg.get("path", "")).strip() if not target_group or not path: self._send({"type": "error", "detail": "Missing group_id or path"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_ROOT_ADD, path, payload={ "group_id": target_group, "path": path, "name": str(msg.get("name", ""))[:128], "kind": str(msg.get("kind", "generic"))[:16], "writable": bool(msg.get("writable", msg.get("upload", False))), "removable": bool(msg.get("removable", False)), }, group_id=target_group) async def _admin_exec_root_add( 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"root_add:{pending['subject'][:24]}") return p = pending["payload"] try: result = await self._run_op( ops.add_root, p["group_id"], p["path"], name=p.get("name", ""), kind=p.get("kind", "generic"), writable=p.get("writable", False), removable=p.get("removable", False)) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return except Exception as e: log.error("root_add failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) return self._audit("root_add", f"{p['path']}→{p['group_id'][:8]}") await self._retarget_indexer(p["group_id"]) self._send({"type": MNP.ROOT_ADD_ACK, "v": MNP_VERSION, **result}) def _do_root_remove(self, msg: dict) -> None: target_group = str(msg.get("group_id", "")).strip() root_name = str(msg.get("root_name", "")).strip() if not target_group or not root_name: self._send({"type": "error", "detail": "Missing group_id or root_name"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_ROOT_REMOVE, root_name, payload={"group_id": target_group, "root_name": root_name}, group_id=target_group) async def _admin_exec_root_remove( 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"root_remove:{pending['subject'][:24]}") return p = pending["payload"] try: result = await self._run_op( ops.remove_root, p["group_id"], p["root_name"]) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return except Exception as e: log.error("root_remove failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) return self._audit("root_remove", f"{p['root_name']}←{p['group_id'][:8]}") await self._retarget_indexer(p["group_id"]) self._send({"type": MNP.ROOT_REMOVE_ACK, "v": MNP_VERSION, **result}) def _do_root_update(self, msg: dict) -> None: target_group = str(msg.get("group_id", self._group_id or "")).strip() root_name = str(msg.get("root_name", "")).strip() if not target_group or not root_name: self._send({"type": "error", "detail": "Missing group_id or root_name"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return updates = [] if "writable" in msg: updates.append(f"rw={'on' if msg['writable'] else 'off'}") if "removable" in msg: updates.append(f"rem={'on' if msg['removable'] else 'off'}") subject = f"{root_name}:{','.join(updates)}" if updates else root_name self._issue_admin_challenge( OP_ROOT_UPDATE, subject, payload={ "group_id": target_group, "root_name": root_name, "writable": msg.get("writable"), "removable": msg.get("removable"), }, group_id=target_group) async def _admin_exec_root_update( 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"root_update:{pending['subject'][:24]}") return p = pending["payload"] try: result = await self._run_op( ops.update_root, p["group_id"], p["root_name"], writable=p.get("writable"), removable=p.get("removable")) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return except Exception as e: log.error("root_update failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) return self._audit("root_update", pending["subject"]) await self._retarget_indexer(p["group_id"]) notice = {"type": MNP.ROOT_UPDATE_ACK, "v": MNP_VERSION, **result} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass def _do_root_eject(self, msg: dict) -> None: target_group = str(msg.get("group_id", self._group_id or "")).strip() root_name = str(msg.get("root_name", "")).strip() if not target_group or not root_name: self._send({"type": "error", "detail": "Missing group_id or root_name"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_ROOT_EJECT, root_name, payload={"group_id": target_group, "root_name": root_name}, group_id=target_group) async def _admin_exec_root_eject( 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"root_eject:{pending['subject'][:24]}") return p = pending["payload"] try: result = await self._run_op( ops.eject_root, p["group_id"], p["root_name"]) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return except Exception as e: log.error("root_eject failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) return self._audit("root_eject", p["root_name"]) notice = {"type": MNP.ROOT_EJECT_ACK, "v": MNP_VERSION, **result} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass def _do_root_plug(self, msg: dict) -> None: target_group = str(msg.get("group_id", self._group_id or "")).strip() root_name = str(msg.get("root_name", "")).strip() if not target_group or not root_name: self._send({"type": "error", "detail": "Missing group_id or root_name"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_ROOT_PLUG, root_name, payload={"group_id": target_group, "root_name": root_name}, group_id=target_group) async def _admin_exec_root_plug( 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"root_plug:{pending['subject'][:24]}") return p = pending["payload"] try: result = await self._run_op( ops.plug_root, p["group_id"], p["root_name"]) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return except Exception as e: log.error("root_plug failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) return self._audit("root_plug", p["root_name"]) notice = {"type": MNP.ROOT_PLUG_ACK, "v": MNP_VERSION, **result} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass 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 _retarget_indexer(self, group_id: str) -> None: """ Pick up a root that was just added to or removed from node.toml. Through the daemon's own reload, which is what the loopback API has always done after the same operations (`ui/app.py`). This used to re-point the indexer at `groups_ctx[gid]["roots"]` instead — the very object the op had just edited — so `retarget` diffed a set against itself, found no new names, scanned nothing, and dropped nothing. A directory added over MNP reached node.toml and was invisible until a restart; one removed kept serving its files. Two front doors doing different things is the shape `ops.py` exists to prevent, and this was it: the loopback path worked and the MNP path did not, which is why it survived until the operator added a directory from a browser. Not awaited: a reload rescans, and a new library is minutes. The ack the caller sends carries the set the node is moving to, and the `index_sync` that follows the scan carries what it found. """ state = self._ctx.get("daemon_state") if not state: return reload_fn = state.get("reload_fn") if reload_fn: self._spawn(reload_fn()) return # No daemon to ask — a test harness, or a context assembled by hand. # Retarget directly, which is correct as long as the caller did not # edit the live set in place. indexer = state.get("indexers", {}).get(group_id) roots = state.get("groups_ctx", {}).get(group_id, {}).get("roots") if indexer and roots: await indexer.retarget(roots) 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 try: result = await self._run_op( ops.revoke_member, user_id, self._group_id or "") await self._new_chat_epoch(self._group_id or "", "member_revoke") except ops.OpError as e: self._send({"type": "error", "detail": e.message}) 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. # Every connection that account holds, not "the" one: with device # linking a person may be connected from several at once, and the # registry is keyed per connection precisely because it cannot hold # only one of them. for peer in self._sessions_of(user_id): try: await peer.close() except Exception: pass self._audit("member_revoke", user_id) self._send({ "type": MNP.MEMBER_REVOKE_ACK, "v": MNP_VERSION, "user_id": user_id, "reminder": result.get("reminder", ""), }) 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) def _on_done(t): self._tasks.discard(t) if not t.cancelled() and t.exception(): log.error("Spawned task failed: %s", t.exception(), exc_info=t.exception()) task.add_done_callback(_on_done) return task def _app_directories_ack(self) -> dict: """ Every application's configured folders, for the handshake ack. Read off the group context rather than from a list of applications kept here, so this cannot name an application the node knows nothing else about — and cannot fail to name one the daemon does. A copy of the daemon's `APP_DIR_KEYS` lived here until 2026-09-10 and had already lost an entry, which made the app that entry belonged to the single one whose directories never reached a client. This module names an application in exactly one place, and it is `ALLOWED_APPS`. `_app_directories_ctx` is the only thing that puts a `*_directories` key in that context, and an absent one reads as none configured — never as "the whole group index". """ return {key: list(value or []) for key, value in self._group_ctx().items() if key.endswith("_directories")} def _group_ctx(self) -> dict: if "groups" in self._ctx and self._group_id: # `.get`, not a bare subscript. A config reload removes a group # from this map (daemon.py's reload does `groups_ctx.pop`) while # sessions connected to it are still open, and the next request # any of them made raised KeyError into _dispatch_message's # catch-all. An absent group now reads the way an unconfigured # one already does — the handlers all test for what they need — # instead of failing every request the session has left. return self._ctx["groups"].get(self._group_id) or {} return self._ctx def _indexing_status(self) -> dict: """ The counters of `_push_index_progress` (daemon.py), for the handshake ack — never a path, a filename or a root name, which stay local to the operator's own admin UI. Absent "progress" (context not loaded, or a group with no indexer at all) reads as idle rather than erroring. """ progress = self._group_ctx().get("progress") if progress is None: return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0, "files_done": 0, "files_total": 0, "kind": "", "root_pos": -1, "queued": 0} return { "scanning": progress.scanning, "scanned_bytes": progress.scanned_bytes, "total_bytes": progress.total_bytes, "files_done": progress.files_done, "files_total": progress.files_total, "kind": progress.kind, "root_pos": progress.root_pos, "queued": len(progress.queued), } # ── Upload ceiling and transfer slots ──────────────────────────────────── def _register_peer(self) -> None: """Add this connection to its group's peer set. One place decides the key, and it is `_registry_key` — per connection, never per account. Written as a method so a test drives the real registration rather than a second copy of this line that agrees with it by construction. """ self._peer_registry()[self._registry_key] = self def _unregister_peer(self) -> None: self._peer_registry().pop(self._registry_key, None) def _sessions_of(self, user_id: str) -> list["WebRTCPeerSession"]: """Every live connection this account holds in this group. Never "the" connection: with device linking a person may be connected from a laptop and a phone at once, and an operation that acts on one of them at random is a revocation that leaves a session running. """ return [s for s in list(self._peer_registry().values()) if s._user_id == user_id] 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_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")}) # ── 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, group_id: str | 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. `group_id` overrides the connection's group for cross-group operations (e.g. root management from a NodePage connection). """ gid = group_id if group_id is not None else (self._group_id or "") 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 {}, "group_id": gid, } 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": gid, }) @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: """ A key this node pinned for the account we just authenticated. `get_identity` returns the account's **oldest** live device, which is a stand-in, not an answer: the handshake never said which device is on this connection. `device_hello` is the answer, and it arrives later — so this must never overwrite a confirmed one. It is spawned from `_complete_handshake` and can therefore finish *after* a fast client has already identified itself, which is exactly the ordering that would put the wrong key back. """ roster = self._ctx.get("roster") if roster is None or not self._user_id or self._device_confirmed: return ident = await roster.get_identity(self._user_id) if ident and not self._device_confirmed: self._pinned_pk = ident["pk_ed25519"] def _is_node_admin(self) -> bool: """ Whether the **account** on this connection is the one the node belongs to. This is a display hint and half of a check — never authority on its own. `self._user_id` is the `sub` of a JWT the hub issued, so read alone it says "the hub says you are the owner", which is the one thing NS4 and M3 rule out: a hub that can name the operator can install itself as node administrator. It rides the handshake ack so a client knows whether to offer the Node page at all, and every operation is gated on `_operator_device()` below. """ node_user_id = self._ctx.get("node_user_id") return bool(node_user_id and self._user_id == node_user_id) async def _operator_device(self) -> bool: """ Whether this connection may run the node's own controls. Two things, and the second is the one that cannot be forged: - the account is the one this node belongs to (`_is_node_admin`), which is what keeps node-wide controls with the machine's owner rather than with every paired operator of every group on it; and - **the device on this connection proved a key the node pinned as an operator**. `device_hello` is signed over a transcript naming this node, this group and this connection's nonce, and `operator_pks()` is rebuilt from the roster on each call, so an unpinned browser and a revoked one are both refused at once. The second clause is the fix for the door this used to leave open. `node_status`, `node_settings_set`, `roster_read`, `denylist_read`, `denylist_clear` and `node_reload` were gated on the account id alone — a value the hub chooses. An active hub that can also reach the group key (which §3.5 concedes it can in an open-join group) could therefore mint a token for the owner's account and read `node_status`, which lists every group on the node with the operator's **absolute paths**, or clear the denylist, which is the persisted revocation H4 exists to keep. It holds no user keys and cannot countersign anything, so it cannot produce a `device_hello` — which is the same property device linking rests on (§3.3), applied to the node's own surface. """ if not self._is_node_admin(): return False if not self._device_confirmed or not self._pinned_pk: return False roster = self._ctx.get("roster") if roster is None: return False return self._pinned_pk in await roster.operator_pks() 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=(pending["group_id"] if pending.get("group_id") is not None else (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_INVITE_LINK_CREATE: self._spawn( self._admin_exec_invite_link_create(pending, transcript, sig_bytes)) elif pending["op"] == OP_INVITE_CANCEL: self._spawn( self._admin_exec_invite_cancel(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)) elif pending["op"] == OP_APPS_ENABLED: self._spawn( self._admin_exec_apps_enabled(pending, transcript, sig_bytes)) elif pending["op"] == OP_TRANSFER_LIMITS: self._spawn( self._admin_exec_transfer_limits(pending, transcript, sig_bytes)) elif pending["op"] == OP_SET_SCAN_SETTINGS: self._spawn( self._admin_exec_set_scan_settings(pending, transcript, sig_bytes)) elif pending["op"] == OP_TMDB_CONFIG: self._spawn( self._admin_exec_tmdb_config(pending, transcript, sig_bytes)) elif pending["op"] == OP_TMDB_ENABLED: self._spawn( self._admin_exec_tmdb_enabled(pending, transcript, sig_bytes)) elif pending["op"] == OP_TMDB_OVERRIDE: self._spawn( self._admin_exec_tmdb_override(pending, transcript, sig_bytes)) elif pending["op"] == OP_TMDB_REMATCH: self._spawn( self._admin_exec_tmdb_rematch(pending, transcript, sig_bytes)) elif pending["op"] == OP_MUSICBRAINZ_ENABLED: self._spawn( self._admin_exec_musicbrainz_enabled(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_ADD: self._spawn( self._admin_exec_root_add(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_REMOVE: self._spawn( self._admin_exec_root_remove(pending, transcript, sig_bytes)) elif pending["op"] == OP_APP_DIRECTORIES: self._spawn( self._admin_exec_app_directories(pending, transcript, sig_bytes)) elif pending["op"] == OP_CHAT_DIRECTORY: self._spawn( self._admin_exec_chat_directory(pending, transcript, sig_bytes)) elif pending["op"] == OP_CHAT_LINK_PREVIEW: self._spawn( self._admin_exec_chat_link_preview(pending, transcript, sig_bytes)) elif pending["op"] == OP_SEARCH_LISTED: self._spawn( self._admin_exec_search_listed(pending, transcript, sig_bytes)) elif pending["op"] == OP_CHAT_EPOCH: self._spawn( self._admin_exec_chat_epoch(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_UPDATE: self._spawn( self._admin_exec_root_update(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_EJECT: self._spawn( self._admin_exec_root_eject(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_PLUG: self._spawn( self._admin_exec_root_plug(pending, transcript, sig_bytes)) elif pending["op"] == OP_GROUP_ATTACH: self._spawn( self._admin_exec_group_attach(pending, transcript, sig_bytes)) elif pending["op"] == OP_GROUP_DETACH: self._spawn( self._admin_exec_group_detach(pending, transcript, sig_bytes)) else: self._send({"type": "error", "detail": "Unknown admin operation"}) def _send(self, obj: dict) -> None: # Stamp the reply with the id of the request being answered, so the # caller never has to guess. Only for this session's own replies: a # handler that also pushes to other peers (a chat broadcast, an index # delta) reaches them through *their* _send, where the owner no longer # matches and nothing is stamped — those messages answer no request. # An explicit req_id already on the object wins, and an unsolicited # push (no request in scope) carries none, exactly as before. owner, req_id = _REPLY_TO.get() if req_id is not None and owner is self and "req_id" not in obj: obj = {**obj, "req_id": req_id} 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() # Before the tasks are cancelled: a lease is not held by a task, so # nothing else would give it back, and this hook is the one place every # way of walking away arrives at (see the connectionstatechange handler, # which calls it for a closed tab, a quit browser and a dead network # alike). self._release_transfers() 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") self._release_transfers() if self._user_id: self._unregister_peer() await self.shutdown_tasks() await self._pc.close() 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, max_concurrent_downloads: int | None = None, max_concurrent_uploads: int | None = None, max_upload_gb: float | None = None, transcode_incompatible_video: bool = True, ): 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, # Read once, when the first transfer builds the pools. None means # the operator said nothing and transfers.py's defaults apply. "max_concurrent_downloads": max_concurrent_downloads, "max_concurrent_uploads": max_concurrent_uploads, # The per-file upload ceiling, in GB. None means the operator said # nothing and MAX_UPLOAD_BYTES stands. "max_upload_gb": max_upload_gb, # Operator opt-out (node.toml) for the HEVC-etc. transcode # fallback in _stream_video_inner — real CPU cost, unlike copy. "transcode_incompatible_video": transcode_incompatible_video, } if groups: self._ctx["groups"] = groups if denylist: self._ctx["denylist"] = denylist from meshbay_node.config import DEFAULT_STUN_SERVERS self._stun = stun_servers or list(DEFAULT_STUN_SERVERS) self._sessions: dict[str, WebRTCPeerSession] = {} self._reapers: set[asyncio.Task] = set() def set_capacity(self, *, max_concurrent_streams: int | None = None, max_concurrent_downloads: int | None = None, max_concurrent_uploads: int | None = None, max_upload_gb: float | None = None) -> dict: """Resize a live pool without restarting the daemon. `ops.set_node_settings` used to do this by assigning `webrtc._stream_sem`, an attribute that has never existed — the pool is `ctx["_transcode_sem"]`, and `hasattr(webrtc, "_stream_sem")` is always False. So the hot-swap was a no-op and **`max_concurrent_streams` has never taken effect from the Node page without a restart**, contrary to docs/MESHBAY_DESIGN.md §6.8. This is the one implementation, on the object that owns the state, so the next two caps do not each grow their own copy of the mistake. What resizing means, stated because it is a decision and not a detail: **the new cap governs new streams; the ones already running are never interrupted.** A slot is held for the length of a film, so lowering the cap below what is in flight cannot take a viewer's film away — it stops the next one starting. The replacement pool is therefore created with the permits that remain (`new - in_flight`, floored at zero), not with a full set, or lowering the cap would briefly allow more viewers than either the old value or the new one. """ changed: dict = {} if max_concurrent_streams is not None: n = int(max_concurrent_streams) if n < 1: raise ValueError("max_concurrent_streams must be positive") before = self._ctx.get("max_concurrent_streams") self._ctx["max_concurrent_streams"] = n if self._ctx.get("_transcode_sem") is not None: in_flight = self._ctx.get("_streams_in_flight", 0) self._ctx["_transcode_sem"] = asyncio.Semaphore( max(0, n - in_flight)) log.info("stream: capacity %s -> %d (%d in flight, %d free now)", before, n, in_flight, max(0, n - in_flight)) else: # Nothing has streamed yet; the pool is built from this value on # first use, so there is nothing to resize. log.info("stream: capacity %s -> %d (no pool built yet)", before, n) changed["max_concurrent_streams"] = n pools = {} if max_concurrent_downloads is not None: pools[transfers_mod.DOWNLOAD] = int(max_concurrent_downloads) if max_concurrent_uploads is not None: pools[transfers_mod.UPLOAD] = int(max_concurrent_uploads) for key, value in pools.items(): if value < 1: raise ValueError(f"max_concurrent_{key}s must be positive") if pools: # Kept on the context whether or not a pool exists yet: the pools # are built on the first transfer, and would otherwise come up with # the defaults after an operator had already changed them. for key, value in pools.items(): self._ctx[f"max_concurrent_{key}s"] = value changed[f"max_concurrent_{key}s"] = value slots = self._ctx.get("_transfer_slots") if slots is not None: granted = slots.set_caps(node=pools) log.info("transfer: capacity now %s (%d started at once)", slots.summary(), len(granted)) # Raising a cap can start queued transfers immediately, and the # peers waiting on them have to be told: a grant nobody hears # about is the "stuck at waiting" report this design exists to # prevent. for lease in granted: self._notify_granted(lease) if max_upload_gb is not None: gb = float(max_upload_gb) if gb <= 0: raise ValueError("max_upload_gb must be greater than zero") self._ctx["max_upload_gb"] = gb changed["max_upload_gb"] = gb log.info("upload: per-file ceiling now %g GB", gb) return changed def _notify_granted(self, lease) -> None: """Tell the connection that owns `lease` it may start. On the transport rather than the session because a cap change has no session behind it — it arrives from the loopback API. """ groups = self._ctx.get("groups") registries = ([g.get("_peers", {}) for g in groups.values()] if groups else [self._ctx.get("_peers", {})]) for reg in registries: session = reg.get(lease.session_key) if session is not None: try: session._send( session._transfer_state_msg(lease, "granted")) except Exception: pass return 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 RTCConfiguration, RTCIceServer # aiortc keeps only the first STUN entry it sees here; the actual # multi-server fan-out is done by transport/stun_multi, which patches # aioice. The full list is still passed so a one-server deploy and the # tests that read `_stun` stay coherent. config = RTCConfiguration( iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else [] ) # Before anything is allocated. Every offer costs an RTCPeerConnection # with its own DTLS and SCTP stacks, and nothing here used to bound how # many a node would hold: the hub meters offers *per account* # (signaling.py), which is a limit on each caller and not on this # machine, so the cost # grew with the number of members in the group. An operator's node must # not be exhaustible by the people they invited. if len(self._sessions) >= MAX_PEER_SESSIONS: log.warning("Refusing WebRTC offer: %d peer sessions already open", len(self._sessions)) raise RuntimeError("Node is at its peer-connection limit") pc = RTCPeerConnection(configuration=config) session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id) self._sessions[peer_id] = session self._reap_if_unauthenticated(peer_id) @pc.on("datachannel") def on_datachannel(channel: RTCDataChannel): log.info("WebRTC DataChannel opened: %s (peer=%s)", channel.label, peer_id) session._setup_channel(channel) if _WEBRTC_TRACE: @pc.on("iceconnectionstatechange") def on_ice_state_change(): log.info("WebRTC ICE state: %s (peer=%s)", pc.iceConnectionState, peer_id) @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. # # And the group's peer set forgets it too, as close() does: # otherwise every later broadcast to the group is written # to a closed channel, and every reconnect leaves one more # dead session held until the node restarts. if gone._user_id: gone._unregister_peer() await gone.shutdown_tasks() offer = RTCSessionDescription(sdp=offer_sdp, type="offer") await pc.setRemoteDescription(offer) answer = await pc.createAnswer() gather_start = time.monotonic() await pc.setLocalDescription(answer) # ICE gathering runs inside setLocalDescription (non-trickle). A slow or # unreachable STUN server shows up here as seconds of wait and zero # srflx lines — the symptom the multi-server fan-out exists to prevent. answer_sdp = pc.localDescription.sdp srflx = answer_sdp.count(" typ srflx") # The host addresses this node put in the answer. When a peer reports # "DataChannel closed" the first question is whether the node offered # anything that peer could route to at all — on a NAT'd host or a VM the # only host candidate is an address no one else can reach, and the log # otherwise looks identical to a working connection. host_addrs: set[str] = set() for line in answer_sdp.splitlines(): if line.startswith("a=candidate:") and " typ host " in line: parts = line.split() if len(parts) > 5: host_addrs.add(parts[4]) log.info( "WebRTC answer ready for peer=%s (ICE gather %.2fs, host: %s, %d srflx)", peer_id, time.monotonic() - gather_start, ", ".join(sorted(host_addrs)) or "none", srflx) return answer_sdp, [] def _reap_if_unauthenticated(self, peer_id: str) -> None: """Close a session that never completes the handshake. A peer that connects and then says nothing is indistinguishable from a working one until it is asked to prove something, and it was never asked: `connectionstatechange` reaps a connection that *fails*, and one that succeeds and stays silent was held for the node's lifetime. That is the cheapest way to spend someone else's memory — no GEK, no token, no group, just an open connection. `_user_id` is set by the GEK proof (`_do_handshake_response`), so it is the one honest test of whether this peer ever became anybody. """ async def reap() -> None: try: await asyncio.sleep(UNAUTHENTICATED_SESSION_TIMEOUT) session = self._sessions.get(peer_id) if session is not None and not session._user_id: log.warning("Closing peer %s: no handshake within %ds", peer_id[:8], UNAUTHENTICATED_SESSION_TIMEOUT) await self.close_peer(peer_id) except asyncio.CancelledError: raise except Exception as e: log.warning("Reaping peer %s failed: %s", peer_id[:8], e) # Held in a set for the same reason every other task here is: asyncio # keeps only a weak reference, and a reaper collected mid-sleep reaps # nothing (see WebRTCPeerSession.__init__). task = asyncio.ensure_future(reap()) self._reapers.add(task) task.add_done_callback(self._reapers.discard) 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 task in list(self._reapers): task.cancel() self._reapers.clear() for session in list(self._sessions.values()): await session.close() self._sessions.clear() @property def active_peers(self) -> int: return len(self._sessions)