refactor(node): move the operator's group controls out of webrtc_server

GroupOpsMixin in transport/webrtc/group_ops.py: member revocation and
unpinning, the group key rotation, apps and their directories with the
allow-list, and Search listing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 379 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/group_ops.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/group_ops.py
new file mode 100644
index 0000000..5f5f9ed
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/group_ops.py
@@ -0,0 +1,379 @@
+"""The operator's controls over one group: members, the group key, which apps
+it shows and where they read from, whether Search lists it."""
+
+from meshbay_common import MNP_VERSION
+from meshbay_common.adminop import (
+    OP_APP_DIRECTORIES,
+    OP_APPS_ENABLED,
+    OP_GEK_ROTATE,
+    OP_MEMBER_REVOKE,
+    OP_MEMBER_UNPIN,
+    OP_SEARCH_LISTED,
+)
+from meshbay_common.groupbox import PURPOSE_ROSTER, seal
+from meshbay_common.protocol import MNP
+
+from meshbay_node import ops
+
+
+class GroupOpsMixin:
+    def _do_member_revoke(self, msg: dict) -> None:
+        """
+        Stop serving the group key to someone, at the operator's request.
+
+        The same authority as an invite, and the same reason: the roster decides
+        who this node serves, so only a key the node pinned as an operator may
+        change it. Membership on the hub is not consulted — the hub can remove
+        someone from a group, and that stops them reaching the node at all, but
+        it cannot make the node forget them.
+        """
+        user_id = str(msg.get("user_id", "")).strip()
+        if not user_id:
+            self._send({"type": "error", "detail": "Missing user_id"})
+            return
+        if user_id == self._user_id:
+            # Removing yourself from your own node is not a member operation;
+            # it would leave the group with nobody able to invite.
+            self._send({"type": "error", "detail": "Cannot revoke yourself"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id)
+
+    def _do_gek_rotate(self, msg: dict) -> None:
+        """
+        Ask for a new group key. Operator only, and signed.
+
+        This is what actually removes a revoked member's access: revocation
+        stops the node serving the *next* key, and they still hold the current
+        one. The node generates the replacement itself — nothing arriving here
+        contributes key material, which is what the C5b rule is about.
+        """
+        group_id = str(msg.get("group_id", "")).strip() or self._group_id
+        if not group_id:
+            self._send({"type": "error", "detail": "No group on this connection"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_GEK_ROTATE, group_id, group_id=group_id)
+
+    async def _admin_exec_gek_rotate(
+        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"gek_rotate:{pending['subject'][:8]}")
+            return
+        try:
+            result = await self._run_op(
+                ops.set_gek, pending["subject"], rotate=True)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        # The operator is rotating because somebody left, and the chat archive
+        # key is not derived from the group key — so rotating that one does not
+        # move this one. Doing both here is what makes "rotate after a removal"
+        # mean the same thing for chat as it does for files.
+        await self._new_chat_epoch(pending["subject"], "gek_rotate")
+        self._audit("gek_rotate", pending["subject"])
+        self._send({
+            "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION,
+            "group_id": pending["subject"],
+            "authorized_members": result.get("authorized_members", 0),
+            # Said plainly, because rotating is the step people skip: content
+            # already downloaded stays readable to whoever holds it.
+            "note": "members re-receive the key on their next connect; content "
+                    "already downloaded is unaffected",
+        })
+
+    def _do_member_unpin(self, msg: dict) -> None:
+        """Forget a pinned identity, so someone can pair again with a new key."""
+        user_id = str(msg.get("user_id", "")).strip()
+        if not user_id:
+            self._send({"type": "error", "detail": "Missing user_id"})
+            return
+        if user_id == self._user_id:
+            # Unpinning yourself over the connection your pin authorizes would
+            # end that connection's authority mid-operation.
+            self._send({"type": "error", "detail": "Cannot unpin yourself"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_MEMBER_UNPIN, user_id)
+
+    async def _admin_exec_member_unpin(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        user_id = pending["subject"]
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"member_unpin:{user_id[:8]}")
+            return
+        try:
+            await self._run_op(ops.unpin_member, user_id)
+            await self._new_chat_epoch(self._group_id or "", "member_unpin")
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("member_unpin", user_id)
+        self._send({"type": MNP.MEMBER_UNPIN_ACK, "v": MNP_VERSION,
+                    "user_id": user_id})
+
+    # Every "application" a group can show. Photos joins this set (and
+    # apps.js's registry, client-side) when it lands; nothing else about
+    # this handler changes. DEFAULT_APPS (roster.py) deliberately does not
+    # include "video" or "music" — both can make outbound third-party
+    # network calls (TMDB, MusicBrainz) once enabled, so an operator opts a
+    # group in explicitly rather than getting it for free
+    # (docs/MESHBAY_DESIGN.md §9.7, §9.8).
+    # `helloworld` is the reference implementation (docs/MESHBAY_DESIGN.md
+    # §9.4), hidden client-side behind `?dev=1`. It is here because the
+    # allow-list is server-side enforcement — a client that names an app this
+    # node does not know is refused — and an app the node refused could not
+    # demonstrate anything. This entry and the client's registry line are the
+    # whole of what adding an application costs.
+    ALLOWED_APPS = frozenset({"chat", "files", "video", "music", "photo",
+                              "helloworld"})
+
+    def _do_apps_enabled(self, msg: dict) -> None:
+        """
+        Turn a group "application" on or off for everyone, for this group.
+
+        Signed like the root ops: this decides what a member sees, and an
+        unsigned message would let any member turn a disabled one back on.
+        """
+        apps = msg.get("apps")
+        if not isinstance(apps, list) or not apps:
+            self._send({"type": "error", "detail": "Missing or empty apps"})
+            return
+        unknown = set(apps) - self.ALLOWED_APPS
+        if unknown:
+            self._send({"type": "error",
+                       "detail": f"Unknown app(s): {', '.join(sorted(unknown))}"})
+            return
+        # Files is not a toggle: MNP permits root exploration regardless of
+        # what this list says, so hiding the tab only ever misled. Added at the
+        # front, the same order ops.set_enabled_apps writes, so the landing-tab
+        # preference sees one list and not two.
+        if "files" not in apps:
+            apps.insert(0, "files")
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        # The subject is what the operator is shown before signing, and what
+        # the client compares its own request against (transport.js) — a
+        # canonical form so both sides build the same transcript.
+        self._issue_admin_challenge(OP_APPS_ENABLED, ",".join(sorted(apps)))
+
+    async def _admin_exec_apps_enabled(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        apps = pending["subject"].split(",") if pending["subject"] else []
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"apps_enabled:{pending['subject']}")
+            return
+        try:
+            await self._run_op(
+                ops.set_enabled_apps, self._group_id or "", apps)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("apps_enabled", pending["subject"])
+
+        # Everyone already connected is told, so a disabled tab disappears
+        # without waiting for a reconnection.
+        notice = {"type": MNP.APPS_ENABLED_ACK, "v": MNP_VERSION, "apps": apps}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
+    # ── App directories (generic) ────────────────────────────────────────
+
+    def _do_app_directories(self, msg: dict) -> None:
+        """
+        Which folder(s) an application works over, for any application.
+
+        One handler for every application, keyed by the app's own name: adding
+        an application adds no message type, and there is no per-app handler
+        differing only in the key it writes and whether it carries a string or
+        a list.
+
+        `app` must be one this node knows (`ALLOWED_APPS`) — a client-supplied
+        key is otherwise a way to write arbitrary rows into `group_settings`.
+        The paths are checked by `ops._validate_app_dirs`, which runs after the
+        signature: this is a settings change, not a capability, so refusing
+        early here would be a courtesy rather than the control.
+        """
+        app = str(msg.get("app", "")).strip()
+        dirs = msg.get("directories")
+        if app not in self.ALLOWED_APPS:
+            self._send({"type": "error", "detail": f"Unknown app {app!r}"})
+            return
+        if not isinstance(dirs, list) or not all(isinstance(d, str) for d in dirs):
+            self._send({"type": "error",
+                        "detail": "Missing or invalid 'directories'"})
+            return
+        clean = sorted({d.strip("/") for d in dirs if d.strip("/")})
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        # The app is in the subject, not only the paths: an operator shown
+        # "Media/Films" alone cannot tell which application is about to be
+        # pointed at it, and two apps' challenges would be indistinguishable.
+        self._issue_admin_challenge(
+            OP_APP_DIRECTORIES, f"{app}:{','.join(clean)}")
+
+    async def _admin_exec_app_directories(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        app, _, joined = pending["subject"].partition(":")
+        dirs = joined.split(",") if joined else []
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"app_directories:{pending['subject']}")
+            return
+        try:
+            result = await self._run_op(
+                ops.set_app_directories, self._group_id or "", app, dirs)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("app_directories", pending["subject"])
+        self._broadcast_to_group({"type": MNP.APP_DIRECTORIES_ACK,
+                                  "v": MNP_VERSION, "app": app,
+                                  "directories": result["directories"]})
+
+    def _do_search_listed(self, msg: dict) -> None:
+        """
+        Whether this group's files appear in members' cross-group Search.
+        Signed because it changes what every member's Search shows, not
+        because it protects anything — see ops.set_search_listed.
+        """
+        listed = msg.get("listed")
+        if not isinstance(listed, bool):
+            self._send({"type": "error", "detail": "Missing or invalid 'listed'"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_SEARCH_LISTED, "on" if listed else "off")
+
+    async def _admin_exec_search_listed(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        listed = pending["subject"] == "on"
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"search_listed:{pending['subject']}")
+            return
+        try:
+            await self._run_op(ops.set_search_listed, self._group_id or "", listed)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("search_listed", pending["subject"])
+        self._broadcast_to_group({"type": MNP.SEARCH_LISTED_ACK,
+                                  "v": MNP_VERSION, "listed": listed})
+
+    async def _do_group_roster_req(self, msg: dict) -> None:
+        """
+        Who is in this group, and which device keys they hold.
+
+        Answers **any member**, not only the operator — that is the whole point.
+        A member verifies for themselves that a message came from a device
+        belonging to the account it claims, instead of taking the node's
+        `sender_id` on trust. What makes that possible is relayed here: each
+        device's key, which already-pinned key countersigned it, and the
+        signature plus the nonce and timestamp needed to rebuild what was
+        signed.
+
+        Sealed under a GEK-derived subkey, for the same reason the index is: it
+        is the group's membership, and a peer that has not completed the
+        handshake has no business reading it.
+
+        What this deliberately does not do is *decide* anything. The node hands
+        over evidence; the client checks the chain and keeps its own pins. A
+        node that lies here is caught by a client that has seen the account
+        before, which is the property Tier 2 buys and the reason the node is not
+        asked to assert trust.
+        """
+        gctx = self._group_ctx()
+        gek = gctx.get("gek")
+        roster = self._ctx.get("roster")
+        if not gek:
+            self._send({"type": "error", "detail": "Group encryption not initialized"})
+            return
+        if roster is None:
+            self._send({"type": "error", "detail": "Roster not available"})
+            return
+
+        devices = await roster.group_devices(self._group_id or "")
+        payload = {"devices": devices,
+                   "node_pk": self._node_pk_b64()}
+        sealed = seal(gek, PURPOSE_ROSTER, MNP.GROUP_ROSTER_RESP,
+                      self._group_id or "", payload)
+        self._send({"type": MNP.GROUP_ROSTER_RESP, "v": MNP_VERSION,
+                    "group_id": self._group_id or "", **sealed})
+
+    async def _admin_exec_member_revoke(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        user_id = pending["subject"]
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"member_revoke:{user_id[:8]}")
+            return
+
+        try:
+            result = await self._run_op(
+                ops.revoke_member, user_id, self._group_id or "")
+            await self._new_chat_epoch(self._group_id or "", "member_revoke")
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+
+        # Anyone connected right now keeps the key they already unwrapped; what
+        # they lose is the next one. Rotating it is the operator's call, and the
+        # ack says so rather than implying this undid anything already read.
+        # Every connection that account holds, not "the" one: with device
+        # linking a person may be connected from several at once, and the
+        # registry is keyed per connection precisely because it cannot hold
+        # only one of them.
+        for peer in self._sessions_of(user_id):
+            try:
+                await peer.close()
+            except Exception:
+                pass
+
+        self._audit("member_revoke", user_id)
+        self._send({
+            "type": MNP.MEMBER_REVOKE_ACK, "v": MNP_VERSION,
+            "user_id": user_id,
+            "reminder": result.get("reminder", ""),
+        })
+
+    def _app_directories_ack(self) -> dict:
+        """
+        Every application's configured folders, for the handshake ack.
+
+        Read off the group context rather than from a list of applications kept
+        here, so this cannot name an application the node knows nothing else
+        about — and cannot fail to name one the daemon does. A copy of the
+        daemon's `APP_DIR_KEYS` lived here until 2026-09-10 and had already lost
+        an entry, which made the app that entry belonged to the single one whose
+        directories never reached a client. This module names an application in
+        exactly one place, and it is `ALLOWED_APPS`.
+
+        `_app_directories_ctx` is the only thing that puts a `*_directories` key
+        in that context, and an absent one reads as none configured — never as
+        "the whole group index".
+        """
+        return {key: list(value or [])
+                for key, value in self._group_ctx().items()
+                if key.endswith("_directories")}