refactor(node): move operator authority and the admin response out of webrtc_server

AdminMixin in transport/webrtc/admin.py: who the operator is, the signed
challenge, signature checks, and _do_admin_response moved unchanged.

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

1 files changed, 2 insertions, 307 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 508d51d..6a34b11 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -23,7 +23,6 @@ Signaling flow (handled externally by the hub):
 """
 
 import asyncio
-import base64
 import logging
 import os
 import time
@@ -33,42 +32,8 @@ from typing import Any
 from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription
 from cryptography.hazmat.primitives.asymmetric.ed25519 import (
     Ed25519PrivateKey,
-    Ed25519PublicKey,
 )
 from meshbay_common import MNP_VERSION
-from meshbay_common.adminop import (
-    ADMIN_CHALLENGE_TTL,
-    OP_APP_DIRECTORIES,
-    OP_APPS_ENABLED,
-    OP_CHAT_DIRECTORY,
-    OP_CHAT_EPOCH,
-    OP_CHAT_LINK_PREVIEW,
-    OP_DIR_DELETE,
-    OP_FILE_DELETE,
-    OP_GEK_ROTATE,
-    OP_GROUP_ATTACH,
-    OP_GROUP_DETACH,
-    OP_INVITE_CANCEL,
-    OP_INVITE_CREATE,
-    OP_INVITE_LINK_CREATE,
-    OP_MEMBER_REVOKE,
-    OP_MEMBER_UNPIN,
-    OP_MUSICBRAINZ_ENABLED,
-    OP_ROOT_ADD,
-    OP_ROOT_EJECT,
-    OP_ROOT_PLUG,
-    OP_ROOT_REMOVE,
-    OP_ROOT_UPDATE,
-    OP_SEARCH_LISTED,
-    OP_SET_SCAN_SETTINGS,
-    OP_TMDB_CONFIG,
-    OP_TMDB_ENABLED,
-    OP_TMDB_OVERRIDE,
-    OP_TMDB_REMATCH,
-    OP_TRANSFER_LIMITS,
-    admin_transcript,
-)
-from meshbay_common.crypto import pk_to_b64
 from meshbay_common.protocol import (
     MNP,
 )
@@ -84,6 +49,7 @@ from meshbay_node.indexer import GroupIndex
 from meshbay_node.roots import (
     RootSet,
 )
+from meshbay_node.transport.webrtc.admin import AdminMixin
 from meshbay_node.transport.webrtc.admission import AdmissionMixin
 from meshbay_node.transport.webrtc.apps.music import MusicMixin
 from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin
@@ -142,7 +108,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0
 
 
 class WebRTCPeerSession(
-    AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin,
+    AdminMixin, AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin,
     HandshakeMixin, NodeOpsMixin, TransferMixin, UploadMixin,
     StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin,
 ):
@@ -628,277 +594,6 @@ class WebRTCPeerSession(
         """
         self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")})
 
-    # ── Admin operation challenge/response (finding H5) ──────────────────────
-
-    def _node_pk_b64(self) -> str:
-        return pk_to_b64(self._ctx["sk_node"].public_key())
-
-    def _issue_admin_challenge(
-        self, op: str, subject: str, payload: dict | None = None,
-        group_id: str | None = None,
-    ) -> None:
-        """
-        Ask the client to authorize `op` on `subject` with its Ed25519 identity key.
-
-        The client is sent the transcript *fields*, not opaque bytes, so it can
-        rebuild and inspect what it signs. The node keeps the authoritative copy and
-        rebuilds the transcript itself at verification time — nothing signed is ever
-        taken from the response message.
-
-        `group_id` overrides the connection's group for cross-group operations
-        (e.g. root management from a NodePage connection).
-        """
-        gid = group_id if group_id is not None else (self._group_id or "")
-        nonce = os.urandom(32)
-        ts = int(time.time())
-        op_id = base64.b64encode(os.urandom(16)).decode()
-        self._admin_ops[op_id] = {
-            "op": op, "subject": subject, "nonce": nonce, "ts": ts,
-            "payload": payload or {}, "group_id": gid,
-        }
-        self._send({
-            "type": MNP.ADMIN_CHALLENGE,
-            "v": MNP_VERSION,
-            "op_id": op_id,
-            "op": op,
-            "subject": subject,
-            "nonce": base64.b64encode(nonce).decode(),
-            "ts": ts,
-            "node_pk": self._node_pk_b64(),
-            "group_id": gid,
-        })
-
-    @staticmethod
-    def _verify_sig(pk: Ed25519PublicKey | None, transcript: bytes, sig: bytes) -> bool:
-        if pk is None:
-            return False
-        try:
-            pk.verify(sig, transcript)
-            return True
-        except Exception:
-            return False
-
-    async def _load_pinned_pk(self) -> None:
-        """
-        A key this node pinned for the account we just authenticated.
-
-        `get_identity` returns the account's **oldest** live device, which is a
-        stand-in, not an answer: the handshake never said which device is on
-        this connection. `device_hello` is the answer, and it arrives later —
-        so this must never overwrite a confirmed one. It is spawned from
-        `_complete_handshake` and can therefore finish *after* a fast client has
-        already identified itself, which is exactly the ordering that would put
-        the wrong key back.
-        """
-        roster = self._ctx.get("roster")
-        if roster is None or not self._user_id or self._device_confirmed:
-            return
-        ident = await roster.get_identity(self._user_id)
-        if ident and not self._device_confirmed:
-            self._pinned_pk = ident["pk_ed25519"]
-
-    def _is_node_admin(self) -> bool:
-        """
-        Whether the **account** on this connection is the one the node belongs to.
-
-        This is a display hint and half of a check — never authority on its own.
-        `self._user_id` is the `sub` of a JWT the hub issued, so read alone it
-        says "the hub says you are the owner", which is the one thing NS4 and
-        M3 rule out: a hub that can name the operator can install itself as
-        node administrator. It rides the handshake ack so a client knows whether
-        to offer the Node page at all, and every operation is gated on
-        `_operator_device()` below.
-        """
-        node_user_id = self._ctx.get("node_user_id")
-        return bool(node_user_id and self._user_id == node_user_id)
-
-    async def _operator_device(self) -> bool:
-        """
-        Whether this connection may run the node's own controls.
-
-        Two things, and the second is the one that cannot be forged:
-
-        - the account is the one this node belongs to (`_is_node_admin`), which
-          is what keeps node-wide controls with the machine's owner rather than
-          with every paired operator of every group on it; and
-        - **the device on this connection proved a key the node pinned as an
-          operator**. `device_hello` is signed over a transcript naming this
-          node, this group and this connection's nonce, and `operator_pks()` is
-          rebuilt from the roster on each call, so an unpinned browser and a
-          revoked one are both refused at once.
-
-        The second clause is the fix for the door this used to leave open.
-        `node_status`, `node_settings_set`, `roster_read`, `denylist_read`,
-        `denylist_clear` and `node_reload` were gated on the account id alone —
-        a value the hub chooses. An active hub that can also reach the group key
-        (which §3.5 concedes it can in an open-join group) could therefore mint
-        a token for the owner's account and read `node_status`, which lists
-        every group on the node with the operator's **absolute paths**, or clear
-        the denylist, which is the persisted revocation H4 exists to keep.
-
-        It holds no user keys and cannot countersign anything, so it cannot
-        produce a `device_hello` — which is the same property device linking
-        rests on (§3.3), applied to the node's own surface.
-        """
-        if not self._is_node_admin():
-            return False
-        if not self._device_confirmed or not self._pinned_pk:
-            return False
-        roster = self._ctx.get("roster")
-        if roster is None:
-            return False
-        return self._pinned_pk in await roster.operator_pks()
-
-    def _has_admin_authority(self) -> bool:
-        """
-        Cheap synchronous pre-check: is there anyone who could authorize this?
-
-        Only decides whether to issue a challenge at all — the gate is
-        `_verify_admin_sig`. The flag is set at startup and refreshed in-process
-        when an operator pairs.
-        """
-        return bool(self._ctx.get("has_admin_authority"))
-
-    async def _verify_admin_sig(self, transcript: bytes, sig: bytes) -> bool:
-        """
-        Check a signature against every key holding node-operator authority.
-
-        Read from the roster on each call rather than cached: revoking a paired
-        browser must take effect immediately, and admin operations are rare enough
-        that a SQLite read costs nothing.
-
-        There is one source of operator authority and this is it. `admin_pk_ed25519`
-        in node.toml used to be honoured alongside the roster; it is gone, and a
-        config that still names it is warned about at startup rather than obeyed.
-        """
-        roster = self._ctx.get("roster")
-        if roster is None:
-            return False
-        for pk_b64 in await roster.operator_pks():
-            try:
-                pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_b64))
-            except Exception:
-                continue
-            if self._verify_sig(pk, transcript, sig):
-                return True
-        return False
-
-    def _do_admin_response(self, msg: dict) -> None:
-        op_id = msg.get("op_id", "")
-        sig_b64 = msg.get("signature", "")
-
-        pending = self._admin_ops.pop(op_id, None)
-        if not pending:
-            self._send({"type": "error", "detail": "No pending admin operation"})
-            return
-
-        if time.time() - pending["ts"] > ADMIN_CHALLENGE_TTL:
-            self._send({"type": "error", "detail": "Admin challenge expired"})
-            return
-
-        try:
-            sig_bytes = base64.b64decode(sig_b64)
-        except Exception:
-            self._send({"type": "error", "detail": "Invalid signature encoding"})
-            return
-
-        transcript = admin_transcript(
-            op=pending["op"],
-            node_pk_b64=self._node_pk_b64(),
-            group_id=(pending["group_id"] if pending.get("group_id") is not None
-                      else (self._group_id or "")),
-            subject=pending["subject"],
-            nonce=pending["nonce"],
-            ts=pending["ts"],
-        )
-
-        if pending["op"] == OP_FILE_DELETE:
-            self._spawn(
-                self._admin_exec_file_delete(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_DIR_DELETE:
-            self._spawn(
-                self._admin_exec_dir_delete(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_MEMBER_REVOKE:
-            self._spawn(
-                self._admin_exec_member_revoke(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_INVITE_CREATE:
-            self._spawn(
-                self._admin_exec_invite_create(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_INVITE_LINK_CREATE:
-            self._spawn(
-                self._admin_exec_invite_link_create(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_INVITE_CANCEL:
-            self._spawn(
-                self._admin_exec_invite_cancel(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_GEK_ROTATE:
-            self._spawn(
-                self._admin_exec_gek_rotate(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_MEMBER_UNPIN:
-            self._spawn(
-                self._admin_exec_member_unpin(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_APPS_ENABLED:
-            self._spawn(
-                self._admin_exec_apps_enabled(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_TRANSFER_LIMITS:
-            self._spawn(
-                self._admin_exec_transfer_limits(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_SET_SCAN_SETTINGS:
-            self._spawn(
-                self._admin_exec_set_scan_settings(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_TMDB_CONFIG:
-            self._spawn(
-                self._admin_exec_tmdb_config(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_TMDB_ENABLED:
-            self._spawn(
-                self._admin_exec_tmdb_enabled(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_TMDB_OVERRIDE:
-            self._spawn(
-                self._admin_exec_tmdb_override(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_TMDB_REMATCH:
-            self._spawn(
-                self._admin_exec_tmdb_rematch(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_MUSICBRAINZ_ENABLED:
-            self._spawn(
-                self._admin_exec_musicbrainz_enabled(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_ROOT_ADD:
-            self._spawn(
-                self._admin_exec_root_add(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_ROOT_REMOVE:
-            self._spawn(
-                self._admin_exec_root_remove(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_APP_DIRECTORIES:
-            self._spawn(
-                self._admin_exec_app_directories(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_CHAT_DIRECTORY:
-            self._spawn(
-                self._admin_exec_chat_directory(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_CHAT_LINK_PREVIEW:
-            self._spawn(
-                self._admin_exec_chat_link_preview(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_SEARCH_LISTED:
-            self._spawn(
-                self._admin_exec_search_listed(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_CHAT_EPOCH:
-            self._spawn(
-                self._admin_exec_chat_epoch(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_ROOT_UPDATE:
-            self._spawn(
-                self._admin_exec_root_update(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_ROOT_EJECT:
-            self._spawn(
-                self._admin_exec_root_eject(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_ROOT_PLUG:
-            self._spawn(
-                self._admin_exec_root_plug(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_GROUP_ATTACH:
-            self._spawn(
-                self._admin_exec_group_attach(pending, transcript, sig_bytes))
-        elif pending["op"] == OP_GROUP_DETACH:
-            self._spawn(
-                self._admin_exec_group_detach(pending, transcript, sig_bytes))
-        else:
-            self._send({"type": "error", "detail": "Unknown admin operation"})
-
     def _send(self, obj: dict) -> None:
         # Stamp the reply with the id of the request being answered, so the
         # caller never has to guess. Only for this session's own replies: a