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, 316 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py
new file mode 100644
index 0000000..1600896
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py
@@ -0,0 +1,316 @@
+"""Operator authority: who counts as the node's operator, the signed challenge
+every operator op goes through, and what runs an op once its signature checks."""
+
+import base64
+import os
+import time
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import 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
+
+
+class AdminMixin:
+    # ── 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"})