"""The operator's controls over a group's roots and scan pacing that MNP carries: removing, ejecting and plugging a root. What widens the sharing, and the node's own status, settings, roster and denylist, are the loopback API's and the CLI's (MNP 6.0).""" import logging from meshbay_common import MNP_VERSION from meshbay_common.adminop import ( OP_ROOT_EJECT, OP_ROOT_PLUG, OP_ROOT_REMOVE, OP_SET_SCAN_SETTINGS, ) from meshbay_common.protocol import MNP from meshbay_node import ops log = logging.getLogger("meshbay_node.transport.webrtc_server") class NodeOpsMixin: # 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 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_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 _retarget_indexer(self, group_id: str) -> None: """ Pick up a root that was just 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 removed over MNP kept serving its files until a restart. 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. Not awaited: a reload can be long. The ack the caller sends carries the set the node is moving to. """ 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)