diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 108 |
1 files changed, 108 insertions, 0 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 b6f572a..fa6c3e9 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -63,6 +63,7 @@ from meshbay_common.adminop import ( OP_MEMBER_UNPIN, OP_MEMBER_UPLOAD, OP_APPS_ENABLED, + OP_SET_SCAN_SETTINGS, OP_ROOT_ADD, OP_ROOT_REMOVE, OP_GROUP_ATTACH, @@ -85,6 +86,7 @@ from meshbay_common.join import ( from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex +from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import ops from meshbay_node.roots import ( RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, @@ -409,6 +411,8 @@ class WebRTCPeerSession: self._do_member_upload(msg) elif mtype == MNP.APPS_ENABLED: self._do_apps_enabled(msg) + elif mtype == MNP.SET_SCAN_SETTINGS: + self._do_set_scan_settings(msg) elif mtype == MNP.MEMBER_UNPIN: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: @@ -659,6 +663,20 @@ class WebRTCPeerSession: # setting (or one whose context has not loaded it yet) hides # nothing. "enabled_apps": list(self._group_ctx().get("enabled_apps") 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: ack["node_user_id"] = node_user_id @@ -668,6 +686,13 @@ class WebRTCPeerSession: 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") @@ -1648,6 +1673,69 @@ class WebRTCPeerSession: 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}") + + 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: @@ -2012,6 +2100,23 @@ class WebRTCPeerSession: return self._ctx["groups"][self._group_id] return self._ctx + def _indexing_status(self) -> dict: + """ + {"scanning": bool, "scanned_bytes": int, "total_bytes": int} for the + handshake ack and INDEX_PROGRESS pushes — never a path or filename, + that stays 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} + return { + "scanning": progress.scanning, + "scanned_bytes": progress.scanned_bytes, + "total_bytes": progress.total_bytes, + } + def _peer_registry(self) -> dict: """ Connected peers for THIS group only. @@ -2626,6 +2731,9 @@ class WebRTCPeerSession: elif pending["op"] == OP_APPS_ENABLED: self._spawn( self._admin_exec_apps_enabled(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_ROOT_ADD: self._spawn( self._admin_exec_root_add(pending, transcript, sig_bytes)) |