refactor(node): move invitations, joins and device linking out of webrtc_server

AdmissionMixin in transport/webrtc/admission.py, with the invite handle
pattern and the join attempt bounds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 3 insertions, 847 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 9264b60..73fefd8 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -26,7 +26,6 @@ import asyncio
 import base64
 import logging
 import os
-import re
 import time
 import uuid
 from typing import Any
@@ -69,13 +68,7 @@ from meshbay_common.adminop import (
     OP_TRANSFER_LIMITS,
     admin_transcript,
 )
-from meshbay_common.crypto import pk_to_b64, wrap_gek_aes
-from meshbay_common.device import (
-    DEVICE_TTL,
-    device_add_transcript,
-    device_hello_transcript,
-    device_request_transcript,
-)
+from meshbay_common.crypto import pk_to_b64
 from meshbay_common.groupbox import (
     PURPOSE_ACK,
     PURPOSE_ROSTER,
@@ -95,12 +88,6 @@ from meshbay_common.handshake import (
     verify_proof,
     webrtc_binding,
 )
-from meshbay_common.join import (
-    JOIN_TTL,
-    ROLE_MEMBER,
-    ROLE_OPERATOR,
-    join_transcript,
-)
 from meshbay_common.protocol import (
     MNP,
 )
@@ -117,7 +104,7 @@ from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node.roots import (
     RootSet,
 )
-from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK
+from meshbay_node.transport.webrtc.admission import AdmissionMixin
 from meshbay_node.transport.webrtc.apps.music import MusicMixin
 from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin
 from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin
@@ -138,9 +125,6 @@ from meshbay_node.transport.webrtc.upload_handlers import UploadMixin
 
 log = logging.getLogger(__name__)
 
-# An invitation link's handle, as `roster.create_link_invite` mints it.
-_INVITE_ID_RE = re.compile(r"[0-9a-f]{32}")
-
 
 # Budget for an unauthenticated peer: enough for a handshake and a bundle fetch,
 # nowhere near enough to be a memory-exhaustion primitive (H6).
@@ -160,21 +144,12 @@ UNAUTHENTICATED_SESSION_TIMEOUT = 60   # seconds
 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited
 # until the native client removes remote keypair bundles entirely.
 MAX_PRE_PROOF_FETCHES = 4
-# Pairing codes carry 40 bits and are single-use, but a connection must not be
-# allowed to sit there guessing. Failures are audited, so a grind is visible.
-MAX_JOIN_ATTEMPTS = 5
-# Per-connection limits alone would not bind an attacker who can open connections
-# at will — and the adversary who can mint tokens for any account is the hub. So
-# failed pairings are also counted node-wide over a window.
-MAX_JOIN_FAILURES_WINDOW = 20
-JOIN_FAILURE_WINDOW = 600  # seconds
 # Everything a member sends lands here: files from the Files panel and
 # attachments from the chat alike. One visible directory the operator can look
 # into, back up or empty — rather than a hidden tree of per-user uuids that
 # nobody could read, or files scattered wherever someone happened to be looking.
 
 
-
 # Opt-in, off by default: a per-session heartbeat log (message count, time
 # since the last message, ICE state) and ICE-state-change logging, on top of
 # the connectionstatechange logging that already runs unconditionally. Added
@@ -190,7 +165,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0
 
 
 class WebRTCPeerSession(
-    BlobsMixin, ChatMixin, FilesMixin, TransferMixin, UploadMixin,
+    AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, TransferMixin, UploadMixin,
     StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin,
 ):
     """One WebRTC peer connection, handling MNP over a DataChannel."""
@@ -856,80 +831,6 @@ class WebRTCPeerSession(
         if note_activity:
             note_activity()
 
-    def _do_invite_create(self, msg: dict) -> None:
-        """
-        Issue a one-time pairing code for someone the operator wants to admit.
-
-        Replaces the old invite path, where the inviter fetched the invitee's
-        public key from the hub and wrapped the group key for whatever came back
-        (H3). The node now needs nothing but a name: it will wrap the key itself,
-        later, for a key the invitee proves they hold.
-        """
-        roster = self._ctx.get("roster")
-        if roster is None:
-            self._send({"type": "error", "detail": "Roster not available"})
-            return
-
-        invitee_id = msg.get("user_id", "")
-        group_id = msg.get("group_id") or self._group_id
-        if not invitee_id or not group_id:
-            self._send({"type": "error", "detail": "Missing user_id or group_id"})
-            return
-        if group_id != self._group_id:
-            self._send({"type": "error", "detail": "Wrong group for this session"})
-            return
-
-        if not self._has_admin_authority():
-            self._send({
-                "type": "error",
-                "detail": "No operator paired — run `meshbay-node operator pair`",
-            })
-            return
-
-        self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, {
-            "group_id": group_id,
-            "user_id": invitee_id,
-            "username": str(msg.get("username", ""))[:64],
-        })
-
-    def _do_invite_link_create(self, msg: dict) -> None:
-        """
-        Issue a code bound to no account, for an invitation link — into the
-        group this connection authenticated to, and no other: a link names its
-        group, so the operator signs for exactly that one (docs/MESHBAY_DESIGN.md §3.4).
-        """
-        group_id = self._group_id or ""
-        if not group_id:
-            self._send({"type": "error", "detail": "No group on this connection"})
-            return
-        if msg.get("group_id") and msg["group_id"] != group_id:
-            self._send({"type": "error", "detail": "Wrong group for this session"})
-            return
-        if not self._has_admin_authority():
-            self._send({
-                "type": "error",
-                "detail": "No operator paired — run `meshbay-node operator pair`",
-            })
-            return
-        self._issue_admin_challenge(
-            OP_INVITE_LINK_CREATE, f"link:{group_id}", {"group_id": group_id})
-
-    def _do_invite_cancel(self, msg: dict) -> None:
-        """Take back an unredeemed link of this group, by its handle."""
-        group_id = self._group_id or ""
-        invite_id = str(msg.get("invite_id", ""))
-        if not group_id:
-            self._send({"type": "error", "detail": "No group on this connection"})
-            return
-        if not _INVITE_ID_RE.fullmatch(invite_id):
-            self._send({"type": "error", "detail": "Not an invitation id"})
-            return
-        if not self._has_admin_authority():
-            self._send({"type": "error", "detail": "No authorized key for this"})
-            return
-        self._issue_admin_challenge(
-            OP_INVITE_CANCEL, invite_id, {"group_id": group_id, "invite_id": invite_id})
-
     # ── Per-account blobs (playlists, docs/playlists.md §8.2) ────────────────
     #
     # The node stores bytes it cannot read and hands them back. `self._user_id`
@@ -939,239 +840,6 @@ class WebRTCPeerSession(
 
     # ── Pairing and join (H3, M3) ────────────────────────────────────────────
 
-    def _join_refuse(self, reason: str, audit_detail: str = "") -> None:
-        self._join_attempts += 1
-        # Node-wide window, shared across connections: reconnecting must not reset
-        # the budget.
-        now = time.time()
-        failures = [t for t in self._ctx.get("join_failures", [])
-                    if now - t < JOIN_FAILURE_WINDOW]
-        failures.append(now)
-        self._ctx["join_failures"] = failures
-        self._audit_join("join_refused", audit_detail or reason)
-        self._send({
-            "type": MNP.JOIN_RESULT,
-            "v": MNP_VERSION,
-            "ok": False,
-            "reason": reason,
-        })
-
-    def _audit_join(self, event: str, detail: str) -> None:
-        audit = self._ctx.get("audit_store")
-        if not audit:
-            return
-        self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
-        self._spawn(audit.log_event(
-            user_id=self._user_id or getattr(self, "_pending_sub", "unknown"),
-            event=event,
-            ip=self._remote_ip,
-            username=self._username or getattr(self, "_pending_username", ""),
-            group_id=self._group_id or getattr(self, "_pending_group", "") or "",
-            detail=detail,
-        ))
-
-    async def _do_join_request(self, msg: dict) -> None:
-        """
-        Pin an identity, or recognise one already pinned.
-
-        The client signs its own Ed25519 and X25519 keys together with the node's
-        nonce, so the identity key vouches for the encryption key — that is what
-        will make it safe for the node to wrap the GEK for a key that arrived over
-        the wire instead of one fetched from the hub's directory (H3).
-
-        A first pairing needs a one-time code, which the hub never sees. Afterwards
-        the pin is the credential and a changed key is refused outright, the same
-        rule the client applies to `pk_node` (11.5.8).
-        """
-        roster = self._ctx.get("roster")
-        if roster is None:
-            self._send({"type": "error", "detail": "Roster not available"})
-            return
-
-        if self._join_attempts >= MAX_JOIN_ATTEMPTS:
-            self._send({"type": "error", "detail": "Too many attempts"})
-            return
-
-        now = time.time()
-        recent = [t for t in self._ctx.get("join_failures", [])
-                  if now - t < JOIN_FAILURE_WINDOW]
-        if len(recent) >= MAX_JOIN_FAILURES_WINDOW:
-            self._audit_join("join_throttled", f"{len(recent)} failures in window")
-            self._send({"type": "error", "detail": "Pairing temporarily locked"})
-            return
-
-        user_id = self._user_id or getattr(self, "_pending_sub", "")
-        username = self._username or getattr(self, "_pending_username", "")
-        if not user_id:
-            self._send({"type": "error", "detail": "Handshake required"})
-            return
-
-        pk_ed_b64 = msg.get("pk_ed25519", "")
-        pk_x_b64 = msg.get("pk_x25519", "")
-        code = msg.get("code", "")
-        ts = msg.get("ts", 0)
-
-        try:
-            pk_ed_raw = base64.b64decode(pk_ed_b64)
-            pk_x_raw = base64.b64decode(pk_x_b64)
-            if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32:
-                raise ValueError
-            pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw)
-        except Exception:
-            self._join_refuse("invalid_keys")
-            return
-
-        if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL:
-            self._join_refuse("stale_request")
-            return
-
-        # An empty group_id means operator pairing, which is node-wide. Anything
-        # else must be the group this connection authenticated to — a signature
-        # obtained for one group must not name another.
-        group_id = msg.get("group_id", "") or ""
-        session_group = self._group_id or getattr(self, "_pending_group", "") or ""
-        if group_id and group_id != session_group:
-            self._join_refuse("group_mismatch")
-            return
-
-        transcript = join_transcript(
-            node_pk_b64=self._node_pk_b64(),
-            group_id=group_id,
-            user_id=user_id,
-            pk_ed25519_b64=pk_ed_b64,
-            pk_x25519_b64=pk_x_b64,
-            nonce_node=self._nonce_node,
-            ts=ts,
-        )
-        try:
-            sig = base64.b64decode(msg.get("sig", ""))
-        except Exception:
-            self._join_refuse("invalid_signature_encoding")
-            return
-        if not self._verify_sig(pk_ed, transcript, sig):
-            self._join_refuse("signature_invalid")
-            return
-
-        # One person may hold several devices here — a browser and a desktop
-        # client are two keys on one account. So the question is not "is this
-        # THE key" but "is this ONE OF this account's live devices".
-        device = await roster.find_device(user_id, pk_ed_b64)
-        if device and device["pk_x25519"] != pk_x_b64:
-            # The Ed25519 key is pinned but arrives with a different encryption
-            # key. The join transcript signs both together, so this is either a
-            # client that regenerated half its identity or something splicing
-            # two messages; either way the pair is not the one admitted.
-            self._join_refuse(
-                "key_changed",
-                f"pinned x25519={device['pk_x25519'][:16]} presented={pk_x_b64[:16]}")
-            return
-        known = device
-        if not known and await roster.list_devices(user_id):
-            # The account is known here but this key is not one of its devices.
-            # Not an error to shout about: it is a second browser or a new
-            # client, and the way in is a device-add approved by a device that
-            # is already trusted — no operator, no new invitation code.
-            self._join_refuse(
-                "unknown_device",
-                f"presented={pk_ed_b64[:16]} — approve it from a device already "
-                f"paired with this node")
-            return
-        if known:
-            # This group's own row first; then the join message's group_id (empty
-            # on the node-wide first connect); then the operator's node-wide row,
-            # which is where an operator opening any group finds their authority.
-            member = (await roster.get_member(session_group, user_id)
-                      or await roster.get_member(group_id, user_id)
-                      or await roster.get_member("", user_id))
-            if not member and self._group_join_policy(session_group) == "open":
-                await roster.set_member(
-                    group_id=session_group, user_id=user_id, role=ROLE_MEMBER,
-                    status="active", approved_by="open-join",
-                )
-                member = await roster.get_member(session_group, user_id)
-
-            # A pending invite means the operator explicitly re-invited this
-            # person — require the code even if they already have a member
-            # row (e.g. they left and were re-invited, or were revoked then
-            # re-invited).  Without this gate a stale roster row lets them
-            # back in without proving they received the new code.
-            pending_invite = any(
-                i["kind"] == KIND_ACCOUNT
-                and i["user_id"] == user_id
-                and i["group_id"] in (session_group, "")
-                for i in await roster.list_invites())
-            # Or they bring a link for this group: somebody already pinned here
-            # through another group, which is the ordinary case for a link, or
-            # somebody removed from it and invited back. Only when they are not
-            # an active member — a member opening the group leaves the link for
-            # whoever it was meant for.
-            active = bool(member) and member.get("status") == "active"
-            if pending_invite or (code and not active):
-                if not code:
-                    self._join_refuse("code_required")
-                    return
-                invite = await roster.consume_invite(code, user_id, session_group)
-                if not invite:
-                    self._join_refuse("code_invalid")
-                    return
-                await roster.set_member(
-                    group_id=invite["group_id"], user_id=user_id,
-                    role=invite["role"], status="active",
-                    approved_by=invite["created_by"],
-                )
-                self._audit_join(
-                    "join_pinned",
-                    f"group={invite['group_id'][:8]} role={invite['role']} "
-                    f"via={'link' if invite['kind'] == KIND_LINK else 'code'} "
-                    "(device already known)")
-                member = (await roster.get_member(session_group, user_id)
-                          or await roster.get_member(invite["group_id"], user_id))
-
-            if not member:
-                self._join_refuse("not_authorized_for_group")
-                return
-            await self._join_ok(
-                user_id, pk_x_raw, session_group,
-                role=member["role"] if member else "",
-                recognised=True,
-            )
-            return
-
-        if not code:
-            if self._group_join_policy(session_group) == "open":
-                # An open-join group admits anyone the hub calls a member, so a
-                # code would protect nothing — the hub can walk in through the
-                # front door. Pin what turns up and say so in the audit log.
-                await self._pin_and_admit(
-                    roster, user_id, username, pk_ed_b64, pk_x_b64,
-                    group_id=session_group, role=ROLE_MEMBER,
-                    approved_by="open-join", via="tofu")
-                await self._join_ok(user_id, pk_x_raw, session_group,
-                                    role=ROLE_MEMBER, recognised=False)
-                return
-            self._join_refuse("code_required")
-            return
-
-        invite = await roster.consume_invite(code, user_id, session_group)
-        if not invite:
-            self._join_refuse("code_invalid")
-            return
-
-        await self._pin_and_admit(
-            # The name comes from the invitation, not from the token: the hub does
-            # not put a username claim in a JWT, so pinning from the session alone
-            # left the roster nameless and `member revoke <name>` unable to match.
-            roster, user_id, invite["username"] or username, pk_ed_b64, pk_x_b64,
-            group_id=invite["group_id"], role=invite["role"],
-            approved_by=invite["created_by"],
-            via="link" if invite["kind"] == KIND_LINK else "code")
-        # The roster row comes from the invitation; the key comes from the
-        # connection. An operator pairing is node-wide (empty group), but they
-        # redeemed the code while opening a group and expect to read it — and
-        # is_authorized() already grants an operator every group on this node.
-        await self._join_ok(user_id, pk_x_raw, session_group or invite["group_id"],
-                            role=invite["role"], recognised=False)
-
 
     # ── Device linking ───────────────────────────────────────────────────────
     #
@@ -1180,443 +848,6 @@ class WebRTCPeerSession(
     # no user keys since 2026-08-14 and therefore cannot countersign anything.
     # See docs/MESHBAY_DESIGN.md §3.3.
 
-    async def _do_device_request(self, msg: dict) -> None:
-        """
-        A new device files itself as pending, bound to a code it displays.
-
-        Served in the pre-proof window: by construction the caller holds no key
-        this node knows, so there is nothing yet to prove. Filing is inert —
-        nothing is admitted until an existing device countersigns.
-        """
-        roster = self._ctx.get("roster")
-        if roster is None or not self._user_id or not self._nonce_node:
-            self._send({"type": "error", "detail": "Not ready for a device request"})
-            return
-
-        if not self._spend_device_attempt():
-            return
-
-        pk_ed_b64 = str(msg.get("pk_ed25519", ""))
-        pk_x_b64 = str(msg.get("pk_x25519", ""))
-        code_hash = str(msg.get("code_hash", ""))
-        if not (pk_ed_b64 and pk_x_b64 and code_hash):
-            self._send({"type": "error", "detail": "Missing device keys or code"})
-            return
-
-        # The account must already be known here. Anti-spam rather than a
-        # security boundary: the filing key is unpinned by construction, so this
-        # bounds the table, not the trust.
-        existing = await roster.list_devices(self._user_id)
-        if not existing:
-            self._send({"type": "error",
-                        "detail": "This account has no device on this node yet — "
-                                  "an invitation code is what admits the first"})
-            return
-        if len(existing) >= roster.MAX_DEVICES_PER_USER:
-            self._send({"type": "error",
-                        "detail": f"Already {len(existing)} devices, which is the "
-                                  f"limit. Revoke one first."})
-            return
-
-        ts = int(msg.get("ts", 0))
-        if abs(time.time() - ts) > DEVICE_TTL:
-            self._send({"type": "error", "detail": "Device request expired"})
-            return
-
-        transcript = device_request_transcript(
-            node_pk_b64=self._node_pk_b64(), user_id=self._user_id,
-            pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64,
-            code_hash=code_hash, nonce_node=self._nonce_node, ts=ts)
-        try:
-            pk_ed = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64))
-            sig = base64.b64decode(msg.get("sig", ""))
-        except Exception:
-            self._send({"type": "error", "detail": "Invalid device key encoding"})
-            return
-        if not self._verify_sig(pk_ed, transcript, sig):
-            # Proof of possession, and nothing more: this says the caller holds
-            # the keys, never that they belong to this account.
-            self._send({"type": "error", "detail": "Device signature invalid"})
-            return
-
-        ttl = int(self._ctx.get("device_request_ttl") or 3600)
-        expires = await roster.file_device_request(
-            user_id=self._user_id, username=self._username or "",
-            pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64,
-            code_hash=code_hash, ttl=ttl)
-        self._audit("device_request", f"{pk_ed_b64[:16]}")
-        log.info("Device request filed for %s (%s)", self._user_id[:8],
-                 pk_ed_b64[:16])
-        self._send({"type": MNP.DEVICE_REQUEST_ACK, "v": MNP_VERSION,
-                    "expires_at": expires})
-
-    async def _do_device_lookup(self, msg: dict) -> None:
-        """
-        List this account's pending device requests, each with its code hash.
-
-        **The node never learns the code**, which is what makes it unable to
-        substitute a key. It answers with candidates; the approver recomputes
-        `sha256(code ‖ keys)` for each and keeps the one that matches. A node
-        offering fabricated keys would have to produce a hash matching
-        `sha256(code ‖ fabricated)` — and it does not know the code.
-
-        An earlier version of this took the hash from the client and looked the
-        request up by it. That is circular: the client cannot compute the hash
-        without already knowing the keys it is asking about.
-        """
-        roster = self._ctx.get("roster")
-        if roster is None or not self._user_id:
-            self._send({"type": "error", "detail": "Roster not available"})
-            return
-
-        pending = await roster.list_device_requests(self._user_id)
-        self._send({
-            "type": MNP.DEVICE_LOOKUP_RESULT, "v": MNP_VERSION,
-            "requests": [
-                {"pk_ed25519": r["pk_ed25519"], "pk_x25519": r["pk_x25519"],
-                 "code_hash": r["code_hash"], "created_at": r["created_at"]}
-                for r in pending
-            ],
-        })
-
-    async def _do_device_add(self, msg: dict) -> None:
-        """
-        Admit a device, countersigned by one this node already pinned.
-
-        The whole control is in `_verify_device_signer`: the signature must
-        verify against a **live device of this same account**. The hub holds no
-        user keys and so cannot produce one.
-        """
-        roster = self._ctx.get("roster")
-        if roster is None or not self._user_id or not self._nonce_node:
-            self._send({"type": "error", "detail": "Not ready to add a device"})
-            return
-        if not self._spend_device_attempt():
-            return
-
-        pk_ed_b64 = str(msg.get("pk_ed25519", ""))
-        pk_x_b64 = str(msg.get("pk_x25519", ""))
-        ts = int(msg.get("ts", 0))
-        if not (pk_ed_b64 and pk_x_b64):
-            self._send({"type": "error", "detail": "Missing device keys"})
-            return
-        if abs(time.time() - ts) > DEVICE_TTL:
-            self._send({"type": "error", "detail": "Approval expired"})
-            return
-
-        transcript = device_add_transcript(
-            node_pk_b64=self._node_pk_b64(), user_id=self._user_id,
-            pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64,
-            nonce_node=self._nonce_node, ts=ts)
-        signer = await self._verify_device_signer(roster, transcript,
-                                                  msg.get("sig", ""))
-        if signer is None:
-            self._audit("device_add_refused", pk_ed_b64[:16])
-            self._send({"type": "error",
-                        "detail": "Not signed by a device already paired here"})
-            return
-
-        devices = await roster.list_devices(self._user_id)
-        if len(devices) >= roster.MAX_DEVICES_PER_USER:
-            self._send({"type": "error", "detail": "Device limit reached"})
-            return
-
-        # Spend the request. Single use: an approval cannot be replayed, and a
-        # code that was used is gone whatever else happens next.
-        code_hash = str(msg.get("code_hash", ""))
-        if code_hash and not await roster.take_device_request(
-                code_hash, self._user_id):
-            self._send({"type": "error",
-                        "detail": "That request is no longer pending"})
-            return
-
-        # The countersignature is **kept**, with the two fields needed to rebuild
-        # what it signed. Until 2026-09-07 it was verified here and thrown away,
-        # leaving only `added_by_pk` — which says *which* key approved and
-        # proves nothing to anyone else. `device_add_transcript` binds
-        # `nonce_node`, this connection's handshake nonce, so a stored signature
-        # without it is still unverifiable; that is why all three go in.
-        #
-        # This is what lets another member check for themselves that this device
-        # belongs to an account whose earlier device they have already pinned,
-        # instead of taking the node's word (Tier 2, docs/MESHBAY_DESIGN.md §3.3).
-        await roster.pin_identity(
-            user_id=self._user_id, username=self._username or "",
-            pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via="device",
-            label=str(msg.get("label", ""))[:64], added_by_pk=signer,
-            add_sig=str(msg.get("sig", "")),
-            add_nonce=base64.b64encode(self._nonce_node).decode(),
-            add_ts=ts)
-        self._audit("device_added", f"{pk_ed_b64[:16]} by {signer[:16]}")
-        log.info("Device added for %s: %s (approved by %s)",
-                 self._user_id[:8], pk_ed_b64[:16], signer[:16])
-        self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION,
-                    "pk_ed25519": pk_ed_b64})
-
-    async def _do_device_hello(self, msg: dict) -> None:
-        """
-        Learn which of this account's devices is on this connection.
-
-        The handshake authenticates a *group membership* (the GEK-HMAC) and an
-        *account* (the hub's token). It has never authenticated a device, and
-        while one account meant one key that was the same statement. It stopped
-        being so on 2026-08-18, and `_load_pinned_pk` — which resolves the
-        account's oldest live device — has been standing in for the real answer
-        ever since, including as the recorded uploader of every file.
-
-        What is checked, in order: the key is a live device *of this account* in
-        the node's own roster (never a token claim — that is
-        `docs/MESHBAY_DESIGN.md` §3.2's rule), the timestamp is fresh, and the
-        signature verifies over a transcript naming this node, this group and
-        this connection's nonce. A key that is merely well-formed proves nothing.
-
-        Idempotent for the same key, refused for a different one: a connection
-        does not get to change device half way through, which would let one
-        session's uploads be attributed to two.
-        """
-        roster = self._ctx.get("roster")
-        if roster is None or not self._user_id:
-            self._send({"type": "error", "detail": "Roster not available"})
-            return
-        if not self._spend_device_attempt():
-            return
-
-        pk_ed_b64 = str(msg.get("pk_ed25519", ""))
-        ts = int(msg.get("ts", 0) or 0)
-        if not pk_ed_b64:
-            self._send({"type": "error", "detail": "Missing device key"})
-            return
-        if self._device_confirmed and pk_ed_b64 != self._pinned_pk:
-            self._send({"type": "error",
-                        "detail": "This connection is already another device"})
-            return
-        if abs(time.time() - ts) > DEVICE_TTL:
-            self._send({"type": "error", "detail": "Stale device_hello"})
-            return
-
-        device = await roster.find_device(self._user_id, pk_ed_b64)
-        if device is None:
-            self._audit("device_hello_refused", pk_ed_b64[:16])
-            self._send({"type": "error",
-                        "detail": "Not a device paired here"})
-            return
-
-        transcript = device_hello_transcript(
-            node_pk_b64=self._node_pk_b64(), group_id=self._group_id or "",
-            user_id=self._user_id, pk_ed25519_b64=pk_ed_b64,
-            nonce_node=self._nonce_node, ts=ts)
-        try:
-            pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64))
-        except Exception:
-            self._send({"type": "error", "detail": "Unreadable device key"})
-            return
-        try:
-            sig = base64.b64decode(msg.get("sig", ""))
-        except Exception:
-            sig = b""
-        if not self._verify_sig(pk, transcript, sig):
-            self._audit("device_hello_refused", pk_ed_b64[:16])
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            return
-
-        self._pinned_pk = pk_ed_b64
-        self._device_confirmed = True
-        log.info("Device identified on connection: user=%s device=%s",
-                 self._user_id[:8], pk_ed_b64[:16])
-        self._send({"type": MNP.DEVICE_HELLO_ACK, "v": MNP_VERSION,
-                    "pk_ed25519": pk_ed_b64})
-
-    async def _do_device_list(self, msg: dict) -> None:
-        """This account's devices. Anyone may read their own, nobody else's."""
-        roster = self._ctx.get("roster")
-        if roster is None or not self._user_id:
-            self._send({"type": "error", "detail": "Roster not available"})
-            return
-        devices = await roster.list_devices(self._user_id)
-        pending = await roster.pending_device_requests(self._user_id)
-        self._send({
-            "type": MNP.DEVICE_LIST_RESULT, "v": MNP_VERSION,
-            "pending": pending,
-            "devices": [
-                {"pk_ed25519": d["pk_ed25519"], "label": d.get("label", ""),
-                 "pinned_at": d["pinned_at"], "pinned_via": d["pinned_via"],
-                 "added_by_pk": d.get("added_by_pk", ""),
-                 "is_this_one": d["pk_ed25519"] == self._pinned_pk}
-                for d in devices
-            ],
-        })
-
-    async def _do_device_revoke(self, msg: dict) -> None:
-        """
-        Retire one of this account's devices — a lost laptop.
-
-        Countersigned like an addition, by a live device of the same account.
-        The last one cannot go: an account with no device on this node can only
-        return through an operator's invitation code, and doing that to yourself
-        by accident is not a mistake worth allowing.
-        """
-        roster = self._ctx.get("roster")
-        if roster is None or not self._user_id or not self._nonce_node:
-            self._send({"type": "error", "detail": "Not ready"})
-            return
-        if not self._spend_device_attempt():
-            return
-
-        target = str(msg.get("pk_ed25519", ""))
-        ts = int(msg.get("ts", 0))
-        if not target:
-            self._send({"type": "error", "detail": "Missing device key"})
-            return
-        if abs(time.time() - ts) > DEVICE_TTL:
-            self._send({"type": "error", "detail": "Request expired"})
-            return
-
-        victim = await roster.find_device(self._user_id, target)
-        if victim is None:
-            self._send({"type": "error", "detail": "No such device"})
-            return
-
-        transcript = device_add_transcript(
-            node_pk_b64=self._node_pk_b64(), user_id=self._user_id,
-            pk_ed25519_b64=target, pk_x25519_b64=victim["pk_x25519"],
-            nonce_node=self._nonce_node, ts=ts)
-        signer = await self._verify_device_signer(roster, transcript,
-                                                  msg.get("sig", ""))
-        if signer is None:
-            self._send({"type": "error",
-                        "detail": "Not signed by a device already paired here"})
-            return
-
-        if len(await roster.list_devices(self._user_id)) <= 1:
-            self._send({"type": "error",
-                        "detail": "This is your only device here — removing it "
-                                  "would need an operator code to come back"})
-            return
-
-        await roster.revoke_device(self._user_id, target)
-        # A revoked device holds every chat key it ever received — a lost laptop
-        # reads the group's chat until the epoch moves.
-        await self._new_chat_epoch(self._group_id or "", "device_revoke")
-        self._audit("device_revoked", f"{target[:16]} by {signer[:16]}")
-        log.info("Device revoked for %s: %s", self._user_id[:8], target[:16])
-        self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION,
-                    "revoked": target})
-
-    async def _verify_device_signer(self, roster, transcript: bytes,
-                                    sig_b64: str) -> str | None:
-        """
-        The pinned key that signed this, or None.
-
-        Every live device of the account is tried, because any of them may
-        approve. A revoked one is not in the list — that is the point of marking
-        rather than deleting: a lost laptop must stop being able to admit its
-        replacement.
-        """
-        try:
-            sig = base64.b64decode(sig_b64)
-        except Exception:
-            return None
-        for device in await roster.list_devices(self._user_id):
-            try:
-                pk = Ed25519PublicKey.from_public_bytes(
-                    base64.b64decode(device["pk_ed25519"]))
-            except Exception:
-                continue
-            if self._verify_sig(pk, transcript, sig):
-                return device["pk_ed25519"]
-        return None
-
-    def _spend_device_attempt(self) -> bool:
-        """
-        Bound guessing on this connection, as the join path does.
-
-        A code is 40 bits, single use and bound to the keys it names, so this is
-        depth rather than the control — but an unbounded loop over the lookup is
-        still a free oracle, and a burst of failures belongs in the audit log.
-        """
-        self._device_attempts = getattr(self, "_device_attempts", 0) + 1
-        if self._device_attempts > 5:
-            self._audit("device_attempts_exceeded", str(self._device_attempts))
-            self._send({"type": "error",
-                        "detail": "Too many device attempts on this connection"})
-            return False
-        return True
-
-    def _group_join_policy(self, group_id: str) -> str:
-        """
-        Admission policy for a group, read from the node's own configuration.
-
-        Never from the hub: a hub that could declare a group open would be handed
-        the key to it (docs/MESHBAY_DESIGN.md §3.4).
-        """
-        gctx = (self._ctx.get("groups") or {}).get(group_id) or {}
-        return gctx.get("join_policy", "invite")
-
-    async def _pin_and_admit(
-        self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str,
-        *, group_id: str, role: str, approved_by: str, via: str,
-    ) -> None:
-        await roster.pin_identity(
-            user_id=user_id, username=username,
-            pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via,
-        )
-        await roster.set_member(
-            group_id=group_id, user_id=user_id, role=role,
-            status="active", approved_by=approved_by,
-        )
-        if role == ROLE_OPERATOR:
-            self._ctx["has_admin_authority"] = True
-
-        log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role)
-        self._audit_join("join_pinned", f"role={role} via={via}")
-
-    async def _join_ok(
-        self, user_id: str, pk_x_raw: bytes, group_id: str,
-        *, role: str, recognised: bool,
-    ) -> None:
-        """
-        Answer a join, wrapping the group key for the key the caller just proved.
-
-        This is the H3 fix. The inviter used to fetch the invitee's public key from
-        the hub and wrap the GEK for whatever came back, so a hub that answered
-        with its own key was handed the group key by an honest member following the
-        protocol exactly. The node now wraps for a key that arrived from its owner
-        over an authenticated channel, bound to a pinned identity.
-        """
-        reply = {
-            "type": MNP.JOIN_RESULT,
-            "v": MNP_VERSION,
-            "ok": True,
-            "recognised": recognised,
-            "role": role,
-        }
-
-        roster = self._ctx["roster"]
-        if group_id and not await roster.is_authorized(group_id, user_id):
-            # Pinned on this node, but not admitted to this group. Hub membership
-            # alone must not produce a key.
-            reply["gek"] = False
-            reply["reason"] = "not_authorized_for_group"
-            self._send(reply)
-            self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized")
-            return
-
-        gctx = (self._ctx.get("groups") or {}).get(group_id) or {}
-        gek = gctx.get("gek")
-        if not gek:
-            reply["gek"] = False
-            reply["reason"] = "no_gek"
-            self._send(reply)
-            return
-
-        bundle = wrap_gek_aes(gek, pk_x_raw)
-        reply["gek"] = True
-        reply["pk_eph_b64"] = bundle["pk_eph_b64"]
-        reply["nonce_b64"] = bundle["nonce_b64"]
-        reply["wrapped_b64"] = bundle["wrapped_b64"]
-        self._send(reply)
-        self._audit_join("gek_wrapped", f"group={group_id[:8]}")
-
     def _do_member_revoke(self, msg: dict) -> None:
         """
         Stop serving the group key to someone, at the operator's request.
@@ -2969,81 +2200,6 @@ class WebRTCPeerSession(
         else:
             self._send({"type": "error", "detail": "Unknown admin operation"})
 
-    async def _admin_exec_invite_create(
-        self, pending: dict, transcript: bytes, sig: bytes,
-    ) -> None:
-        # Node operator only. A group admin who does not run the node has no
-        # authority over who this node admits (deny by default). Delegation is
-        # designed but deferred — see docs/MESHBAY_DESIGN.md §3.4.
-        if not await self._verify_admin_sig(transcript, sig):
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}")
-            return
-
-        payload = pending["payload"]
-        try:
-            result = await self._run_op(
-                ops.create_invite,
-                payload["group_id"],
-                payload.get("username", ""),
-                user_id=payload["user_id"],
-                created_by=self._user_id or "",
-            )
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-
-        self._audit("invite_create", f"target={payload['user_id'][:8]}")
-        self._send({
-            "type": MNP.INVITE_RESULT,
-            "v": MNP_VERSION,
-            "code": result["code"],
-            "expires_at": result["expires_at"],
-            "user_id": result["user_id"],
-            "username": result.get("username", ""),
-        })
-
-    async def _admin_exec_invite_link_create(
-        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", "invite_link_create")
-            return
-        try:
-            result = await self._run_op(
-                ops.create_link_invite, pending["payload"]["group_id"],
-                created_by=self._user_id or "")
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-        self._audit("invite_link_create", f"invite={result['invite_id'][:8]}")
-        self._send({
-            "type": MNP.INVITE_LINK_RESULT,
-            "v": MNP_VERSION,
-            "code": result["code"],
-            "invite_id": result["invite_id"],
-            "expires_at": result["expires_at"],
-            "group_id": result["group_id"],
-        })
-
-    async def _admin_exec_invite_cancel(
-        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"invite_cancel:{pending['subject'][:8]}")
-            return
-        payload = pending["payload"]
-        try:
-            await self._run_op(ops.cancel_invite, payload["group_id"], payload["invite_id"])
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-        self._audit("invite_cancel", f"invite={payload['invite_id'][:8]}")
-        self._send({"type": "ack", "v": MNP_VERSION, "detail": "invite_cancelled",
-                    "invite_id": payload["invite_id"]})
-
     def _send(self, obj: dict) -> None:
         # Stamp the reply with the id of the request being answered, so the
         # caller never has to guess. Only for this session's own replies: a