feat(node)!: the node wraps the group key — closes H3 and M3

The invite flow fetched the invitee's pk_x25519 from the hub and wrapped the
GEK for whatever came back (app.js:1466, and gek-init did the same server-side).
The hub is the key directory, so a hub answering with its own key was handed the
group key by an honest member following the protocol exactly. No forgery, no
injection, nothing for the client to notice. That was H3.

The fix is not safety numbers. Nobody reads the directory any more:

  - the node holds the GEK and wraps it itself, on every connection, for the
    X25519 key the joiner signed with their Ed25519 identity in one transcript
    (meshbay:join:v1), so the identity key vouches for the encryption key;
  - identities are bound to accounts by a one-time code the hub never sees —
    40 bits, single use, one account, bounded per connection AND node-wide;
  - the node's own roster decides who may receive the key. Hub membership lets
    someone reach a node; it no longer gets them anything. A hub that invents an
    account and mints it a token is answered not_authorized_for_group.

Safety numbers would have made substitution detectable by a human who checks, at
the moment there is nothing to check against — first contact. Removing the lookup
makes it impossible, and costs the user one code to pass along.

M3 falls out of the same work. The daemon auto-pinned its own keystore key as
admin_pk_ed25519 while the browser signs with the user identity key, so every
privileged operation failed closed with a signature error that looked like a bug
somewhere else; the demo only worked because a deploy script overwrote the value.
Authority now comes from the roster, established locally by `operator pair`.
Asking the hub for the operator's key — the obvious-looking fix — would have let
the hub install itself as node administrator.

BREAKING: gek_bundle_store is deleted, not gated. No member hands the node key
material at all, so C5b becomes structural rather than an authorization to check.
Existing stored bundles are still served, so current deployments keep working.

Also:
  - join_policy (invite|open) is read from node.toml, never from the hub — a hub
    able to declare a group open would be handed its key. Unknown group ⇒ invite.
  - admin signatures are verified against the roster on every check, so unpinning
    takes effect without a restart. admin_pk_ed25519 stays readable as legacy.
  - two C5b tests were rewritten, deliberately: they asserted that
    gek_bundle_store demanded an operator signature, and the message is gone. They
    now assert the stronger property. The file says not to fix these tests, so
    this is the record of why they changed.
  - a slice-1 bug found while writing slice 2: connect() never passed skEdB64, so
    pairing would have failed at runtime with no test able to catch it.

Tests: 152 node+common here, including an end-to-end DataChannel run where a
member who has never held the group key redeems a code in the pre-proof window
and receives the key wrapped for a key only they can open.

Design: docs/invite-pairing-v1.md

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

1 files changed, 356 insertions, 55 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 34a96bd..fe3ee2e 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -57,10 +57,16 @@ from meshbay_common.handshake import (
 from meshbay_common.adminop import (
     ADMIN_CHALLENGE_TTL,
     OP_FILE_DELETE,
-    OP_GEK_BUNDLE_STORE,
+    OP_INVITE_CREATE,
     admin_transcript,
 )
-from meshbay_common.crypto import pk_to_b64
+from meshbay_common.crypto import pk_to_b64, wrap_gek_aes
+from meshbay_common.join import (
+    JOIN_TTL,
+    ROLE_MEMBER,
+    ROLE_OPERATOR,
+    join_transcript,
+)
 from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
 from meshbay_common.protocol import MNP
 from meshbay_node.indexer import GroupIndex
@@ -85,6 +91,14 @@ MAX_CONCURRENT_TRANSCODES = 2
 # 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
 UPLOAD_DIR_NAME = ".uploads"
 # Conservative allowlist: also what keeps markup out of filenames, which the node admin
 # UI used to render unescaped (finding H2).
@@ -218,6 +232,11 @@ class WebRTCPeerSession:
         self._username: str = ""
         self._pk_user: str = ""
         self._gek_challenge: bytes | None = None
+        # Same value as the GEK challenge, but kept for the life of the connection:
+        # a join_request is signed over it, and it must stay verifiable after the
+        # handshake clears the challenge (an operator pairs while already connected).
+        self._nonce_node: bytes = b""
+        self._join_attempts = 0
         self._nonce_client: bytes = b""
         self._admin_ops: dict[str, dict] = {}   # op_id → pending admin operation
         self._uploads: dict[str, dict] = {}   # filename → {next_index, bytes}
@@ -259,6 +278,12 @@ class WebRTCPeerSession:
                     asyncio.ensure_future(self._do_gek_bundle_fetch())
                 else:
                     asyncio.ensure_future(self._do_keypair_bundle_fetch())
+            elif mtype == MNP.JOIN_REQUEST and self._nonce_node:
+                # Valid both before the GEK proof (a new member has no GEK to prove
+                # with) and after it (an operator pairing a browser is already
+                # connected). Authority comes from the pairing code and the
+                # signature, never from the session state.
+                asyncio.ensure_future(self._do_join_request(msg))
             elif self._user_id is None:
                 self._send({"type": "error", "detail": "Handshake required"})
             elif mtype == MNP.INDEX_SYNC:
@@ -277,8 +302,8 @@ class WebRTCPeerSession:
                 self._do_file_delete(msg)
             elif mtype == MNP.ADMIN_RESPONSE:
                 self._do_admin_response(msg)
-            elif mtype == MNP.GEK_BUNDLE_STORE:
-                self._do_gek_bundle_store(msg)
+            elif mtype == MNP.INVITE_CREATE:
+                self._do_invite_create(msg)
             elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
                 asyncio.ensure_future(self._do_keypair_bundle_store(msg))
             elif mtype == MNP.STREAM_REQUEST:
@@ -359,6 +384,7 @@ class WebRTCPeerSession:
             return
 
         self._gek_challenge = os.urandom(NONCE_LEN)
+        self._nonce_node = self._gek_challenge
         self._send({
             "type": MNP.HANDSHAKE_CHALLENGE,
             "v": MNP_VERSION,
@@ -472,49 +498,40 @@ class WebRTCPeerSession:
         else:
             self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
 
-    def _do_gek_bundle_store(self, msg: dict) -> None:
+    def _do_invite_create(self, msg: dict) -> None:
         """
-        Request to store a wrapped GEK bundle for a target user.
-
-        Finding C5b: this used to write whatever any authenticated member sent, with
-        INSERT OR REPLACE semantics, and then auto-activate the bundle if it was
-        addressed to the node operator. Since the operator's X25519 public key is
-        public — the node even hands it out in handshake_ack — any member could wrap
-        a GEK of their own choosing for the operator and make the node adopt it,
-        locking every legitimate member out of the group and taking over the key.
+        Issue a one-time pairing code for someone the operator wants to admit.
 
-        Storing a bundle is now a node-operator operation gated by an Ed25519
-        challenge, and nothing arriving over MNP can activate a GEK: activation
-        happens only through the local admin UI or the CLI.
+        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.
         """
-        bundle_store = self._ctx.get("bundle_store")
-        if not bundle_store:
-            self._send({"type": "error", "detail": "Bundle store not available"})
+        roster = self._ctx.get("roster")
+        if roster is None:
+            self._send({"type": "error", "detail": "Roster not available"})
             return
 
-        target_user_id = msg.get("user_id", "")
+        invitee_id = msg.get("user_id", "")
         group_id = msg.get("group_id") or self._group_id
-        pk_eph = msg.get("pk_eph_b64", "")
-        nonce = msg.get("nonce_b64", "")
-        wrapped = msg.get("wrapped_b64", "")
-
-        if not target_user_id or not pk_eph or not nonce or not wrapped or not group_id:
-            self._send({"type": "error", "detail": "Missing bundle fields"})
+        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._ctx.get("admin_pk_ed25519"):
+        if not self._has_admin_authority():
             self._send({
                 "type": "error",
-                "detail": "No admin key pinned — bundle storage refused",
+                "detail": "No operator paired — run `meshbay-node operator pair`",
             })
             return
 
-        self._issue_admin_challenge(OP_GEK_BUNDLE_STORE, target_user_id, {
+        self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, {
             "group_id": group_id,
-            "user_id": target_user_id,
-            "pk_eph_b64": pk_eph,
-            "nonce_b64": nonce,
-            "wrapped_b64": wrapped,
+            "user_id": invitee_id,
+            "username": str(msg.get("username", ""))[:64],
         })
 
     async def _do_keypair_bundle_fetch(self) -> None:
@@ -560,6 +577,240 @@ class WebRTCPeerSession:
             "detail": "keypair_bundle_stored",
         })
 
+    # ── 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)
+        asyncio.ensure_future(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
+
+        known = await roster.get_identity(user_id)
+        if known:
+            if known["pk_ed25519"] != pk_ed_b64 or known["pk_x25519"] != pk_x_b64:
+                # The blocking warning, raised where it matters: whoever this is
+                # holds a different key than the person the operator paired.
+                self._join_refuse(
+                    "key_changed",
+                    f"pinned={known['pk_ed25519'][:16]} presented={pk_ed_b64[:16]}")
+                return
+            member = await roster.get_member(group_id, user_id)
+            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)
+        if not invite:
+            self._join_refuse("code_invalid")
+            return
+
+        await self._pin_and_admit(
+            roster, user_id, username, pk_ed_b64, pk_x_b64,
+            group_id=invite["group_id"], role=invite["role"],
+            approved_by=invite["created_by"], via="code")
+        await self._join_ok(user_id, pk_x_raw, invite["group_id"],
+                            role=invite["role"], recognised=False)
+
+    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 (§3.4 of docs/invite-pairing-v1.md).
+        """
+        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 _audit_pre_proof_fetch(self, mtype: str) -> None:
         """Record bundle access made before the GEK proof (C4)."""
         audit = self._ctx.get("audit_store")
@@ -903,9 +1154,8 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not found"})
             return
 
-        admin_pk = self._ctx.get("admin_pk_ed25519")
         has_uploader_pk = bool(entry.uploader_pk)
-        if not admin_pk and not has_uploader_pk:
+        if not self._has_admin_authority() and not has_uploader_pk:
             self._send({"type": "error", "detail": "No authorized key for deletion"})
             return
 
@@ -956,6 +1206,43 @@ class WebRTCPeerSession:
         except Exception:
             return False
 
+    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("admin_pk_ed25519")
+                    or 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. `admin_pk_ed25519` in node.toml is still
+        honoured so an existing deployment keeps working until its operator pairs
+        (M3) — it is the legacy form of the same statement.
+        """
+        legacy = self._ctx.get("admin_pk_ed25519")
+        if self._verify_sig(legacy, transcript, sig):
+            return True
+
+        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", "")
@@ -985,14 +1272,15 @@ class WebRTCPeerSession:
         )
 
         if pending["op"] == OP_FILE_DELETE:
-            self._admin_exec_file_delete(pending, transcript, sig_bytes)
-        elif pending["op"] == OP_GEK_BUNDLE_STORE:
             asyncio.ensure_future(
-                self._admin_exec_bundle_store(pending, transcript, sig_bytes))
+                self._admin_exec_file_delete(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_INVITE_CREATE:
+            asyncio.ensure_future(
+                self._admin_exec_invite_create(pending, transcript, sig_bytes))
         else:
             self._send({"type": "error", "detail": "Unknown admin operation"})
 
-    def _admin_exec_file_delete(
+    async def _admin_exec_file_delete(
         self, pending: dict, transcript: bytes, sig: bytes,
     ) -> None:
         file_id = pending["subject"]
@@ -1012,7 +1300,7 @@ class WebRTCPeerSession:
 
         # Node operator, or the user who uploaded this file — verified by the key
         # recorded at upload time, never by a JWT claim (the hub controls those).
-        if not (self._verify_sig(self._ctx.get("admin_pk_ed25519"), transcript, sig)
+        if not (await self._verify_admin_sig(transcript, sig)
                 or self._verify_sig(uploader_pk, transcript, sig)):
             self._send({"type": "error", "detail": "Signature verification failed"})
             self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}")
@@ -1020,33 +1308,46 @@ class WebRTCPeerSession:
 
         self._exec_file_delete(ctx, file_id, entry)
 
-    async def _admin_exec_bundle_store(
+    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 what this node stores (draft-v4 §4.2.x, deny by default).
-        if not self._verify_sig(self._ctx.get("admin_pk_ed25519"), transcript, sig):
+        # authority over who this node admits (deny by default). Delegation is
+        # designed but deferred — see §6.2 of docs/invite-pairing-v1.md.
+        if not await self._verify_admin_sig(transcript, sig):
             self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed", f"gek_bundle_store:{pending['subject'][:16]}")
+            self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}")
             return
 
-        payload = pending["payload"]
-        bundle_store = self._ctx.get("bundle_store")
-        if not bundle_store:
-            self._send({"type": "error", "detail": "Bundle store not available"})
+        roster = self._ctx.get("roster")
+        if roster is None:
+            self._send({"type": "error", "detail": "Roster not available"})
             return
 
-        await bundle_store.store(
-            payload["group_id"], payload["user_id"],
-            payload["pk_eph_b64"], payload["nonce_b64"], payload["wrapped_b64"],
+        payload = pending["payload"]
+        code = await roster.create_invite(
+            group_id=payload["group_id"],
+            user_id=payload["user_id"],
+            role=ROLE_MEMBER,
+            created_by=self._user_id or "",
         )
-        log.info("GEK bundle stored: group=%s user=%s",
+        invites = await roster.list_invites()
+        expires = next(
+            (i["expires_at"] for i in invites
+             if i["user_id"] == payload["user_id"]
+             and i["group_id"] == payload["group_id"]), "")
+
+        log.info("Invite created: group=%s user=%s",
                  payload["group_id"][:8], payload["user_id"][:8])
-        self._audit("gek_bundle_store", f"target={payload['user_id'][:8]}")
+        self._audit("invite_create", f"target={payload['user_id'][:8]}")
+        # The code exists in the clear exactly here and in the operator's hands.
         self._send({
-            "type": "ack", "v": MNP_VERSION,
-            "detail": "gek_bundle_stored",
+            "type": MNP.INVITE_RESULT,
+            "v": MNP_VERSION,
+            "code": code,
+            "expires_at": expires,
             "user_id": payload["user_id"],
+            "username": payload.get("username", ""),
         })
 
     def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: