1 files changed, 95 insertions, 0 deletions
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()