"""Which handler answers which MNP message, and the guards that come before any of them.""" import logging 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 # After the handshake: which method answers each message type, and whether it # runs as a task of this session (`_spawn`) or before the next message is read. # That choice is made per type, deliberately; the golden master in the tests # records it for every type. SPAWNED, INLINE = True, False _HANDLERS = { MNP.INDEX_SYNC: ("_do_index_sync", INLINE), # 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. MNP.FILE_REQUEST: ("_do_file_request", SPAWNED), MNP.CHAT_MESSAGE: ("_do_chat_message", INLINE), MNP.CHAT_HISTORY: ("_do_chat_history", INLINE), MNP.LINK_PREVIEW_REQ: ("_do_link_preview_request", SPAWNED), MNP.PING: ("_do_ping", INLINE), MNP.TRANSFER_OPEN: ("_do_transfer_open", INLINE), MNP.TRANSFER_CLOSE: ("_do_transfer_close", INLINE), MNP.FILE_UPLOAD: ("_do_file_upload", SPAWNED), MNP.DIR_CREATE: ("_do_dir_create", SPAWNED), MNP.DIR_DELETE: ("_do_dir_delete", SPAWNED), MNP.FILE_DELETE: ("_do_file_delete", INLINE), MNP.ADMIN_RESPONSE: ("_do_admin_response", INLINE), MNP.INVITE_CREATE: ("_do_invite_create", INLINE), MNP.INVITE_LINK_CREATE: ("_do_invite_link_create", INLINE), MNP.INVITE_CANCEL: ("_do_invite_cancel", INLINE), MNP.MEMBER_REVOKE: ("_do_member_revoke", INLINE), MNP.DEVICE_REQUEST: ("_do_device_request", SPAWNED), MNP.DEVICE_LOOKUP: ("_do_device_lookup", SPAWNED), MNP.DEVICE_ADD: ("_do_device_add", SPAWNED), MNP.DEVICE_LIST: ("_do_device_list", SPAWNED), MNP.DEVICE_REVOKE: ("_do_device_revoke", SPAWNED), MNP.DEVICE_HELLO: ("_do_device_hello", SPAWNED), MNP.APPS_ENABLED: ("_do_apps_enabled", INLINE), MNP.TRANSFER_LIMITS: ("_do_transfer_limits", INLINE), MNP.SET_SCAN_SETTINGS: ("_do_set_scan_settings", INLINE), MNP.TMDB_CONFIG: ("_do_tmdb_config", INLINE), MNP.TMDB_ENABLED: ("_do_tmdb_enabled", INLINE), MNP.APP_DIRECTORIES: ("_do_app_directories", INLINE), MNP.CHAT_DIRECTORY: ("_do_chat_directory", INLINE), MNP.CHAT_LINK_PREVIEW: ("_do_chat_link_preview", INLINE), MNP.SEARCH_LISTED: ("_do_search_listed", INLINE), MNP.CHAT_EPOCH: ("_do_chat_epoch", INLINE), MNP.CHAT_KEYS_REQ: ("_do_chat_keys_req", SPAWNED), MNP.GROUP_ROSTER_REQ: ("_do_group_roster_req", SPAWNED), MNP.MEDIA_META_REQ: ("_do_media_meta_request", SPAWNED), MNP.SEASON_META_REQ: ("_do_season_meta_request", SPAWNED), MNP.TMDB_SEARCH_REQ: ("_do_tmdb_search_request", SPAWNED), MNP.TMDB_OVERRIDE: ("_do_tmdb_override", INLINE), MNP.TMDB_REMATCH: ("_do_tmdb_rematch", INLINE), MNP.MUSICBRAINZ_ENABLED: ("_do_musicbrainz_enabled", INLINE), MNP.MUSIC_META_REQ: ("_do_music_meta_request", SPAWNED), MNP.AUDIO_TRANSCODE_REQ: ("_do_audio_transcode_request", SPAWNED), MNP.SUBTITLE_REQ: ("_do_subtitle_request", SPAWNED), MNP.MEMBER_UNPIN: ("_do_member_unpin", INLINE), MNP.GEK_ROTATE: ("_do_gek_rotate", INLINE), MNP.NODE_STATUS: ("_do_node_status", SPAWNED), MNP.ROOT_ADD: ("_do_root_add", INLINE), MNP.ROOT_REMOVE: ("_do_root_remove", INLINE), MNP.ROOT_UPDATE: ("_do_root_update", INLINE), MNP.ROOT_EJECT: ("_do_root_eject", INLINE), MNP.ROOT_PLUG: ("_do_root_plug", INLINE), MNP.ROSTER_READ: ("_do_roster_read", SPAWNED), MNP.DENYLIST_READ: ("_do_denylist_read", SPAWNED), MNP.DENYLIST_CLEAR: ("_do_denylist_clear", SPAWNED), MNP.GROUP_ATTACH: ("_do_group_attach", INLINE), MNP.GROUP_DETACH: ("_do_group_detach", INLINE), MNP.NODE_SETTINGS_SET: ("_do_node_settings_set", SPAWNED), MNP.NODE_RELOAD: ("_do_node_reload", SPAWNED), MNP.KEYPAIR_BUNDLE_STORE: ("_do_keypair_bundle_store", SPAWNED), MNP.KEYPAIR_BUNDLE_DELETE: ("_do_keypair_bundle_delete", SPAWNED), MNP.USER_BLOB_STORE: ("_do_user_blob_store", SPAWNED), MNP.USER_BLOB_FETCH: ("_do_user_blob_fetch", SPAWNED), MNP.USER_BLOB_LIST: ("_do_user_blob_list", SPAWNED), MNP.USER_BLOB_DELETE: ("_do_user_blob_delete", SPAWNED), MNP.STREAM_REQUEST: ("_do_stream_request", INLINE), MNP.STREAM_MORE: ("_grant_stream_credit", INLINE), 'client_diag': ("_do_client_diag", INLINE), MNP.STREAM_STOP: ("_do_stream_stop", INLINE), } # The handlers that are called without the message. _TAKES_NO_MESSAGE = frozenset({ "_do_index_sync", "_do_keypair_bundle_delete", "_do_stream_stop", "_do_user_blob_list", }) 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"}) else: try: entry = _HANDLERS.get(mtype) except TypeError: # A list or a map is a legal msgpack value for `type` and not a # key: it names no handler, as it matched no comparison before. entry = None if entry is None: log.warning("Unknown MNP message type on DataChannel: %s", mtype) else: name, spawned = entry method = getattr(self, name) result = method() if name in _TAKES_NO_MESSAGE else method(msg) if spawned: self._spawn(result) 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"})