"""The operator's controls over the node itself: status and settings, roster and denylist, roots, hosted groups, reload, scan pacing and transfer limits.""" import logging from meshbay_common import MNP_VERSION from meshbay_common.adminop import ( OP_GROUP_ATTACH, OP_GROUP_DETACH, OP_ROOT_ADD, OP_ROOT_EJECT, OP_ROOT_PLUG, OP_ROOT_REMOVE, OP_ROOT_UPDATE, OP_SET_SCAN_SETTINGS, OP_TRANSFER_LIMITS, ) 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}") MIN_TRANSFER_LIMIT = 1 MAX_TRANSFER_LIMIT = 32 def _do_transfer_limits(self, msg: dict) -> None: """How many transfers one member may run at once in this group. Zero is not "unlimited" and is refused: a member who may not transfer at all is a member the operator revokes, and reading 0 as no-limit would make the most dangerous value the easiest to type by accident. """ try: downloads = int(msg.get("downloads")) uploads = int(msg.get("uploads")) except (TypeError, ValueError): self._send({"type": "error", "detail": "Invalid transfer limits"}) return for value in (downloads, uploads): if not (self.MIN_TRANSFER_LIMIT <= value <= self.MAX_TRANSFER_LIMIT): self._send({"type": "error", "detail": f"transfer limits must be between " f"{self.MIN_TRANSFER_LIMIT} and " f"{self.MAX_TRANSFER_LIMIT}"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_TRANSFER_LIMITS, f"d={downloads},u={uploads}") async def _admin_exec_transfer_limits( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: try: parts = dict(p.split("=") for p in pending["subject"].split(",")) downloads, uploads = int(parts["d"]), int(parts["u"]) except (ValueError, KeyError): self._send({"type": "error", "detail": "Invalid transfer limits"}) return if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"transfer_limits:{pending['subject']}") return try: result = await self._run_op( ops.set_transfer_limits, self._group_id or "", downloads, uploads) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("transfer_limits", pending["subject"]) notice = {"type": MNP.TRANSFER_LIMITS_ACK, "v": MNP_VERSION, "limits": result["limits"]} for session in list(self._peer_registry().values()): try: session._send(notice) except Exception: pass 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: """All groups, roots, peers — the operator's overview. Including every root's absolute path, which is why this is gated on a proved operator device and not on an account the hub named. """ node_uid = self._ctx.get("node_user_id") log.info("node_status: user=%s node_user=%s owner=%s device=%s", self._user_id, node_uid, self._is_node_admin(), "confirmed" if self._device_confirmed else "unidentified") if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return try: result = await self._run_op(ops.list_groups) self._send({"type": MNP.NODE_STATUS_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("node_status failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) async def _do_node_settings_set(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return settings = msg.get("settings", {}) if not settings: self._send({"type": "error", "detail": "No settings provided"}) return try: result = await self._run_op(ops.set_node_settings, settings) self._send({"type": MNP.NODE_SETTINGS_SET_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("node_settings_set failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) async def _do_roster_read(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return group_id = str(msg.get("group_id", "")).strip() try: result = await self._run_op(ops.read_roster, group_id) self._send({"type": MNP.ROSTER_READ_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("roster_read failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) async def _do_denylist_read(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return try: result = await self._run_op(ops.read_denylist) self._send({"type": MNP.DENYLIST_READ_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("denylist_read failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) async def _do_denylist_clear(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return subject = str(msg.get("subject", "")).strip() try: result = await self._run_op(ops.clear_denylist, subject=subject) self._audit("denylist_clear", subject or "all") self._send({"type": MNP.DENYLIST_CLEAR_ACK, "v": MNP_VERSION, **result}) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) except Exception as e: log.error("denylist_clear failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) def _do_group_attach(self, msg: dict) -> None: name = str(msg.get("name", "")).strip() shared_dir = str(msg.get("shared_dir", "")).strip() if not name or not shared_dir: self._send({"type": "error", "detail": "Missing name or shared_dir"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return # `upload_dir` is not read here any more, and a client still sending it # is ignored rather than obeyed: on load it forces every other root # read-only, which is the model the RO/RW one replaced. A second # writable directory is `root_add` with `writable`. self._issue_admin_challenge( OP_GROUP_ATTACH, name, payload={"name": name, "shared_dir": shared_dir, "writable": bool(msg.get("writable", True))}, group_id="") async def _admin_exec_group_attach( 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"group_attach:{pending['subject'][:16]}") return p = pending.get("payload") or {} try: result = await self._run_op( ops.attach_group, p["name"], p["shared_dir"], writable=bool(p.get("writable", True))) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("group_attach", pending["subject"]) self._send({"type": MNP.GROUP_ATTACH_ACK, "v": MNP_VERSION, **result}) state = self._ctx.get("daemon_state") reload_fn = state.get("reload_fn") if state else None if reload_fn: try: await reload_fn() except Exception as e: log.error("Reload after group_attach failed: %s", e) def _do_group_detach(self, msg: dict) -> None: name = str(msg.get("name", "")).strip() if not name: self._send({"type": "error", "detail": "Missing group name or id"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_GROUP_DETACH, name, payload={"name": name}, group_id="") async def _admin_exec_group_detach( 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"group_detach:{pending['subject'][:16]}") return p = pending.get("payload") or {} try: result = await self._run_op(ops.detach_group, p["name"]) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("group_detach", pending["subject"]) self._send({"type": MNP.GROUP_DETACH_ACK, "v": MNP_VERSION, **result}) state = self._ctx.get("daemon_state") reload_fn = state.get("reload_fn") if state else None if reload_fn: try: await reload_fn() except Exception as e: log.error("Reload after group_detach failed: %s", e) async def _do_node_reload(self, msg: dict) -> None: if not await self._operator_device(): self._send({"type": "error", "detail": "Not the node operator", "code": "not_operator"}) return state = self._ctx.get("daemon_state") reload_fn = state.get("reload_fn") if state else None if not reload_fn: self._send({"type": "error", "detail": "Reload not available"}) return try: await reload_fn() self._send({"type": MNP.NODE_RELOAD_ACK, "v": MNP_VERSION, "status": "reloaded"}) except Exception as e: log.error("node_reload failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Reload failed"}) def _do_root_add(self, msg: dict) -> None: target_group = str(msg.get("group_id", "")).strip() path = str(msg.get("path", "")).strip() if not target_group or not path: self._send({"type": "error", "detail": "Missing group_id or path"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge( OP_ROOT_ADD, path, payload={ "group_id": target_group, "path": path, "name": str(msg.get("name", ""))[:128], "kind": str(msg.get("kind", "generic"))[:16], "writable": bool(msg.get("writable", msg.get("upload", False))), "removable": bool(msg.get("removable", False)), }, group_id=target_group) async def _admin_exec_root_add( 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_add:{pending['subject'][:24]}") return p = pending["payload"] try: result = await self._run_op( ops.add_root, p["group_id"], p["path"], name=p.get("name", ""), kind=p.get("kind", "generic"), writable=p.get("writable", False), removable=p.get("removable", False)) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return except Exception as e: log.error("root_add failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) return self._audit("root_add", f"{p['path']}→{p['group_id'][:8]}") await self._retarget_indexer(p["group_id"]) self._send({"type": MNP.ROOT_ADD_ACK, "v": MNP_VERSION, **result}) 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_update(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 updates = [] if "writable" in msg: updates.append(f"rw={'on' if msg['writable'] else 'off'}") if "removable" in msg: updates.append(f"rem={'on' if msg['removable'] else 'off'}") subject = f"{root_name}:{','.join(updates)}" if updates else root_name self._issue_admin_challenge( OP_ROOT_UPDATE, subject, payload={ "group_id": target_group, "root_name": root_name, "writable": msg.get("writable"), "removable": msg.get("removable"), }, group_id=target_group) async def _admin_exec_root_update( 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_update:{pending['subject'][:24]}") return p = pending["payload"] try: result = await self._run_op( ops.update_root, p["group_id"], p["root_name"], writable=p.get("writable"), removable=p.get("removable")) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return except Exception as e: log.error("root_update failed: %s", e, exc_info=True) self._send({"type": "error", "detail": "Internal error"}) return self._audit("root_update", pending["subject"]) await self._retarget_indexer(p["group_id"]) notice = {"type": MNP.ROOT_UPDATE_ACK, "v": MNP_VERSION, **result} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass 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 added to or 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 added over MNP reached node.toml and was invisible until a restart; one removed kept serving its files. 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, which is why it survived until the operator added a directory from a browser. Not awaited: a reload rescans, and a new library is minutes. The ack the caller sends carries the set the node is moving to, and the `index_sync` that follows the scan carries what it found. """ 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)