diff options
Diffstat (limited to 'packages/meshbay-node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py | 586 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 565 |
2 files changed, 588 insertions, 563 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py new file mode 100644 index 0000000..9d58d8c --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py @@ -0,0 +1,586 @@ +"""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) 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 1393b6e..bb5ba46 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -103,6 +103,7 @@ from meshbay_node.transport.webrtc.channel import ( from meshbay_node.transport.webrtc.chat import ChatMixin from meshbay_node.transport.webrtc.files import FilesMixin from meshbay_node.transport.webrtc.handshake import HandshakeMixin +from meshbay_node.transport.webrtc.node_ops import NodeOpsMixin from meshbay_node.transport.webrtc.transfer_handlers import TransferMixin from meshbay_node.transport.webrtc.upload_handlers import UploadMixin @@ -145,7 +146,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0 class WebRTCPeerSession( AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, HandshakeMixin, - TransferMixin, UploadMixin, + NodeOpsMixin, TransferMixin, UploadMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, ): """One WebRTC peer connection, handling MNP over a DataChannel.""" @@ -844,532 +845,6 @@ 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}") - - 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 _run_op(self, fn, *args, **kwargs): """ Call an operation from `meshbay_node.ops` with the daemon's own view. @@ -1385,42 +860,6 @@ class WebRTCPeerSession( raise ops.OpError("Node state not available", status=503) return await fn(state, *args, **kwargs) - 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) - async def _admin_exec_member_revoke( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: |