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

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