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