"""The operator's controls over one group: members, the group key, which apps it shows and where they read from, whether Search lists it.""" from meshbay_common import MNP_VERSION from meshbay_common.adminop import ( OP_APP_DIRECTORIES, OP_APPS_ENABLED, OP_GEK_ROTATE, OP_MEMBER_REVOKE, OP_MEMBER_UNPIN, OP_SEARCH_LISTED, ) from meshbay_common.groupbox import PURPOSE_ROSTER, seal from meshbay_common.protocol import MNP from meshbay_node import ops class GroupOpsMixin: def _do_member_revoke(self, msg: dict) -> None: """ Stop serving the group key to someone, at the operator's request. The same authority as an invite, and the same reason: the roster decides who this node serves, so only a key the node pinned as an operator may change it. Membership on the hub is not consulted — the hub can remove someone from a group, and that stops them reaching the node at all, but it cannot make the node forget them. """ user_id = str(msg.get("user_id", "")).strip() if not user_id: self._send({"type": "error", "detail": "Missing user_id"}) return if user_id == self._user_id: # Removing yourself from your own node is not a member operation; # it would leave the group with nobody able to invite. self._send({"type": "error", "detail": "Cannot revoke yourself"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id) def _do_gek_rotate(self, msg: dict) -> None: """ Ask for a new group key. Operator only, and signed. This is what actually removes a revoked member's access: revocation stops the node serving the *next* key, and they still hold the current one. The node generates the replacement itself — nothing arriving here contributes key material, which is what the C5b rule is about. """ group_id = str(msg.get("group_id", "")).strip() or self._group_id if not group_id: self._send({"type": "error", "detail": "No group on this connection"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_GEK_ROTATE, group_id, group_id=group_id) async def _admin_exec_gek_rotate( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"gek_rotate:{pending['subject'][:8]}") return try: result = await self._run_op( ops.set_gek, pending["subject"], rotate=True) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return # The operator is rotating because somebody left, and the chat archive # key is not derived from the group key — so rotating that one does not # move this one. Doing both here is what makes "rotate after a removal" # mean the same thing for chat as it does for files. await self._new_chat_epoch(pending["subject"], "gek_rotate") self._audit("gek_rotate", pending["subject"]) self._send({ "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION, "group_id": pending["subject"], "authorized_members": result.get("authorized_members", 0), # Said plainly, because rotating is the step people skip: content # already downloaded stays readable to whoever holds it. "note": "members re-receive the key on their next connect; content " "already downloaded is unaffected", }) def _do_member_unpin(self, msg: dict) -> None: """Forget a pinned identity, so someone can pair again with a new key.""" user_id = str(msg.get("user_id", "")).strip() if not user_id: self._send({"type": "error", "detail": "Missing user_id"}) return if user_id == self._user_id: # Unpinning yourself over the connection your pin authorizes would # end that connection's authority mid-operation. self._send({"type": "error", "detail": "Cannot unpin yourself"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_MEMBER_UNPIN, user_id) async def _admin_exec_member_unpin( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: user_id = pending["subject"] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"member_unpin:{user_id[:8]}") return try: await self._run_op(ops.unpin_member, user_id) await self._new_chat_epoch(self._group_id or "", "member_unpin") except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("member_unpin", user_id) self._send({"type": MNP.MEMBER_UNPIN_ACK, "v": MNP_VERSION, "user_id": user_id}) # Every "application" a group can show. Photos joins this set (and # apps.js's registry, client-side) when it lands; nothing else about # this handler changes. DEFAULT_APPS (roster.py) deliberately does not # include "video" or "music" — both can make outbound third-party # network calls (TMDB, MusicBrainz) once enabled, so an operator opts a # group in explicitly rather than getting it for free # (docs/MESHBAY_DESIGN.md §9.7, §9.8). # `helloworld` is the reference implementation (docs/MESHBAY_DESIGN.md # §9.4), hidden client-side behind `?dev=1`. It is here because the # allow-list is server-side enforcement — a client that names an app this # node does not know is refused — and an app the node refused could not # demonstrate anything. This entry and the client's registry line are the # whole of what adding an application costs. ALLOWED_APPS = frozenset({"chat", "files", "video", "music", "photo", "helloworld"}) def _do_apps_enabled(self, msg: dict) -> None: """ Turn a group "application" on or off for everyone, for this group. Signed like the root ops: this decides what a member sees, and an unsigned message would let any member turn a disabled one back on. """ apps = msg.get("apps") if not isinstance(apps, list) or not apps: self._send({"type": "error", "detail": "Missing or empty apps"}) return unknown = set(apps) - self.ALLOWED_APPS if unknown: self._send({"type": "error", "detail": f"Unknown app(s): {', '.join(sorted(unknown))}"}) return # Files is not a toggle: MNP permits root exploration regardless of # what this list says, so hiding the tab only ever misled. Added at the # front, the same order ops.set_enabled_apps writes, so the landing-tab # preference sees one list and not two. if "files" not in apps: apps.insert(0, "files") if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return # The subject is what the operator is shown before signing, and what # the client compares its own request against (transport.js) — a # canonical form so both sides build the same transcript. self._issue_admin_challenge(OP_APPS_ENABLED, ",".join(sorted(apps))) async def _admin_exec_apps_enabled( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: apps = pending["subject"].split(",") if pending["subject"] else [] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"apps_enabled:{pending['subject']}") return try: await self._run_op( ops.set_enabled_apps, self._group_id or "", apps) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("apps_enabled", pending["subject"]) # Everyone already connected is told, so a disabled tab disappears # without waiting for a reconnection. notice = {"type": MNP.APPS_ENABLED_ACK, "v": MNP_VERSION, "apps": apps} for uid, session in list(self._peer_registry().items()): try: session._send(notice) except Exception: pass # ── App directories (generic) ──────────────────────────────────────── def _do_app_directories(self, msg: dict) -> None: """ Which folder(s) an application works over, for any application. One handler for every application, keyed by the app's own name: adding an application adds no message type, and there is no per-app handler differing only in the key it writes and whether it carries a string or a list. `app` must be one this node knows (`ALLOWED_APPS`) — a client-supplied key is otherwise a way to write arbitrary rows into `group_settings`. The paths are checked by `ops._validate_app_dirs`, which runs after the signature: this is a settings change, not a capability, so refusing early here would be a courtesy rather than the control. """ app = str(msg.get("app", "")).strip() dirs = msg.get("directories") if app not in self.ALLOWED_APPS: self._send({"type": "error", "detail": f"Unknown app {app!r}"}) return if not isinstance(dirs, list) or not all(isinstance(d, str) for d in dirs): self._send({"type": "error", "detail": "Missing or invalid 'directories'"}) return clean = sorted({d.strip("/") for d in dirs if d.strip("/")}) if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return # The app is in the subject, not only the paths: an operator shown # "Media/Films" alone cannot tell which application is about to be # pointed at it, and two apps' challenges would be indistinguishable. self._issue_admin_challenge( OP_APP_DIRECTORIES, f"{app}:{','.join(clean)}") async def _admin_exec_app_directories( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: app, _, joined = pending["subject"].partition(":") dirs = joined.split(",") if joined else [] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"app_directories:{pending['subject']}") return try: result = await self._run_op( ops.set_app_directories, self._group_id or "", app, dirs) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("app_directories", pending["subject"]) self._broadcast_to_group({"type": MNP.APP_DIRECTORIES_ACK, "v": MNP_VERSION, "app": app, "directories": result["directories"]}) def _do_search_listed(self, msg: dict) -> None: """ Whether this group's files appear in members' cross-group Search. Signed because it changes what every member's Search shows, not because it protects anything — see ops.set_search_listed. """ listed = msg.get("listed") if not isinstance(listed, bool): self._send({"type": "error", "detail": "Missing or invalid 'listed'"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return self._issue_admin_challenge(OP_SEARCH_LISTED, "on" if listed else "off") async def _admin_exec_search_listed( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: listed = pending["subject"] == "on" if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"search_listed:{pending['subject']}") return try: await self._run_op(ops.set_search_listed, self._group_id or "", listed) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("search_listed", pending["subject"]) self._broadcast_to_group({"type": MNP.SEARCH_LISTED_ACK, "v": MNP_VERSION, "listed": listed}) async def _do_group_roster_req(self, msg: dict) -> None: """ Who is in this group, and which device keys they hold. Answers **any member**, not only the operator — that is the whole point. A member verifies for themselves that a message came from a device belonging to the account it claims, instead of taking the node's `sender_id` on trust. What makes that possible is relayed here: each device's key, which already-pinned key countersigned it, and the signature plus the nonce and timestamp needed to rebuild what was signed. Sealed under a GEK-derived subkey, for the same reason the index is: it is the group's membership, and a peer that has not completed the handshake has no business reading it. What this deliberately does not do is *decide* anything. The node hands over evidence; the client checks the chain and keeps its own pins. A node that lies here is caught by a client that has seen the account before, which is the property Tier 2 buys and the reason the node is not asked to assert trust. """ gctx = self._group_ctx() gek = gctx.get("gek") roster = self._ctx.get("roster") if not gek: self._send({"type": "error", "detail": "Group encryption not initialized"}) return if roster is None: self._send({"type": "error", "detail": "Roster not available"}) return devices = await roster.group_devices(self._group_id or "") payload = {"devices": devices, "node_pk": self._node_pk_b64()} sealed = seal(gek, PURPOSE_ROSTER, MNP.GROUP_ROSTER_RESP, self._group_id or "", payload) self._send({"type": MNP.GROUP_ROSTER_RESP, "v": MNP_VERSION, "group_id": self._group_id or "", **sealed}) async def _admin_exec_member_revoke( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: user_id = pending["subject"] if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"member_revoke:{user_id[:8]}") return try: result = await self._run_op( ops.revoke_member, user_id, self._group_id or "") await self._new_chat_epoch(self._group_id or "", "member_revoke") except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return # Anyone connected right now keeps the key they already unwrapped; what # they lose is the next one. Rotating it is the operator's call, and the # ack says so rather than implying this undid anything already read. # Every connection that account holds, not "the" one: with device # linking a person may be connected from several at once, and the # registry is keyed per connection precisely because it cannot hold # only one of them. for peer in self._sessions_of(user_id): try: await peer.close() except Exception: pass self._audit("member_revoke", user_id) self._send({ "type": MNP.MEMBER_REVOKE_ACK, "v": MNP_VERSION, "user_id": user_id, "reminder": result.get("reminder", ""), }) def _app_directories_ack(self) -> dict: """ Every application's configured folders, for the handshake ack. Read off the group context rather than from a list of applications kept here, so this cannot name an application the node knows nothing else about — and cannot fail to name one the daemon does. A copy of the daemon's `APP_DIR_KEYS` lived here until 2026-09-10 and had already lost an entry, which made the app that entry belonged to the single one whose directories never reached a client. This module names an application in exactly one place, and it is `ALLOWED_APPS`. `_app_directories_ctx` is the only thing that puts a `*_directories` key in that context, and an absent one reads as none configured — never as "the whole group index". """ return {key: list(value or []) for key, value in self._group_ctx().items() if key.endswith("_directories")}