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, 3 insertions, 367 deletions

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 bb5ba46..508d51d 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -69,10 +69,6 @@ from meshbay_common.adminop import (
     admin_transcript,
 )
 from meshbay_common.crypto import pk_to_b64
-from meshbay_common.groupbox import (
-    PURPOSE_ROSTER,
-    seal,
-)
 from meshbay_common.protocol import (
     MNP,
 )
@@ -102,6 +98,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.group_ops import GroupOpsMixin
 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
@@ -145,8 +142,8 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0
 
 
 class WebRTCPeerSession(
-    AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, HandshakeMixin,
-    NodeOpsMixin, TransferMixin, UploadMixin,
+    AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin,
+    HandshakeMixin, NodeOpsMixin, TransferMixin, UploadMixin,
     StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin,
 ):
     """One WebRTC peer connection, handling MNP over a DataChannel."""
@@ -527,310 +524,6 @@ class WebRTCPeerSession(
                 detail=detail,
             ))
 
-    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})
-
     def _broadcast_to_group(self, notice: dict) -> None:
         """
         Tell everyone connected to this group about a setting that changed.
@@ -860,43 +553,6 @@ class WebRTCPeerSession(
             raise ops.OpError("Node state not available", status=503)
         return await fn(state, *args, **kwargs)
 
-    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 _spawn(self, coro) -> asyncio.Task:
         """Run a coroutine in the background and hold on to it.
 
@@ -914,26 +570,6 @@ class WebRTCPeerSession(
         task.add_done_callback(_on_done)
         return task
 
-    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")}
-
     def _group_ctx(self) -> dict:
         if "groups" in self._ctx and self._group_id:
             # `.get`, not a bare subscript. A config reload removes a group