"""Which handler answers which MNP message, and the guards that come before any of them.""" import logging import time from meshbay_common.protocol import MNP log = logging.getLogger("meshbay_node.transport.webrtc_server") # 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 class DispatchMixin: 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"})