""" 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 re import time import uuid from pathlib import Path from typing import Any import blake3 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.chatbox import ( NONCE_LEN as CHAT_NONCE_LEN, ) from meshbay_common.chatbox import ( SIG_LEN as CHAT_SIG_LEN, ) from meshbay_common.crypto import pk_to_b64, wrap_gek_aes from meshbay_common.device import ( DEVICE_TTL, device_add_transcript, device_hello_transcript, device_request_transcript, ) from meshbay_common.groupbox import ( PURPOSE_ACK, PURPOSE_CHAT_KEYS, 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.join import ( JOIN_TTL, ROLE_MEMBER, ROLE_OPERATOR, join_transcript, ) from meshbay_common.protocol import ( MNP, UPLOAD_PROBE_INDEX, chunk_ciphertext, file_chunk_wire, file_upload_ack_wire, file_upload_payload, ) from meshbay_node import hwaccel, linkpreview, ops, platform from meshbay_node import transfers as transfers_mod from meshbay_node import uploads as uploads_mod from meshbay_node.chat import FORMAT_SEALED_V1, ReplayedMessage 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.media_probe import ( BROWSER_INCOMPATIBLE_VIDEO_CODECS, ) from meshbay_node.media_probe import ( probe_video as _probe_video, ) from meshbay_node.roots import ( ROOT_NOT_SERVED, SAFE_UPLOAD_NAME, RootSet, _free_name, off_disk, safe_subdir, ) from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK from meshbay_node.transfers import TransferSlots from meshbay_node.transport.webrtc.apps.music import MusicMixin from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin from meshbay_node.transport.webrtc.channel import ( _REPLY_TO, _DataChannelBuffer, _extract_dtls_fingerprint, _get_remote_ip, _pack, ) from meshbay_node.transport.webrtc.disk import _locate from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, MAX_MSG from meshbay_node.transport.webrtc.media_tools import ( _seek_lands_at, ) from meshbay_node.transport.wire import index_sync_message log = logging.getLogger(__name__) # Per-account blobs (docs/playlists.md §4.3). These are an unbounded write # primitive pointed at somebody else's disk, so every one of them is checked — # and every check **refuses**, never truncates. A truncating cap silently loses # tracks, which is the one failure the whole playlist design exists to prevent. # # The numbers are sized against the measured shape: ~300 bytes per track before # compression, deflate worth about three on a payload this repetitive. A 1 MB # body is therefore roughly ten thousand tracks in one playlist, and the # manifest holds names and revisions only. USER_BLOB_MANIFEST_MAX = 64 * 1024 USER_BLOB_BODY_MAX = 1024 * 1024 USER_BLOB_ACCOUNT_MAX = 8 * 1024 * 1024 # "playlists" is the manifest; "playlist:" is one playlist's tracks. A # pattern rather than a set, because the ids are client-generated — but a # pattern, not anything at all, or the table becomes a key/value store for # whatever a client feels like writing. # # The character class is deliberately wider than a UUID: the reserved id is the # word "favorites" (docs/playlists.md §5.1), so a hex-only pattern refuses the # one playlist every account has. It stays narrow enough to carry no structure # of its own — no "/", no ".", no second ":" — so a kind can never be read as a # path or as anything but one name in one namespace. _USER_BLOB_KIND_RE = re.compile( r"^(playlists|playlist:[A-Za-z0-9_-]{1,64})$") # An invitation link's handle, as `roster.create_link_invite` mints it. _INVITE_ID_RE = re.compile(r"[0-9a-f]{32}") # Chat link-preview results, kept in memory only (docs/MESHBAY_DESIGN.md §6.5: # the node # produces enrichment on demand and keeps nothing durable — the asking device # caches). Bounded and time-limited so a busy group cannot grow it without end # and a page that changed its card is picked up within the hour. _LINK_PREVIEW_TTL = 3600 _LINK_PREVIEW_MAX = 256 _link_preview_cache: dict[str, tuple[float, dict]] = {} def _link_preview_cache_get(url: str) -> dict | None: hit = _link_preview_cache.get(url) if hit is None: return None ts, value = hit if time.time() - ts > _LINK_PREVIEW_TTL: _link_preview_cache.pop(url, None) return None return value def _link_preview_cache_put(url: str, value: dict) -> None: if not url: return if len(_link_preview_cache) >= _LINK_PREVIEW_MAX: oldest = min(_link_preview_cache, key=lambda k: _link_preview_cache[k][0]) _link_preview_cache.pop(oldest, None) _link_preview_cache[url] = (time.time(), value) # A member pasting a link is normal; a member — or a hub minting tokens for many # accounts — firing hundreds is amplification/DoS and a way to make the node # reach arbitrary hosts on demand (finding M3). Only a real outbound fetch is # counted (a cache hit costs nothing), and the ceilings are generous enough that # ordinary chat never meets them. _LINK_PREVIEW_RATE_WINDOW = 60.0 _LINK_PREVIEW_RATE_PER_CONN = 15 _LINK_PREVIEW_RATE_NODE = 60 # A free-text TMDB search spends the *operator's* credential, which is rated by # TMDB and shared by everyone in the group: one member typing in the search box # can exhaust what every other member's automatic matching depends on, and the # operator is the one who has to notice. §6.5's rule is a bound and a named # adversary in the same commit; this one arrived without either. # # Per member rather than per connection, unlike link previews above: three tabs # is one person, and a ceiling a tab can multiply is not a ceiling. Kept in the # group context so it survives a reconnect, which is the other thing a per-session # count cannot do. # # Generous next to what a person types — ten searches a minute is a search every # six seconds, sustained — and small next to a loop. _TMDB_SEARCH_WINDOW = 60.0 _TMDB_SEARCH_PER_MEMBER = 10 _TMDB_SEARCH_NODE = 30 # Chat limits. A message is a member-supplied write onto the operator's disk # (`chat.db`, where retention is a manual CLI command — §6.6), relayed from there # to every other connected member and turned into a notification for every member # of the group. Nothing bounded any of it: the only ceiling was the frame size, # 64 MB once the handshake is done, so one member in a loop could fill the # operator's disk and saturate everyone else's connection. Uploads — the other # member-supplied write — have carried four protections and a size cap since # C5a; this is the same question asked of the path nobody had asked it of. # # 64 KB of ciphertext is about sixty thousand characters. The sealed payload is # the text, a thread id, a display name and a timestamp: an attachment is a file # on a root and travels as a reference (§4.5), so nothing legitimate comes close. MAX_CHAT_CIPHERTEXT = 64 * 1024 # Per account per group, not per connection: a second tab does not make a person # type faster, and keying on the session would hand a script one budget per # socket. Sixty a minute is far above a human and far below a flood. _CHAT_RATE_WINDOW = 60.0 _CHAT_RATE_PER_ACCOUNT = 60 # When the map of senders grows past this, the stale entries are dropped. A node # with more live chatters than this in one window is not the case being bounded. _CHAT_RATE_MAX_TRACKED = 1000 # Upload limits (finding C5a). Uploads used to land directly in the shared root under # a name the client chose, overwriting whatever was already there — which both violated # node sovereignty and defeated the delete authorization (overwrite a file, become its # recorded uploader, then delete it legitimately). # The ceiling is the operator's to set (`max_upload_gb` in node.toml, the Node # page and `meshbay-node transfers max-size`) because it is their disk that # fills: this is only the default a node starts from when they have said # nothing. It is read from the transport context on every chunk, so a change # applies to an upload already in flight. MAX_UPLOAD_BYTES = 8 * 1024 * 1024 * 1024 # 8 GB per file GB_BYTES = 1024 * 1024 * 1024 # What the `tr` on a chunk request turned out to be (see `_lease_of`). LEASE_GRANTED = "granted" LEASE_QUEUED = "queued" LEASE_NONE = "none" # 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 # ffmpeg is spawned per stream request; without a cap any member can fork-bomb # the node by requesting many streams at once (H6). # # Two was sized when a stream was a burst: the client took segments as fast as # it could append them, so a slot was held for the minute it took to push the # file and then came back. Now that the client only pulls ninety seconds ahead # of the playhead, a slot is held for as long as the film runs — so two slots # means two people can watch anything at all, and the third is refused for the # next hour and a half. The work behind a slot has not changed and is small: # ffmpeg runs `-c copy`, a remux with no encoding in it, and spends most of the # film blocked on a pipe nobody is reading. # # This is the default, not the policy: the right number depends on the machine, # so the operator sets `max_concurrent_streams` under [node] in node.toml. This # value applies when they have said nothing. MAX_CONCURRENT_TRANSCODES = 8 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited # until the native client removes remote keypair bundles entirely. MAX_PRE_PROOF_FETCHES = 4 # Pairing codes carry 40 bits and are single-use, but a connection must not be # allowed to sit there guessing. Failures are audited, so a grind is visible. MAX_JOIN_ATTEMPTS = 5 # Per-connection limits alone would not bind an attacker who can open connections # at will — and the adversary who can mint tokens for any account is the hub. So # failed pairings are also counted node-wide over a window. MAX_JOIN_FAILURES_WINDOW = 20 JOIN_FAILURE_WINDOW = 600 # seconds # Everything a member sends lands here: files from the Files panel and # attachments from the chat alike. One visible directory the operator can look # into, back up or empty — rather than a hidden tree of per-user uuids that # nobody could read, or files scattered wherever someone happened to be looking. STREAM_SEGMENT_SIZE = 256 * 1024 # A chunk is a megabyte and the browser keeps eight in flight, so answering them # as they arrive queues 8 MB on the channel with nothing watching. On a LAN that # drains before anyone notices; on a phone that is also uploading, it is minutes # of head-of-line delay for the reader. Above this, wait for room. DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 1024 # What a client may ask for in one go, and how long the node waits for it to ask # again before deciding nobody is watching any more. STREAM_MAX_CREDIT = 256 STREAM_CREDIT_TIMEOUT = 120 # How often that budget is re-examined. A viewer who left stops being # charged for a slot within this, rather than within the timeout. STREAM_CREDIT_POLL = 3 # How often transfer leases are swept. Nothing depends on it being # prompt -- the session teardown is the reclaim that matters and is # immediate; this catches peers that vanished without the connection # noticing, so it trades latency for a timer that hardly ever runs. TRANSFER_SWEEP_SECS = 15 # 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(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() async def _do_gek_bundle_fetch(self) -> None: """Serve the caller's wrapped GEK bundle during the handshake window.""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) return group_id = getattr(self, "_pending_group", "") user_id = getattr(self, "_pending_sub", "") if not group_id or not user_id: self._send({"type": "error", "detail": "No pending handshake"}) return bundle = await bundle_store.fetch(group_id, user_id) if bundle: self._send({ "type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": True, "pk_eph_b64": bundle["pk_eph_b64"], "nonce_b64": bundle["nonce_b64"], "wrapped_b64": bundle["wrapped_b64"], }) else: self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) def _do_invite_create(self, msg: dict) -> None: """ Issue a one-time pairing code for someone the operator wants to admit. Replaces the old invite path, where the inviter fetched the invitee's public key from the hub and wrapped the group key for whatever came back (H3). The node now needs nothing but a name: it will wrap the key itself, later, for a key the invitee proves they hold. """ roster = self._ctx.get("roster") if roster is None: self._send({"type": "error", "detail": "Roster not available"}) return invitee_id = msg.get("user_id", "") group_id = msg.get("group_id") or self._group_id if not invitee_id or not group_id: self._send({"type": "error", "detail": "Missing user_id or group_id"}) return if group_id != self._group_id: self._send({"type": "error", "detail": "Wrong group for this session"}) return if not self._has_admin_authority(): self._send({ "type": "error", "detail": "No operator paired — run `meshbay-node operator pair`", }) return self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, { "group_id": group_id, "user_id": invitee_id, "username": str(msg.get("username", ""))[:64], }) def _do_invite_link_create(self, msg: dict) -> None: """ Issue a code bound to no account, for an invitation link — into the group this connection authenticated to, and no other: a link names its group, so the operator signs for exactly that one (docs/MESHBAY_DESIGN.md §3.4). """ group_id = self._group_id or "" if not group_id: self._send({"type": "error", "detail": "No group on this connection"}) return if msg.get("group_id") and msg["group_id"] != group_id: self._send({"type": "error", "detail": "Wrong group for this session"}) return if not self._has_admin_authority(): self._send({ "type": "error", "detail": "No operator paired — run `meshbay-node operator pair`", }) return self._issue_admin_challenge( OP_INVITE_LINK_CREATE, f"link:{group_id}", {"group_id": group_id}) def _do_invite_cancel(self, msg: dict) -> None: """Take back an unredeemed link of this group, by its handle.""" group_id = self._group_id or "" invite_id = str(msg.get("invite_id", "")) if not group_id: self._send({"type": "error", "detail": "No group on this connection"}) return if not _INVITE_ID_RE.fullmatch(invite_id): self._send({"type": "error", "detail": "Not an invitation id"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_INVITE_CANCEL, invite_id, {"group_id": group_id, "invite_id": invite_id}) async def _do_keypair_bundle_fetch(self) -> None: """Serve the caller's encrypted keypair bundle during the handshake window.""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) return user_id = getattr(self, "_pending_sub", "") if not user_id: self._send({"type": "error", "detail": "No pending handshake"}) return kp = await bundle_store.fetch_keypair(user_id) if kp and kp.get("bundle_enc"): resp = { "type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": True, "bundle_enc": kp["bundle_enc"], } # The recovery-wrapped copy (MNP 0.14) rides along when present, so a # client holding the recovery key can re-wrap it under a new # passphrase — docs/MESHBAY_DESIGN.md §3.6. if kp.get("bundle_enc_recovery"): resp["bundle_enc_recovery"] = kp["bundle_enc_recovery"] self._send(resp) else: self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) async def _do_keypair_bundle_store(self, msg: dict) -> None: """Store an encrypted keypair bundle (user backs up their own keys on node).""" bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": "error", "detail": "Bundle store not available"}) return bundle_enc = msg.get("bundle_enc", "") if not bundle_enc: self._send({"type": "error", "detail": "Missing bundle_enc"}) return # Optional second copy wrapped under the recovery key (MNP 0.14). Omitted # by an older client and by a plain re-backup; the store keeps any # existing recovery copy when this is absent. recovery = msg.get("bundle_enc_recovery") or None await bundle_store.store_keypair(self._user_id, bundle_enc, recovery) log.info("Keypair bundle stored for user=%s (recovery=%s)", self._user_id[:8], bool(recovery)) self._audit("keypair_bundle_store") self._send({ "type": "ack", "v": MNP_VERSION, "detail": "keypair_bundle_stored", }) # ── 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. def _user_blob_refuse(self, detail: str, kind: str = "") -> None: """ Refuse, and say so in the audit log. A refusal used to be invisible here: the audit line was written only after a store *succeeded*, so a client whose writes were all being turned away looked exactly like a client that never wrote — which is how a wedged sync went unnoticed for two hours. """ self._audit("user_blob_refused", f"{kind} {detail}".strip()) self._send({"type": "error", "detail": detail}) def _user_blob_kind(self, msg: dict) -> str | None: """The validated `kind`, or None having already refused.""" kind = msg.get("kind") if not isinstance(kind, str) or not _USER_BLOB_KIND_RE.match(kind): self._user_blob_refuse("Unknown blob kind", str(kind)[:40]) return None return kind def _user_blob_store_ok(self): store = self._ctx.get("bundle_store") if not store: self._send({"type": "error", "detail": "Bundle store not available"}) return None if not self._user_id: self._send({"type": "error", "detail": "Not authenticated"}) return None return store async def _do_user_blob_store(self, msg: dict) -> None: store = self._user_blob_store_ok() if not store: return kind = self._user_blob_kind(msg) if not kind: return blob = msg.get("blob_enc") if not isinstance(blob, (bytes, bytearray)) or not blob: self._user_blob_refuse("Missing blob_enc", kind) return blob = bytes(blob) rev = msg.get("rev") if not isinstance(rev, int) or rev < 0: self._user_blob_refuse("Missing rev", kind) return limit = (USER_BLOB_MANIFEST_MAX if kind == "playlists" else USER_BLOB_BODY_MAX) if len(blob) > limit: # A stated reason, not a bare error: the client turns this into a # sentence the reader can act on ("this playlist is too large"), # and a refusal nobody can read is a support case. self._user_blob_refuse(f"Blob too large ({len(blob)} > {limit})", kind) return # What the account already uses, minus whatever this call replaces. used = await store.user_blob_total_bytes(self._user_id) existing = await store.fetch_user_blob(self._user_id, kind) if existing: used -= len(existing["blob_enc"]) if used + len(blob) > USER_BLOB_ACCOUNT_MAX: self._user_blob_refuse( f"Account blob quota exceeded " f"({used + len(blob)} > {USER_BLOB_ACCOUNT_MAX})", kind) return await store.store_user_blob(self._user_id, kind, rev, blob) self._audit("user_blob_store", f"{kind} rev={rev} bytes={len(blob)}") self._send({"type": "ack", "v": MNP_VERSION, "detail": "user_blob_stored"}) async def _do_user_blob_fetch(self, msg: dict) -> None: store = self._user_blob_store_ok() if not store: return kind = self._user_blob_kind(msg) if not kind: return row = await store.fetch_user_blob(self._user_id, kind) self._audit("user_blob_fetch", kind) self._send({ "type": MNP.USER_BLOB_RESP, "v": MNP_VERSION, "kind": kind, # A kind this account has never written is `null`, not an error: # "no playlist here yet" is the ordinary state of a fresh node and # the client must not read it as a failure. "rev": row["rev"] if row else None, "blob_enc": row["blob_enc"] if row else None, }) async def _do_user_blob_list(self) -> None: store = self._user_blob_store_ok() if not store: return blobs = await store.list_user_blobs(self._user_id) self._audit("user_blob_list", f"{len(blobs)} blobs") self._send({"type": MNP.USER_BLOB_LIST_RESP, "v": MNP_VERSION, "blobs": blobs}) async def _do_user_blob_delete(self, msg: dict) -> None: store = self._user_blob_store_ok() if not store: return kind = self._user_blob_kind(msg) if not kind: return removed = await store.delete_user_blob(self._user_id, kind) self._audit("user_blob_delete", kind) self._send({"type": "ack", "v": MNP_VERSION, "detail": "user_blob_deleted" if removed else "user_blob_absent"}) # ── Pairing and join (H3, M3) ──────────────────────────────────────────── def _join_refuse(self, reason: str, audit_detail: str = "") -> None: self._join_attempts += 1 # Node-wide window, shared across connections: reconnecting must not reset # the budget. now = time.time() failures = [t for t in self._ctx.get("join_failures", []) if now - t < JOIN_FAILURE_WINDOW] failures.append(now) self._ctx["join_failures"] = failures self._audit_join("join_refused", audit_detail or reason) self._send({ "type": MNP.JOIN_RESULT, "v": MNP_VERSION, "ok": False, "reason": reason, }) def _audit_join(self, event: str, detail: str) -> None: audit = self._ctx.get("audit_store") if not audit: return self._remote_ip = self._remote_ip or _get_remote_ip(self._pc) self._spawn(audit.log_event( user_id=self._user_id or getattr(self, "_pending_sub", "unknown"), event=event, ip=self._remote_ip, username=self._username or getattr(self, "_pending_username", ""), group_id=self._group_id or getattr(self, "_pending_group", "") or "", detail=detail, )) async def _do_join_request(self, msg: dict) -> None: """ Pin an identity, or recognise one already pinned. The client signs its own Ed25519 and X25519 keys together with the node's nonce, so the identity key vouches for the encryption key — that is what will make it safe for the node to wrap the GEK for a key that arrived over the wire instead of one fetched from the hub's directory (H3). A first pairing needs a one-time code, which the hub never sees. Afterwards the pin is the credential and a changed key is refused outright, the same rule the client applies to `pk_node` (11.5.8). """ roster = self._ctx.get("roster") if roster is None: self._send({"type": "error", "detail": "Roster not available"}) return if self._join_attempts >= MAX_JOIN_ATTEMPTS: self._send({"type": "error", "detail": "Too many attempts"}) return now = time.time() recent = [t for t in self._ctx.get("join_failures", []) if now - t < JOIN_FAILURE_WINDOW] if len(recent) >= MAX_JOIN_FAILURES_WINDOW: self._audit_join("join_throttled", f"{len(recent)} failures in window") self._send({"type": "error", "detail": "Pairing temporarily locked"}) return user_id = self._user_id or getattr(self, "_pending_sub", "") username = self._username or getattr(self, "_pending_username", "") if not user_id: self._send({"type": "error", "detail": "Handshake required"}) return pk_ed_b64 = msg.get("pk_ed25519", "") pk_x_b64 = msg.get("pk_x25519", "") code = msg.get("code", "") ts = msg.get("ts", 0) try: pk_ed_raw = base64.b64decode(pk_ed_b64) pk_x_raw = base64.b64decode(pk_x_b64) if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32: raise ValueError pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw) except Exception: self._join_refuse("invalid_keys") return if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL: self._join_refuse("stale_request") return # An empty group_id means operator pairing, which is node-wide. Anything # else must be the group this connection authenticated to — a signature # obtained for one group must not name another. group_id = msg.get("group_id", "") or "" session_group = self._group_id or getattr(self, "_pending_group", "") or "" if group_id and group_id != session_group: self._join_refuse("group_mismatch") return transcript = join_transcript( node_pk_b64=self._node_pk_b64(), group_id=group_id, user_id=user_id, pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, nonce_node=self._nonce_node, ts=ts, ) try: sig = base64.b64decode(msg.get("sig", "")) except Exception: self._join_refuse("invalid_signature_encoding") return if not self._verify_sig(pk_ed, transcript, sig): self._join_refuse("signature_invalid") return # One person may hold several devices here — a browser and a desktop # client are two keys on one account. So the question is not "is this # THE key" but "is this ONE OF this account's live devices". device = await roster.find_device(user_id, pk_ed_b64) if device and device["pk_x25519"] != pk_x_b64: # The Ed25519 key is pinned but arrives with a different encryption # key. The join transcript signs both together, so this is either a # client that regenerated half its identity or something splicing # two messages; either way the pair is not the one admitted. self._join_refuse( "key_changed", f"pinned x25519={device['pk_x25519'][:16]} presented={pk_x_b64[:16]}") return known = device if not known and await roster.list_devices(user_id): # The account is known here but this key is not one of its devices. # Not an error to shout about: it is a second browser or a new # client, and the way in is a device-add approved by a device that # is already trusted — no operator, no new invitation code. self._join_refuse( "unknown_device", f"presented={pk_ed_b64[:16]} — approve it from a device already " f"paired with this node") return if known: # This group's own row first; then the join message's group_id (empty # on the node-wide first connect); then the operator's node-wide row, # which is where an operator opening any group finds their authority. member = (await roster.get_member(session_group, user_id) or await roster.get_member(group_id, user_id) or await roster.get_member("", user_id)) if not member and self._group_join_policy(session_group) == "open": await roster.set_member( group_id=session_group, user_id=user_id, role=ROLE_MEMBER, status="active", approved_by="open-join", ) member = await roster.get_member(session_group, user_id) # A pending invite means the operator explicitly re-invited this # person — require the code even if they already have a member # row (e.g. they left and were re-invited, or were revoked then # re-invited). Without this gate a stale roster row lets them # back in without proving they received the new code. pending_invite = any( i["kind"] == KIND_ACCOUNT and i["user_id"] == user_id and i["group_id"] in (session_group, "") for i in await roster.list_invites()) # Or they bring a link for this group: somebody already pinned here # through another group, which is the ordinary case for a link, or # somebody removed from it and invited back. Only when they are not # an active member — a member opening the group leaves the link for # whoever it was meant for. active = bool(member) and member.get("status") == "active" if pending_invite or (code and not active): if not code: self._join_refuse("code_required") return invite = await roster.consume_invite(code, user_id, session_group) if not invite: self._join_refuse("code_invalid") return await roster.set_member( group_id=invite["group_id"], user_id=user_id, role=invite["role"], status="active", approved_by=invite["created_by"], ) self._audit_join( "join_pinned", f"group={invite['group_id'][:8]} role={invite['role']} " f"via={'link' if invite['kind'] == KIND_LINK else 'code'} " "(device already known)") member = (await roster.get_member(session_group, user_id) or await roster.get_member(invite["group_id"], user_id)) if not member: self._join_refuse("not_authorized_for_group") return await self._join_ok( user_id, pk_x_raw, session_group, role=member["role"] if member else "", recognised=True, ) return if not code: if self._group_join_policy(session_group) == "open": # An open-join group admits anyone the hub calls a member, so a # code would protect nothing — the hub can walk in through the # front door. Pin what turns up and say so in the audit log. await self._pin_and_admit( roster, user_id, username, pk_ed_b64, pk_x_b64, group_id=session_group, role=ROLE_MEMBER, approved_by="open-join", via="tofu") await self._join_ok(user_id, pk_x_raw, session_group, role=ROLE_MEMBER, recognised=False) return self._join_refuse("code_required") return invite = await roster.consume_invite(code, user_id, session_group) if not invite: self._join_refuse("code_invalid") return await self._pin_and_admit( # The name comes from the invitation, not from the token: the hub does # not put a username claim in a JWT, so pinning from the session alone # left the roster nameless and `member revoke ` unable to match. roster, user_id, invite["username"] or username, pk_ed_b64, pk_x_b64, group_id=invite["group_id"], role=invite["role"], approved_by=invite["created_by"], via="link" if invite["kind"] == KIND_LINK else "code") # The roster row comes from the invitation; the key comes from the # connection. An operator pairing is node-wide (empty group), but they # redeemed the code while opening a group and expect to read it — and # is_authorized() already grants an operator every group on this node. await self._join_ok(user_id, pk_x_raw, session_group or invite["group_id"], role=invite["role"], recognised=False) # ── 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. async def _do_device_request(self, msg: dict) -> None: """ A new device files itself as pending, bound to a code it displays. Served in the pre-proof window: by construction the caller holds no key this node knows, so there is nothing yet to prove. Filing is inert — nothing is admitted until an existing device countersigns. """ roster = self._ctx.get("roster") if roster is None or not self._user_id or not self._nonce_node: self._send({"type": "error", "detail": "Not ready for a device request"}) return if not self._spend_device_attempt(): return pk_ed_b64 = str(msg.get("pk_ed25519", "")) pk_x_b64 = str(msg.get("pk_x25519", "")) code_hash = str(msg.get("code_hash", "")) if not (pk_ed_b64 and pk_x_b64 and code_hash): self._send({"type": "error", "detail": "Missing device keys or code"}) return # The account must already be known here. Anti-spam rather than a # security boundary: the filing key is unpinned by construction, so this # bounds the table, not the trust. existing = await roster.list_devices(self._user_id) if not existing: self._send({"type": "error", "detail": "This account has no device on this node yet — " "an invitation code is what admits the first"}) return if len(existing) >= roster.MAX_DEVICES_PER_USER: self._send({"type": "error", "detail": f"Already {len(existing)} devices, which is the " f"limit. Revoke one first."}) return ts = int(msg.get("ts", 0)) if abs(time.time() - ts) > DEVICE_TTL: self._send({"type": "error", "detail": "Device request expired"}) return transcript = device_request_transcript( node_pk_b64=self._node_pk_b64(), user_id=self._user_id, pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, code_hash=code_hash, nonce_node=self._nonce_node, ts=ts) try: pk_ed = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64)) sig = base64.b64decode(msg.get("sig", "")) except Exception: self._send({"type": "error", "detail": "Invalid device key encoding"}) return if not self._verify_sig(pk_ed, transcript, sig): # Proof of possession, and nothing more: this says the caller holds # the keys, never that they belong to this account. self._send({"type": "error", "detail": "Device signature invalid"}) return ttl = int(self._ctx.get("device_request_ttl") or 3600) expires = await roster.file_device_request( user_id=self._user_id, username=self._username or "", pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, code_hash=code_hash, ttl=ttl) self._audit("device_request", f"{pk_ed_b64[:16]}") log.info("Device request filed for %s (%s)", self._user_id[:8], pk_ed_b64[:16]) self._send({"type": MNP.DEVICE_REQUEST_ACK, "v": MNP_VERSION, "expires_at": expires}) async def _do_device_lookup(self, msg: dict) -> None: """ List this account's pending device requests, each with its code hash. **The node never learns the code**, which is what makes it unable to substitute a key. It answers with candidates; the approver recomputes `sha256(code ‖ keys)` for each and keeps the one that matches. A node offering fabricated keys would have to produce a hash matching `sha256(code ‖ fabricated)` — and it does not know the code. An earlier version of this took the hash from the client and looked the request up by it. That is circular: the client cannot compute the hash without already knowing the keys it is asking about. """ roster = self._ctx.get("roster") if roster is None or not self._user_id: self._send({"type": "error", "detail": "Roster not available"}) return pending = await roster.list_device_requests(self._user_id) self._send({ "type": MNP.DEVICE_LOOKUP_RESULT, "v": MNP_VERSION, "requests": [ {"pk_ed25519": r["pk_ed25519"], "pk_x25519": r["pk_x25519"], "code_hash": r["code_hash"], "created_at": r["created_at"]} for r in pending ], }) async def _do_device_add(self, msg: dict) -> None: """ Admit a device, countersigned by one this node already pinned. The whole control is in `_verify_device_signer`: the signature must verify against a **live device of this same account**. The hub holds no user keys and so cannot produce one. """ roster = self._ctx.get("roster") if roster is None or not self._user_id or not self._nonce_node: self._send({"type": "error", "detail": "Not ready to add a device"}) return if not self._spend_device_attempt(): return pk_ed_b64 = str(msg.get("pk_ed25519", "")) pk_x_b64 = str(msg.get("pk_x25519", "")) ts = int(msg.get("ts", 0)) if not (pk_ed_b64 and pk_x_b64): self._send({"type": "error", "detail": "Missing device keys"}) return if abs(time.time() - ts) > DEVICE_TTL: self._send({"type": "error", "detail": "Approval expired"}) return transcript = device_add_transcript( node_pk_b64=self._node_pk_b64(), user_id=self._user_id, pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, nonce_node=self._nonce_node, ts=ts) signer = await self._verify_device_signer(roster, transcript, msg.get("sig", "")) if signer is None: self._audit("device_add_refused", pk_ed_b64[:16]) self._send({"type": "error", "detail": "Not signed by a device already paired here"}) return devices = await roster.list_devices(self._user_id) if len(devices) >= roster.MAX_DEVICES_PER_USER: self._send({"type": "error", "detail": "Device limit reached"}) return # Spend the request. Single use: an approval cannot be replayed, and a # code that was used is gone whatever else happens next. code_hash = str(msg.get("code_hash", "")) if code_hash and not await roster.take_device_request( code_hash, self._user_id): self._send({"type": "error", "detail": "That request is no longer pending"}) return # The countersignature is **kept**, with the two fields needed to rebuild # what it signed. Until 2026-09-07 it was verified here and thrown away, # leaving only `added_by_pk` — which says *which* key approved and # proves nothing to anyone else. `device_add_transcript` binds # `nonce_node`, this connection's handshake nonce, so a stored signature # without it is still unverifiable; that is why all three go in. # # This is what lets another member check for themselves that this device # belongs to an account whose earlier device they have already pinned, # instead of taking the node's word (Tier 2, docs/MESHBAY_DESIGN.md §3.3). await roster.pin_identity( user_id=self._user_id, username=self._username or "", pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via="device", label=str(msg.get("label", ""))[:64], added_by_pk=signer, add_sig=str(msg.get("sig", "")), add_nonce=base64.b64encode(self._nonce_node).decode(), add_ts=ts) self._audit("device_added", f"{pk_ed_b64[:16]} by {signer[:16]}") log.info("Device added for %s: %s (approved by %s)", self._user_id[:8], pk_ed_b64[:16], signer[:16]) self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION, "pk_ed25519": pk_ed_b64}) async def _do_device_hello(self, msg: dict) -> None: """ Learn which of this account's devices is on this connection. The handshake authenticates a *group membership* (the GEK-HMAC) and an *account* (the hub's token). It has never authenticated a device, and while one account meant one key that was the same statement. It stopped being so on 2026-08-18, and `_load_pinned_pk` — which resolves the account's oldest live device — has been standing in for the real answer ever since, including as the recorded uploader of every file. What is checked, in order: the key is a live device *of this account* in the node's own roster (never a token claim — that is `docs/MESHBAY_DESIGN.md` §3.2's rule), the timestamp is fresh, and the signature verifies over a transcript naming this node, this group and this connection's nonce. A key that is merely well-formed proves nothing. Idempotent for the same key, refused for a different one: a connection does not get to change device half way through, which would let one session's uploads be attributed to two. """ roster = self._ctx.get("roster") if roster is None or not self._user_id: self._send({"type": "error", "detail": "Roster not available"}) return if not self._spend_device_attempt(): return pk_ed_b64 = str(msg.get("pk_ed25519", "")) ts = int(msg.get("ts", 0) or 0) if not pk_ed_b64: self._send({"type": "error", "detail": "Missing device key"}) return if self._device_confirmed and pk_ed_b64 != self._pinned_pk: self._send({"type": "error", "detail": "This connection is already another device"}) return if abs(time.time() - ts) > DEVICE_TTL: self._send({"type": "error", "detail": "Stale device_hello"}) return device = await roster.find_device(self._user_id, pk_ed_b64) if device is None: self._audit("device_hello_refused", pk_ed_b64[:16]) self._send({"type": "error", "detail": "Not a device paired here"}) return transcript = device_hello_transcript( node_pk_b64=self._node_pk_b64(), group_id=self._group_id or "", user_id=self._user_id, pk_ed25519_b64=pk_ed_b64, nonce_node=self._nonce_node, ts=ts) try: pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64)) except Exception: self._send({"type": "error", "detail": "Unreadable device key"}) return try: sig = base64.b64decode(msg.get("sig", "")) except Exception: sig = b"" if not self._verify_sig(pk, transcript, sig): self._audit("device_hello_refused", pk_ed_b64[:16]) self._send({"type": "error", "detail": "Signature verification failed"}) return self._pinned_pk = pk_ed_b64 self._device_confirmed = True log.info("Device identified on connection: user=%s device=%s", self._user_id[:8], pk_ed_b64[:16]) self._send({"type": MNP.DEVICE_HELLO_ACK, "v": MNP_VERSION, "pk_ed25519": pk_ed_b64}) async def _do_device_list(self, msg: dict) -> None: """This account's devices. Anyone may read their own, nobody else's.""" roster = self._ctx.get("roster") if roster is None or not self._user_id: self._send({"type": "error", "detail": "Roster not available"}) return devices = await roster.list_devices(self._user_id) pending = await roster.pending_device_requests(self._user_id) self._send({ "type": MNP.DEVICE_LIST_RESULT, "v": MNP_VERSION, "pending": pending, "devices": [ {"pk_ed25519": d["pk_ed25519"], "label": d.get("label", ""), "pinned_at": d["pinned_at"], "pinned_via": d["pinned_via"], "added_by_pk": d.get("added_by_pk", ""), "is_this_one": d["pk_ed25519"] == self._pinned_pk} for d in devices ], }) async def _new_chat_epoch(self, group_id: str, reason: str) -> None: """ Open a chat epoch because the set of devices that may read future messages just shrank. Called on every removal — a member, a device, an unpin — and on group key rotation, because the operator rotates precisely when someone has left. It is the exact counterpart of "still rotate the GEK, the ex-member holds the current one": revocation stops the node handing over the *next* key, and nothing else takes the current one away. Best effort by design: a failure here must never turn a successful revocation into a refused one — the revocation is the control, and this is the follow-through. It is logged loudly instead, because an operator who removed someone needs to know if the chat key did not move. """ if not group_id: return try: result = await self._run_op(ops.open_chat_epoch, group_id) except Exception as e: log.error("chat: could not open a new epoch for group %s after " "%s (%s) — the removed party still holds the current " "chat key", group_id[:8], reason, e) self._audit("chat_epoch_failed", reason) return self._audit("chat_epoch", f"{reason}:{result['epoch']}") # Everyone still connected picks the new key up without reconnecting. for session in list( (self._ctx.get("groups") or {}).get(group_id, {}) .get("_peers", {}).values()): try: session._send({"type": MNP.CHAT_EPOCH_ACK, "v": MNP_VERSION, "epoch": result["epoch"]}) except Exception: pass async def _do_device_revoke(self, msg: dict) -> None: """ Retire one of this account's devices — a lost laptop. Countersigned like an addition, by a live device of the same account. The last one cannot go: an account with no device on this node can only return through an operator's invitation code, and doing that to yourself by accident is not a mistake worth allowing. """ roster = self._ctx.get("roster") if roster is None or not self._user_id or not self._nonce_node: self._send({"type": "error", "detail": "Not ready"}) return if not self._spend_device_attempt(): return target = str(msg.get("pk_ed25519", "")) ts = int(msg.get("ts", 0)) if not target: self._send({"type": "error", "detail": "Missing device key"}) return if abs(time.time() - ts) > DEVICE_TTL: self._send({"type": "error", "detail": "Request expired"}) return victim = await roster.find_device(self._user_id, target) if victim is None: self._send({"type": "error", "detail": "No such device"}) return transcript = device_add_transcript( node_pk_b64=self._node_pk_b64(), user_id=self._user_id, pk_ed25519_b64=target, pk_x25519_b64=victim["pk_x25519"], nonce_node=self._nonce_node, ts=ts) signer = await self._verify_device_signer(roster, transcript, msg.get("sig", "")) if signer is None: self._send({"type": "error", "detail": "Not signed by a device already paired here"}) return if len(await roster.list_devices(self._user_id)) <= 1: self._send({"type": "error", "detail": "This is your only device here — removing it " "would need an operator code to come back"}) return await roster.revoke_device(self._user_id, target) # A revoked device holds every chat key it ever received — a lost laptop # reads the group's chat until the epoch moves. await self._new_chat_epoch(self._group_id or "", "device_revoke") self._audit("device_revoked", f"{target[:16]} by {signer[:16]}") log.info("Device revoked for %s: %s", self._user_id[:8], target[:16]) self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION, "revoked": target}) async def _verify_device_signer(self, roster, transcript: bytes, sig_b64: str) -> str | None: """ The pinned key that signed this, or None. Every live device of the account is tried, because any of them may approve. A revoked one is not in the list — that is the point of marking rather than deleting: a lost laptop must stop being able to admit its replacement. """ try: sig = base64.b64decode(sig_b64) except Exception: return None for device in await roster.list_devices(self._user_id): try: pk = Ed25519PublicKey.from_public_bytes( base64.b64decode(device["pk_ed25519"])) except Exception: continue if self._verify_sig(pk, transcript, sig): return device["pk_ed25519"] return None def _spend_device_attempt(self) -> bool: """ Bound guessing on this connection, as the join path does. A code is 40 bits, single use and bound to the keys it names, so this is depth rather than the control — but an unbounded loop over the lookup is still a free oracle, and a burst of failures belongs in the audit log. """ self._device_attempts = getattr(self, "_device_attempts", 0) + 1 if self._device_attempts > 5: self._audit("device_attempts_exceeded", str(self._device_attempts)) self._send({"type": "error", "detail": "Too many device attempts on this connection"}) return False return True def _group_join_policy(self, group_id: str) -> str: """ Admission policy for a group, read from the node's own configuration. Never from the hub: a hub that could declare a group open would be handed the key to it (docs/MESHBAY_DESIGN.md §3.4). """ gctx = (self._ctx.get("groups") or {}).get(group_id) or {} return gctx.get("join_policy", "invite") async def _pin_and_admit( self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str, *, group_id: str, role: str, approved_by: str, via: str, ) -> None: await roster.pin_identity( user_id=user_id, username=username, pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via, ) await roster.set_member( group_id=group_id, user_id=user_id, role=role, status="active", approved_by=approved_by, ) if role == ROLE_OPERATOR: self._ctx["has_admin_authority"] = True log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role) self._audit_join("join_pinned", f"role={role} via={via}") async def _join_ok( self, user_id: str, pk_x_raw: bytes, group_id: str, *, role: str, recognised: bool, ) -> None: """ Answer a join, wrapping the group key for the key the caller just proved. This is the H3 fix. The inviter used to fetch the invitee's public key from the hub and wrap the GEK for whatever came back, so a hub that answered with its own key was handed the group key by an honest member following the protocol exactly. The node now wraps for a key that arrived from its owner over an authenticated channel, bound to a pinned identity. """ reply = { "type": MNP.JOIN_RESULT, "v": MNP_VERSION, "ok": True, "recognised": recognised, "role": role, } roster = self._ctx["roster"] if group_id and not await roster.is_authorized(group_id, user_id): # Pinned on this node, but not admitted to this group. Hub membership # alone must not produce a key. reply["gek"] = False reply["reason"] = "not_authorized_for_group" self._send(reply) self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized") return gctx = (self._ctx.get("groups") or {}).get(group_id) or {} gek = gctx.get("gek") if not gek: reply["gek"] = False reply["reason"] = "no_gek" self._send(reply) return bundle = wrap_gek_aes(gek, pk_x_raw) reply["gek"] = True reply["pk_eph_b64"] = bundle["pk_eph_b64"] reply["nonce_b64"] = bundle["nonce_b64"] reply["wrapped_b64"] = bundle["wrapped_b64"] self._send(reply) self._audit_join("gek_wrapped", f"group={group_id[:8]}") async def _do_dir_create(self, msg: dict) -> None: """ Create a directory, for any member of the group. Same confinement as an upload: every segment passes the name allowlist and the result must resolve under the shared root. Making a directory is not a privileged act — a member who can add a file can organise where it goes — but it writes to the operator's disk, so it is audited like one. """ ctx = self._group_ctx() roots: RootSet | None = ctx.get("roots") if not roots: self._send({"type": "error", "detail": "No shared directory"}) return name = str(msg.get("name", "")).strip() if not SAFE_UPLOAD_NAME.match(name): self._send({"type": "error", "detail": "Invalid directory name"}) return # The virtual root is not a directory on anyone's disk, so a member # cannot create one there — that would be adding a root, which is the # operator's configuration and not a file operation. parent_rel = (msg.get("dir") or "").strip("/") if not parent_rel: self._send({"type": "error", "detail": "Choose a folder to create this in"}) return # Read-only means read-only, and creating a folder writes to the # operator's disk. `_do_file_upload` gained this check with the RO/RW # model and this one did not — so a member could not add a file to a # published library but could still leave empty directories in it. owner = roots.split(parent_rel) if owner is None: self._send({"type": "error", "detail": "Invalid directory"}) return parent_root, _tail = owner if not parent_root.writable: self._send({"type": "error", "detail": f"Directory '{parent_root.name}' is read-only", "code": "root_read_only"}) self._audit("dir_create_refused", parent_rel[:64]) return if not parent_root.available: self._send({"type": "error", "detail": f"Directory '{parent_root.name}' is " f"currently unavailable", "code": "root_unavailable"}) return parent = await off_disk(roots, safe_subdir, roots, parent_rel) if parent is None or not await off_disk(roots, parent.is_dir): self._send({"type": "error", "detail": "Invalid directory"}) return target = await off_disk(roots, safe_subdir, roots, f"{parent_rel}/{name}") if target is None: self._send({"type": "error", "detail": "Invalid directory"}) return refusal = await off_disk(roots, _mkdir_if_absent, target) if refusal is not None: self._send({"type": "error", "detail": refusal}) return virtual = roots.virtual_of(target) or f"{parent_rel}/{name}" log.info("Directory created by %s: %s", self._user_id[:8], virtual) self._audit("dir_create", virtual) self._send({ "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION, "dir": virtual, }) @staticmethod def _names_a_root(roots: RootSet, rel: str) -> bool: """True when `rel` is a bare root name rather than something inside one.""" found = roots.split(rel or "") return found is not None and not found[1] async def _do_dir_delete(self, msg: dict) -> None: """ Remove an empty directory, for the node operator. Creating one is not privileged — a member who can add a file may organise where it goes — but removing one is: it acts on a name other members are using, and on the operator's disk. Empty is the whole safety property here. Nothing recursive: refusing a directory with anything in it means this can never destroy content, whatever the caller intended, so the operator deletes the files first and sees what they are losing. """ ctx = self._group_ctx() roots: RootSet | None = ctx.get("roots") if not roots: self._send({"type": "error", "detail": "No shared directory"}) return rel = (msg.get("dir") or "").strip("/") target = await off_disk(roots, safe_subdir, roots, rel) # A root itself is not deletable here: removing one is a configuration # change, and doing it through a file operation would leave the group # config naming a directory nobody can reach. if target is None or self._names_a_root(roots, rel): self._send({"type": "error", "detail": "Invalid directory"}) return if not await off_disk(roots, target.is_dir): self._send({"type": "error", "detail": "Not a directory"}) return if not await off_disk(roots, _is_empty_dir, target): self._send({"type": "error", "detail": "Directory is not empty"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for deletion"}) return self._issue_admin_challenge( OP_DIR_DELETE, roots.virtual_of(target) or rel) async def _admin_exec_dir_delete( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: rel = pending["subject"] ctx = self._group_ctx() roots: RootSet | None = ctx.get("roots") target = await off_disk(roots, safe_subdir, roots, rel) if roots else None if (target is None or self._names_a_root(roots, rel) or not await off_disk(roots, target.is_dir)): self._send({"type": "error", "detail": "Not a directory"}) return # Operator only. A file has an uploader who may remove their own; a # directory has none, so there is no second key to accept here. if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"dir_delete:{rel}") return # Checked again after the signature: the emptiness test that let this # through happened before a round trip to the operator's browser, and a # file could have landed in the meantime. if not await off_disk(roots, _rmdir_if_empty, target): self._send({"type": "error", "detail": "Directory is not empty"}) return log.info("Directory removed by %s: %s", self._user_id[:8], rel) self._audit("dir_delete", rel) self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel}) def _do_member_revoke(self, msg: dict) -> None: """ Stop serving the group key to someone, at the operator's request. The same authority as an invite, and the same reason: the roster decides who this node serves, so only a key the node pinned as an operator may change it. Membership on the hub is not consulted — the hub can remove someone from a group, and that stops them reaching the node at all, but it cannot make the node forget them. """ user_id = str(msg.get("user_id", "")).strip() if not user_id: self._send({"type": "error", "detail": "Missing user_id"}) return if user_id == self._user_id: # Removing yourself from your own node is not a member operation; # it would leave the group with nobody able to invite. self._send({"type": "error", "detail": "Cannot revoke yourself"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id) def _do_gek_rotate(self, msg: dict) -> None: """ Ask for a new group key. Operator only, and signed. This is what actually removes a revoked member's access: revocation stops the node serving the *next* key, and they still hold the current one. The node generates the replacement itself — nothing arriving here contributes key material, which is what the C5b rule is about. """ 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 def _do_tmdb_config(self, msg: dict) -> None: """ Optionally set (or clear) a custom TMDB API token, and optionally set the language TMDB is queried in (e.g. "fr-FR") — one for the whole node, since both are one operator's shared credential/cache, not a per-group concern (see _do_tmdb_enabled for the per-group on/off switch). Signed like the rest: this changes outbound third-party network traffic the node did not have before the Videos app (docs/MESHBAY_DESIGN.md §9.7, §6.5) — an unsigned change would let any member alter egress the operator never agreed to. """ token = msg.get("token") if token is not None and not isinstance(token, str): self._send({"type": "error", "detail": "Invalid 'token'"}) return language = msg.get("language") if language is not None and not isinstance(language, str): self._send({"type": "error", "detail": "Invalid 'language'"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return # The subject is the signed, audited, human-shown string — it must # never contain the token itself (it would end up in the audit log # in plaintext). The actual token travels only in `payload`, which # is node-side context, never re-sent or re-verified from the wire. # The language is not a secret, so it travels in the subject itself. subject = f"custom_token={'yes' if token else 'no'},language={language or 'default'}" self._issue_admin_challenge( OP_TMDB_CONFIG, subject, payload={"token": token, "language": language}, group_id="") async def _admin_exec_tmdb_config( 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"tmdb_config:{pending['subject']}") return p = pending.get("payload") or {} try: result = await self._run_op( ops.set_tmdb_config, p.get("token"), p.get("language")) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("tmdb_config", pending["subject"]) # Node-wide setting: every connected peer in every group is told, not # just this group's peers (unlike apps_enabled/the root ops/the # per-group tmdb_enabled below). notice = { "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION, "token_customized": result["token_customized"], "language": result["language"], } for gctx in self._ctx.get("groups", {}).values(): for session in list(gctx.get("_peers", {}).values()): try: session._send(notice) except Exception: pass def _do_tmdb_enabled(self, msg: dict) -> None: """ Whether TMDB lookups run for this group at all. Per-group, unlike tmdb_config's token/language — see ops.set_tmdb_enabled. Signed like app_directories: it decides whether this group's members' Videos tab ever makes outbound TMDB traffic. """ enabled = msg.get("enabled") if not isinstance(enabled, bool): self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_TMDB_ENABLED, str(enabled)) async def _admin_exec_tmdb_enabled( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: enabled = pending["subject"] == "True" if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"tmdb_enabled:{pending['subject']}") return try: await self._run_op(ops.set_tmdb_enabled, self._group_id or "", enabled) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("tmdb_enabled", pending["subject"]) notice = {"type": MNP.TMDB_ENABLED_ACK, "v": MNP_VERSION, "enabled": enabled} 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_chat_directory(self, msg: dict) -> None: """ Where chat attachments are written. Unlike every other app directory this one is a destination, so it has to be on a read-write root — checked by `ops.set_chat_directory` after the signature, which is where the refusal actually lives. """ path = msg.get("path") if not isinstance(path, str): self._send({"type": "error", "detail": "Missing or invalid 'path'"}) return path = path.strip("/") if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_CHAT_DIRECTORY, path) async def _admin_exec_chat_directory( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: path = 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"chat_directory:{path}") return try: await self._run_op( ops.set_chat_directory, self._group_id or "", path) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("chat_directory", path) self._broadcast_to_group( {"type": MNP.CHAT_DIRECTORY_ACK, "v": MNP_VERSION, "path": path}) def _do_chat_link_preview(self, msg: dict) -> None: """ Whether the node fetches a page's title and image when a member posts a link — outbound traffic on the operator's connection, from a message they did not write, so it is signed like everything else that decides what leaves this machine. """ enabled = msg.get("enabled") if not isinstance(enabled, bool): self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_CHAT_LINK_PREVIEW, "on" if enabled else "off") async def _admin_exec_chat_link_preview( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: enabled = 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"chat_link_preview:{pending['subject']}") return try: await self._run_op( ops.set_chat_link_preview, self._group_id or "", enabled) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("chat_link_preview", pending["subject"]) self._broadcast_to_group({"type": MNP.CHAT_LINK_PREVIEW_ACK, "v": MNP_VERSION, "enabled": enabled}) 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}) def _do_chat_epoch(self, msg: dict) -> None: """ Open a new chat epoch by hand. Operator only, and signed. There is no switch to turn chat encryption on: MNP 2.0 has no plaintext chat to fall back to. What an operator may want to do deliberately is move the key on — the same instruction as `gek_rotate`, and signed for the same reason. The removals that matter (member revoke, member unpin, device revoke, `gek_rotate`) already open one by themselves. """ 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_CHAT_EPOCH, group_id, group_id=group_id) async def _admin_exec_chat_epoch( 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"chat_epoch:{pending['subject'][:8]}") return try: result = await self._run_op(ops.open_chat_epoch, pending["subject"]) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("chat_epoch", f"manual:{result['epoch']}") self._broadcast_to_group({"type": MNP.CHAT_EPOCH_ACK, "v": MNP_VERSION, "epoch": result["epoch"]}) 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}) async def _do_chat_keys_req(self, msg: dict) -> None: """ Hand this member every chat epoch key the group has, sealed. Sealed under a group-derived subkey rather than sent in clear: the same reasoning as the index and the handshake ack, and one step stronger here, because the payload *is* key material. A peer that has completed the handshake holds the group key and can open it; anything short of that gets a ciphertext. **Every** live epoch, not just the current one, which is what keeps the history readable to a member who joined after it was written and to a device linked this morning. Whether a new member should receive the back catalogue at all is a policy question with a per-group answer; the shape is here so that answer can be given without a wire change. """ gctx = self._group_ctx() gek = gctx.get("gek") if not gek: self._send({"type": "error", "detail": "Group encryption not initialized"}) return try: keys = await self._run_op(ops.chat_epoch_keys, self._group_id or "") except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return payload = {"epochs": [{"epoch": k["epoch"], "key": k["key"]} for k in keys], "current": keys[-1]["epoch"] if keys else 0} sealed = seal(gek, PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP, self._group_id or "", payload) self._send({"type": MNP.CHAT_KEYS_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 def _do_musicbrainz_enabled(self, msg: dict) -> None: """ Whether MusicBrainz lookups run for this group at all. Per-group from the start (docs/MESHBAY_DESIGN.md §9.8) — signed like tmdb_enabled: it decides whether this group's members' Music tab ever makes outbound MusicBrainz traffic. """ enabled = msg.get("enabled") if not isinstance(enabled, bool): self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_MUSICBRAINZ_ENABLED, str(enabled)) async def _admin_exec_musicbrainz_enabled( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: enabled = pending["subject"] == "True" if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"musicbrainz_enabled:{pending['subject']}") return try: await self._run_op(ops.set_musicbrainz_enabled, self._group_id or "", enabled) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("musicbrainz_enabled", pending["subject"]) notice = {"type": MNP.MUSICBRAINZ_ENABLED_ACK, "v": MNP_VERSION, "enabled": enabled} 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", ""), }) async def _do_keypair_bundle_delete(self) -> None: """ Withdraw our own key backup from this node. Only ever our own: the user_id comes from the authenticated session, never from the message. Someone who does not want a second browser should not be leaving a PBKDF2-protected blob on every node they have ever joined (C4), and turning the setting off has to remove what is already there — not just stop adding to it. """ bundle_store = self._ctx.get("bundle_store") if not bundle_store: self._send({"type": "error", "detail": "Bundle store not available"}) return removed = await bundle_store.delete_keypair(self._user_id) if removed: log.info("Keypair bundle withdrawn by user=%s", self._user_id[:8]) self._audit("keypair_bundle_delete") self._send({"type": "ack", "v": MNP_VERSION, "detail": "keypair_bundle_deleted", "removed": removed}) def _audit_pre_proof_fetch(self, mtype: str) -> None: """Record bundle access made before the GEK proof (C4).""" audit = self._ctx.get("audit_store") if not audit: return self._remote_ip = self._remote_ip or _get_remote_ip(self._pc) self._spawn(audit.log_event( user_id=getattr(self, "_pending_sub", "unknown"), event="pre_proof_fetch", ip=self._remote_ip, username=self._username or getattr(self, "_pending_username", ""), group_id=getattr(self, "_pending_group", "") or "", detail=mtype, )) def _audit_auth_failed(self, group_id: str, reason: str) -> None: audit = self._ctx.get("audit_store") if audit: self._remote_ip = _get_remote_ip(self._pc) self._spawn(audit.log_event( user_id="unknown", event="auth_failed", ip=self._remote_ip, group_id=group_id, detail=reason, )) def _spawn(self, coro) -> asyncio.Task: """Run a coroutine in the background and hold on to it. The reference is what keeps the task alive; the done callback is what stops the set growing. Anything that owns a resource for its lifetime — a transcode slot, an ffmpeg process — must go through here rather than `asyncio.ensure_future`. """ task = asyncio.ensure_future(coro) self._tasks.add(task) 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 _max_upload_bytes(self) -> int: """The per-file upload ceiling this node is running with, in bytes. Read from the transport context rather than captured once, for the same reason the transfer pools are refreshed there: the operator can change it from the Node page or the CLI while an upload is running, and a ceiling that only applies after a restart is not the one they were shown. `None` means they have said nothing and the default stands. """ gb = self._ctx.get("max_upload_gb") if not gb: return MAX_UPLOAD_BYTES return max(1, int(float(gb) * GB_BYTES)) def _slots(self) -> "TransferSlots": """The node's transfer pools, shared across every peer and every group. On the transport context, not the session: it counts the node's transfers, not one browser's. Built once, for the same reason the transcode semaphore is — rebuilding it per call would hand every caller its own budget and cap nothing at all. """ slots = self._ctx.get("_transfer_slots") if slots is None: slots = TransferSlots() n = self._ctx.get("max_concurrent_downloads") u = self._ctx.get("max_concurrent_uploads") if n: slots.caps[transfers_mod.DOWNLOAD] = int(n) if u: slots.caps[transfers_mod.UPLOAD] = int(u) self._ctx["_transfer_slots"] = slots log.info("transfer: %s", slots.summary()) # Refreshed from the group context rather than only at construction: a # node serves several groups, each with its own signed cap, and the # pools are built by whichever group happens to transfer first. limits = self._group_ctx().get("transfer_limits") if limits and self._group_id: slots.group_limits[self._group_id] = dict(limits) return slots def _lease_of(self, tr) -> str: """What the `tr` on a request actually is, from the node's own record. `tr` is drawn by the client (§5.5) and arrives on every chunk request and every upload chunk. It was read as a boolean: *present* meant "this is a leased transfer", and nothing asked whether the node had ever granted such a lease — so any non-empty string skipped the leaseless ceiling and every cap the operator set. `touch()` has always answered exactly this question (`False` if it is not granted) and its answer was discarded. Three outcomes, because they deserve different treatment: - **`granted`** — a live lease of *this session*, and the transfer is under the caps it was granted against. The session is checked as well as the id: a lease belongs to a connection, and touching somebody else's would refresh their idle timer. - **`queued`** — the node has this lease and has not granted it. The client is jumping its own queue; refused, and no shipped client does it (the transfer store awaits the grant before it reads a byte). - **`none`** — the node has no such lease. Deliberately *not* a refusal: it is what a reconnect looks like from here, where the session's leases died with the old connection and the client is re-opening them, and it is what a client that never asked looks like. Both are then bounded by the leaseless ceiling instead — which is the residual §5.5 already states: a client that lies gets that bound's worth of files at a time, not the whole library. """ slots = self._ctx.get("_transfer_slots") if slots is None: return LEASE_NONE lease = slots.leases.get(tr) if lease is None or lease.session_key != self._registry_key: return LEASE_NONE if lease.state != "granted": return LEASE_QUEUED slots.touch(tr) return LEASE_GRANTED def _note_unleased(self, tr: str) -> None: """Say once that this connection transferred outside its lease. Once per session, not per chunk: the interesting fact is that it happened, and a per-chunk line would bury it under itself. The bound is the leaseless ceiling either way; this is what makes the residual visible to the operator rather than merely stated in a document. """ if self._unleased_noted: return self._unleased_noted = True log.info("transfer: %s sent chunk requests under an unknown lease %s " "— bounded by the leaseless ceiling", (self._user_id or "?")[:8], str(tr)[:8]) self._audit("transfer_unleased", str(tr)[:16]) def _transfer_state_msg(self, lease, state: str, reason: str = "") -> dict: slots = self._slots() out = { "type": MNP.TRANSFER_STATE, "v": MNP_VERSION, "tr": lease.tr, "state": state, "kind": lease.kind, "used": slots.member_in_use(lease.kind, lease.member), # `member_cap`, not the node-wide default: this group's own limit is # what `_has_room` enforces and what the handshake ack announces, so # reading the default here would have the widget contradicting both # — "1 of 2" in a group where the operator signed 5, or two slots # offered in a group limited to one. "cap": slots.member_cap(lease.kind, lease.member), "node_used": slots.in_use(lease.kind), "node_cap": slots.caps.get(lease.kind, transfers_mod.DEFAULT_MAX_CONCURRENT), } if state == "queued": out["ahead"] = slots.ahead_of(lease) if reason: out["reason"] = reason return out def _notify_transfer(self, lease, state: str, reason: str = "") -> None: """Push a lease's state to the connection that owns it. By session key, never by account: a lease belongs to one connection, and telling a member's other device that *its* transfer was granted is how a queue starts lying. """ session = self._peer_registry().get(lease.session_key) for candidate in ([session] if session else self._sessions_everywhere(lease.session_key)): try: candidate._send(self._transfer_state_msg(lease, state, reason)) except Exception: pass def _sessions_everywhere(self, session_key: str) -> list["WebRTCPeerSession"]: """The session with this key, whichever group it is in. `_peer_registry` is per group (finding H1) and the pools are node-wide, so a slot freed in one group can grant one in another: the peer to tell is not necessarily in this session's own registry. """ groups = self._ctx.get("groups") registries = ([g.get("_peers", {}) for g in groups.values()] if groups else [self._ctx.get("_peers", {})]) return [reg[session_key] for reg in registries if session_key in reg] def _announce(self, granted: list, ended: list | None = None) -> None: for lease, reason in (ended or []): self._notify_transfer( lease, "queued" if lease.state == "queued" else "closed", reason) for lease in granted: self._notify_transfer(lease, "granted") def _do_transfer_open(self, msg: dict) -> None: tr = str(msg.get("tr") or "")[:64] kind = str(msg.get("kind") or transfers_mod.DOWNLOAD) if not tr: self._send({"type": "error", "detail": "Missing transfer id", "code": "bad_transfer_id"}) return slots = self._slots() try: nbytes = int(msg.get("bytes") or 0) chunks = int(msg.get("chunks") or 0) except (TypeError, ValueError): self._send({"type": "error", "detail": "Invalid transfer size", "code": "bad_transfer_size", "tr": tr}) return lease, err = slots.open( tr=tr, kind=kind, session_key=self._registry_key, user_id=self._user_id or "", group_id=self._group_id or "", bytes=nbytes, chunks=chunks) if err: self._send({"type": "error", "detail": err, "code": err, "tr": tr}) return self._send(self._transfer_state_msg(lease, lease.state)) # INFO, not DEBUG. This is the line that answers "did the client ever # ask for a slot, and what was it told" when somebody reports a stuck # transfer — and a whole afternoon was spent concluding "the node saw # nothing" from a journal that could not have shown it. One line per # transfer is not a volume problem; turning the root logger up to DEBUG # to see it is, because aiortc logs every SCTP chunk. log.info("transfer: open %s %s -> %s (%s)", kind, tr[:8], lease.state, slots.summary()) self._ensure_transfer_sweeper() def _do_transfer_close(self, msg: dict) -> None: tr = str(msg.get("tr") or "")[:64] reason = str(msg.get("reason") or transfers_mod.REASON_DONE)[:32] slots = self._slots() held = slots.leases.get(tr) if held is not None and held.session_key != self._registry_key: # Closing somebody else's transfer would be a denial of service one # random id away. self._send({"type": "error", "detail": "not_your_transfer", "code": "not_your_transfer", "tr": tr}) return lease, granted = slots.close(tr, reason) if lease is not None: self._send(self._transfer_state_msg(lease, "closed", reason)) self._announce(granted) def _release_transfers(self) -> None: """Give back everything this connection held. Called from teardown.""" slots = self._ctx.get("_transfer_slots") if slots is None: return gone, granted = slots.release_session(self._registry_key) if gone: log.info("transfer: session gone, released %d (%s)", len(gone), slots.summary()) self._announce(granted) def _ensure_transfer_sweeper(self) -> None: """Start the maintenance task, once, and only while it has work. It reclaims what a session teardown cannot see — a grant nobody took up, a transfer that went quiet — and logs the one line that answers "was this peer ever in a queue" when somebody reports a stuck transfer. It stops when the last lease goes, so an idle node runs no timer. """ running = self._ctx.get("_transfer_sweeper") if running is not None and not running.done(): return ctx = self._ctx async def _sweep_loop() -> None: while True: await asyncio.sleep(TRANSFER_SWEEP_SECS) slots = ctx.get("_transfer_slots") if slots is None or not slots.leases: return ended, granted = slots.sweep() for lease, reason in ended: log.info("transfer: reclaimed %s (%s)", lease.tr[:8], reason) self._announce(granted, ended) log.debug("transfer: %s", slots.summary()) # Deliberately NOT `self._spawn`, which is otherwise the only way to # start a task here. `_spawn` ties a task to *this session's* set, and # `shutdown_tasks` cancels those when the peer leaves — so the sweeper # would die with whichever connection happened to open the first # transfer, and every other peer's abandoned lease would then never be # reclaimed. It belongs to the node, so the strong reference that keeps # it off the garbage collector lives on the transport context; the rule # `_spawn` exists for (asyncio holds only a weak reference) is satisfied # by that reference, not by which set it is in. task = asyncio.ensure_future(_sweep_loop()) ctx["_transfer_sweeper"] = task def _finished(done: asyncio.Task) -> None: if ctx.get("_transfer_sweeper") is done: ctx["_transfer_sweeper"] = None if not done.cancelled() and done.exception() is not None: # Nothing awaits this task, so an exception here would otherwise # be swallowed and idle leases would silently stop being # reclaimed — the failure mode is a node that fills up over days. log.error("transfer: sweeper died: %r", done.exception()) task.add_done_callback(_finished) 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_index_sync(self) -> None: ctx = self._group_ctx() self._send(index_sync_message(ctx["index"], ctx.get("roots"))) async def _try_serve_thumbnail( self, thumb_hash: str, chunk_index: int, gek: bytes | None, ) -> dict | None: """ docs/MESHBAY_DESIGN.md §6.5: a thumbnail is served through the same chunked file_req path as a real file, resolved against the media cache instead of the index when the id doesn't match a file. Sliced by `chunk_index` like a real file's chunks, not just handed back whole: a thumbnail/poster/cover never approached CHUNK_SIZE so this used to be equivalent to "only chunk 0 exists", but an audio transcode result (docs/MESHBAY_DESIGN.md §9.8, the WMA/Musepack exception) is cached in the same media_cache blob store and can be several MB — genuinely multi-chunk, same as a file read straight off disk. """ media_cache = self._ctx.get("media_cache") if media_cache is None: return None blob = await media_cache.get_thumb(thumb_hash) if blob is None: return None start = chunk_index * CHUNK_SIZE if start > len(blob) or (start == len(blob) and chunk_index != 0): return None piece = blob[start:start + CHUNK_SIZE] return file_chunk_wire( gek, piece, chunk_index, bytes.fromhex(thumb_hash), thumb_hash) async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() # A chunk request is what "this transfer is alive" looks like. Nothing # marked a lease used, so `used` stayed False for the whole download and # the sweeper revoked the grant every 30 s as never-taken-up — while the # file was transferring at 20 MB/s. Found in the node's own log, which # repeated the same two reclaims every 30 s for as long as the daemon # ran. # # And it is also what makes the caps real: `tr` names a lease or it # does not, and `_lease_of` is the only thing that decides which. tr = str(msg.get("tr") or "")[:64] leased = False if tr: state = self._lease_of(tr) if state == LEASE_QUEUED: self._send({ "type": "error", "detail": "This transfer is waiting for a slot.", "code": "lease_not_granted", "tr": tr, }) return leased = state == LEASE_GRANTED if not leased: self._note_unleased(tr) file_id = msg["file_id"] chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) if not entry: thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek")) if thumb is not None: log.debug("file_req file_id=%s chunk=%s: served as thumbnail", file_id[:16], chunk_index) self._send(thumb) return log.warning("File not found: %s", file_id[:16]) self._send({"type": "error", "detail": "File not found"}) return file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) if refusal is not None: self._send({"type": "error", "detail": refusal}) return # A real index entry, asked for without a lease: browsing, or a client # helping itself to the whole library outside every cap. # # Both look identical here — which is why the bound is a small count of # files rather than a judgement about what the read is for. Thumbnails, # posters and cover art never reach this line: they resolve through # `_try_serve_thumbnail` above, out of a cache the node built itself, # and are never leased, never counted, never queued. # # `not leased`, not `not tr`: a `tr` the node cannot match to a granted # lease of this session is not a transfer, whatever the client calls it, # and reading the field as a boolean is what let any string at all # bypass this ceiling and the member cap behind it. if not leased: if not self._leaseless.admit(str(file_id)): self._send({ "type": "error", "detail": "Too many files open at once without a transfer. " "Download this one instead of previewing it.", "code": "transfer_required", "file_id": file_id, }) return log.debug("dl: req file=%s chunk=%s buffered=%s", file_id[:12], chunk_index, getattr(self._channel, "bufferedAmount", "?")) file_hash = bytes.fromhex(entry.id) chunk_data = await off_disk( ctx["roots"], _read_and_encrypt, ctx["gek"], file_path, chunk_index, file_hash, entry.id) # Backpressure. Without it the node hands the whole window to the # channel at once and the reader sees the first chunk, then nothing for # as long as the link takes to drain the rest. waited = 0.0 while (self._channel is not None and getattr(self._channel, "bufferedAmount", 0) > DOWNLOAD_BUFFER_HIGH and self._channel.readyState == "open" and waited < 60): await asyncio.sleep(0.05) waited += 0.05 if self._channel is None or self._channel.readyState != "open": return self._send(chunk_data) log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s", file_id[:12], chunk_index, len(chunk_data.get("ct") or b""), getattr(self._channel, "bufferedAmount", "?")) if chunk_index == 0: self._audit("file_download", entry.name) # The last chunk is the only "close" a leaseless read has. Without this # the session carries the entry until it goes idle, and the person who # just looked at two photos cannot look at a third for a minute. if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size: self._leaseless.finish(str(file_id)) @staticmethod async def _fetch_and_cache_poster(media_cache, tmdb_client, poster_path: str | None) -> str | None: """ Downloads a TMDB poster/backdrop once, caches it under its own blake3 like a video thumbnail (docs/MESHBAY_DESIGN.md §9.7), and returns the hash a client then fetches via the normal file_req/ chunk path (§5.3) — no client ever contacts image.tmdb.org directly. Checked by the synthetic `tmdb:{poster_path}` id *before* touching the network: without this, every `media_meta_req` for an already-cached file re-downloaded the same poster from TMDB (found live — a poster grid re-fetched both a show's poster and backdrop from TMDB on every single visit, real added latency and needless outbound traffic for an image that never changes). """ if not poster_path: return None synthetic_id = f"tmdb:{poster_path}" cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id) if cached_hash is not None: return cached_hash content = await tmdb_client.fetch_image(tmdb_client.poster_url(poster_path)) if content is None: return None thumb_hash = blake3.blake3(content).hexdigest() await media_cache.put_thumb(thumb_hash, synthetic_id, content) return thumb_hash async def _do_media_meta_request(self, msg: dict) -> None: """ docs/MESHBAY_DESIGN.md §9.7: TMDB metadata for one file, resolved from the group's index by its content id (root+relpath the client already knows from index_sync/index_delta identify the entry; its own `id` is what actually names one file — never a raw filesystem path off the wire). Keyed by `file_id`, not `path`: `IndexEntry.path` is the *folder* a file is in (indexer.py's `_virtual_dir`), so two files in the same folder — any multi-episode season, routinely — shared the same `.path`, and a lookup by it could silently resolve to the wrong entry (found live via the Music app's identical bug, 2026-08-25). """ file_id = msg.get("file_id") log.debug("media_meta_req file_id=%r", file_id) if not isinstance(file_id, str) or not file_id: self._send({"type": "error", "detail": "Missing file_id"}) return ctx = self._group_ctx() entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return media_cache = self._ctx.get("media_cache") tmdb_client = self._ctx.get("tmdb_client") # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24): # treated exactly like "no client configured" — same silent, no-error # degradation, since a member's Videos tab already has to handle "no # TMDB match" as the ordinary case. if media_cache is None or tmdb_client is None or not ctx.get("tmdb_enabled", True): self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "file_id": file_id, "confidence": 0}) return # A video the indexer has seen but not yet *enriched* has no # display_title (enrich.py always sets one) and season/episode still # None — so the movie/show split reads "movie" and would hand its raw # filename to TMDB's movie search. During a slow initial scan with a # browser on the Videos tab that is a storm of # `search/movie?query=` (found live 2026-08-29, an # 8-minute scan). While un-enriched we never *search*: we serve a # cached match if there is one (§ below), else confidence 0 and the # client refetches once the index delta carries the enriched fields. enriched = bool(entry.display_title) is_show = entry.season is not None and entry.episode is not None media_type = "tv" if is_show else "movie" cached = await media_cache.get_file_tmdb(entry.id) meta = None tmdb_id = None if cached is not None: cached_tmdb_id, cached_media_type = cached # Serve the cached match when its kind still agrees with the # entry's current classification — OR when the entry is not # enriched yet: its season/episode aren't populated, so the # movie/show split above is not meaningful, and the cached kind # (set when this file WAS enriched) is the reliable one. This is # what keeps a restart from re-querying TMDB for everything # already resolved: the storm was an un-enriched show episode # looking like a "movie" and treating its own valid "tv" match # as stale. # # Once enriched, the strict `cached_media_type == media_type` # check still stands: an enrichment fix that reclassifies a # folder movie->tv must drop the stale movie-era match and # re-resolve (found live — a Specials-folder fix left hundreds # of files answering with their wrong-kind match forever). if cached_media_type == media_type or not enriched: media_type = cached_media_type is_show = media_type == "tv" tmdb_id = cached_tmdb_id meta = await media_cache.get_tmdb_meta(tmdb_id, media_type) if meta is None: if not enriched: self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "file_id": file_id, "confidence": 0}) return result, ratio = await self._tmdb_search(tmdb_client, entry, is_show) if result is None or ratio < 0.6: self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "file_id": file_id, "confidence": 0}) return tmdb_id = str(result["id"]) meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, result) await media_cache.set_file_tmdb(entry.id, tmdb_id, media_type) await media_cache.set_tmdb_meta(tmdb_id, media_type, meta) poster_thumb_hash = await self._fetch_and_cache_poster( media_cache, tmdb_client, meta.get("poster_path")) backdrop_thumb_hash = await self._fetch_and_cache_poster( media_cache, tmdb_client, meta.get("backdrop_path")) log.debug("media_meta_req file_id=%r: replying tmdb_id=%s poster=%s backdrop=%s", file_id, tmdb_id, poster_thumb_hash, backdrop_thumb_hash) resp = { "type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "file_id": file_id, "tmdb_id": tmdb_id, "title": meta.get("title"), "original_title": meta.get("original_title"), "overview": meta.get("overview"), "poster_thumb_hash": poster_thumb_hash, "backdrop_thumb_hash": backdrop_thumb_hash, "release_date": meta.get("release_date"), "first_air_date": meta.get("first_air_date"), "genres": meta.get("genres", []), "vote_average": meta.get("vote_average"), "runtime": meta.get("runtime"), "cast": meta.get("cast", []), "director": meta.get("director"), "confidence": meta.get("confidence", 1.0), } if is_show: resp["season"] = entry.season resp["episode"] = entry.episode self._send(resp) async def _do_season_meta_request(self, msg: dict) -> None: """ Per-season TMDB overview/poster/air_date for a multi-season show — found live: `media_meta_resp`'s one static show-level overview does not necessarily describe every season alike (a season-3-specific promotional summary applied to all three seasons of a show). `tmdb_id` is whatever the client's own prior `media_meta_resp` already resolved — never re-derived from a path here, so this never re-runs a TMDB search of its own. """ tmdb_id = msg.get("tmdb_id") season = msg.get("season") if not isinstance(tmdb_id, str) or not tmdb_id or not isinstance(season, int): self._send({"type": "error", "detail": "Missing tmdb_id or season"}) return media_cache = self._ctx.get("media_cache") tmdb_client = self._ctx.get("tmdb_client") # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24) — # same silent zero-confidence degradation as "no client configured". if (media_cache is None or tmdb_client is None or not self._group_ctx().get("tmdb_enabled", True)): self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION, "tmdb_id": tmdb_id, "season": season, "confidence": 0}) return details = await media_cache.get_season_meta(tmdb_id, season) if details is None: fetched = await tmdb_client.tv_season(tmdb_id, season) if fetched is None: self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION, "tmdb_id": tmdb_id, "season": season, "confidence": 0}) return # Same per-field English fallback as _tmdb_build_meta: TMDB # returns "" for an untranslated field rather than falling back # itself. if not fetched.get("overview"): fallback = await tmdb_client.tv_season(tmdb_id, season, language="en-US") or {} fetched = {**fallback, **{k: v for k, v in fetched.items() if v not in (None, "", [])}} await media_cache.set_season_meta(tmdb_id, season, fetched) details = fetched poster_thumb_hash = await self._fetch_and_cache_poster( media_cache, tmdb_client, details.get("poster_path")) self._send({ "type": MNP.SEASON_META_RESP, "v": MNP_VERSION, "tmdb_id": tmdb_id, "season": season, "confidence": 1.0, "name": details.get("name"), "overview": details.get("overview"), "air_date": details.get("air_date"), "poster_thumb_hash": poster_thumb_hash, }) async def _do_tmdb_search_request(self, msg: dict) -> None: """ Candidate TMDB matches for an operator correcting a wrong automatic match (docs/MESHBAY_DESIGN.md §9.7, §V-whatever this becomes) — a plain lookup, not a mutation, so unlike `tmdb_override` this needs no admin authority: any member can see what TMDB itself would offer, the same as the automatic search already silently does on their behalf. Only `tmdb_override` actually changes what everyone sees. """ query = msg.get("query") media_type = msg.get("media_type") if not isinstance(query, str) or not query.strip() or media_type not in ("movie", "tv"): self._send({"type": "error", "detail": "Missing query or media_type"}) return media_cache = self._ctx.get("media_cache") tmdb_client = self._ctx.get("tmdb_client") # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24) — # same silent empty-results degradation as "no client configured": # a member with TMDB off for this group sees the same "type it in # yourself" affordance either way, never an error. if (media_cache is None or tmdb_client is None or not self._group_ctx().get("tmdb_enabled", True)): self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION, "query": query, "media_type": media_type, "results": []}) return # Refused out loud, not as an empty result: "no matches" is what the # client draws for an empty list, and telling somebody their film is # unknown when the node simply declined to ask is a worse answer than # the truth. `video-app.js`'s `runSearch` puts `detail` on screen. if not self._tmdb_search_rate_ok(): log.info("tmdb_search_req: rate-limited (user=%s)", (self._user_id or "")[:8]) self._send({ "type": "error", "detail": "Too many searches in the last minute. This spends the " "operator's search quota, which everyone in the group " "shares — try again shortly.", "code": "tmdb_search_rate_limited", }) return raw = (await tmdb_client.search_movie_results(query) if media_type == "movie" else await tmdb_client.search_tv_results(query)) results = [] for r in raw: poster_thumb_hash = await self._fetch_and_cache_poster( media_cache, tmdb_client, r.get("poster_path")) results.append({ "tmdb_id": str(r.get("id")), "title": r.get("title") or r.get("name"), "year": (r.get("release_date") or r.get("first_air_date") or "")[:4], "poster_thumb_hash": poster_thumb_hash, }) # media_type echoed back, not just query: a client can fire a "war" # tv search and a "war" movie search close together, and without it # the two responses are indistinguishable for keyed matching # (transport.js's tmdb_search_resp handler). self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION, "query": query, "media_type": media_type, "results": results}) def _do_tmdb_override(self, msg: dict) -> None: """ An operator correcting a wrong automatic TMDB match. Signed like app_directories/tmdb_config: it replaces what every member sees for a show/movie, node-wide (media_cache is shared, not per-viewer). For a **show**, applied to every entry sharing the representative file's display_title — the same grouping the poster grid uses (§3.4/§V6) — so the correction sticks regardless of which episode a future render picks as representative. For a **movie** it is applied to that one file only: guessit gives a whole franchise the same display_title, and a fan-out there corrected the wrong films (found live, 2026-08-29). See `_admin_exec_tmdb_override`. Keyed by `file_id`, not `path` — see `_do_media_meta_request`'s docstring for why a folder-level path cannot name one file. """ file_id = msg.get("file_id") tmdb_id = msg.get("tmdb_id") media_type = msg.get("media_type") if not isinstance(file_id, str) or not file_id: self._send({"type": "error", "detail": "Missing file_id"}) return if not isinstance(tmdb_id, str) or not tmdb_id or media_type not in ("movie", "tv"): self._send({"type": "error", "detail": "Missing tmdb_id or media_type"}) return ctx = self._group_ctx() entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return subject = f"file_id={file_id},tmdb_id={tmdb_id},media_type={media_type}" self._issue_admin_challenge(OP_TMDB_OVERRIDE, subject) async def _admin_exec_tmdb_override( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: subject = 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"tmdb_override:{subject}") return fields = dict(part.split("=", 1) for part in subject.split(",")) file_id, tmdb_id, media_type = fields["file_id"], fields["tmdb_id"], fields["media_type"] ctx = self._group_ctx() entry = ctx["index"].get_entry(file_id) media_cache = self._ctx.get("media_cache") if entry is None or media_cache is None: self._send({"type": "error", "detail": "File or media cache not available"}) return # This only ever recorded the file->tmdb_id mapping, never the # metadata tmdb_id names — _do_media_meta_request's cache check # (entry, cache) both agree on media_type, so it trusted the # mapping — but found nothing under this *new* id in tmdb_meta # (nothing had ever fetched it), and silently fell through to a # fresh search using the file's own title, exactly the one that # produced the wrong match in the first place. Confirmed live: an # override "stuck" for shows only because their own title happened # to be enough for that fallback search to land on the right # answer anyway, coincidentally — never because the override itself # was actually being honored — and was invisible until a movie # whose own title search kept landing on the same wrong result # exposed it. Fetching and storing the real metadata up front is # what makes the *override* the thing a later lookup finds. tmdb_client = self._ctx.get("tmdb_client") if tmdb_client is not None: meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, {}) await media_cache.set_tmdb_meta(tmdb_id, media_type, meta) # A show's episodes are many files that legitimately share one match, # and which episode a render picks as representative rotates — so a # show override fans out across every entry with the same # display_title. A *movie* is one file: fanning out by display_title # there is a bug — guessit gives every # " - - .mkv" the same display_title, so # "Fix match" on one entry rewrote the whole franchise (found live, # 2026-08-29). Each corrected file is also marked as a manual # override so ops.rematch_video / a rename never wipe it. is_show = entry.season is not None and entry.episode is not None if is_show: target_title = entry.display_title or entry.name matched = [e for e in ctx["index"].entries if e.type == "video" and (e.display_title or e.name) == target_title] else: matched = [entry] for e in matched: await media_cache.set_file_tmdb(e.id, tmdb_id, media_type) await media_cache.mark_tmdb_override(e.id) self._audit("tmdb_override", subject) notice = {"type": MNP.TMDB_OVERRIDE_ACK, "v": MNP_VERSION, "file_id": file_id, "tmdb_id": tmdb_id, "media_type": media_type} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass def _do_tmdb_rematch(self, msg: dict) -> None: """ An operator dropping one file's cached TMDB match so it re-resolves with the current matcher (V13) — the one-click alternative to the full search-and-pick "Fix match" flow, and reachable without SSH (`meshbay-node video rematch` clears a whole group). Signed like `tmdb_override`: `media_cache` is shared node-wide. """ file_id = msg.get("file_id") if not isinstance(file_id, str) or not file_id: self._send({"type": "error", "detail": "Missing file_id"}) return if not self._group_ctx()["index"].get_entry(file_id): self._send({"type": "error", "detail": "File not found"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_TMDB_REMATCH, f"file_id={file_id}") async def _admin_exec_tmdb_rematch( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: subject = 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"tmdb_rematch:{subject}") return file_id = dict(part.split("=", 1) for part in subject.split(","))["file_id"] media_cache = self._ctx.get("media_cache") if media_cache is None: self._send({"type": "error", "detail": "Media cache not available"}) return await media_cache.drop_tmdb_match(file_id) self._audit("tmdb_rematch", subject) notice = {"type": MNP.TMDB_REMATCH_ACK, "v": MNP_VERSION, "file_id": file_id} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass async def _tmdb_search(self, tmdb_client, entry, is_show: bool): """ docs/MESHBAY_DESIGN.md §9.7's scored ladder — same shape for movies and shows (V8). TMDB's own top result is still trusted per query (§3.3's last row — no local re-ranking of *its* list); what the ladder adds is that it *scores every candidate query* and keeps the best, instead of returning the first that merely clears 0.6. The bare parsed title is the weakest query: guessit drops a "Volume 2", strips a real subtitle into `alternative_title`, renders a sequel index where TMDB spells it differently, and a show's folder name can carry a year or release-group noise. A wrong entry that scored ~0.7 against that weak query — a same-year making-of documentary, a franchise entry whose localized TMDB title *is* the franchise name, a season-specific promo entry standing in for a whole show — used to win outright before a stronger candidate was ever tried. Found live (2026-08-29). """ from meshbay_node.indexer import title_parse if is_show: title = entry.display_title or title_parse.naive_title(entry.name) name_naive = title_parse.naive_title(entry.name) year = title_parse.year_in(title) or title_parse.year_in(entry.name) extra = [c for c in (name_naive, title_parse.clean_query(title)) if c and c != title] return await self._tmdb_ladder( tmdb_client.search_tv, title, extra, year, strong_extra=False) parsed = title_parse.parse_movie_filename(entry.name) title = entry.display_title or parsed.display_title or parsed.naive_title strong = [c for c in (parsed.alt_title, *title_parse.sequel_variants(title)) if c] extra = [c for c in (*strong, parsed.naive_title) if c and c != title] return await self._tmdb_ladder( tmdb_client.search_movie, title, extra, parsed.year, strong_extra=bool(strong)) @staticmethod async def _tmdb_ladder(search_fn, primary: str, extra: list[str], year: int | None, strong_extra: bool): """ `search_fn(query, year) -> (result|None, ratio)`. Try `primary`, return at once on a confident hit (ratio >= 0.85 — the common case, one request). Otherwise score each `extra` candidate and keep the best. `strong_extra` says whether `extra` contains anything more specific than a punctuation-normalised restatement of `primary` (an alternative_title, a sequel variant); when it does not and the primary hit is already decent, the remaining calls are skipped (V11 — they almost never win and cost a round trip each). """ def _year_of(res: dict) -> int | None: d = str(res.get("release_date") or res.get("first_air_date") or "") return int(d[:4]) if d[:4].isdigit() else None def _rescue(res: dict, r: float) -> float: # A sub-0.6 hit whose result lands on the exact requested year: # TMDB already year-filtered the search, so this is a hard # corroboration that the low ratio is a localised/rearranged # title, not a wrong entry. Never overrides a confident hit. if r < 0.6 and year and _year_of(res) == year: return max(r, 0.6) return r result, ratio = await search_fn(primary, year) if result is not None and ratio >= 0.85: return result, ratio best_result, best_score = (result, ratio) if result is not None else (None, 0.0) if best_result is not None: best_score = _rescue(best_result, best_score) if best_score >= 0.6 and not strong_extra: return best_result, best_score for candidate in extra: if candidate == primary: continue r2, ratio2 = await search_fn(candidate, year) if r2 is None and year: # A year-filtered search that finds nothing: the year tag # may be an edition/regional year TMDB doesn't carry. Retry # the candidate unconstrained before dropping it. r2, ratio2 = await search_fn(candidate, None) if r2 is None: continue score2 = _rescue(r2, ratio2) if score2 > best_score: best_result, best_score = r2, score2 if best_score >= 0.85: break return best_result, best_score @staticmethod async def _tmdb_build_meta(tmdb_client, tmdb_id: str, media_type: str, result: dict) -> dict: """ `result` (the search hit) only carries `genre_ids` and no `runtime` at all — the full details endpoint is the actual source for those, falling back to the search result for anything details somehow lacks (never expected in practice, just avoids a KeyError-shaped surprise if TMDB's response ever varies). """ details = (await tmdb_client.tv_details(tmdb_id) if media_type == "tv" else await tmdb_client.movie_details(tmdb_id)) or result # TMDB doesn't fall back server-side for a field with no translation # in the configured language — it returns "" (or an empty list) for # it, not the English text (confirmed live: a French query left # `overview` empty for a title TMDB has no French translation for). # The TMDB website covers exactly this gap client-side, by falling # back to English per field rather than discarding an otherwise-good # localized response over one empty one — mirrored here the same # way, at field granularity, not by abandoning the whole response. if (not details.get("overview") or not details.get("poster_path") or not details.get("genres")): fallback = (await tmdb_client.tv_details(tmdb_id, language="en-US") if media_type == "tv" else await tmdb_client.movie_details(tmdb_id, language="en-US")) or {} details = {**fallback, **{k: v for k, v in details.items() if v not in (None, "", [])}} credits = (await tmdb_client.tv_credits(tmdb_id) if media_type == "tv" else await tmdb_client.movie_credits(tmdb_id)) cast = [{"name": c.get("name"), "character": c.get("character")} for c in (credits or {}).get("cast", [])[:10]] director = None if media_type == "movie": director = next( (c.get("name") for c in (credits or {}).get("crew", []) if c.get("job") == "Director"), None) else: # A series has no single director, and `tv_credits`' crew is the # aggregate one — routinely empty, and never a "Director" job. # TMDB models the equivalent credit as `created_by` on the show # itself, which is what its own page shows; several creators are # ordinary, and they read as one line in the detail modal. director = ", ".join( name for name in (c.get("name") for c in details.get("created_by") or []) if name) or None runtime = details.get("runtime") if runtime is None and media_type == "tv": episode_run_times = details.get("episode_run_time") or [] runtime = episode_run_times[0] if episode_run_times else None return { "title": details.get("title") or details.get("name"), "original_title": details.get("original_title") or details.get("original_name"), "overview": details.get("overview"), "poster_path": details.get("poster_path"), "backdrop_path": details.get("backdrop_path"), "release_date": details.get("release_date"), "first_air_date": details.get("first_air_date"), "genres": [g.get("name") for g in details.get("genres", []) if g.get("name")], "vote_average": details.get("vote_average"), "runtime": runtime, "cast": cast, "director": director, } def _do_chat_message(self, msg: dict) -> None: """ Store one message and hand it to everyone else in this group. The node is a relay and an archive here, not a reader: once a group has chat encryption on, `payload` is a ciphertext it cannot open, and every decision below is made from fields that stay in clear — which is why those fields are the ones that must be *authenticated* rather than merely present. `sender_id` comes from the authenticated session and never from the wire (NS6). What the wire may now assert is the sending *device*, and that is checked against this connection rather than believed: a member who could name any device could sign as anyone once receivers verify signatures. """ # Per-group store — see _peer_registry() and finding H1. Reading # chat_store off the shared transport context sent every group's # messages to the first group's database, and served them back to # anyone on the node. gctx = self._group_ctx() chat_store = gctx.get("chat_store") sender_name = msg.get("sender_name", "") # Two shapes, and keeping them apart is what makes this deployable. # # A plaintext message is exactly what it has always been: a string in # `payload`. A sealed one carries its ciphertext in `ct`, beside the # `nonce`/`device`/`sig` that authenticate it. Putting the ciphertext in # `payload` instead would have been tidier and wrong: `payload` reaches # older clients — the UI ships inside the desktop package now, so it can # be months behind the node — and they would render bytes where they # expect text. A field they have never heard of is ignored instead. fmt = int(msg.get("format", 0) or 0) epoch = int(msg.get("epoch", 0) or 0) device = msg.get("device") nonce = msg.get("nonce") sig = msg.get("sig") if fmt == FORMAT_SEALED_V1: payload = "" raw = bytes(msg.get("ct") or b"") else: payload = msg.get("payload", "") raw = (payload.encode() if isinstance(payload, str) else bytes(payload or b"")) refusal = self._check_chat_envelope(gctx, fmt, raw, device, nonce, sig) if refusal: self._send({"type": "error", "detail": refusal}) self._audit("chat_refused", refusal) return # Separate from the envelope check above, and deliberately: that one asks # whether the message is well formed and authentic, this one asks what it # costs everyone else. Before either is written to disk or relayed. detail, code = self._chat_bounds_refusal(raw) if detail: self._send({"type": "error", "detail": detail, "code": code}) self._audit("chat_refused", code) return if sender_name: self._user_names()[self._user_id] = sender_name if chat_store: self._spawn(self._store_chat_message( chat_store, iteration=msg.get("iteration", 0), payload=raw, thread_id=msg.get("thread_id"), sender_name=sender_name, format=fmt, epoch=epoch, device=device, nonce=nonce, sig=sig, )) peers = self._peer_registry() broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "sender_id": self._user_id, "sender_name": sender_name, "payload": payload, "thread_id": msg.get("thread_id"), "timestamp": time.time(), "format": fmt, "epoch": epoch, "device": device, "nonce": nonce, "sig": sig, } if fmt == FORMAT_SEALED_V1: broadcast["ct"] = raw # Excludes this connection, not this account. The sender's other # devices are ordinary recipients: they did not compose the message and # have no local echo of it, so skipping them by user_id left a person's # second device silently missing everything they said from the first. for session in list(peers.values()): if session is not self: try: session._send(broadcast) except Exception: pass hub_ws = self._ctx.get("hub_ws") if hub_ws and self._group_id: try: import json as _json self._spawn(hub_ws.send(_json.dumps({ "type": "chat_notify", "group_id": self._group_id, # No sender_name. The body is unreadable to the hub the # moment a group turns encryption on, and shipping the # author's display name beside it would leave the hub a # per-message record of who spoke where — the metadata the # feature is otherwise about not producing. The hub renders # "New message in ". # # `sender_user_id` stays: the hub needs it to not notify # the author of their own message, and it already knows the # group's membership. "sender_user_id": self._user_id, }))) except Exception: pass self._send({"type": "ack", "v": MNP_VERSION}) self._audit("chat_message") def _check_chat_envelope(self, gctx: dict, fmt: int, ct: bytes, device, nonce, sig) -> str: """ Why this message is refused, or "" to accept it. Two rules, and the first is the one that matters: **A device may only send as itself.** `device` is what receivers verify a signature against, so a member free to name another member's key could be that member to everyone — worse than the node-asserted attribution it replaces (NS6), not better. The connection has proved which device it is (`device_hello`), and this must match it. **Plaintext is refused, always.** Not "accepts and marks", and not "unless a switch says otherwise": a member who can post in clear into a group whose members believe their chat is encrypted is a downgrade, and C6 is the standing lesson that the bypass left open is the one that gets used. There is no switch to leave open — MNP 2.0 refuses a 1.x peer at the handshake, so nothing that reaches here is unable to seal. `FORMAT_PLAIN` still exists, because rows written before 2.0 are still in `chat.db` and still served. It is a *storage* state, never something this accepts from the wire. """ if fmt != FORMAT_SEALED_V1: return "Chat messages must be encrypted" if not (isinstance(device, (bytes, bytearray)) and isinstance(nonce, (bytes, bytearray)) and isinstance(sig, (bytes, bytearray))): return "Sealed chat message is missing its envelope" if len(nonce) != CHAT_NONCE_LEN or len(sig) != CHAT_SIG_LEN: return "Sealed chat message has a malformed envelope" if not ct: return "Sealed chat message has no ciphertext" claimed = base64.b64encode(bytes(device)).decode() if not self._device_confirmed: return ("Identify this device before sending chat (device_hello)") if claimed != self._pinned_pk: return "That is not the device on this connection" return "" def _chat_bounds_refusal(self, ct: bytes) -> tuple[str, str]: """What this message would cost the others, or ("", "") to accept it. Two bounds, and each answers a different half of "who pays". A message is written to `chat.db` on the operator's disk and kept — retention is a manual command (§6.6) — then relayed to every other connected member and turned into a notification for every member of the group. So **size** bounds what one message costs, and **rate** bounds how often one member may impose it. There is deliberately no node-wide ceiling to go with the per-account one. The link-preview limiter has both because a preview spends the *node's* egress and its third-party quota, which is one shared thing; a chat message spends the sender's own group. A node-wide chat ceiling would let a busy group silence a quiet one, which is the same class of defect this bound exists to close, one level up. """ if len(ct) > MAX_CHAT_CIPHERTEXT: return ("This message is too large to send in chat — " "send a large file as an attachment instead.", "chat_too_large") if not self._chat_rate_ok(): return ("Too many messages just now — wait a moment.", "chat_rate_limited") return ("", "") def _chat_rate_ok(self) -> bool: """True when this sender is within their window; records it when so. Keyed by (group, account) on the transport context rather than on the session: the sender is authenticated, so this is the one identifier a second tab — or fifty of them — cannot multiply. The window is trimmed on every call, and the map of senders is swept when it grows, so neither can be the memory leak the bound was added to prevent. """ now = time.monotonic() hits: dict = self._ctx.setdefault("chat_hits", {}) if len(hits) > _CHAT_RATE_MAX_TRACKED: for key, times in list(hits.items()): if not times or now - times[-1] >= _CHAT_RATE_WINDOW: hits.pop(key, None) key = (self._group_id or "", self._user_id or "") mine = [t for t in hits.get(key, ()) if now - t < _CHAT_RATE_WINDOW] if len(mine) >= _CHAT_RATE_PER_ACCOUNT: hits[key] = mine return False mine.append(now) hits[key] = mine return True async def _store_chat_message(self, chat_store, **kwargs) -> None: """ Persist one message, treating a replay as already-done. A replayed message is a *validly signed* copy of a real one, so nothing about the signature refuses it; the unique `(device, nonce)` does. It is logged and dropped rather than raised at the sender: the message it duplicates is already stored, so there is nothing for anyone to retry. """ try: await chat_store.save_message(sender_id=self._user_id, **kwargs) except ReplayedMessage: log.warning("Replayed chat message from %s dropped", (self._user_id or "?")[:8]) self._audit("chat_replay_dropped") def _do_ping(self, msg: dict) -> None: """Answer a liveness probe on an open channel, echoing the caller's token. Echoed rather than bare so a client can match the answer to the probe it sent and measure a round trip, instead of being reassured by a reply to some earlier one. """ self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")}) def _do_chat_history(self, msg: dict) -> None: chat_store = self._group_ctx().get("chat_store") if not chat_store: self._send({ "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "messages": [], "has_more": False, }) return # `before` pages backwards from the newest, which is the direction a chat # is actually read. `since` remains for callers that want everything # after a point in time; the browser no longer uses it. before = msg.get("before") limit = max(1, min(int(msg.get("limit", 100)), 200)) self._spawn(self._send_chat_history(chat_store, before, limit)) async def _send_chat_history(self, chat_store, before, limit: int) -> None: if before: msgs = await chat_store.get_before(int(before), limit=limit) else: msgs = await chat_store.get_recent(limit=limit) # Whether the "load older" control has anything left to fetch. Asked # about the oldest row returned, so an empty page correctly says no. has_more = await chat_store.has_before(msgs[0].id) if msgs else False names = self._user_names() self._send({ "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "has_more": has_more, # `payload` goes out as **bytes**, never decoded here. It used to be # `.decode("utf-8", errors="replace")`, which substitutes U+FFFD for # every byte that is not valid UTF-8 — fine while chat was text, and # silent destruction of a ciphertext. Live messages would have kept # working (they are relayed, not re-read), so the symptom would have # been "history won't decrypt", which is the hardest possible place # to look. msgpack carries `bin` on both sides; the client decides # how to read it from `format`. "messages": [self._history_row(m, names) for m in msgs], }) @staticmethod def _history_row(m, names: dict) -> dict: """One stored message on the wire. A plaintext row goes out under `payload` as a string, exactly as it always has — an older client reads this response and must keep working. A sealed row's ciphertext goes out under `ct` as bytes and `payload` stays empty: decoding a ciphertext as UTF-8 (which is what this did, with `errors="replace"`) substitutes U+FFFD for most of it, and the symptom would have been history that will not decrypt while live messages worked — the hardest possible place to look. """ row = { "id": m.id, "sender_id": m.sender_id, "sender_name": m.sender_name or names.get(m.sender_id, ""), "timestamp": m.timestamp, "thread_id": m.thread_id, "format": m.format, "epoch": m.epoch, "device": m.device, "nonce": m.nonce, "sig": m.sig, } if m.format == FORMAT_SEALED_V1: row["payload"] = "" row["ct"] = m.payload else: row["payload"] = (m.payload.decode("utf-8", errors="replace") if isinstance(m.payload, bytes) else m.payload) return row def _tmdb_search_rate_ok(self) -> bool: """ True when this search is within both the member's window and the node's; records it when so, and trims both to the window on every call so neither list can grow without bound. Both are checked because they answer different questions: the member's keeps one person from spending everyone's quota, and the node's keeps a group of them from doing it together. """ now = time.monotonic() w = _TMDB_SEARCH_WINDOW ctx = self._group_ctx() by_member = ctx.setdefault("tmdb_search_hits", {}) who = self._user_id or "" mine = [t for t in by_member.get(who, []) if now - t < w] node = [t for t in self._ctx.get("tmdb_search_hits_node", []) if now - t < w] if len(mine) >= _TMDB_SEARCH_PER_MEMBER or len(node) >= _TMDB_SEARCH_NODE: by_member[who] = mine self._ctx["tmdb_search_hits_node"] = node return False mine.append(now) node.append(now) by_member[who] = mine self._ctx["tmdb_search_hits_node"] = node return True def _link_preview_rate_ok(self) -> bool: """ True when this preview fetch is within both the per-connection and the node-wide window; records it when so, and both counts are trimmed to the window on every call so the lists cannot grow without bound. """ now = time.monotonic() w = _LINK_PREVIEW_RATE_WINDOW mine = [t for t in getattr(self, "_link_preview_hits", []) if now - t < w] node = [t for t in self._ctx.get("link_preview_hits", []) if now - t < w] if (len(mine) >= _LINK_PREVIEW_RATE_PER_CONN or len(node) >= _LINK_PREVIEW_RATE_NODE): self._link_preview_hits = mine self._ctx["link_preview_hits"] = node return False mine.append(now) node.append(now) self._link_preview_hits = mine self._ctx["link_preview_hits"] = node return True async def _do_link_preview_request(self, msg: dict) -> None: """ Unfurl a URL a member pasted into chat (docs/MESHBAY_DESIGN.md §6.5's enrichment rule: the client asks, the node produces on demand, the asking device caches — nothing durable here). `linkpreview.safe_url` is the SSRF gate: the URL a *member* chose decides an outbound request from the operator's machine, so http(s) only and the resolved address must be globally routable. Failure of any kind — blocked, unreachable, not HTML, nothing worth showing — comes back as `ok: false`, the way a TMDB miss does; the client then just shows the bare link. """ url = msg.get("url") key = url if isinstance(url, str) else "" # Checked before the cache, not after: the operator turning previews # off has to stop serving the ones already fetched too, or the setting # takes effect only for links nobody has posted yet. Refused as an # ordinary miss — the client shows the bare link, which is exactly what # "no preview" looks like for a page that has none. if not self._group_ctx().get("chat_link_preview", True): self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION, "url": key, "ok": False}) return cached = _link_preview_cache_get(key) if cached is not None: self._send({**cached, "type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION}) return if not self._link_preview_rate_ok(): # Same shape as any other miss — the client shows the bare link. A # rate-limited result is not cached, so it is retried once the # window clears rather than pinned as "no preview". log.debug("link_preview_req: rate-limited (peer=%s)", self._peer_id) self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION, "url": key, "ok": False}) return resp: dict = {"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION, "url": key, "ok": False} try: meta = await linkpreview.fetch_preview(url) if meta is not None: resp.update(ok=True, title=meta["title"], description=meta["description"], site_name=meta["site_name"]) image_url = meta.get("image_url") media_cache = self._ctx.get("media_cache") if image_url and media_cache is not None: synthetic_id = f"linkpreview:{image_url}" thumb_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id) if thumb_hash is None: jpeg = await linkpreview.fetch_image(image_url) if jpeg: thumb_hash = blake3.blake3(jpeg).hexdigest() await media_cache.put_thumb(thumb_hash, synthetic_id, jpeg) if thumb_hash: resp["image_thumb_hash"] = thumb_hash except Exception as e: log.debug("link_preview_req %s: %s", key[:80], e) _link_preview_cache_put(key, {k: v for k, v in resp.items() if k not in ("type", "v")}) self._send(resp) def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads: """This group's uploads in progress, created on first use. In the group context rather than on the session, so a client that reconnects finds its own upload where it left it — and so the reaper has something to ask "is anyone still writing this?". """ store = ctx.get("partial_uploads") if store is None: store = uploads_mod.PartialUploads() ctx["partial_uploads"] = store return store @staticmethod def _upload_lock(ctx: dict) -> asyncio.Lock: """ One lock per group, beside the state it protects. Not per session: `partial_uploads` lives in the group context so a client that reconnects finds its upload where it left it, which means two sessions of the same member share the position of one `.part` file. A lock on the session would let them interleave — and the loop no longer serializes them for free now that a chunk write is awaited. """ lock = ctx.get("upload_lock") if lock is None: lock = asyncio.Lock() ctx["upload_lock"] = lock return lock async def _do_file_upload(self, msg: dict) -> None: """ One chunk of an upload, in the order it arrived. The chunk ordering rule — `chunk_index != state.next_index` is refused — used to hold for free: the handler was synchronous, so nothing could run between the check and the `advance` that answers it. Awaiting the write opens that gap, and two chunks of one upload racing through it is a `.part` file with a hole in it or a chunk refused for arriving on time. So the check, the write and the advance are one critical section again. The order is the arrival order: `_dispatch_message` runs per message as it arrives and creates these tasks in that order, tasks start in creation order, and this lock is the first thing each one waits on, so its queue of waiters is in arrival order too. """ async with self._upload_lock(self._group_ctx()): await self._upload_chunk(msg) async def _upload_chunk(self, msg: dict) -> None: """ One chunk of an upload, sealed under the group key (MNP 2.0). Sealing this direction is not symmetry for its own sake. Downloads have been under a GEK-derived key since the beginning; uploads carried the filename and the raw bytes in plain msgpack, so the same file was ciphertext leaving a node and plaintext arriving at one. The node holds the GEK for its own group, so it opens the payload here — before it decides a destination, before it touches the disk — and refuses a chunk that does not open. `upload_id` is the correlation key and stays in clear; `filename`, `dir` and `root` moved inside the seal, which is why every refusal below names the upload rather than the file. A `code` says which refusal it is, and the client already knows what it sent. """ ctx = self._group_ctx() upload_id = str(msg.get("upload_id") or "")[:64] # Say the slot is being used, chunk by chunk, exactly as `_do_file_req` # does for a download. # # A grant nobody takes up is reclaimed after GRANT_DEADLINE_SECS and, on # the third miss, abandoned. Uploads were not gated by the lease, so the # file still arrived — but the widget follows the lease, so a 3.5 GB # upload showed "waiting, 0 ahead" for a minute and a half while it was # in fact transferring, and the node logged three reclaims against a # transfer that never stopped. Measured, from the journal: # # 11:52:49 open upload 919ebf54 -> granted # 11:53:19 reclaimed 919ebf54 (not_taken_up) # 11:54:19 reclaimed 919ebf54 (abandoned) # 11:55:48 Upload complete: ... (3 522 297 517 bytes) # # `_do_file_request` marks a lease alive for exactly the same reason; # both sides of a transfer have to say they are still moving, or the # sweeper reclaims whichever one forgot. # # And a queued lease is refused here rather than written to disk. There # is no leaseless fallback on this side — a write is never "browsing" — # so the two cases part company: a lease this node has not granted is a # member taking a slot they were told to wait for, and an unknown `tr` # is the reconnect case, where the client is re-opening leases that died # with the old connection and the four upload protections (§6.4) are # what bound it meanwhile. tr = str(msg.get("tr") or "")[:64] if tr: state = self._lease_of(tr) if state == LEASE_QUEUED: self._send({ "type": "error", "detail": "This upload is waiting for a slot.", "code": "lease_not_granted", "upload_id": upload_id, "tr": tr, }) return if state == LEASE_NONE: self._note_unleased(tr) gek = ctx.get("gek") if not gek: self._send({"type": "error", "detail": "Group encryption not initialized", "code": "no_group_key", "upload_id": upload_id}) return try: payload = file_upload_payload(gek, self._group_id or "", msg) except Exception: # Deliberately one answer for "not sealed at all" and "sealed wrong": # distinguishing them tells a peer which of the two it got right. # An MNP 1.x client lands here, which is the whole of the upgrade # story — everything else it does still works. self._audit("upload_refused", "unsealed") self._send({ "type": "error", "detail": "This upload did not open under the group key — the " "client may be running an older version", "code": "upload_not_sealed", "upload_id": upload_id, }) return filename = payload.get("filename") or "" data = payload.get("data") # From the clear part of the message, so peer-controlled and unchecked # by the AEAD. Everything below compares and adds to them. try: chunk_index = int(msg.get("chunk_index", 0)) total_chunks = int(msg.get("total_chunks", 1)) except (TypeError, ValueError): self._send({"type": "error", "detail": "Invalid chunk index", "code": "bad_chunk_index", "upload_id": upload_id}) return def _refuse(detail: str, code: str = "") -> None: """A refusal names the upload, never the file: the name is sealed.""" out = {"type": "error", "detail": detail, "upload_id": upload_id} if code: out["code"] = code self._send(out) # Types first, and before any state is created. What comes out of a # sealed payload is authenticated, not validated: it is msgpack a # member wrote, and `SAFE_UPLOAD_NAME.match(123)` raises where a # refusal was meant. if not isinstance(filename, str) or not filename: _refuse("Missing filename or data", "upload_incomplete") return # Bytes, always: base64 was the shape of the old plaintext `data` field # and there is no sealed message that can carry a string here. if not isinstance(data, (bytes, bytearray)): _refuse("Invalid chunk encoding", "bad_chunk_encoding") return chunk_bytes = bytes(data) if not SAFE_UPLOAD_NAME.match(filename): _refuse("Invalid filename", "invalid_filename") return roots: RootSet | None = ctx.get("roots") if not roots: _refuse("No directories configured for this group", "no_roots") return # The client names the root it is uploading into — it is browsing one, # and with several writable roots any other choice is a guess. It names # a root, never a path: the destination inside it is decided below and # is not negotiable, which is what keeps C5a closed. # # An unknown name is refused rather than falling back to a writable # root, because "the file went somewhere else" is discovered weeks # later — the same reason the old single upload root was never guessed. # `dir` is the folder being browsed, as a virtual path # (`Media/Films/1999`); `root` is the older, coarser form and is what # its first segment means on its own. Both are sealed now, so a refusal # below can no longer quote them back. target_rel = str(payload.get("dir") or "").strip().strip("/") target_root_name = (target_rel.split("/")[0] if target_rel else str(payload.get("root") or "").strip()) upload_root = None if target_root_name: upload_root = roots.by_name(target_root_name) if upload_root is None: _refuse("No such directory in this group", "no_such_root") return else: writable = roots.writable_roots upload_root = writable[0] if writable else None if upload_root is None: _refuse("No writable directory in this group", "no_writable_root") return if not upload_root.writable: _refuse("That directory is read-only", "root_read_only") self._audit("upload_refused", filename[:64]) return if not upload_root.available: _refuse("That directory is currently unavailable", "root_unavailable") return # The folder the sender is looking at, and no subdirectory of the node's # invention. # # Uploads used to be confined to `/uploads/`, created on demand. # That was the last of v5's quarantine (the per-user layer went on # 2026-08-14, for the same reason): a shared directory nobody can # organise is not a shared directory, and a folder appearing beside the # operator's library because somebody sent a file is the node deciding # how their disk is arranged. # # What made the quarantine worth having is not the subdirectory — it is # the filename allowlist, the size cap, the chunk ordering, and the # no-overwrite rule below. All four are unchanged. # # `resolve()` and not a join: it refuses `..`, absolute segments and # anything whose resolved form escapes its root, symlinks included. The # client names *where among the group's own folders*, never a path on # the operator's filesystem. if target_rel: target_dir = await off_disk(roots, roots.resolve, target_rel) if target_dir is None or not await off_disk(roots, target_dir.is_dir): _refuse("Not a directory in this group", "no_such_directory") return rel_dir = target_rel else: # A client that names nothing: the first writable root is where its # one destination is. target_dir = upload_root.path rel_dir = upload_root.name if not await off_disk(roots, target_dir.is_dir): _refuse("That directory is currently unavailable", "root_unavailable") return # Held by the group, not by this connection. # # This used to be `self._uploads`, on the session. A dropped link threw # the position away and the next chunk was refused with `not_started`: # an upload interrupted at 99% could only be started again from zero, on # a connection flaky enough to have interrupted it once. And the state # it lost was the only thing that knew about the `.part` file left # behind — see `uploads.orphaned_parts`, which is the other half of this. # # Keyed by member as well as by name, because a shared directory means # two people can be sending IMG_1234.jpg at the same moment and neither # may inherit the other's position. uploads = self._partial_uploads(ctx) user_id = self._user_id or "" state = uploads.get(user_id, rel_dir, filename) # A shared directory means two people can send the same name. Refusing the # second is safe but silly — everyone's camera produces IMG_1234.jpg — so # a free name is found instead. Never a replacement. stored_name = (state.stored_name if state else await off_disk(roots, _free_name, target_dir, filename)) tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}" final_path = target_dir / stored_name if chunk_index == UPLOAD_PROBE_INDEX: # "Where am I?", asked inside the seal rather than on a clear # message, because the answer is about a file whose name is exactly # what sealing this path was for. # # It writes nothing, creates no state and reserves no name: a client # that asks and then goes away has cost this node one reply. Every # check above has already run, so it cannot be used to ask questions # about a directory the caller may not write to. self._send(file_upload_ack_wire( gek, self._group_id or "", upload_id=upload_id, chunk_index=UPLOAD_PROBE_INDEX, filename=filename, # Only what is really on disk. Without state, `_free_name` above # picked a name nothing has claimed yet, and reporting it would # promise a destination the real chunk 0 may not choose. stored_as=state.stored_name if state else "", dir=rel_dir, resume_from=state.next_index if state else 0, )) return if chunk_index == 0: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. if await off_disk(roots, final_path.exists): _refuse("File already exists", "already_exists") return state = uploads.start(user_id, rel_dir, filename, stored_name, part_path=tmp_path) elif state is None: _refuse("Upload not started", "not_started") return # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends # blindly to whatever .part file is already on disk. if chunk_index != state.next_index: _refuse("Unexpected chunk index", "bad_chunk_index") return if state.bytes + len(chunk_bytes) > self._max_upload_bytes(): uploads.drop(user_id, rel_dir, filename) await off_disk(roots, tmp_path.unlink, True) _refuse("Upload exceeds size limit", "too_large") return await off_disk(roots, _append_chunk, tmp_path, chunk_bytes, chunk_index == 0) uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes)) self._send(file_upload_ack_wire( gek, self._group_id or "", upload_id=upload_id, chunk_index=chunk_index, filename=filename, # What it is actually called on disk, which a chat attachment has to # reference and the uploader deserves to be told. stored_as=stored_name, dir=rel_dir, )) if chunk_index + 1 >= total_chunks: uploads.drop(user_id, rel_dir, filename) await off_disk(roots, tmp_path.rename, final_path) log.info("Upload complete: %s (%d chunks, %d bytes)", stored_name, total_chunks, state.bytes) self._audit("file_upload", f"{rel_dir}/{stored_name}") self._register_uploader(ctx, final_path) def _register_uploader(self, ctx: dict, file_path: Path) -> None: """ Record who sent this file, for the index entry that does not exist yet. The key recorded is the one this node pinned, not the one the token carried. `pk_user` was a hub-chosen claim, and it decided who could later delete the file: a hub issuing a token naming its own key could delete anyone's uploads on any node. Deletion is supposed to be authorized by the node, and this closes the last place where it was not. **The entry is not here to be tagged.** This used to walk `ctx["index"]` for the name just written and set the fields on it; at this point the watchdog has not fired (it debounces for two seconds and then hashes) and the file was a `.part` until the line above, which is not indexable — so the walk matched nothing, every time, and said nothing about it. The indexer stamps the entry from this record when it creates it. """ record = ctx.get("record_upload") if record is None: return self._spawn(record(file_path, self._user_id or "", self._pinned_pk or "")) def _do_file_delete(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg.get("file_id", "") if not file_id: self._send({"type": "error", "detail": "Missing file_id"}) return entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return # An owner is an *account* now, so an entry that records one is # challengeable even if the device that uploaded it is gone. has_uploader = bool(entry.uploader_pk or getattr(entry, "uploader_id", "")) if not self._has_admin_authority() and not has_uploader: self._send({"type": "error", "detail": "No authorized key for deletion"}) return self._issue_admin_challenge(OP_FILE_DELETE, file_id) # ── Admin operation challenge/response (finding H5) ────────────────────── def _node_pk_b64(self) -> str: return pk_to_b64(self._ctx["sk_node"].public_key()) def _issue_admin_challenge( self, op: str, subject: str, payload: dict | None = None, 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 _verify_uploader_sig(self, entry, transcript: bytes, sig: bytes) -> bool: """ Whether this signature comes from a live device of the file's uploader. Every non-revoked device of `entry.uploader_id` is tried, the same way `_verify_device_signer` tries every device that may approve a new one. Two properties worth keeping straight: - **Ownership survives device revocation.** A retired laptop's uploads keep their owner, because the account is what owns them; the revoked key simply is not among the ones that may act. - **Ownership survives the account losing every device**, where nothing verifies here and the operator remains able to delete — which is the behaviour a group needs when someone leaves. Falls back to the recorded `uploader_pk` only when the roster cannot answer at all (no roster wired, or no `uploader_id` on an entry written before that field existed). That is the pre-device-linking behaviour, so an old index does not become undeletable. """ roster = self._ctx.get("roster") uploader_id = getattr(entry, "uploader_id", "") or "" if roster is not None and uploader_id: for device in await roster.list_devices(uploader_id): try: pk = Ed25519PublicKey.from_public_bytes( base64.b64decode(device["pk_ed25519"])) except Exception: continue if self._verify_sig(pk, transcript, sig): return True return False if not entry.uploader_pk: return False try: pk = Ed25519PublicKey.from_public_bytes( base64.b64decode(entry.uploader_pk)) except Exception: return False return self._verify_sig(pk, transcript, sig) 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"}) async def _admin_exec_file_delete( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: file_id = pending["subject"] ctx = self._group_ctx() entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return # Node operator, or the account that uploaded this file — **any of its # non-revoked devices**, resolved through the node's own roster. # # This used to verify against `entry.uploader_pk` alone, the exact key # that uploaded. Device linking broke that on 2026-08-18 without # anything failing loudly: a file uploaded from a phone could not be # deleted from the same person's laptop, and the only symptom was # "Signature verification failed" on their own file # (docs/MESHBAY_DESIGN.md §3.3). # # `uploader_pk` is kept, and stops being the authorization key: it is # now the audit record of *which device* did it. Authorization is by # account, through the roster — never through a token claim, which is # the protection per-node identity keys give (docs/MESHBAY_DESIGN.md # §3.2) and which a lookup by `uploader_id` in the hub's world would # give straight back. if not (await self._verify_admin_sig(transcript, sig) or await self._verify_uploader_sig(entry, transcript, sig)): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") return await self._exec_file_delete(ctx, file_id, entry) async def _admin_exec_invite_create( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: # Node operator only. A group admin who does not run the node has no # authority over who this node admits (deny by default). Delegation is # designed but deferred — see docs/MESHBAY_DESIGN.md §3.4. if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}") return payload = pending["payload"] try: result = await self._run_op( ops.create_invite, payload["group_id"], payload.get("username", ""), user_id=payload["user_id"], created_by=self._user_id or "", ) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("invite_create", f"target={payload['user_id'][:8]}") self._send({ "type": MNP.INVITE_RESULT, "v": MNP_VERSION, "code": result["code"], "expires_at": result["expires_at"], "user_id": result["user_id"], "username": result.get("username", ""), }) async def _admin_exec_invite_link_create( 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", "invite_link_create") return try: result = await self._run_op( ops.create_link_invite, pending["payload"]["group_id"], created_by=self._user_id or "") except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("invite_link_create", f"invite={result['invite_id'][:8]}") self._send({ "type": MNP.INVITE_LINK_RESULT, "v": MNP_VERSION, "code": result["code"], "invite_id": result["invite_id"], "expires_at": result["expires_at"], "group_id": result["group_id"], }) async def _admin_exec_invite_cancel( 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"invite_cancel:{pending['subject'][:8]}") return payload = pending["payload"] try: await self._run_op(ops.cancel_invite, payload["group_id"], payload["invite_id"]) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("invite_cancel", f"invite={payload['invite_id'][:8]}") self._send({"type": "ack", "v": MNP_VERSION, "detail": "invite_cancelled", "invite_id": payload["invite_id"]}) async def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) if refusal == ROOT_NOT_SERVED: # Frozen, not gone: removing the entry would lose a file that is # still on a drive the node cannot read right now. self._send({"type": "error", "detail": ROOT_NOT_SERVED}) return if file_path is not None: await off_disk(ctx["roots"], file_path.unlink) log.info("File deleted: %s", entry.name) self._audit("file_delete", entry.name) ctx["index"].remove_entry(file_id) self._send({ "type": MNP.FILE_DELETE_ACK, "v": MNP_VERSION, "file_id": file_id, }) def _grant_stream_credit(self, msg: dict) -> None: """ The client has room for more segments. `n` of zero is a keepalive, not a no-op: a viewer whose buffer is already a minute and a half ahead of the playhead deliberately grants nothing, and must still be able to say it is there. Without that, the stall timeout below cannot tell a paused film from a closed tab. """ log.debug("stream credit +%s (had %d, sent %d)", msg.get("n"), self._stream_credit, self._stream_segments) try: n = int(msg.get("n", 1)) except (TypeError, ValueError): n = 1 if n == 0: # The fingerprint of a client that bounds its read-ahead. A client # that never sends one is granting credit per append — which is # what fills the browser's buffer ceiling and wedges the player. self._stream_keepalives += 1 if self._stream_keepalives == 1: log.info("stream: peer is pacing itself (first keepalive at " "%d segments)", self._stream_segments) self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT)) self._stream_heard_at = time.monotonic() self._stream_credit_evt.set() def _stop_stream(self) -> None: """ The viewer was closed. Stop transcoding and let go of the slot. Without this the only thing that ended a stream was the credit timeout, so ffmpeg kept running and held one of the node's two transcode slots for two minutes after nobody was watching — which is how closing a video made the next one answer "server busy". """ self._stream_stopped = True self._stream_credit_evt.set() async def _await_stream_credit(self) -> bool: """ Block until the client has room. False if it stopped asking. Without this the node hands ffmpeg's entire output to the channel as fast as it is produced, and the browser holds a four gigabyte film in a JavaScript array while MediaSource consumes it a segment at a time. """ # Measured from the last thing the peer said, not from the start of the # wait: a viewer that is buffered well ahead sends keepalives and grants # nothing for minutes at a time, and that is a watched film, not a # stalled one. self._stream_heard_at = time.monotonic() waiting_since = 0.0 while self._stream_credit <= 0: if waiting_since == 0.0: waiting_since = time.monotonic() # Debug: a paced viewer runs out of credit between every # window, so this is one line per eight segments — hundreds # per film. It is worth having, but not by default. log.debug("stream: out of credit at %d segments (%.0f MB) — " "waiting for the peer", self._stream_segments, self._stream_segments * STREAM_SEGMENT_SIZE / 1048576) if self._stream_stopped: return False # Checked before the wait as well as after it: a peer that vanishes # sends no credit and fires no event, so waiting the full timeout # on a channel that is already shut is pure dead time on a slot. if self._channel is None or self._channel.readyState != "open": return False self._stream_credit_evt.clear() try: # In slices rather than one long sleep, so a connection that # dies mid-wait is noticed in seconds instead of minutes. The # total budget is unchanged. await asyncio.wait_for(self._stream_credit_evt.wait(), timeout=STREAM_CREDIT_POLL) except TimeoutError: silent = time.monotonic() - self._stream_heard_at if silent >= STREAM_CREDIT_TIMEOUT: log.info("Stream stalled: nothing from peer=%s for %.0fs", (self._user_id or "?")[:8], silent) return False continue if self._stream_stopped: return False if self._channel is None or self._channel.readyState != "open": return False if waiting_since: waited_for = time.monotonic() - waiting_since # Only a wait long enough to be a symptom. Normal pacing puts a # gap of a few seconds between windows; a minute means the viewer # is buffered right up and playing, or has stopped watching. level = log.info if waited_for >= 10 else log.debug level("stream: credit arrived after %.1fs", waited_for) self._stream_credit -= 1 return True async def _replace_stream(self, msg: dict) -> None: """Retire this session's previous stream before starting another. A viewer plays one film at a time, so a second request means the first one is finished whatever the client managed to tell us. Relying on `stream_stop` alone was not enough: a browser that is backgrounded, reloaded or simply loses the message never sends it, and the only other thing that ends a stream is STREAM_CREDIT_TIMEOUT — two minutes during which ffmpeg keeps running and holds one of the node's two transcode slots. That is the reported failure exactly: first video fine, second fine, third answered "Server busy" because the first two were still holding both slots. The client shows that as "buffering" forever. Waiting for the old task is what makes the slot available: it is the exit of its `async with sem` that releases it. """ prev = self._stream_task if prev is not None and not prev.done(): t0 = time.monotonic() log.info("stream: retiring previous stream") self._stop_stream() try: await asyncio.wait_for(asyncio.shield(prev), timeout=15) log.info("stream: previous stream ended in %.1fs", time.monotonic() - t0) except TimeoutError: log.warning("stream: previous stream STILL RUNNING after 15s") except Exception: pass # it failed on its own; the slot is free either way self._stream_task = asyncio.current_task() await self._stream_video(msg) def _transcode_semaphore(self) -> asyncio.Semaphore: """The node's stream budget, shared across every peer. One ffmpeg per request with no cap lets any member exhaust the node's CPU and process table (H6). The semaphore lives on the transport context rather than the session so that it counts the node's viewers and not one browser's, and it is created once: rebuilding it per call would hand every caller its own budget and cap nothing at all. """ sem = self._ctx.get("_transcode_sem") if sem is None: n = self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES sem = asyncio.Semaphore(n) self._ctx["_transcode_sem"] = sem log.info("stream: %d concurrent viewers allowed", n) return sem async def _stream_video(self, msg: dict) -> None: """Stream a video file as fMP4 segments via MSE-compatible output.""" sem = self._transcode_semaphore() if sem.locked() and sem._value <= 0: self._send({"type": "error", "detail": "Server busy, retry shortly"}) return ctx = self._ctx log.info("stream: waiting for a slot (%d of %d in use)", ctx.get("_streams_in_flight", 0), self._stream_capacity()) async with sem: # Counted here rather than read back out of the semaphore's private # `_value`: `set_capacity` needs to know how many slots are held in # order to resize without letting the pool overshoot, and a number # this code maintains itself is one that survives the semaphore # object being replaced underneath it. ctx["_streams_in_flight"] = ctx.get("_streams_in_flight", 0) + 1 log.info("stream: slot acquired (%d of %d in use)", ctx["_streams_in_flight"], self._stream_capacity()) try: await self._stream_video_inner(msg) finally: ctx["_streams_in_flight"] = max( 0, ctx.get("_streams_in_flight", 1) - 1) log.info("stream: slot released (%d of %d in use)", ctx["_streams_in_flight"], self._stream_capacity()) def _stream_capacity(self) -> int: return self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES async def _stream_video_inner(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg.get("file_id", "") entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) if refusal is not None: self._send({"type": "error", "detail": refusal}) return gek = ctx.get("gek") file_hash = bytes.fromhex(entry.id) try: probe = await _probe_video(str(file_path)) except Exception as e: self._send({"type": "error", "detail": f"Probe failed: {e}"}) return codec_str = probe.codec duration = probe.duration has_audio = probe.has_audio raw_video_codec = probe.raw_codec_name # No video stream at all is the only thing this path cannot serve, and # it is the only thing refused here. A source with no MSE codec string # is emphatically not that — it is the case the re-encode below exists # for, and refusing it here (as "Unsupported video codec") is what this # fixed. if not raw_video_codec: self._send({"type": "error", "detail": "No video stream in this file"}) return # Where to begin. Seeking is a stream restarted somewhere else: the # viewer moves the scrubber, this session's previous stream is retired # by _replace_stream, and ffmpeg is spawned again with -ss. try: start = float(msg.get("start", 0) or 0) except (TypeError, ValueError): start = 0.0 # Past the end would produce an empty stream and a player waiting for # segments that are never coming. if duration and start >= duration - 1: start = max(0.0, duration - 5) start = max(0.0, start) # Video is copied whenever the browser can decode it directly — # re-encoding it is the expensive thing this pipeline exists to avoid, # and H264/VP9/AV1 already decode fine in-browser. HEVC is the one # exception (BROWSER_INCOMPATIBLE_VIDEO_CODECS, media_probe.py): found # live, a real HEVC/EAC3 WEB-DL reported "Codec not supported for # streaming" from MediaSource.isTypeSupported even though ffprobe/VLC # play it fine — Chrome has no HEVC decoder on most non-Apple # platforms. The operator can turn this fallback off (node.toml # transcode_incompatible_video = false) for a client fleet they know # already decodes HEVC, since it is real CPU cost, unlike the copy # path. Audio is always transcoded to AAC, never copied — see # _probe_video for why "copy" there is not an option, not even for a # codec that sounds close enough (plain AC-3 has the same in-browser # decode problem as E-AC-3, just without ffmpeg also refusing to mux # it). Transcoding audio is cheap; it does not change the cost model # the transcode-slot semaphore is sized around. # # Two kinds of source cannot be copied, and both re-encode: # # - one whose MSE codec string is real but that no mainstream browser # decodes — HEVC, BROWSER_INCOMPATIBLE_VIDEO_CODECS; # - one with **no MSE codec string at all**: MPEG-4 Part 2 (Xvid, # DivX), MPEG-2, VC-1, WMV, Theora. `stream_init` has to carry a # string the client puts through MediaSource.isTypeSupported, and # `probe_video` returns None for these precisely because no browser # has a MediaSource decoder for them, so there is none to carry. # This second kind used to be refused outright with "Unsupported # video codec" — which named the source's problem and not the # node's answer to it, since ffmpeg re-encodes these in real time on # any machine that can run this daemon. Reported live against an # Xvid/MP3 .avi. `transcode_incompatible_video`'s own documentation # (docs/MESHBAY_DESIGN.md §6.8) already said "HEVC *and other browser- # incompatible video codecs*"; only HEVC was ever wired up. can_copy = (bool(codec_str) and raw_video_codec not in BROWSER_INCOMPATIBLE_VIDEO_CODECS) allow_transcode = self._ctx.get("transcode_incompatible_video", True) transcode_video = not can_copy and allow_transcode if not can_copy and not allow_transcode and not codec_str: # The operator turned the fallback off and there is nothing to fall # back *to*: a stream_init with no codec string is one the client # refuses before the first byte arrives. Which of the two it is # matters — "unsupported codec" sends the reader to look at the # file, and the file is fine. log.info("stream: %s is %s, which needs a re-encode, and " "transcode_incompatible_video is off — refusing", entry.name, raw_video_codec) self._send({"type": "error", "detail": "This video needs transcoding, which the " "operator has turned off"}) return # Seeking, and the trap that made a seek on a copied stream unwatchable. # # -ss BEFORE -i seeks by the container index rather than by decoding up # to the point: milliseconds on a 500 MB film instead of tens of # seconds. It lands on the keyframe at or before `start`, so the # picture can begin a few seconds earlier than asked — which is what # every streaming player does. # # **`-accurate_seek` is on by default, and it trims what it can.** It # cannot trim copied video, which has to begin on a keyframe; it does # trim the re-encoded audio, to exactly `start`. So the output began # with video from the keyframe and audio from `start` — correct # timestamps, both streams honestly placed, and **a hole in the audio # one whole GOP wide**. Measured on a real film with a 10 s keyframe # interval: seeking to 609 s against a keyframe at 599.104 s left # 9.979 s of silence, after which sound and picture were a GOP apart # for the rest of the film. # # Nothing downstream could see it. Every timestamp check passes — the # first PTS of each stream, their durations, their spans, the browser's # own A/V delta through MediaSource — because the timestamps were never # wrong. Only the *content* at a given instant was, which is why this # was found by decoding the output and comparing it against the source: # the first frame is byte-identical to the source frame at the # keyframe, and with the fix the audio's energy envelope matches the # source at that same instant (r = 0.97) instead of one GOP later. # # This is also why re-encoded video never showed the fault, and why a # library's HEVC files looked like the only ones that worked: video # that is re-encoded *can* start exactly at `start`, so accurate # seeking is right there and stays on. # # **`start` is rewritten on the copy path to where the seek actually # lands**, which is measured below rather than predicted — see # `_seek_lands_at`. From the rewrite on, it is where the picture really # begins and not where the viewer dragged to. The distinction was # invisible while only the scrubber read the number; it stopped being # invisible when subtitles did, since their cues carry the source's # absolute times and every second of disagreement puts a line on # screen a second away from the voice saying it. requested = start seek_args: list[str] = [] if requested > 0: seek_args = ["-ss", f"{requested:.3f}"] if transcode_video else [ "-noaccurate_seek", "-ss", f"{requested:.3f}"] map_args = ["-map", "0:v:0"] if transcode_video: log.info("stream: re-encoding %s (%s) to H264", entry.name, raw_video_codec) # Where the re-encode runs, and with which arguments — both live # in hwaccel.py now, including the 8-bit downsampling a 10-bit HDR # source needs before either encoder will take it. `modes_for` has # measured this machine by encoding on it and returns the ladder to # try, always ending in libx264: a node with no usable VA-API does # exactly what it did before this existed, and a Celeron with an # iGPU stops being a machine where `transcode_incompatible_video` # has to be turned off to keep streaming watchable. modes = await hwaccel.modes_for(raw_video_codec) hw = await hwaccel.encoder() # Must match "-profile:v high -level 4.1" byte-for-byte (avc1.) — the client checks this string with # MediaSource.isTypeSupported before trusting a single byte of the # stream, so a mismatch here fails exactly the check this exists to # pass. Both encoders get those two arguments, spelled the same way, # from hwaccel._PROFILE_ARGS — one place, so they cannot drift. codec_str = "avc1.640029,mp4a.40.2" if has_audio else "avc1.640029" else: modes, hw = [hwaccel.SW], None def video_args(mode: str) -> list[str]: return (hwaccel.codec_args(mode, hw) if transcode_video else ["-c:v", "copy"]) # The audio half does not change with the video encoder, and is never a # copy — see _probe_video for why. audio_args: list[str] = [] # Which audio track. A dubbed film carries several and the first one is # not a neutral default — it is whatever the person who muxed the file # happened to put first, which across a real library is overwhelmingly # one language. Out of range falls back to the first rather than # refusing: the client's list comes from a `stream_init` that may # predate the file being replaced on disk, and a viewer who asked for # the second track of a file that now has one wants the film, not an # error. `stream_init` says which track was actually used, the same way # it says which `start` was actually used and for the same reason. try: audio_track = int(msg.get("audio_track", 0) or 0) except (TypeError, ValueError): audio_track = 0 if not 0 <= audio_track < len(probe.audio_tracks): audio_track = 0 if has_audio: map_args += ["-map", f"0:a:{audio_track}"] # Downmixed to stereo: a WEB-DL's 5.1 track becomes 6-channel AAC # with no "-ac", which ffprobe and VLC accept fine but which some # browsers' MSE decoder rejects outright once real fragments are # appended — isTypeSupported() only checks the codec string, so # the failure doesn't surface until playback, as a SourceBuffer # forced out of its MediaSource with no further explanation. audio_args = ["-c:a", "aac", "-ac", "2", "-b:a", "192k"] # Where that seek lands, measured with the mapping this stream will # use. It has to be here rather than beside `seek_args` above: the # landing point depends on which audio track is mapped, because the # container is seeked to a position that serves *every* mapped stream # — on a real title, video alone landed at 4909.863 s and the same # seek with the second audio track landed at 4907.236 s. The `-ss` # argument is deliberately left at the request, so the bytes served # are exactly the ones served before; only the number naming them # changes. if requested > 0 and not transcode_video: landed = await _seek_lands_at(file_path, requested, map_args) if landed is not None: start = landed # One spawn per mode, and only ever more than one when hwaccel.py found # a working GPU. **What a mode is tried against is the file itself**: # a test encode proves the encoder, and nothing proves the GPU can # decode *this* source until it is asked to — iHD has no MPEG-4 Part 2 # decoder at all, so an Xvid .avi fails the full-hardware mode and # nothing about the machine could have predicted it. # # The failure is silent and instant: ffmpeg writes its complaint to # stderr and exits, so stdout reaches EOF with nothing on it. That is # the signal read here, before `stream_init` is sent and therefore # before the client has been told anything it would have to be told # again. The first segment is kept and handed to the loop below rather # than re-read, since the process it came from is still running. # # The last mode is spawned and trusted, which is what keeps the # single-mode path — every node without a GPU, and every copied stream # — byte-for-byte what it was: no extra read, no extra wait. first_segment = b"" for attempt, mode in enumerate(modes): proc = await asyncio.create_subprocess_exec( platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", *hwaccel.input_args(mode, hw), *seek_args, "-i", str(file_path), *map_args, *video_args(mode), *audio_args, "-movflags", "frag_keyframe+empty_moov+default_base_moof", "-f", "mp4", "pipe:1", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) if attempt == len(modes) - 1: break first_segment = await proc.stdout.read(STREAM_SEGMENT_SIZE) if first_segment: break err = (await proc.stderr.read()).decode("utf-8", "replace").strip() await proc.wait() hwaccel.demote(raw_video_codec, mode, err.splitlines()[0] if err else "no output") self._send({ "type": MNP.STREAM_INIT, "v": MNP_VERSION, "file_id": file_id, "codec": codec_str, "duration": duration, # ffmpeg restarts its timestamps at zero whatever we seek to, so # this is what the client adds back (`SourceBuffer.timestampOffset`) # to put the fragments where they belong on the timeline. "start": start, # The track list is how a client discovers that this node can # switch language at all — there is no version check anywhere in # the player. A node that does not send it gets no selector, and # the client then never sends `audio_track` to a peer that would # ignore it and serve the wrong language without saying so. "audio_tracks": [ { "i": tr.ordinal, "lang": tr.language, "title": tr.title, "codec": tr.codec_name, "ch": tr.channels, } for tr in probe.audio_tracks ], "audio_track": audio_track if has_audio else None, # Same discovery-from-the-answer shape as `audio_tracks`: a node # too old to enumerate sends no list, the client shows no selector # and never sends `subtitle_req` to a peer that would answer # "unknown message type". Text tracks only — a bitmap one has no # WebVTT to offer (media_probe.py), so it is absent here rather # than present and unplayable. "subtitle_tracks": [ { "i": tr.ordinal, "lang": tr.language, "title": tr.title, "codec": tr.codec_name, # What tells a full translation from signage-only. Without # it the two are the same menu entry, and picking the # forced one shows nothing for minutes at a time — which # reads as a broken feature and was reported as one. "forced": tr.forced, "sdh": tr.hearing_impaired, } for tr in probe.subtitle_tracks ], }) # A client that says nothing gets the old behaviour, which is why this # defaults to unlimited rather than to zero: a stream that waits for # credit from a peer that will never send any is a stream that hangs. try: self._stream_credit = int(msg.get("credits", 0) or 0) except (TypeError, ValueError): self._stream_credit = 0 paced = self._stream_credit > 0 self._stream_stopped = False index = 0 self._stream_started_at = time.monotonic() self._stream_segments = 0 reason = "eof" log.info("stream: stream_init sent file=%s paced=%s credits=%d start=%.1fs " "audio=%s/%d", file_id[:12], paced, self._stream_credit, start, audio_track if has_audio else "-", len(probe.audio_tracks)) try: while True: if paced and not await self._await_stream_credit(): reason = "no-credit-or-gone" break if self._stream_stopped: reason = "stopped-by-peer" log.info("Stream stopped by peer=%s after %d segments", (self._user_id or "?")[:8], index) break if first_segment: data, first_segment = first_segment, b"" else: data = await proc.stdout.read(STREAM_SEGMENT_SIZE) if not data: break # Same derivation as a file chunk, indexed by segment: one # implementation, in `meshbay_common.protocol`. nonce, ct = chunk_ciphertext(gek, data, index, file_hash) self._send({ "type": MNP.STREAM_DATA, "v": MNP_VERSION, "file_id": file_id, "segment_index": index, "nonce": nonce, "ct": ct, "plaintext_size": len(data), }) index += 1 self._stream_segments = index if index % 100 == 0: # A stream that stops shows up here as a last line, and the # numbers on it say which side stopped it. log.info("stream: %d segments (%.0f MB), credit=%d, " "keepalives=%d, %.0fs in", index, index * STREAM_SEGMENT_SIZE / 1048576, self._stream_credit, self._stream_keepalives, time.monotonic() - self._stream_started_at) await asyncio.sleep(0) except Exception as e: log.error("Stream error: %s", e) finally: try: proc.kill() except ProcessLookupError: pass # `await proc.wait()` on its own is the deadlock the asyncio docs # warn about: ffmpeg fills the stdout pipe we have stopped reading, # and the transport cannot finish closing until that buffer is # drained. Measured on 2026-08-16 with stream: — a viewer closed # the player after 99 segments (25 MB) and the task sat here past # the 15 s handover timeout, holding a transcode slot. The node has # two, so the next video waited and the one after was refused. # # Drain first, then wait with a bound. The slot must come back even # if the process is being stubborn: it has already had SIGKILL, and # the OS will reap it whether or not we are still watching. stderr_output = b"" for pipe in (proc.stdout, proc.stderr): if pipe is None: continue try: drained = await asyncio.wait_for(pipe.read(), timeout=2) if pipe is proc.stderr: stderr_output = drained except Exception: pass try: await asyncio.wait_for(proc.wait(), timeout=5) except Exception: log.warning("stream: ffmpeg did not reap in 5s — " "releasing the slot regardless") # A positive returncode is ffmpeg exiting on its own with an error, # before we ever killed it (a kill shows up as a negative signal # number instead) — zero segments in that case is a real failure, # not a normal end, and saying nothing here is indistinguishable # from "the file is just this short". Found live against a real # 5.1 E-AC-3 WEB-DL that ffmpeg refused to even start muxing. # Detail stays server-side (L3: never hand a peer raw stderr). if index == 0 and proc.returncode is not None and proc.returncode > 0: log.error("stream: ffmpeg exited rc=%s before any output — %s", proc.returncode, stderr_output.decode(errors="replace").strip().splitlines()[-1:] or "(no stderr)") if not self._stream_stopped: self._send({"type": "error", "detail": "Could not stream this file"}) elif not self._stream_stopped: self._send({ "type": MNP.STREAM_END, "v": MNP_VERSION, "file_id": file_id, }) log.info("stream: stream ended reason=%s segments=%d after %.1fs", reason, index, time.monotonic() - self._stream_started_at) log.info("Streamed %s: %d segments", entry.name, index) self._audit("stream_video", entry.name) def _send(self, obj: dict) -> None: # 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() def _mkdir_if_absent(target: Path) -> str | None: """ Create a directory unless it is already there, or say why not. Both in one call, not a check awaited and then an act: the disk thread is one worker, so nothing can slip between them. Split across two awaits, two members creating the same name would both find nothing there and the second `mkdir` would raise where a refusal was meant. Blocking; called through `off_disk`. """ if target.exists(): return "Already exists" target.mkdir(parents=False) return None def _is_empty_dir(target: Path) -> bool: """Blocking; called through `off_disk`.""" return not any(target.iterdir()) def _rmdir_if_empty(target: Path) -> bool: """ Remove a directory if nothing is in it. False if something is. The emptiness test and the removal are one call for the reason the caller re-tests at all: the first test happened before a round trip to the operator's browser, and a file can land in between. Two awaits here would reopen the same window one size smaller. Blocking; called through `off_disk`. """ if any(target.iterdir()): return False target.rmdir() return True def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None: """Add one chunk to a partial upload. Blocking; called through `off_disk`.""" with open(tmp_path, "wb" if first else "ab") as f: f.write(chunk_bytes) def _read_and_encrypt( gek: bytes, file_path: Path, chunk_index: int, file_hash: bytes, file_id: str = "", ) -> dict: """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`.""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) 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)