diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 245 |
1 files changed, 2 insertions, 243 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 92e1cd4..20636fa 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -31,9 +31,6 @@ from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, ) -from meshbay_common.protocol import ( - MNP, -) from meshbay_node import transfers as transfers_mod from meshbay_node.indexer import GroupIndex @@ -54,6 +51,7 @@ from meshbay_node.transport.webrtc.apps.video_meta import VideoMetaMixin from meshbay_node.transport.webrtc.blobs import BlobsMixin from meshbay_node.transport.webrtc.chat import ChatMixin from meshbay_node.transport.webrtc.core import _WEBRTC_TRACE, SessionCore +from meshbay_node.transport.webrtc.dispatch import DispatchMixin from meshbay_node.transport.webrtc.files import FilesMixin from meshbay_node.transport.webrtc.group_ops import GroupOpsMixin from meshbay_node.transport.webrtc.handshake import HandshakeMixin @@ -75,254 +73,15 @@ log = logging.getLogger(__name__) MAX_PEER_SESSIONS = 64 UNAUTHENTICATED_SESSION_TIMEOUT = 60 # seconds -# Bundle fetches are served in the pre-proof window (C4). Bounded and audited -# until the native client removes remote keypair bundles entirely. -MAX_PRE_PROOF_FETCHES = 4 - class WebRTCPeerSession( - AdminMixin, AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin, + DispatchMixin, AdminMixin, AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin, HandshakeMixin, NodeOpsMixin, TransferMixin, UploadMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, SessionCore, ): """One WebRTC peer connection, handling MNP over a DataChannel.""" - 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"}) - class WebRTCTransport: """ |