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

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