fix(node): the roster, not the key alone, decides who gets a session

The handshake opened a session for anyone holding the group key with a hub
token naming the group; the roster was consulted only when wrapping the key
in a join. A member revoked or unpinned on the node but still a member on
the hub kept a full session with the key they already held — and was handed
the chat epoch their removal had just opened, since chat keys go to any
session. An honest client never met this (it asks for the key through
join_request every time); one that kept the key did not have to.

- After the proof, the node asks the roster and refuses with
  `not_authorized_for_group` unless the account is an active member of the
  group or the node's operator.
- A removal from any door — MNP, the node page, the CLI — now opens a new
  chat epoch in each group the person could read, broadcasts it, and closes
  every connection they hold (`ops.members._after_removal`). The CLI and the
  node page did neither.
- Design §5.2, protocol §6.1, §6.3, §14.2.

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

6 files changed, 200 insertions, 21 deletions

diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md
index 78a3261..d2bec10 100644
--- a/docs/MESHBAY_DESIGN.md
+++ b/docs/MESHBAY_DESIGN.md
@@ -1046,6 +1046,14 @@ requester and it therefore grants nothing across accounts.
 - The denylist is consulted for user, `jti` **and** group.
 - The node **refuses connections when it holds no group key** — there is no
   `gek_required: false` bypass (**NS8**).
+- **The roster must admit the account for the group** before a session opens,
+  after the proof and whatever the token says (`not_authorized_for_group`). The
+  key proves possession, the token proves the hub's view of membership, and
+  neither is the node's own answer: without this, a member revoked or unpinned
+  here but still on the hub kept a full session with the key they held, and was
+  handed the chat epoch their removal had just opened. A removal, from any door
+  (MNP, the node page, the CLI), also opens a new chat epoch in each group the
+  person could read and closes every connection they hold.
 
 **Refusals carry a code**, not only a sentence, because a client can act on a code.
 `not_a_member` means the hub did not count the account a member when it minted the
diff --git a/docs/MESHBAY_NODE_PROTOCOL.md b/docs/MESHBAY_NODE_PROTOCOL.md
index 70047b2..47f508b 100644
--- a/docs/MESHBAY_NODE_PROTOCOL.md
+++ b/docs/MESHBAY_NODE_PROTOCOL.md
@@ -413,7 +413,8 @@ fails if a transport skips a step.
   |-------------------------------------------------------------->|
   |                                     8. rebuild binding;       |
   |                                        refuse if empty;       |
-  |                                        compare_digest(proof)  |
+  |                                        compare_digest(proof); |
+  |                                        roster admits the user |
   |                                     9. session authenticated: |
   |                                        frame limit -> 64 MiB, |
   |                                        peer registry, audit   |
@@ -480,6 +481,7 @@ absent. `verify_proof` compares with `hmac.compare_digest`.
 | not on the denylist for `user_id`, `jti` **or** `group_id` | `Token revoked` | all three targets, and persisted to disk: a revocation that a restart forgets is not one |
 | `group_id ∈ token.groups` | `Not a member of this group`, code `not_a_member` | the membership check itself — a token is proof of an account, never of a group |
 | `group_id ∈ node.hosted_groups` | `Group not hosted on this node`, code `not_hosted` | the hub may hand a client several nodes for one group, and only some of them host it |
+| *(after the proof)* the roster admits `sub` for `group_id` — an active member row, or the node-wide operator row | `This node has not admitted you to this group`, code `not_authorized_for_group` | the key proves possession and the token the hub's view; the node's own answer is the roster. Without it, someone revoked here but still a hub member kept a session with the key they held |
 
 `AuthorizedPeer` carries `user_id`, `group_id`, `username`, `jti` — and deliberately
 **no user public key**. `username` is read from a `username` claim that neither the MNP
@@ -2294,8 +2296,9 @@ walks through the gate meant to stop it.
   filename and no path anywhere in them (§11.1a).
 * **Rotation is the only thing that removes access.** Revoking a member stops the node
   serving the next key; the current key and anything already downloaded stay readable.
-  A chat epoch is opened at the same time, which stops them reading what is said next —
-  not what was said before, which they could already read.
+  A chat epoch is opened at the same time, and their connections are closed and refused
+  from then on (the handshake consults the roster), which stops them reading what is
+  said next — not what was said before, which they could already read.
 
 ---
 
diff --git a/packages/meshbay-node/src/meshbay_node/ops/members.py b/packages/meshbay-node/src/meshbay_node/ops/members.py
index b5da9c0..2562dd6 100644
--- a/packages/meshbay-node/src/meshbay_node/ops/members.py
+++ b/packages/meshbay-node/src/meshbay_node/ops/members.py
@@ -262,6 +262,52 @@ async def cancel_link_invitation(state: dict, group_id: str, invite_id: str) ->
             "node": node_cancelled, "hub": hub_cancelled}
 
 
+async def _after_removal(state: dict, user_id: str, group_ids: list[str]) -> None:
+    """
+    What a removal has to do beyond the roster, whichever door it came through.
+
+    - **A new chat epoch in every group the person could read.** They keep the
+      key of the epoch they were in; the next one is what they must not get.
+    - **Every live connection of theirs closed.** The handshake now refuses them
+      (it consults the roster), so this is what makes the removal immediate
+      rather than "at their next reconnection".
+
+    It used to live in the MNP handlers only, so a removal from the CLI or the
+    node page — the doors an operator at their own machine uses — left the
+    epoch where it was and the person connected. Best effort: a failure here
+    must not turn a done removal into a refused one, and is logged loudly.
+    """
+    from meshbay_common import MNP_VERSION
+    from meshbay_common.protocol import MNP
+
+    from meshbay_node.ops.chat import open_chat_epoch
+
+    groups_ctx = state.get("groups_ctx") or {}
+    for gid in sorted({g for g in group_ids if g}):
+        try:
+            result = await open_chat_epoch(state, gid)
+        except Exception as e:
+            log.error("Removal of %s: no new chat epoch for group %s (%s) — "
+                      "they still hold the current chat key", user_id[:8], gid[:8], e)
+            continue
+        notice = {"type": MNP.CHAT_EPOCH_ACK, "v": MNP_VERSION, "epoch": result["epoch"]}
+        for session in list((groups_ctx.get(gid) or {}).get("_peers", {}).values()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
+    webrtc = state.get("webrtc")
+    if webrtc is not None:
+        for session in list(getattr(webrtc, "_sessions", {}).values()):
+            if session._user_id == user_id and (
+                    not group_ids or session._group_id in group_ids):
+                try:
+                    await session.close()
+                except Exception:
+                    pass
+
+
 async def revoke_member(state: dict, user_id: str, group_id: str) -> dict:
     """
     Stop serving the group key to someone.
@@ -285,6 +331,7 @@ async def revoke_member(state: dict, user_id: str, group_id: str) -> dict:
         raise OpError("No such member in that group", status=404)
     log.info("Member revoked: user=%s group=%s member=%s invites_dropped=%d",
              user_id[:8], group_id[:8], revoked, dropped)
+    await _after_removal(state, user_id, [group_id] if revoked else [])
     return {"status": "revoked", "user_id": user_id, "group_id": group_id,
             "was_member": revoked, "invites_dropped": dropped,
             # Only what is true: somebody who never redeemed a code never held
@@ -297,6 +344,8 @@ async def revoke_member(state: dict, user_id: str, group_id: str) -> dict:
 async def unpin_member(state: dict, user_id: str) -> dict:
     """Forget a pinned identity, so the person can pair again with a new key."""
     roster = _roster(state)
+    groups = [m["group_id"] for m in await roster.list_members()
+              if m.get("user_id") == user_id and m.get("group_id")]
     if not await roster.unpin(user_id):
         raise OpError("No such pinned identity", status=404)
     # Drop the stored keypair bundle too. Left behind, it is served to the next
@@ -310,4 +359,5 @@ async def unpin_member(state: dict, user_id: str) -> dict:
         except Exception:
             log.warning("unpin: could not drop keypair bundle for %s", user_id[:8])
     log.info("Identity unpinned: user=%s", user_id[:8])
+    await _after_removal(state, user_id, groups)
     return {"status": "unpinned", "user_id": user_id}
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/group_ops.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/group_ops.py
index 5f5f9ed..3756eac 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/group_ops.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/group_ops.py
@@ -113,8 +113,9 @@ class GroupOpsMixin:
             self._audit("admin_auth_failed", f"member_unpin:{user_id[:8]}")
             return
         try:
+            # The new chat epochs and the closed sessions are the op's own
+            # (`ops.members._after_removal`), for every door alike.
             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
@@ -331,26 +332,17 @@ class GroupOpsMixin:
             return
 
         try:
+            # The op opens the new chat epoch and closes every connection the
+            # account holds in this group (`ops.members._after_removal`), for
+            # the loopback API and the CLI as much as for this door. They keep
+            # the key they already unwrapped; rotating it is the operator's
+            # call, and the ack says so.
             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,
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/handshake.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/handshake.py
index f701072..7822186 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/handshake.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/handshake.py
@@ -137,7 +137,8 @@ class HandshakeMixin:
         return {"sig": base64.b64encode(self._ctx["sk_node"].sign(transcript)).decode()}
 
     def _do_handshake_response(self, msg: dict) -> None:
-        if not self._gek_challenge or not hasattr(self, "_pending_sub"):
+        if not self._gek_challenge or not hasattr(self, "_pending_sub") \
+                or getattr(self, "_admitting", False):
             self._send({"type": "error", "detail": "No pending handshake challenge"})
             return
 
@@ -170,8 +171,38 @@ class HandshakeMixin:
             self._audit_auth_failed(group_id, "GEK HMAC mismatch")
             return
 
-        self._complete_handshake(gek, binding)
-        self._gek_challenge = None
+        # One answer at a time: the roster is asked asynchronously, and a
+        # second response arriving meanwhile must not be taken as a new one.
+        self._admitting = True
+        self._spawn(self._admit(gek, binding, group_id))
+
+    async def _admit(self, gek: bytes, binding: bytes, group_id: str) -> None:
+        """
+        Open the session only for someone this node's roster admits.
+
+        The group-key proof says the peer holds the key, and the token says the
+        hub counts the account a member. Neither is the node's own answer —
+        and the roster is supposed to be the authority (§6.1). Without this, a
+        member revoked here, or unpinned, but still a member on the hub, kept
+        full access with the key they already held: files, uploads, and — since
+        chat keys are handed to any session — the very epoch their removal had
+        just opened. An honest client never met the gap, because it asks for
+        the key through `join_request`, which does consult the roster; a
+        client that kept the key did not have to.
+        """
+        try:
+            roster = self._ctx.get("roster")
+            if roster is not None and not await roster.is_authorized(
+                    group_id, self._pending_sub):
+                self._send({"type": "error",
+                            "detail": "This node has not admitted you to this group",
+                            "code": "not_authorized_for_group"})
+                self._audit_auth_failed(group_id, "not admitted by the roster")
+                return
+            self._complete_handshake(gek, binding)
+        finally:
+            self._gek_challenge = None
+            self._admitting = False
 
     def _complete_handshake(self, gek: bytes, binding: bytes) -> None:
         # Authenticated peers may send large frames (file uploads); unauthenticated
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index 7a6f517..803c143 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -1546,6 +1546,8 @@ async def test_a_link_needs_the_operator(sk_node, sk_hub, gek, shared_dir, tmp_p
     )
     transport._ctx["roster"] = roster
     transport._ctx["has_admin_authority"] = False
+    # A member this node admitted — the handshake consults the roster.
+    await roster.set_member(TEST_GROUP, "user-001", ROLE_MEMBER, "active", "test")
     pc, ch, q = await _setup_peer(transport, sk_hub, gek, "peer-member")
     try:
         ch.send(_pack({"type": MNP.INVITE_LINK_CREATE, "v": MNP_VERSION}))
@@ -2167,3 +2169,96 @@ async def test_a_large_user_blob_round_trips_whole(
         assert resp["type"] == MNP.USER_BLOB_RESP
         assert len(resp["blob_enc"]) == len(body), "truncated"
         assert resp["blob_enc"] == body
+
+
+# ── The roster decides who gets a session, not the key alone ────────────────
+
+async def _roster_transport(sk_node, sk_hub, gek, shared_dir, tmp_path):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=_hub_pk_pem(sk_hub), gek=gek,
+        roots=one_root(shared_dir), index=indexer.index, stun_servers=[])
+    transport._ctx["roster"] = roster
+    return transport, roster
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("state", ["revoked", "absent"])
+async def test_holding_the_key_is_not_enough_without_the_roster(
+        sk_node, sk_hub, gek, shared_dir, tmp_path, state):
+    """
+    A member revoked on this node — or never admitted by it — who still holds
+    the group key and is still a member on the hub. The proof verifies; the
+    node must refuse anyway, or revocation waits for a key rotation and the
+    new chat epoch it opened is handed straight back.
+    """
+    transport, roster = await _roster_transport(sk_node, sk_hub, gek, shared_dir, tmp_path)
+    if state == "revoked":
+        await roster.set_member(TEST_GROUP, "user-001", ROLE_MEMBER, "revoked", "test")
+    pc, ch, q = await _open_channel(transport, "peer-kept-key")
+    try:
+        token = _token(sk_hub, "user-001", "peer-kept-key", TEST_GROUP)
+        reply = await _do_mnp_handshake(ch, q, token, gek, pc, TEST_GROUP)
+        assert reply.get("type") == "error", reply
+        assert reply.get("code") == "not_authorized_for_group"
+        ch.send(_pack({"type": MNP.INDEX_SYNC, "v": MNP_VERSION}))
+        after = await asyncio.wait_for(q.get(), timeout=5.0)
+        assert after.get("detail") == "Handshake required"
+    finally:
+        await roster.close()
+        await pc.close()
+        await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_an_admitted_member_still_gets_a_session(
+        sk_node, sk_hub, gek, shared_dir, tmp_path):
+    transport, roster = await _roster_transport(sk_node, sk_hub, gek, shared_dir, tmp_path)
+    await roster.set_member(TEST_GROUP, "user-001", ROLE_MEMBER, "active", "test")
+    try:
+        pc, ch, q = await _setup_peer(transport, sk_hub, gek, "peer-admitted")
+        await pc.close()
+    finally:
+        await roster.close()
+        await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_revoking_from_the_node_closes_the_session_and_moves_the_chat_epoch(
+        sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """The loopback door (CLI, node page) did neither before."""
+    from meshbay_node import ops
+    from meshbay_node.bundle_store import BundleStore
+
+    transport, roster = await _roster_transport(sk_node, sk_hub, gek, shared_dir, tmp_path)
+    await roster.set_member(TEST_GROUP, "user-001", ROLE_MEMBER, "active", "test")
+    store = BundleStore(tmp_path / "bundles.db")
+    await store.open()
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+    sk_x = X25519PrivateKey.generate()
+    state = {"roster": roster, "bundle_store": store, "webrtc": transport,
+             "groups_ctx": {},
+             "pk_x25519_raw": sk_x.public_key().public_bytes(
+                 serialization.Encoding.Raw, serialization.PublicFormat.Raw),
+             "sk_x25519_raw": sk_x.private_bytes(
+                 serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
+                 serialization.NoEncryption())}
+    try:
+        pc, ch, q = await _setup_peer(transport, sk_hub, gek, "peer-to-remove")
+        assert "peer-to-remove" in transport._sessions
+        before = await store.latest_chat_epoch(TEST_GROUP)
+
+        await ops.revoke_member(state, "user-001", TEST_GROUP)
+
+        assert await store.latest_chat_epoch(TEST_GROUP) == before + 1
+        assert "peer-to-remove" not in transport._sessions or \
+            transport._sessions["peer-to-remove"]._pc.connectionState == "closed"
+        await pc.close()
+    finally:
+        await store.close()
+        await roster.close()
+        await transport.close_all()