"""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 # 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_GEK_ROTATE: "_admin_exec_gek_rotate", OP_MEMBER_UNPIN: "_admin_exec_member_unpin", OP_APPS_ENABLED: "_admin_exec_apps_enabled", OP_TRANSFER_LIMITS: "_admin_exec_transfer_limits", 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_ADD: "_admin_exec_root_add", 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_UPDATE: "_admin_exec_root_update", OP_ROOT_EJECT: "_admin_exec_root_eject", OP_ROOT_PLUG: "_admin_exec_root_plug", OP_GROUP_ATTACH: "_admin_exec_group_attach", OP_GROUP_DETACH: "_admin_exec_group_detach", } 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"], ) 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))