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

import msgpack
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_INVITE_CANCEL,
    OP_INVITE_CREATE,
    OP_INVITE_LINK_CREATE,
    OP_MEMBER_REVOKE,
    OP_MUSICBRAINZ_ENABLED,
    OP_ROOT_EJECT,
    OP_ROOT_PLUG,
    OP_ROOT_REMOVE,
    OP_SEARCH_LISTED,
    OP_SET_SCAN_SETTINGS,
    OP_TMDB_CONFIG,
    OP_TMDB_ENABLED,
    OP_TMDB_OVERRIDE,
    OP_TMDB_REMATCH,
    admin_transcript,
)
from meshbay_common.crypto import pk_to_b64
from meshbay_common.protocol import MNP

# What one connection may have waiting for a signature. Anyone authenticated can
# ask for a challenge — the signature is what is checked, and it comes later — so
# without a bound a member who never answers makes the node keep every request,
# payload and all, for the life of the connection (§13.5b). A person signs one
# operation at a time; a handful covers a settings page saving several at once.
MAX_PENDING_ADMIN_OPS = 8
# The subject and payload of one pending operation, packed. A path is at most a
# few KiB, and the largest field a legitimate request carries is a directory list.
MAX_ADMIN_OP_BYTES = 64 * 1024

# Which executor runs each signed operation once its signature has been
# checked. Every one runs as a task of the session.
_ADMIN_EXECUTORS = {
    OP_FILE_DELETE: "_admin_exec_file_delete",
    OP_DIR_DELETE: "_admin_exec_dir_delete",
    OP_MEMBER_REVOKE: "_admin_exec_member_revoke",
    OP_INVITE_CREATE: "_admin_exec_invite_create",
    OP_INVITE_LINK_CREATE: "_admin_exec_invite_link_create",
    OP_INVITE_CANCEL: "_admin_exec_invite_cancel",
    OP_APPS_ENABLED: "_admin_exec_apps_enabled",
    OP_SET_SCAN_SETTINGS: "_admin_exec_set_scan_settings",
    OP_TMDB_CONFIG: "_admin_exec_tmdb_config",
    OP_TMDB_ENABLED: "_admin_exec_tmdb_enabled",
    OP_TMDB_OVERRIDE: "_admin_exec_tmdb_override",
    OP_TMDB_REMATCH: "_admin_exec_tmdb_rematch",
    OP_MUSICBRAINZ_ENABLED: "_admin_exec_musicbrainz_enabled",
    OP_ROOT_REMOVE: "_admin_exec_root_remove",
    OP_APP_DIRECTORIES: "_admin_exec_app_directories",
    OP_CHAT_DIRECTORY: "_admin_exec_chat_directory",
    OP_CHAT_LINK_PREVIEW: "_admin_exec_chat_link_preview",
    OP_SEARCH_LISTED: "_admin_exec_search_listed",
    OP_CHAT_EPOCH: "_admin_exec_chat_epoch",
    OP_ROOT_EJECT: "_admin_exec_root_eject",
    OP_ROOT_PLUG: "_admin_exec_root_plug",
}


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 "")
        now = time.time()
        for op_id, pending in list(self._admin_ops.items()):
            if now - pending["ts"] > ADMIN_CHALLENGE_TTL:
                del self._admin_ops[op_id]
        if len(self._admin_ops) >= MAX_PENDING_ADMIN_OPS:
            self._send({"type": "error", "detail": "Too many operations waiting for a "
                        "signature", "code": "too_many_pending"})
            self._audit("admin_pending_flood", op)
            return
        if len(msgpack.packb([subject, payload or {}], use_bin_type=True)) \
                > MAX_ADMIN_OP_BYTES:
            self._send({"type": "error", "detail": "Request too large",
                        "code": "too_large"})
            return
        nonce = os.urandom(32)
        ts = int(now)
        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 operator controls at all; every operation is gated on a
        signature (`_verify_admin_sig`).
        """
        node_user_id = self._ctx.get("node_user_id")
        return bool(node_user_id and self._user_id == node_user_id)

    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"],
        )

        executor = _ADMIN_EXECUTORS.get(pending["op"])
        if executor is None:
            self._send({"type": "error", "detail": "Unknown admin operation"})
        else:
            self._spawn(getattr(self, executor)(pending, transcript, sig_bytes))