path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py

"""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)