merge: Phase 11.5 security remediation, invite redesign, per-node identity

Brings in the security remediation branch. Three bodies of work, and what they
changed about what this project may claim.

Phase 11.5 closed the gap between the documents and the code: the unauthenticated
node HTTP API and the TCP transport deleted, one handshake shared by the
remaining two transports, mutual authentication, structured admin transcripts,
upload confinement, group isolation, revocation that reaches nodes. Six critical
and seven high findings closed, bounded, or deferred by decision.

The invite redesign closed H3 and M3 — the last open High. The hub was the key
directory: an inviter fetched the invitee's key from it and wrapped the group key
for whatever came back, so a hub answering with its own key was handed the group
key by an honest member following the protocol exactly. That lookup is gone. The
node holds the group key and wraps it itself, for a key its recipient proves
possession of, bound to an account by a one-time code the hub never sees. M3 fell
out of the same work: node authority comes from a local roster, never from the
hub.

Per-node identity cut what remains of C4 down to one operator. A single keypair
used to be copied to every node its owner joined; each node now gets its own, so
cracking the bundle on one machine yields a key that is a stranger everywhere
else — and on that machine, one that unlocks nothing its holder did not already
serve. The bundle KDF moved to Argon2id 128 MB, and the hub stopped storing or
publishing user keys at all.

What this project may now say: the hub cannot read your content unless it ships
you malicious client code. T3 remains, accepted (D1), and is what the native
client removes. C4 is reduced, not closed, until 13.3. Chat is still plaintext at
rest until Phase 15. Draft-v5 §2 states each claim against the adversary it holds
against, which is the convention this branch exists to keep.

Four defects were found by deploying it and using a browser, none by the test
suite: a node going deaf on its hub socket, a token that predated group
membership, a client reading values before they were assigned, and identity keys
a browser held but never re-read. The lessons are recorded in CLAUDE.md.

Tests: 343 across the three packages, plus QE/deploy/e2e.py — register, pair,
invite, join, download, stream, second browser, revoke — run against the live
deployment on a wiped hub and node.

7 files changed, 1684 insertions, 571 deletions

diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py
index 1c5a07e..60ef11f 100644
--- a/packages/meshbay-node/tests/test_daemon.py
+++ b/packages/meshbay-node/tests/test_daemon.py
@@ -21,7 +21,6 @@ from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, Keys
 from meshbay_node.daemon import NodeDaemon
 from meshbay_node.indexer import DirectoryIndexer
 
-
 def _mock_keystore_keys(sk_ed):
     """Create a mock keystore with real Ed25519 + X25519 key material."""
     sk_x = X25519PrivateKey.generate()
@@ -35,23 +34,33 @@ def _mock_keystore_keys(sk_ed):
     mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode()
     return mock_keys
 
+def _free_port() -> int:
+    """
+    A port nobody else in the session is on.
+
+    These tests start the real admin UI server. Hardcoding 28000 made them fail
+    with EADDRINUSE whenever another test file had a node running — which is why
+    the full suite failed while each file passed on its own.
+    """
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
 
 @pytest.fixture
 def sk_hub():
     return Ed25519PrivateKey.generate()
 
-
 @pytest.fixture
 def hub_pk_pem(sk_hub):
     return sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
-
 @pytest.fixture
 def gek():
     return generate_gek()
 
-
 @pytest.fixture
 def shared_dir(tmp_path):
     d = tmp_path / "shared"
@@ -60,26 +69,22 @@ def shared_dir(tmp_path):
     (d / "hello.txt").write_bytes(b"hello daemon test " * 50)
     return d
 
-
 @pytest.fixture
 def node_config(tmp_path, shared_dir):
     return Config(
         hub=HubConfig(url="http://localhost:9999", username="testuser"),
-        node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
         groups=[GroupConfig(
             id="g" * 32,
             name="test-group",
             shared_dir=str(shared_dir),
             visibility="private",
-            port=29000,
             quic_port=29010,
-            http_port=29001,
         )],
         keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
         data_dir=tmp_path / "data",
     )
 
-
 @pytest.mark.asyncio
 async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem):
     """Daemon creates ChatStore for each group and shuts down cleanly."""
@@ -130,7 +135,14 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem)
         assert daemon._chat_stores[group_id]._db is not None
 
         if daemon._webrtc:
-            assert "chat_store" in daemon._webrtc._ctx
+            # Finding H1: chat_store must live in the per-group context, never on
+            # the shared transport context. Hoisting the first group's store
+            # transport-wide sent every group's chat to one database and served it
+            # back to members of every other group.
+            assert "chat_store" not in daemon._webrtc._ctx
+            groups_ctx = daemon._webrtc._ctx["groups"]
+            assert groups_ctx[group_id]["chat_store"] is daemon._chat_stores[group_id]
+
             assert "hub_ws" in daemon._webrtc._ctx
             assert "node_user_id" in daemon._webrtc._ctx
             assert daemon._webrtc._ctx["node_user_id"] == "user123"
@@ -146,7 +158,6 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem)
         for store in daemon._chat_stores.values():
             assert store._db is None
 
-
 @pytest.mark.asyncio
 async def test_daemon_no_groups_exits(tmp_path):
     """Daemon with no valid groups exits cleanly."""
@@ -188,19 +199,18 @@ async def test_daemon_no_groups_exits(tmp_path):
 
     assert len(daemon._chat_stores) == 0
 
-
 @pytest.mark.asyncio
 async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hub_pk_pem):
     """Index change callback pushes updated index to WebRTC peers."""
     config = Config(
         hub=HubConfig(url="http://localhost:9999", username="testuser"),
-        node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
         groups=[GroupConfig(
             id="a" * 32,
             name="test-group",
             shared_dir=str(shared_dir),
             visibility="private",
-            port=29000, quic_port=29010, http_port=29001,
+            quic_port=29010,
         )],
         keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
         data_dir=tmp_path / "data",
@@ -232,10 +242,45 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu
     assert msg["group_id"] == "a" * 32
     assert len(msg["entries"]) == indexer.index.count
 
+    # Finding H7: this group is private, so its content hashes must NOT be
+    # registered with the hub. The test previously asserted the opposite —
+    # publishing a fingerprint of every private file was treated as expected
+    # behaviour. Index push to members is unaffected (asserted above).
+    await asyncio.sleep(0.1)
+    daemon._hub.register_swarm.assert_not_called()
+
+@pytest.mark.asyncio
+async def test_daemon_index_change_registers_swarm_for_public_group(
+        tmp_path, shared_dir, gek, hub_pk_pem):
+    """Public groups still register content hashes with the hub swarm (H7)."""
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
+        groups=[GroupConfig(
+            id="a" * 32,
+            name="public-group",
+            shared_dir=str(shared_dir),
+            visibility="public",
+            quic_port=29010,
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._hub = AsyncMock()
+    daemon._hub.register_swarm = AsyncMock(return_value=2)
+    daemon._state["endpoint_hint"] = "node123"
+
+    indexer = DirectoryIndexer(
+        root=shared_dir, group_id="a" * 32,
+        sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer.initial_scan()
+
+    await daemon._on_index_change(indexer)
+
     await asyncio.sleep(0.1)
     daemon._hub.register_swarm.assert_called_once()
-    call_args = daemon._hub.register_swarm.call_args
-    assert len(call_args[0][0]) == indexer.index.count
+    assert len(daemon._hub.register_swarm.call_args[0][0]) == indexer.index.count
 
 
 @pytest.mark.asyncio
diff --git a/packages/meshbay-node/tests/test_http_server.py b/packages/meshbay-node/tests/test_http_server.py
deleted file mode 100644
index d4ccc32..0000000
--- a/packages/meshbay-node/tests/test_http_server.py
+++ /dev/null
@@ -1,227 +0,0 @@
-"""Tests for the node HTTP file API."""
-
-import asyncio
-import base64
-import json
-import os
-import time
-import pytest
-import jwt
-import httpx
-from pathlib import Path
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
-from cryptography.hazmat.primitives import serialization
-
-from meshbay_common.crypto import generate_gek, pk_to_b64
-from meshbay_node.indexer import DirectoryIndexer
-from meshbay_node.transport.http_server import create_http_app
-
-
-@pytest.fixture
-def sk_node():
-    return Ed25519PrivateKey.generate()
-
-@pytest.fixture
-def sk_hub():
-    return Ed25519PrivateKey.generate()
-
-@pytest.fixture
-def hub_pk_pem(sk_hub):
-    return sk_hub.public_key().public_bytes(
-        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
-
-@pytest.fixture
-def gek():
-    return generate_gek()
-
-@pytest.fixture
-def shared_dir(tmp_path):
-    d = tmp_path / "shared"
-    d.mkdir()
-    (d / "video.mp4").write_bytes(os.urandom(3 * 1024 * 1024))  # 3MB
-    (d / "doc.pdf").write_bytes(os.urandom(512 * 1024))
-    (d / "song.mp3").write_bytes(os.urandom(256 * 1024))
-    return d
-
-def make_token(sk_hub, pk_node_b64, ttl=3600):
-    sk_pem = sk_hub.private_bytes(
-        serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
-        serialization.NoEncryption())
-    now = int(time.time())
-    return jwt.encode({
-        "iss": "test-hub", "sub": "user-001",
-        "pk_user": pk_node_b64, "hub_id": "test-hub",
-        "jti": "test-jti", "iat": now, "exp": now + ttl,
-    }, sk_pem, algorithm="EdDSA")
-
-
-@pytest.mark.asyncio
-async def test_node_info(sk_node, sk_hub, hub_pk_pem, gek, shared_dir):
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
-    await indexer.initial_scan()
-
-    app = create_http_app(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem,
-        shared_root=shared_dir, index=indexer.index,
-        group_id="test-group", group_name="Test Group",
-    )
-    async with httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app), base_url="http://test"
-    ) as c:
-        r = await c.get("/")
-        assert r.status_code == 200
-        data = r.json()
-        assert data["group_id"] == "test-group"
-        assert data["file_count"] == 3
-        assert "pk_node" in data
-
-
-@pytest.mark.asyncio
-async def test_public_index(sk_node, sk_hub, hub_pk_pem, shared_dir):
-    """Public group: index accessible without auth."""
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
-    await indexer.initial_scan()
-
-    app = create_http_app(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem,
-        shared_root=shared_dir, index=indexer.index,
-        group_id="pub-group", group_name="Public Group",
-        gek=None,
-    )
-    async with httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app), base_url="http://test"
-    ) as c:
-        r = await c.get("/index")
-        assert r.status_code == 200
-        data = r.json()
-        assert len(data["entries"]) == 3
-        names = {e["name"] for e in data["entries"]}
-        assert "video.mp4" in names
-        assert "doc.pdf" in names
-
-
-@pytest.mark.asyncio
-async def test_file_download(sk_node, sk_hub, hub_pk_pem, shared_dir):
-    """Full file download via HTTP."""
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
-    await indexer.initial_scan()
-
-    app = create_http_app(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem,
-        shared_root=shared_dir, index=indexer.index,
-        group_id="g", group_name="G",
-    )
-    entry = next(e for e in indexer.index.entries if e.name == "doc.pdf")
-    original = (shared_dir / "doc.pdf").read_bytes()
-
-    async with httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app), base_url="http://test"
-    ) as c:
-        r = await c.get(f"/file/{entry.id}")
-        assert r.status_code == 200
-        assert r.content == original
-
-
-@pytest.mark.asyncio
-async def test_chunk_public_group(sk_node, sk_hub, hub_pk_pem, shared_dir):
-    """Public group chunk: plaintext, signed, auth required."""
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
-    await indexer.initial_scan()
-
-    app = create_http_app(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem,
-        shared_root=shared_dir, index=indexer.index,
-        group_id="g", group_name="G", gek=None,
-    )
-    entry = next(e for e in indexer.index.entries if e.name == "video.mp4")
-    token = make_token(sk_hub, pk_to_b64(sk_node.public_key()))
-
-    async with httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app), base_url="http://test"
-    ) as c:
-        r = await c.get(f"/file/{entry.id}/0",
-                        headers={"Authorization": f"Bearer {token}"})
-        assert r.status_code == 200
-        chunk = r.json()
-        assert chunk["encrypted"] is False
-        assert chunk["chunk_index"] == 0
-        assert "data_b64" in chunk
-
-        # Verify the chunk data matches original
-        original = (shared_dir / "video.mp4").read_bytes()
-        data = base64.b64decode(chunk["data_b64"])
-        assert data == original[:len(data)]
-
-
-@pytest.mark.asyncio
-async def test_chunk_private_group(sk_node, sk_hub, hub_pk_pem, gek, shared_dir):
-    """Private group chunk: encrypted with GEK."""
-    from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk
-    import blake3
-
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
-    await indexer.initial_scan()
-
-    app = create_http_app(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem,
-        shared_root=shared_dir, index=indexer.index,
-        group_id="g", group_name="G", gek=gek,
-    )
-    entry = next(e for e in indexer.index.entries if e.name == "doc.pdf")
-    token = make_token(sk_hub, pk_to_b64(sk_node.public_key()))
-
-    async with httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app), base_url="http://test"
-    ) as c:
-        r = await c.get(f"/file/{entry.id}/0",
-                        headers={"Authorization": f"Bearer {token}"})
-        assert r.status_code == 200
-        chunk = r.json()
-        assert chunk["encrypted"] is True
-
-        # Decrypt and verify
-        file_hash = base64.b64decode(chunk["file_hash_b64"])
-        nonce     = base64.b64decode(chunk["nonce_b64"])
-        ct        = base64.b64decode(chunk["ct_b64"])
-        ckey      = derive_chunk_key(gek, file_hash, 0)
-        plaintext = decrypt_chunk(ckey, nonce, ct)
-        original  = (shared_dir / "doc.pdf").read_bytes()
-        assert plaintext == original[:len(plaintext)]
-
-
-@pytest.mark.asyncio
-async def test_chunk_requires_auth(sk_node, hub_pk_pem, shared_dir):
-    """Chunk endpoint rejects unauthenticated requests."""
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
-    await indexer.initial_scan()
-    app = create_http_app(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem,
-        shared_root=shared_dir, index=indexer.index,
-        group_id="g", group_name="G",
-    )
-    entry = indexer.index.entries[0]
-
-    async with httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app), base_url="http://test"
-    ) as c:
-        r = await c.get(f"/file/{entry.id}/0")   # no token
-        assert r.status_code == 401
-
-
-@pytest.mark.asyncio
-async def test_unknown_file_404(sk_node, hub_pk_pem, sk_hub, shared_dir):
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
-    await indexer.initial_scan()
-    app = create_http_app(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem,
-        shared_root=shared_dir, index=indexer.index,
-        group_id="g", group_name="G",
-    )
-    token = make_token(sk_hub, pk_to_b64(sk_node.public_key()))
-
-    async with httpx.AsyncClient(
-        transport=httpx.ASGITransport(app=app), base_url="http://test"
-    ) as c:
-        r = await c.get("/file/nonexistent-hash/0",
-                        headers={"Authorization": f"Bearer {token}"})
-        assert r.status_code == 404
diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py
index 0c1a1cd..93ab1b0 100644
--- a/packages/meshbay-node/tests/test_quic_transport.py
+++ b/packages/meshbay-node/tests/test_quic_transport.py
@@ -49,7 +49,8 @@ def make_jwt(sk_hub, pk_node_b64, ttl=3600, groups=None):
         "iss": "test-hub", "sub": "user-001",
         "pk_user": pk_node_b64, "hub_id": "test-hub",
         "jti": "test-jti", "iat": now, "exp": now + ttl,
-        "groups": groups or [],
+        # group_id is mandatory (M1), so default tokens are members of "g".
+        "groups": groups if groups is not None else ["g"],
     }, sk_pem, algorithm="EdDSA")
 
 
@@ -80,6 +81,7 @@ async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path):
         host="127.0.0.1", port=19100,
         jwt_token=token, gek=gek,
         pk_node_b64=pk_to_b64(sk_node.public_key()),
+        group_id="g",
     ) as client:
         chunk0 = await client.fetch_chunk(entry.id, chunk_index=0)
         chunk1 = await client.fetch_chunk(entry.id, chunk_index=1)
@@ -116,6 +118,7 @@ async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path):
         host="127.0.0.1", port=19101,
         jwt_token=token, gek=gek,
         pk_node_b64=pk_to_b64(sk_node.public_key()),
+        group_id="g",
     ) as client:
         wire = await client.fetch_index()
         recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
@@ -220,10 +223,12 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat
     async with QuicChunkClient(
         host="127.0.0.1", port=19104,
         jwt_token=token, gek=gek, pk_node_b64=pk_b64,
+            group_id="g",
     ) as client:
         wire = await client.fetch_index()
         assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2
         saved_ticket = client.session_ticket
+        saved_cert = client.peer_cert_der
 
     # Allow server to process the close
     await asyncio.sleep(0.1)
@@ -233,6 +238,10 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat
         host="127.0.0.1", port=19104,
         jwt_token=token, gek=gek, pk_node_b64=pk_b64,
         session_ticket=saved_ticket,
+        # A resumed session carries no certificate, so the binding anchor from the
+        # original handshake travels with the ticket (11.5.6).
+        peer_cert_der=saved_cert,
+            group_id="g",
     ) as client:
         wire = await client.fetch_index()
         assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2
@@ -269,6 +278,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p
     async with QuicChunkClient(
         host="127.0.0.1", port=19105,
         jwt_token=token, gek=gek, pk_node_b64=pk_b64,
+            group_id="g",
     ) as client:
         wire = await client.fetch_index()
         assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2
@@ -281,6 +291,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p
         async with QuicChunkClient(
             host="127.0.0.1", port=19105,
             jwt_token=token, gek=gek, pk_node_b64=pk_b64,
+            group_id="g",
         ) as client:
             await client.fetch_index()
 
diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py
new file mode 100644
index 0000000..665c060
--- /dev/null
+++ b/packages/meshbay-node/tests/test_roster_pairing.py
@@ -0,0 +1,744 @@
+"""
+Roster and operator pairing (M3, and the mechanism that will close H3).
+
+Negative assertions, per the posture set in Phase 11.5: each test states an attack
+or a mistake that must not work. The one to keep an eye on is
+`test_daemon_does_not_auto_pin_keystore_key` — the auto-pin is what made node
+sovereignty inert as shipped, and it fails closed, so nothing else in the suite
+notices if it comes back.
+
+See `docs/invite-pairing-v1.md`.
+"""
+
+import base64
+import time
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+
+from meshbay_common.crypto import generate_gek, pk_to_b64, unwrap_gek_aes
+from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import Roster, hash_code, normalize_code
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+@pytest.fixture
+async def roster(tmp_path):
+    r = Roster(db_path=tmp_path / "roster.db")
+    await r.open()
+    yield r
+    await r.close()
+
+
+def _keypair_full():
+    """(sk_ed, pk_ed_b64, pk_x_b64, sk_x) — the X25519 secret is needed to unwrap."""
+    sk_ed = Ed25519PrivateKey.generate()
+    sk_x = X25519PrivateKey.generate()
+    pk_ed_b64 = pk_to_b64(sk_ed.public_key())
+    pk_x_b64 = base64.b64encode(
+        sk_x.public_key().public_bytes(
+            encoding=serialization.Encoding.Raw,
+            format=serialization.PublicFormat.Raw,
+        )
+    ).decode()
+    return sk_ed, pk_ed_b64, pk_x_b64, sk_x
+
+
+def _keypair():
+    sk_ed, pk_ed_b64, pk_x_b64, _ = _keypair_full()
+    return sk_ed, pk_ed_b64, pk_x_b64
+
+
+def _session(tmp_path: Path, roster, user_id: str = "grenet",
+             group_id: str | None = None, gek: bytes | None = None,
+             join_policy: str = "invite") -> WebRTCPeerSession:
+    """A peer session with the join path wired and sending stubbed out."""
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {
+        "shared_root": shared_root,
+        "index": index,
+        "sk_node": index.sk_node,
+        "roster": roster,
+    }
+    if group_id:
+        session._ctx["groups"] = {
+            group_id: {
+                "gek": gek,
+                "shared_root": shared_root,
+                "index": index,
+                "join_policy": join_policy,
+            },
+        }
+    session._group_id = group_id
+    session._user_id = user_id
+    session._username = user_id
+    session._pk_user = ""
+    session._uploads = {}
+    session._join_attempts = 0
+    session._nonce_node = b"\x11" * 32
+    session._remote_ip = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="", user_id="grenet",
+              group_id="", nonce=None, ts=None):
+    ts = int(time.time()) if ts is None else ts
+    transcript = join_transcript(
+        node_pk_b64=session._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=nonce if nonce is not None else session._nonce_node,
+        ts=ts,
+    )
+    return {
+        "type": "join_request",
+        "group_id": group_id,
+        "pk_ed25519": pk_ed_b64,
+        "pk_x25519": pk_x_b64,
+        "code": code,
+        "ts": ts,
+        "sig": base64.b64encode(sk_ed.sign(transcript)).decode(),
+    }
+
+
+def _last(session):
+    return session.sent[-1] if session.sent else {}
+
+
+# ── Roster ────────────────────────────────────────────────────────────────────
+
+async def test_invite_is_single_use(roster):
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+    assert await roster.consume_invite(code, "grenet") is not None
+    assert await roster.consume_invite(code, "grenet") is None, (
+        "a pairing code must not be redeemable twice")
+
+
+async def test_invite_is_bound_to_one_account(roster):
+    """A leaked code must be useless to whoever finds it."""
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+    assert await roster.consume_invite(code, "eve") is None
+    assert await roster.consume_invite(code, "grenet") is not None
+
+
+async def test_expired_invite_is_refused(roster):
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli", ttl=-1)
+    assert await roster.consume_invite(code, "grenet") is None
+
+
+async def test_reinvite_supersedes_the_previous_code(roster):
+    first = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+    second = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+    assert await roster.consume_invite(first, "grenet") is None
+    assert await roster.consume_invite(second, "grenet") is not None
+
+
+async def test_codes_are_not_stored_in_the_clear(roster, tmp_path):
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+    rows = await roster.list_invites()
+    assert rows and rows[0]["code_hash"] != normalize_code(code)
+    assert rows[0]["code_hash"] == hash_code(code)
+
+
+def test_code_normalization_absorbs_human_error():
+    """Someone reading a code aloud must not be able to get it wrong."""
+    assert normalize_code("k7m2-qx4p") == normalize_code("K7M2QX4P")
+    assert normalize_code("O1IL") == "0111"
+    assert normalize_code(" k7m2 qx4p ") == "K7M2QX4P"
+
+
+async def test_operator_pks_reflect_unpinning(roster):
+    _, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
+    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
+    assert await roster.operator_pks() == [pk_ed_b64]
+
+    await roster.unpin("grenet")
+    assert await roster.operator_pks() == [], (
+        "authority must disappear with the pin, without a daemon restart")
+
+
+# ── Join / pairing over MNP ───────────────────────────────────────────────────
+
+async def test_pairing_with_a_valid_code_pins_the_identity(tmp_path, roster):
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code))
+
+    assert _last(session).get("ok") is True
+    pinned = await roster.get_identity("grenet")
+    assert pinned["pk_ed25519"] == pk_ed_b64
+    assert await roster.operator_pks() == [pk_ed_b64]
+
+
+async def test_pairing_without_a_code_is_refused(tmp_path, roster):
+    """Fails closed: an unknown identity gets nothing until someone authorizes it."""
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+
+    await session._do_join_request(_join_msg(session, sk_ed, pk_ed_b64, pk_x_b64))
+
+    assert _last(session).get("ok") is False
+    assert _last(session).get("reason") == "code_required"
+    assert await roster.get_identity("grenet") is None
+
+
+async def test_wrong_code_pins_nothing(tmp_path, roster):
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="ZZZZ-ZZZZ"))
+
+    assert _last(session).get("reason") == "code_invalid"
+    assert await roster.get_identity("grenet") is None
+
+
+async def test_signature_must_cover_the_presented_keys(tmp_path, roster):
+    """
+    The heart of it: the X25519 key is only trustworthy because the Ed25519
+    identity signed it. Swapping in another encryption key after signing must fail.
+    """
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+
+    msg = _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code)
+    _, _, attacker_pk_x = _keypair()
+    msg["pk_x25519"] = attacker_pk_x
+
+    await session._do_join_request(msg)
+
+    assert _last(session).get("reason") == "signature_invalid"
+    assert await roster.get_identity("grenet") is None
+
+
+async def test_join_cannot_be_replayed_onto_another_connection(tmp_path, roster):
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+
+    # Signed against a nonce this connection never issued.
+    msg = _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
+                    nonce=b"\x99" * 32)
+    await session._do_join_request(msg)
+
+    assert _last(session).get("reason") == "signature_invalid"
+    assert await roster.get_identity("grenet") is None
+
+
+async def test_pinned_identity_presenting_a_new_key_is_refused(tmp_path, roster):
+    """
+    11.5.8's rule, applied to people: a changed key is refused outright rather
+    than warned about, and clearing it is a deliberate operator action.
+    """
+    session = _session(tmp_path, roster)
+    _, old_pk_ed, old_pk_x = _keypair()
+    await roster.pin_identity("grenet", "grenet", old_pk_ed, old_pk_x, "code")
+
+    sk_ed2, new_pk_ed, new_pk_x = _keypair()
+    await session._do_join_request(
+        _join_msg(session, sk_ed2, new_pk_ed, new_pk_x, code="ANY-CODE"))
+
+    assert _last(session).get("reason") == "key_changed"
+    assert (await roster.get_identity("grenet"))["pk_ed25519"] == old_pk_ed
+
+
+async def test_attempts_are_bounded(tmp_path, roster):
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+
+    for _ in range(6):
+        await session._do_join_request(
+            _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA"))
+
+    assert any(m.get("detail") == "Too many attempts" for m in session.sent), (
+        "a connection must not be able to sit there guessing codes")
+
+
+async def test_failures_are_counted_across_connections(tmp_path, roster):
+    """
+    The adversary who can mint a token for any account is the hub, and it can
+    reconnect at will — so a per-connection budget alone would bound nothing.
+    """
+    shared_ctx = None
+    for _ in range(6):
+        session = _session(tmp_path, roster)
+        if shared_ctx is None:
+            shared_ctx = session._ctx
+        else:
+            session._ctx = shared_ctx          # same node, new connection
+        sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+        for _ in range(4):
+            await session._do_join_request(
+                _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA"))
+
+    assert any(m.get("detail") == "Pairing temporarily locked"
+               for m in session.sent), (
+        "reconnecting must not reset the pairing budget")
+
+
+async def test_group_id_cannot_name_another_group(tmp_path, roster):
+    session = _session(tmp_path, roster)
+    session._group_id = "a" * 32
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, group_id="b" * 32))
+
+    assert _last(session).get("reason") == "group_mismatch"
+
+
+# ── H3: the node wraps the group key, and only for people it admitted ─────────
+
+GROUP = "g" * 32
+
+
+async def test_node_wraps_the_gek_for_the_key_the_member_proved(tmp_path, roster):
+    """
+    The H3 fix. Nobody fetches a public key from the hub: the node encrypts the
+    group key for the X25519 key the joiner signed with their pinned identity, so
+    a hub substituting a key of its own has nothing to substitute into.
+    """
+    gek = generate_gek()
+    session = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek)
+    sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full()
+
+    code = await roster.create_invite(GROUP, "bob", ROLE_MEMBER, "grenet")
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
+                  user_id="bob", group_id=GROUP))
+
+    reply = _last(session)
+    assert reply["ok"] is True and reply["gek"] is True
+
+    pk_x_raw = base64.b64decode(pk_x_b64)
+    sk_x_raw = sk_x.private_bytes(
+        encoding=serialization.Encoding.Raw,
+        format=serialization.PrivateFormat.Raw,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    assert unwrap_gek_aes(reply, sk_x_raw, pk_x_raw) == gek
+
+
+async def test_hub_membership_alone_yields_no_key(tmp_path, roster):
+    """
+    A hub can invent an account, add it to a group and mint it a token. What it
+    cannot do is put it on the node's roster — so the key never leaves.
+    """
+    gek = generate_gek()
+    session = _session(tmp_path, roster, user_id="eve", group_id=GROUP, gek=gek)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+
+    # Pinned on this node (say, for another group) but never admitted to this one.
+    await roster.pin_identity("eve", "eve", pk_ed_b64, pk_x_b64, "code")
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
+                  user_id="eve", group_id=GROUP))
+
+    reply = _last(session)
+    assert reply.get("gek") is False
+    assert reply.get("reason") == "not_authorized_for_group"
+    assert "wrapped_b64" not in reply
+
+
+async def test_open_join_group_admits_without_a_code(tmp_path, roster):
+    """§3.4: where anyone may join, a code protects nothing and is not required."""
+    gek = generate_gek()
+    session = _session(tmp_path, roster, user_id="newcomer", group_id=GROUP,
+                       gek=gek, join_policy="open")
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
+                  user_id="newcomer", group_id=GROUP))
+
+    reply = _last(session)
+    assert reply["ok"] is True and reply["gek"] is True
+    pinned = await roster.get_identity("newcomer")
+    assert pinned["pinned_via"] == "tofu"
+
+
+async def test_invite_only_group_still_demands_a_code(tmp_path, roster):
+    """Being public (discoverable) is not being open (admitting anyone)."""
+    gek = generate_gek()
+    session = _session(tmp_path, roster, user_id="newcomer", group_id=GROUP,
+                       gek=gek, join_policy="invite")
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
+                  user_id="newcomer", group_id=GROUP))
+
+    assert _last(session).get("reason") == "code_required"
+    assert await roster.get_identity("newcomer") is None
+
+
+async def test_unknown_group_is_invite_only(tmp_path, roster):
+    """
+    Fail closed: a group whose policy the node cannot read is treated as
+    invite-only, never as open.
+    """
+    session = _session(tmp_path, roster, user_id="newcomer")
+    session._group_id = "unconfigured-group"
+    assert session._group_join_policy("unconfigured-group") == "invite"
+    assert session._group_join_policy("") == "invite"
+
+
+def test_join_policy_is_carried_from_node_config():
+    """
+    The policy reaches the transport from node.toml. If it ever came from the hub
+    instead, a hub could declare any group open and be handed its key.
+    """
+    daemon_src = (Path(__file__).parent.parent
+                  / "src" / "meshbay_node" / "daemon.py").read_text()
+    assert '"join_policy": group_cfg.join_policy' in daemon_src
+
+    config_src = (Path(__file__).parent.parent
+                  / "src" / "meshbay_node" / "config.py").read_text()
+    assert "join_policy" in config_src, "GroupConfig must carry the admission policy"
+
+
+async def test_revoked_member_stops_receiving_the_key(tmp_path, roster):
+    """
+    Wrapping on demand is what makes revocation work. A stored bundle survived
+    revocation; this does not. (Rotating the GEK is still required — the
+    ex-member has the old one.)
+    """
+    gek = generate_gek()
+    session = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code")
+    await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "grenet")
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
+                  user_id="bob", group_id=GROUP))
+    assert _last(session)["gek"] is True
+
+    await roster.set_status(GROUP, "bob", "revoked")
+    session.sent.clear()
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
+                  user_id="bob", group_id=GROUP))
+    assert _last(session).get("gek") is False
+
+
+# ── What a first-time joiner can know ─────────────────────────────────────────
+
+def test_challenge_carries_node_pk_in_source():
+    """
+    Belt and braces for the above: the field must be in the message the node
+    builds, whatever the surrounding handshake does.
+    """
+    source = (Path(__file__).parent.parent
+              / "src" / "meshbay_node" / "transport" / "webrtc_server.py").read_text()
+    challenge = source[source.find("MNP.HANDSHAKE_CHALLENGE,"):]
+    challenge = challenge[:challenge.find("})")]
+    assert "node_pk" in challenge, (
+        "the challenge must announce the node key — a first-time joiner cannot "
+        "learn it any other way, and join_request signs it")
+
+
+async def test_a_key_pinned_by_one_node_is_worthless_at_another(tmp_path, roster):
+    """
+    The whole point of per-node identity: node A's operator who cracks the bundle
+    on their own disk holds a key node B has never seen. Presenting it there is a
+    first contact like any other — it needs a code from B's operator.
+    """
+    gek = generate_gek()
+    node_b = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek)
+
+    # The key bob uses at node A. Node B's roster knows nothing about it.
+    sk_ed_a, pk_ed_a, pk_x_a = _keypair()
+
+    await node_b._do_join_request(
+        _join_msg(node_b, sk_ed_a, pk_ed_a, pk_x_a,
+                  user_id="bob", group_id=GROUP))
+
+    assert _last(node_b).get("reason") == "code_required"
+    assert await roster.get_identity("bob") is None
+
+
+async def test_the_stolen_key_cannot_be_forced_in_with_someone_elses_code(
+        tmp_path, roster):
+    """And a code issued for another account does not help either."""
+    gek = generate_gek()
+    session = _session(tmp_path, roster, user_id="eve", group_id=GROUP, gek=gek)
+    sk_ed, pk_ed, pk_x = _keypair()
+    code = await roster.create_invite(GROUP, "bob", ROLE_MEMBER, "grenet")
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed, pk_x, code=code,
+                  user_id="eve", group_id=GROUP))
+
+    assert _last(session).get("reason") == "code_invalid"
+    assert await roster.get_identity("eve") is None
+
+
+# ── Code lifetimes ────────────────────────────────────────────────────────────
+
+async def test_invitations_outlive_pairing_codes(roster):
+    """
+    An invitation crosses a human conversation; a pairing code crosses an SSH
+    session. A day was long enough for the second and not for the first — a code
+    that dies over a weekend means someone has to be at a browser to reissue it.
+    """
+    from meshbay_node.roster import DEFAULT_INVITE_TTL, DEFAULT_PAIR_TTL
+
+    assert DEFAULT_INVITE_TTL == 7 * 24 * 3600
+    assert DEFAULT_PAIR_TTL == 24 * 3600
+    assert DEFAULT_INVITE_TTL > DEFAULT_PAIR_TTL
+
+
+def test_code_lifetimes_are_configurable(tmp_path):
+    """The operator decides, not the default."""
+    from meshbay_node.config import load_config
+
+    path = tmp_path / "node.toml"
+    path.write_text(
+        '[hub]\nurl = "https://example.org"\nusername = "grenet"\n'
+        "[node]\ninvite_ttl_hours = 72\npair_ttl_hours = 2\n"
+    )
+    cfg = load_config(path)
+    assert cfg.node.invite_ttl_hours == 72
+    assert cfg.node.pair_ttl_hours == 2
+
+    default = load_config(tmp_path / "missing.toml")
+    assert default.node.invite_ttl_hours == 168
+    assert default.node.pair_ttl_hours == 24
+
+
+async def test_expiry_is_enforced_at_redemption(tmp_path, roster):
+    """Purging is housekeeping; the check that matters happens on use."""
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli", ttl=-1)
+
+    await session._do_join_request(
+        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code))
+
+    assert _last(session).get("reason") == "code_invalid"
+    assert await roster.get_identity("grenet") is None
+
+
+# ── M3: where node authority comes from ───────────────────────────────────────
+
+async def test_admin_signature_verified_against_the_paired_key(tmp_path, roster):
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
+    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
+
+    transcript = b"meshbay:admin:v1 whatever"
+    assert await session._verify_admin_sig(transcript, sk_ed.sign(transcript))
+
+    stranger = Ed25519PrivateKey.generate()
+    assert not await session._verify_admin_sig(
+        transcript, stranger.sign(transcript))
+
+
+async def test_unpinned_operator_loses_authority_immediately(tmp_path, roster):
+    """No caching: revoking a paired browser must not need a daemon restart."""
+    session = _session(tmp_path, roster)
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
+    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
+
+    transcript = b"meshbay:admin:v1 whatever"
+    assert await session._verify_admin_sig(transcript, sk_ed.sign(transcript))
+
+    await roster.unpin("grenet")
+    assert not await session._verify_admin_sig(transcript, sk_ed.sign(transcript))
+
+
+# ── Operator surface (slice 3) ────────────────────────────────────────────────
+
+def _ui_client(tmp_path, roster, **extra):
+    from fastapi.testclient import TestClient
+
+    from meshbay_node.config import Config
+    from meshbay_node.ui.app import create_ui_app
+
+    state = {
+        "status": "running", "groups_ctx": {GROUP: {"gek": b"k" * 32}},
+        "indexes": {}, "ui_token": "tok", "roster": roster,
+        "node_user_id": "grenet", "config": Config(),
+    }
+    state.update(extra)
+    return TestClient(create_ui_app(state)), state
+
+
+async def test_revoke_endpoint_stops_authorization(tmp_path, roster):
+    client, _ = _ui_client(tmp_path, roster)
+    _, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code")
+    await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "grenet")
+    assert await roster.is_authorized(GROUP, "bob")
+
+    resp = client.post(f"/api/members/bob/revoke?group_id={GROUP}&t=tok")
+    assert resp.status_code == 200
+    assert "gek-init" in resp.json()["reminder"], (
+        "revocation must remind the operator to rotate the key they still hold")
+    assert not await roster.is_authorized(GROUP, "bob")
+
+
+async def test_unpin_endpoint_allows_repairing(tmp_path, roster):
+    client, _ = _ui_client(tmp_path, roster)
+    _, pk_ed_b64, pk_x_b64 = _keypair()
+    await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code")
+
+    assert client.post("/api/members/bob/unpin?t=tok").status_code == 200
+    assert await roster.get_identity("bob") is None
+    assert client.post("/api/members/bob/unpin?t=tok").status_code == 404
+
+
+async def test_operator_surface_needs_the_session_token(tmp_path, roster):
+    """11.5.3 applies to every one of these: they change who may hold the key."""
+    client, _ = _ui_client(tmp_path, roster)
+    for path in ("/api/roster",
+                 "/api/operator/pair",
+                 f"/api/members/bob/revoke?group_id={GROUP}",
+                 "/api/members/bob/unpin",
+                 f"/api/groups/{GROUP}/invites?username=bob"):
+        method = client.get if path == "/api/roster" else client.post
+        assert method(path).status_code == 403, f"{path} reachable without a token"
+
+
+async def test_cli_invite_asks_the_hub_for_an_account_never_a_key(tmp_path, roster):
+    """
+    The CLI resolves a username to an account id through the hub, and stops there.
+    A key fetched from the hub is what H3 was; an account id is not a secret and
+    a wrong one produces an invite whose code the hub never learns.
+    """
+    class _Hub:
+        _session = object()
+
+        async def get_user_pubkeys(self, username):
+            return {"user_id": f"id-of-{username}",
+                    "pk_x25519": "SHOULD-NOT-BE-USED",
+                    "pk_ed25519": "SHOULD-NOT-BE-USED"}
+
+    client, _ = _ui_client(tmp_path, roster, hub=_Hub())
+    resp = client.post(f"/api/groups/{GROUP}/invites?username=bob&t=tok")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["user_id"] == "id-of-bob"
+
+    invites = await roster.list_invites()
+    assert [i["user_id"] for i in invites] == ["id-of-bob"]
+    # Whatever the hub said about keys was never stored anywhere.
+    assert "SHOULD-NOT-BE-USED" not in str(invites)
+    assert await roster.get_identity("id-of-bob") is None
+
+
+def _run_cli(monkeypatch, tmp_path, argv, responses):
+    """Drive the real CLI with the daemon API stubbed, capturing the calls."""
+    import sys as _sys
+
+    from meshbay_node import daemon as _daemon
+
+    calls = []
+
+    def fake_api(cfg, path, method="GET", timeout=30):
+        calls.append((method, path))
+        for key, value in responses.items():
+            if key in path:
+                return value
+        return {}
+
+    monkeypatch.setattr(_daemon, "_daemon_api", fake_api)
+
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'data_dir = "{tmp_path}"\n'
+        '[hub]\nurl = "https://example.org"\nusername = "grenet"\n'
+        f'[[groups]]\nid = "{GROUP}"\nname = "demo"\n'
+        f'shared_dir = "{tmp_path}"\n'
+    )
+    monkeypatch.setattr(_sys, "argv",
+                        ["meshbay-node", *argv, "--config", str(conf)])
+    try:
+        _daemon.main()
+    except SystemExit as e:
+        calls.append(("exit", e.code))
+    return calls
+
+
+def test_cli_member_commands_reach_the_right_endpoints(monkeypatch, tmp_path, capsys):
+    resolved = {"user_id": "u-bob", "source": "roster"}
+
+    calls = _run_cli(monkeypatch, tmp_path, ["member", "revoke", "bob"],
+                     {"/api/resolve": resolved,
+                      "revoke": {"status": "revoked", "reminder": "gek-init"}})
+    assert ("POST", f"/api/members/u-bob/revoke?group_id={GROUP}") in calls
+    # The operator is told the revocation does not take back the key they hold.
+    assert "rotate" in capsys.readouterr().out.lower()
+
+    calls = _run_cli(monkeypatch, tmp_path, ["member", "unpin", "bob"],
+                     {"/api/resolve": resolved, "unpin": {"status": "unpinned"}})
+    assert ("POST", "/api/members/u-bob/unpin") in calls
+
+
+def test_cli_resolves_a_name_before_acting(monkeypatch, tmp_path):
+    """
+    The name has to be turned into an account first, and the node's own roster is
+    asked before the hub. A JWT carries no username, so an identity pinned without
+    an invitation has none — the hub fallback is what keeps it manageable.
+    """
+    calls = _run_cli(monkeypatch, tmp_path, ["member", "revoke", "bob"],
+                     {"/api/resolve": {"user_id": "u-bob", "source": "hub"},
+                      "revoke": {"status": "revoked", "reminder": "gek-init"}})
+
+    assert ("GET", "/api/resolve?username=bob") == calls[0], (
+        "the CLI must resolve the name before acting on anyone")
+    assert ("POST", f"/api/members/u-bob/revoke?group_id={GROUP}") in calls
+
+def test_daemon_does_not_auto_pin_keystore_key():
+    """
+    M3: the daemon used to auto-pin its own keystore key as the admin key, while
+    the browser signs with the user's identity key. Different keys, so every
+    privileged operation failed closed with a signature error that looked like a
+    bug elsewhere — and the demo only worked because a deploy script overwrote it.
+
+    Authority now comes from the roster, or from an explicit node.toml value.
+    """
+    source = (Path(__file__).parent.parent
+              / "src" / "meshbay_node" / "daemon.py").read_text()
+    assert "Auto-pinning admin key" not in source
+    assert "_resolve_admin_pk" not in source, (
+        "the auto-pin resolver is back — node authority must be established "
+        "locally by pairing, never inferred from the node's own keystore (M3)")
+
+
+def test_admin_authority_is_never_fetched_from_the_hub():
+    """
+    The fix M3 invites: ask the hub which key belongs to the operator. That would
+    hand a malicious hub the node — the same substitution as H3, one level deeper.
+    """
+    source = (Path(__file__).parent.parent
+              / "src" / "meshbay_node" / "daemon.py").read_text()
+    admin_region = source[source.find("_legacy_admin_pk"):]
+    assert "pubkeys" not in admin_region.split("def ")[1], (
+        "node authority must never be resolved through a hub lookup")
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
new file mode 100644
index 0000000..dcd9cf6
--- /dev/null
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -0,0 +1,604 @@
+"""
+Phase 11.5 security regression tests.
+
+Each test here encodes a finding from `second-review.md`. They are negative tests:
+they assert that an attack does NOT work. The pre-11.5 code passed 209 feature
+tests while every one of these attacks succeeded — the suite only ever exercised
+happy paths, never an authorization boundary.
+
+If one of these starts failing, a fix has been reverted. Do not "fix" the test.
+"""
+
+import base64
+import struct
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.protocol import IndexEntry
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+
+def _safe_name_re():
+    """
+    Imported lazily so that a missing allowlist fails the two tests that need it,
+    rather than aborting collection of the whole module and hiding every other
+    finding's result.
+    """
+    from meshbay_node.transport.webrtc_server import SAFE_UPLOAD_NAME
+    return SAFE_UPLOAD_NAME
+
+
+# ── C1: the unauthenticated HTTP file API must stay deleted ───────────────────
+
+def test_http_file_api_is_gone():
+    """
+    C1: transport/http_server.py served GET /index and GET /file/{id} on 0.0.0.0
+    with no authentication, for private groups too. It was deleted rather than
+    patched. Re-adding any module that serves file bytes outside the MNP handshake
+    reintroduces a full confidentiality bypass.
+    """
+    with pytest.raises(ImportError):
+        import meshbay_node.transport.http_server  # noqa: F401
+
+    import meshbay_node.transport as transport
+    assert not hasattr(transport, "create_http_app")
+
+
+def test_tcp_transport_is_gone():
+    """C6: the TCP+TLS server accepted a bare JWT with no GEK proof."""
+    with pytest.raises(ImportError):
+        import meshbay_node.transport.server  # noqa: F401
+
+    import meshbay_node.transport as transport
+    assert not hasattr(transport, "ChunkServer")
+
+
+def test_daemon_exposes_no_plaintext_listener():
+    """
+    C1: the daemon must not bind anything that serves content without a handshake.
+    NodeConfig no longer carries an HTTP port at all.
+    """
+    from meshbay_node.config import NodeConfig, GroupConfig
+
+    assert "http_port" not in NodeConfig.__dataclass_fields__
+    assert "http_port" not in GroupConfig.__dataclass_fields__
+    assert "port" not in NodeConfig.__dataclass_fields__
+
+
+# ── C5a: upload filename allowlist ───────────────────────────────────────────
+
+@pytest.mark.parametrize("name", [
+    "../../etc/passwd",
+    "..\\windows\\system32",
+    "/absolute/path",
+    "<img src=x onerror=alert(1)>",          # the H2 stored-XSS vector
+    'name";DROP TABLE x;--',
+    ".hidden",
+    "",
+    "a" * 200,
+    "file\x00.mp4",
+    "sub/dir/file.mp4",
+])
+def test_upload_rejects_unsafe_filenames(name):
+    """C5a/H2: only a conservative allowlist may reach the filesystem."""
+    assert not _safe_name_re().match(name), f"should be rejected: {name!r}"
+
+
+@pytest.mark.parametrize("name", [
+    "movie.mp4",
+    "My Holiday Video.mkv",
+    "report-2026.pdf",
+    "track_01.flac",
+])
+def test_upload_accepts_ordinary_filenames(name):
+    """The allowlist must not break normal use."""
+    assert _safe_name_re().match(name), f"should be accepted: {name!r}"
+
+
+def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
+    """A peer session wired to a real shared root, with sending stubbed out."""
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {"shared_root": shared_root, "index": index, "sk_node": index.sk_node}
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = None
+    session._user_id = user_id
+    session._pk_user = ""
+    session._uploads = {}
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def test_upload_cannot_overwrite_another_members_file(tmp_path):
+    """
+    C5a: uploads used to land in the shared root under a client-chosen name and
+    overwrite whatever was there. That let any member destroy the operator's files,
+    and — by becoming the recorded uploader of the replaced file — delete them
+    through the uploader path, bypassing the Ed25519 admin challenge entirely.
+    """
+    victim = _session(tmp_path, "victim-user")
+    shared_root = victim._ctx["shared_root"]
+
+    original = shared_root / "important.mp4"
+    original.write_bytes(b"operator's original content")
+
+    attacker = _session(tmp_path, "attacker-user")
+    attacker._do_file_upload({
+        "filename": "important.mp4",
+        "chunk_index": 0,
+        "total_chunks": 1,
+        "data": base64.b64encode(b"attacker content").decode(),
+    })
+
+    assert original.read_bytes() == b"operator's original content"
+    uploaded = shared_root / ".uploads" / "attacker-user" / "important.mp4"
+    assert uploaded.exists(), "upload should be quarantined, not dropped"
+    assert uploaded.read_bytes() == b"attacker content"
+
+
+def test_upload_rejects_out_of_order_chunks(tmp_path):
+    """C5a: chunk_index > 0 used to append blindly to any .part file on disk."""
+    session = _session(tmp_path, "user-1")
+    session._do_file_upload({
+        "filename": "movie.mp4", "chunk_index": 3, "total_chunks": 5,
+        "data": base64.b64encode(b"spliced").decode(),
+    })
+    assert any(m.get("type") == "error" for m in session.sent)
+
+
+def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path):
+    """C5a: even the original uploader goes through a fresh name, not an overwrite."""
+    session = _session(tmp_path, "user-1")
+    payload = {"filename": "movie.mp4", "chunk_index": 0, "total_chunks": 1,
+               "data": base64.b64encode(b"first").decode()}
+    session._do_file_upload(dict(payload))
+    session.sent.clear()
+
+    session._do_file_upload(dict(payload))
+    assert any(m.get("type") == "error" for m in session.sent)
+    stored = session._ctx["shared_root"] / ".uploads" / "user-1" / "movie.mp4"
+    assert stored.read_bytes() == b"first"
+
+
+# ── H1: group isolation ──────────────────────────────────────────────────────
+
+def test_chat_store_and_peers_are_per_group(tmp_path):
+    """
+    H1: chat_store and the peer registry were read from the shared transport
+    context, so on a multi-group node every group's messages went to the first
+    group's database and were served back to members of every other group.
+    """
+    index_a = GroupIndex(group_id="a" * 32, sk_node=Ed25519PrivateKey.generate())
+    index_b = GroupIndex(group_id="b" * 32, sk_node=Ed25519PrivateKey.generate())
+    groups = {
+        "a" * 32: {"chat_store": "STORE_A", "index": index_a, "shared_root": tmp_path},
+        "b" * 32: {"chat_store": "STORE_B", "index": index_b, "shared_root": tmp_path},
+    }
+    ctx = {"groups": groups}
+
+    sess_a = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    sess_a._ctx, sess_a._group_id, sess_a._user_id = ctx, "a" * 32, "alice"
+
+    sess_b = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    sess_b._ctx, sess_b._group_id, sess_b._user_id = ctx, "b" * 32, "bob"
+
+    assert sess_a._group_ctx()["chat_store"] == "STORE_A"
+    assert sess_b._group_ctx()["chat_store"] == "STORE_B"
+
+    sess_a._peer_registry()["alice"] = sess_a
+    sess_b._peer_registry()["bob"] = sess_b
+
+    # Alice's broadcast target set must not contain Bob, who is in another group.
+    assert "bob" not in sess_a._peer_registry()
+    assert "alice" not in sess_b._peer_registry()
+
+    sess_a._user_names()["alice"] = "Alice"
+    assert "alice" not in sess_b._user_names()
+
+
+def test_daemon_sets_no_global_chat_store(tmp_path):
+    """H1: the daemon must not hoist one group's chat store onto the transport."""
+    source = (Path(__file__).parent.parent
+              / "src" / "meshbay_node" / "daemon.py").read_text()
+    assert '_ctx["chat_store"]' not in source, (
+        "daemon must not assign a transport-wide chat_store — it leaks chat "
+        "across groups (H1)"
+    )
+
+
+# ── H2: node admin UI escaping ───────────────────────────────────────────────
+
+def test_no_member_can_hand_the_node_key_material(tmp_path):
+    """
+    C5b, strengthened by the invite redesign (docs/invite-pairing-v1.md).
+
+    This test used to assert that `gek_bundle_store` answered with an admin
+    challenge and stored nothing without an operator signature. The message is now
+    gone entirely: the node holds the GEK and wraps it itself, so no member ever
+    submits key material, authorized or not. Deleting the path is a stronger
+    guarantee than gating it, which is why the assertion changed rather than the
+    behaviour regressing.
+    """
+    from meshbay_common.protocol import MNP as _MNP
+
+    assert not hasattr(_MNP, "GEK_BUNDLE_STORE"), (
+        "the member-supplied bundle message is back — the node must never accept "
+        "key material over MNP (C5b)"
+    )
+
+    source = (Path(__file__).parent.parent
+              / "src" / "meshbay_node" / "transport" / "webrtc_server.py").read_text()
+    assert "_do_gek_bundle_store" not in source
+    assert "_admin_exec_bundle_store" not in source
+
+
+def test_unknown_message_stores_nothing(tmp_path):
+    """A peer sending the retired message must not reach any storage path."""
+    session = _session(tmp_path, "ordinary-member")
+    session._group_id = None
+    session._admin_ops = {}
+
+    stored = []
+
+    class _Store:
+        async def store(self, *args):
+            stored.append(args)
+
+    session._ctx["bundle_store"] = _Store()
+    session._handle_message({
+        "type": "gek_bundle_store",
+        "user_id": "victim", "group_id": "g" * 32,
+        "pk_eph_b64": "AA==", "nonce_b64": "AA==", "wrapped_b64": "AA==",
+    })
+
+    assert stored == [], "a retired message type still reached the bundle store"
+
+
+def test_gek_auto_activation_is_gone():
+    """
+    C5b: the node used to unwrap and adopt any bundle addressed to the operator.
+    Since the operator's X25519 public key is public, any member could hand the
+    node a GEK of their choosing. Nothing arriving over MNP may set a live GEK.
+    """
+    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
+              / "transport" / "webrtc_server.py").read_text()
+    assert "_try_activate_gek" not in source
+    assert 'unwrap_gek_aes' not in source, (
+        "the MNP path must not unwrap a GEK — activation is local-admin only"
+    )
+
+
+# ── H5: admin challenge is bound, not a blind signing oracle ─────────────────
+
+def _transcript(**kw):
+    from meshbay_common.adminop import admin_transcript
+    base = dict(op="file_delete", node_pk_b64="NODEPK", group_id="g" * 32,
+                subject="file-1", nonce=b"\x01" * 32, ts=1_700_000_000)
+    base.update(kw)
+    return admin_transcript(**base)
+
+
+def test_admin_transcript_is_domain_separated():
+    """H5: signatures here can never be valid in another MeshBay protocol."""
+    assert _transcript().startswith(b"meshbay:admin:v1")
+
+
+@pytest.mark.parametrize("field,value", [
+    ("op", "invite_create"),
+    ("subject", "file-2"),
+    ("node_pk_b64", "OTHERNODE"),
+    ("group_id", "h" * 32),
+    ("nonce", b"\x02" * 32),
+    ("ts", 1_700_000_001),
+])
+def test_admin_transcript_binds_every_field(field, value):
+    """
+    H5: a signature must not carry over to another operation, subject, node,
+    group, challenge or moment in time.
+    """
+    assert _transcript() != _transcript(**{field: value}), (
+        f"transcript ignores {field} — signature would be reusable"
+    )
+
+
+def test_admin_transcript_is_unambiguous():
+    """
+    H5/L4: fields are length-prefixed. With plain concatenation a crafted subject
+    could impersonate the following field and two different operations would
+    produce identical signed bytes.
+    """
+    a = _transcript(subject="file-1", group_id="g")
+    b = _transcript(subject="1", group_id="gfile-")
+    assert a != b, "concatenation is ambiguous — length prefixes missing"
+
+
+def test_admin_signature_does_not_transfer_between_operations(tmp_path):
+    """
+    H5: the concrete attack. A signature collected to delete a file must not
+    authorize storing a GEK bundle.
+    """
+    from meshbay_common.adminop import OP_FILE_DELETE, OP_INVITE_CREATE
+
+    sk_admin = Ed25519PrivateKey.generate()
+    delete_transcript = _transcript(op=OP_FILE_DELETE)
+    signature = sk_admin.sign(delete_transcript)
+
+    invite_transcript = _transcript(op=OP_INVITE_CREATE)
+    with pytest.raises(Exception):
+        sk_admin.public_key().verify(signature, invite_transcript)
+
+
+def test_admin_challenge_expires(tmp_path):
+    """H5: a stale challenge must not be usable."""
+    import time as _time
+    from meshbay_common.adminop import ADMIN_CHALLENGE_TTL, OP_FILE_DELETE
+
+    session = _session(tmp_path, "operator")
+    session._group_id = None
+    session._admin_ops = {
+        "op-1": {
+            "op": OP_FILE_DELETE, "subject": "file-1", "nonce": b"\x00" * 32,
+            "ts": int(_time.time()) - ADMIN_CHALLENGE_TTL - 5, "payload": {},
+        }
+    }
+    session._do_admin_response({"op_id": "op-1", "signature": ""})
+    assert any(m.get("type") == "error" and "expired" in m.get("detail", "").lower()
+               for m in session.sent)
+
+
+def test_denylist_persists_and_honours_groups(tmp_path):
+    """
+    H4: revocations lived only in memory, so a node restart silently un-revoked
+    everyone, and 'group' targets were dropped entirely — the hub signed and
+    broadcast them, the node's handler understood only 'user' and 'jti'.
+    """
+    from meshbay_node.transport import Denylist
+
+    path = tmp_path / "denylist.json"
+    first = Denylist(path=path)
+    first.deny_group("g-revoked")
+    first.deny_user("u-revoked")
+    first.deny_jti("j-revoked")
+
+    # A fresh instance stands in for a daemon restart.
+    reloaded = Denylist(path=path)
+    assert reloaded.is_denied("", "", "g-revoked"), "group revocation not honoured"
+    assert reloaded.is_denied("u-revoked", "")
+    assert reloaded.is_denied("", "j-revoked")
+    assert not reloaded.is_denied("someone", "other", "g-allowed")
+
+
+def test_swarm_registration_skips_private_groups():
+    """
+    H7: the daemon registered content hashes for every group, private included,
+    handing the hub a fingerprint of every private file. The bug was masked by a
+    mis-mounted route, so fixing the route without this filter would have turned a
+    dormant leak into a live one.
+    """
+    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
+              / "daemon.py").read_text()
+    assert 'visibility' in source and '_register_swarm' in source
+    # Both registration sites must gate on public visibility.
+    for marker in ['gctx.get("visibility") != "public"',
+                   'group_cfg.visibility == "public"']:
+        assert marker in source, f"swarm registration not gated: {marker}"
+
+
+def test_keystore_argon2_is_production_strength():
+    """M2: the keystore protects the node's private keys and sat at 64 MB."""
+    from meshbay_common.crypto import ARGON2_MEMORY_COST
+    assert ARGON2_MEMORY_COST >= 262144
+
+
+def test_keystore_records_argon2_params_for_migration(tmp_path):
+    """
+    M2: raising the parameters must not orphan existing keystores, so each
+    envelope records the parameters it was written with.
+    """
+    import json
+    from meshbay_node.keystore import create_keystore, load_keystore
+
+    path = tmp_path / "keystore.enc"
+    created = create_keystore(path=path, password="correct horse battery")
+    envelope = json.loads(path.read_text())
+    assert envelope["argon2"]["memory_cost"] >= 262144
+
+    reopened = load_keystore(path=path, password="correct horse battery")
+    assert reopened.pk_ed25519_b64 == created.pk_ed25519_b64
+
+
+def test_legacy_keystore_still_opens(tmp_path):
+    """M2: a keystore written under the 64 MB profile must still unlock."""
+    import base64 as _b64
+    import json
+    import msgpack
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+    from meshbay_common.crypto import (
+        LEGACY_ARGON2_ITERATIONS, LEGACY_ARGON2_LANES, LEGACY_ARGON2_MEMORY_COST,
+        derive_keystore_key, encrypt_keystore, pk_to_b64, sk_to_b64,
+    )
+    from meshbay_node.keystore import load_keystore
+
+    sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate()
+    payload = msgpack.packb({
+        "sk_ed25519_b64": sk_to_b64(sk_ed),
+        "sk_x25519_b64": sk_to_b64(sk_x),
+    }, use_bin_type=True)
+
+    salt = b"\x01" * 16
+    key = derive_keystore_key(
+        "legacy-pass", salt,
+        iterations=LEGACY_ARGON2_ITERATIONS,
+        memory_cost=LEGACY_ARGON2_MEMORY_COST,
+        lanes=LEGACY_ARGON2_LANES,
+    )
+    iv, ct, tag = encrypt_keystore(payload, key)
+
+    path = tmp_path / "legacy.enc"
+    # No "argon2" key — exactly how pre-M2 envelopes look.
+    path.write_text(json.dumps({
+        "version": 1,
+        "argon2_salt_b64": _b64.b64encode(salt).decode(),
+        "iv_b64": _b64.b64encode(iv).decode(),
+        "tag_b64": _b64.b64encode(tag).decode(),
+        "ciphertext_b64": _b64.b64encode(ct).decode(),
+    }))
+
+    keys = load_keystore(path=path, password="legacy-pass")
+    assert keys.pk_ed25519_b64 == pk_to_b64(sk_ed.public_key())
+    assert keys.pk_x25519_b64 == pk_to_b64(sk_x.public_key())
+
+
+def test_dead_gek_protocol_constants_removed():
+    """L1: the node never serves a GEK; the message types should not suggest it."""
+    from meshbay_common.protocol import MNP
+    assert not hasattr(MNP, "GEK_REQUEST")
+    assert not hasattr(MNP, "GEK_RESPONSE")
+
+
+def test_peer_errors_do_not_leak_internals():
+    """
+    L3: arbitrary exception text carries filesystem paths and internal state, so
+    the catch-all handler must not relay it.
+
+    Deliberately narrow: HandshakeError messages ARE sent to the peer, because a
+    client needs to know why it was refused, and those strings are authored for
+    that purpose. The check targets the generic `except Exception as e` path.
+    """
+    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
+              / "transport" / "webrtc_server.py").read_text()
+    assert '"detail": str(e)' not in source, (
+        "generic exception text relayed to peer — use a fixed message"
+    )
+    # And the catch-all must still exist, sending something opaque.
+    assert '"detail": "Request failed"' in source
+
+
+def test_pre_handshake_message_budget_is_small():
+    """
+    H6: the frame limit was a flat 64 MB applied before authentication, so an
+    unauthenticated peer could announce a huge frame and dribble bytes into it.
+    """
+    from meshbay_node.transport.webrtc_server import (
+        MAX_MSG, PRE_HANDSHAKE_MAX_MSG, _DataChannelBuffer,
+    )
+    assert PRE_HANDSHAKE_MAX_MSG <= 1024 * 1024
+    assert PRE_HANDSHAKE_MAX_MSG < MAX_MSG
+
+    buf = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG)
+    buf.feed(struct.pack(">I", PRE_HANDSHAKE_MAX_MSG + 1) + b"x")
+    with pytest.raises(ValueError):
+        list(buf.messages())
+
+
+def test_stream_segment_is_not_synchronous():
+    """
+    H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop,
+    stalling every peer on the node for up to thirty seconds per request.
+
+    Asserts the property (the worker is a coroutine, ffmpeg is spawned through
+    asyncio) rather than grepping for "subprocess.run" — which also matches the
+    comment that documents the old behaviour.
+    """
+    import ast
+    import inspect
+    from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+    assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async)
+
+    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
+              / "transport" / "webrtc_server.py").read_text()
+    tree = ast.parse(source)
+    blocking = [
+        node for node in ast.walk(tree)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Attribute)
+        and node.func.attr == "run"
+        and isinstance(node.func.value, ast.Name)
+        and node.func.value.id == "subprocess"
+    ]
+    assert not blocking, "blocking subprocess.run() in the event loop"
+    assert "_transcode_sem" in source, "ffmpeg spawns must be capped"
+
+
+def test_pre_proof_fetches_are_bounded():
+    """C4: the pre-proof bundle window is a disclosure surface; bound it."""
+    from meshbay_node.transport.webrtc_server import MAX_PRE_PROOF_FETCHES
+    assert 0 < MAX_PRE_PROOF_FETCHES <= 10
+
+
+def test_node_admin_ui_requires_token():
+    """
+    11.5.3: "localhost only" is not authentication. Any local process — or a
+    rebound browser page — could re-initialise a group's GEK and read the audit log.
+    """
+    from fastapi.testclient import TestClient
+    from meshbay_node.ui.app import create_ui_app
+
+    app = create_ui_app({"status": "running", "groups_ctx": {},
+                         "indexes": {}, "ui_token": "secret-token"})
+    client = TestClient(app)
+
+    assert client.get("/api/status").status_code == 403
+    assert client.get("/api/status?t=wrong").status_code == 403
+    assert client.get("/api/config?t=wrong").status_code == 403
+    assert client.get("/api/status?t=secret-token").status_code == 200
+    assert client.get(
+        "/api/status", headers={"X-MeshBay-Token": "secret-token"}
+    ).status_code == 200
+
+
+def test_admin_ui_escapes_filenames(tmp_path):
+    """
+    H2: filenames are chosen by any group member and were rendered into the
+    localhost admin UI unescaped, giving script execution against an
+    unauthenticated admin API.
+    """
+    from meshbay_node.ui.app import _render_page
+
+    payload = '<img src=x onerror="fetch(1)">'
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    index.add_entry(IndexEntry(
+        id="0" * 64, name=payload, path="", size=1, type="video", added_at=0,
+    ))
+
+    html = _render_page({
+        "status": "running",
+        "groups_ctx": {"g" * 32: {"index": index, "shared_root": tmp_path}},
+        "indexes": {"g" * 32: index},
+    })
+
+    assert payload not in html, "filename rendered unescaped — stored XSS (H2)"
+    assert "&lt;img" in html, "filename should appear escaped"
+
+
+def test_admin_ui_escapes_roster_usernames(tmp_path):
+    """
+    H2 again, for the roster: usernames originate at the hub and land on the
+    operator's own admin page, which can re-key groups and read the audit log.
+    """
+    from meshbay_node.ui.app import _render_page
+
+    payload = '<img src=x onerror="fetch(1)">'
+    html = _render_page(
+        {"status": "running", "groups_ctx": {}, "indexes": {}},
+        {
+            "identities": {"u1": {"user_id": "u1", "username": payload,
+                                  "pk_ed25519": "AAA", "pinned_at": "now",
+                                  "pinned_via": "code"}},
+            "members": [{"group_id": "", "user_id": "u1", "role": "operator",
+                         "status": "active"}],
+            "invites": [],
+        },
+    )
+
+    assert payload not in html, "username rendered unescaped — stored XSS (H2)"
+    assert "&lt;img" in html
diff --git a/packages/meshbay-node/tests/test_transport.py b/packages/meshbay-node/tests/test_transport.py
deleted file mode 100644
index 0e70d72..0000000
--- a/packages/meshbay-node/tests/test_transport.py
+++ /dev/null
@@ -1,222 +0,0 @@
-"""
-Integration test: ChunkServer ↔ ChunkClient over TLS.
-
-Starts a real TLS server on localhost, connects a client,
-fetches index and a chunk, verifies signature+hash+decryption.
-"""
-
-import asyncio
-import base64
-import os
-import time
-import jwt
-import pytest
-from pathlib import Path
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
-from cryptography.hazmat.primitives import serialization
-
-from meshbay_common.crypto import generate_gek, pk_to_b64
-from meshbay_node.indexer import DirectoryIndexer, GroupIndex
-from meshbay_node.transport.server import ChunkServer
-from meshbay_node.transport.client import ChunkClient
-
-
-@pytest.fixture
-def sk_node():
-    return Ed25519PrivateKey.generate()
-
-@pytest.fixture
-def sk_hub():
-    return Ed25519PrivateKey.generate()
-
-@pytest.fixture
-def gek():
-    return generate_gek()
-
-@pytest.fixture
-def shared_dir(tmp_path):
-    d = tmp_path / "shared"
-    d.mkdir()
-    (d / "test.mp4").write_bytes(os.urandom(2 * 1024 * 1024))   # 2 MB
-    (d / "small.txt").write_bytes(b"hello meshbay " * 100)
-    return d
-
-def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600, groups=None):
-    sk_pem = sk_hub.private_bytes(
-        serialization.Encoding.PEM,
-        serialization.PrivateFormat.PKCS8,
-        serialization.NoEncryption(),
-    )
-    now = int(time.time())
-    return jwt.encode({
-        "iss": "test-hub", "sub": user_id,
-        "pk_user": pk_node_b64, "hub_id": "test-hub",
-        "jti": "test-jti",
-        "iat": now, "exp": now + ttl,
-        "groups": groups or [],
-    }, sk_pem, algorithm="EdDSA")
-
-
-@pytest.mark.asyncio
-async def test_chunk_server_client_roundtrip(
-        sk_node, sk_hub, gek, shared_dir, tmp_path):
-    """Full integration: server serves a chunk, client verifies and decrypts."""
-
-    # Build index
-    indexer = DirectoryIndexer(
-        root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
-    await indexer.initial_scan()
-    assert indexer.index.count == 2
-
-    # Hub PK for JWT verification
-    hub_pk_pem = sk_hub.public_key().public_bytes(
-        serialization.Encoding.PEM,
-        serialization.PublicFormat.SubjectPublicKeyInfo)
-
-    # TLS cert in tmp dir
-    cert_path = tmp_path / "node.crt"
-    key_path  = tmp_path / "node.key"
-
-    server = ChunkServer(
-        sk_node=sk_node,
-        hub_pk_pem=hub_pk_pem,
-        gek=gek,
-        shared_root=shared_dir,
-        index=indexer.index,
-        host="127.0.0.1",
-        port=0,   # OS picks a free port
-        cert_path=cert_path,
-        key_path=key_path,
-    )
-    await server.start()
-    port = server._server.sockets[0].getsockname()[1]
-
-    token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()))
-
-    # Find the large test file in the index
-    entry = next(e for e in indexer.index.entries if e.name == "test.mp4")
-
-    async with ChunkClient(
-        host="127.0.0.1",
-        port=port,
-        jwt_token=token,
-        gek=gek,
-        pk_node_b64=pk_to_b64(sk_node.public_key()),
-    ) as client:
-        # Fetch first chunk
-        chunk0 = await client.fetch_chunk(entry.id, chunk_index=0)
-        assert len(chunk0) == 1024 * 1024   # first 1MB of 2MB file
-
-        # Fetch second chunk
-        chunk1 = await client.fetch_chunk(entry.id, chunk_index=1)
-        assert len(chunk1) == 1024 * 1024   # second 1MB
-
-        # Reassembled file matches original
-        original = (shared_dir / "test.mp4").read_bytes()
-        assert chunk0 + chunk1 == original
-
-    await server.stop()
-
-
-@pytest.mark.asyncio
-async def test_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path):
-    hub_pk_pem = sk_hub.public_key().public_bytes(
-        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
-
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g",
-                               sk_node=sk_node, gek=gek)
-    await indexer.initial_scan()
-
-    cert_path = tmp_path / "node.crt"
-    key_path  = tmp_path / "node.key"
-
-    server = ChunkServer(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
-        host="127.0.0.1", port=0,
-        cert_path=cert_path, key_path=key_path,
-    )
-    await server.start()
-    port = server._server.sockets[0].getsockname()[1]
-
-    # Use a different hub key to sign the token
-    sk_other_hub = Ed25519PrivateKey.generate()
-    bad_token = make_jwt(sk_other_hub, pk_to_b64(sk_node.public_key()))
-
-    with pytest.raises(Exception):
-        async with ChunkClient(
-            host="127.0.0.1", port=port,
-            jwt_token=bad_token, gek=gek,
-            pk_node_b64=pk_to_b64(sk_node.public_key()),
-        ) as client:
-            pass
-
-    await server.stop()
-
-
-@pytest.mark.asyncio
-async def test_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path):
-    """TCP+TLS server rejects a client whose JWT groups don't include the requested group_id."""
-    hub_pk_pem = sk_hub.public_key().public_bytes(
-        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
-
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g",
-                               sk_node=sk_node, gek=gek)
-    await indexer.initial_scan()
-
-    cert_path = tmp_path / "node.crt"
-    key_path  = tmp_path / "node.key"
-
-    server = ChunkServer(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
-        host="127.0.0.1", port=0,
-        cert_path=cert_path, key_path=key_path,
-    )
-    await server.start()
-    port = server._server.sockets[0].getsockname()[1]
-
-    token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"])
-
-    with pytest.raises(ConnectionError, match="rejected"):
-        async with ChunkClient(
-            host="127.0.0.1", port=port,
-            jwt_token=token, gek=gek,
-            pk_node_b64=pk_to_b64(sk_node.public_key()),
-            group_id="group-b",
-        ) as client:
-            pass
-
-    await server.stop()
-
-
-@pytest.mark.asyncio
-async def test_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path):
-    hub_pk_pem = sk_hub.public_key().public_bytes(
-        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g",
-                               sk_node=sk_node, gek=gek)
-    await indexer.initial_scan()
-
-    cert_path = tmp_path / "node.crt"
-    key_path  = tmp_path / "node.key"
-
-    server = ChunkServer(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
-        host="127.0.0.1", port=0,
-        cert_path=cert_path, key_path=key_path,
-    )
-    await server.start()
-    port = server._server.sockets[0].getsockname()[1]
-    token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()))
-
-    async with ChunkClient(
-        host="127.0.0.1", port=port, jwt_token=token,
-        gek=gek, pk_node_b64=pk_to_b64(sk_node.public_key()),
-    ) as client:
-        wire = await client.fetch_index()
-        recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
-        assert recovered.count == 2
-
-    await server.stop()
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index 693a68b..93cd3fd 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -19,7 +19,9 @@ import jwt
 import msgpack
 import pytest
 from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey, Ed25519PublicKey,
+)
 from aiortc import RTCPeerConnection, RTCSessionDescription
 
 from meshbay_common import MNP_VERSION
@@ -29,10 +31,24 @@ from meshbay_common.crypto import (
     wrap_gek,
     wrap_gek_aes,
     unwrap_gek,
+    unwrap_gek_aes,
 )
 from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
 from meshbay_common.protocol import MNP
+TEST_GROUP = "g"
+
+from meshbay_common.handshake import (
+    NONCE_LEN, ROLE_CLIENT, ROLE_NODE, handshake_transcript,
+    make_proof, verify_proof, webrtc_binding,
+)
+from meshbay_common.adminop import (
+    OP_FILE_DELETE,
+    OP_INVITE_CREATE,
+    admin_transcript,
+)
+from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript
 from meshbay_node.bundle_store import BundleStore
+from meshbay_node.roster import Roster
 from meshbay_node.indexer import DirectoryIndexer
 from meshbay_node.transport.webrtc_server import WebRTCTransport
 
@@ -72,6 +88,9 @@ def _hub_pk_pem(sk_hub):
 
 
 def _make_jwt(sk_hub, groups=None, pk_user="test"):
+    # group_id is mandatory now (M1), so the default token must be a member
+    # of the group the tests connect to. Tests that exercise refusal pass
+    # groups=[...] explicitly.
     sk_pem = sk_hub.private_bytes(
         serialization.Encoding.PEM,
         serialization.PrivateFormat.PKCS8,
@@ -82,10 +101,25 @@ def _make_jwt(sk_hub, groups=None, pk_user="test"):
         "iss": "test-hub", "sub": "user-001",
         "pk_user": pk_user, "hub_id": "test-hub",
         "jti": "test-jti-webrtc", "iat": now, "exp": now + 3600,
-        "groups": groups or [],
+        "groups": groups if groups is not None else [TEST_GROUP],
     }, sk_pem, algorithm="EdDSA")
 
 
+def _transcript_from(challenge_msg: dict) -> bytes:
+    """
+    Rebuild the signed transcript from an admin_challenge, the way a real client
+    does — from the announced fields, never from opaque bytes on the wire (H5).
+    """
+    return admin_transcript(
+        op=challenge_msg["op"],
+        node_pk_b64=challenge_msg["node_pk"],
+        group_id=challenge_msg["group_id"],
+        subject=challenge_msg["subject"],
+        nonce=base64.b64decode(challenge_msg["nonce"]),
+        ts=challenge_msg["ts"],
+    )
+
+
 def _pack(obj: dict) -> bytes:
     data = msgpack.packb(obj, use_bin_type=True)
     return struct.pack(">I", len(data)) + data
@@ -103,36 +137,80 @@ def _extract_dtls_fp(sdp: str) -> bytes:
     return b""
 
 
-async def _handshake_with_gek_proof(channel, received, sk_hub, gek, groups=None,
-                                    browser_pc=None):
-    """Send handshake, handle GEK challenge, return handshake_ack."""
-    token = _make_jwt(sk_hub, groups=groups)
+async def _do_mnp_handshake(channel, received, token, gek, pc, group_id):
+    """
+    Client half of the unified handshake (11.5.4): client nonce, length-prefixed
+    role-bound transcript, and verification of the node's own proof + signature.
+    """
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
-        "type": MNP.HANDSHAKE,
-        "v": MNP_VERSION,
-        "token": token,
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
+        "token": token, "group_id": group_id,
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
-    if msg["type"] == MNP.HANDSHAKE_CHALLENGE:
-        nonce = base64.b64decode(msg["nonce"])
-        offer_fp = b""
-        answer_fp = b""
-        if browser_pc:
-            offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp)
-            answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp)
-        proof = hmac.new(gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest()
-        channel.send(_pack({
-            "type": MNP.HANDSHAKE_RESPONSE,
-            "v": MNP_VERSION,
-            "proof": base64.b64encode(proof).decode(),
-        }))
-        msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    if msg["type"] != MNP.HANDSHAKE_CHALLENGE:
+        return msg
+
+    nonce_s = base64.b64decode(msg["nonce"])
+    binding = webrtc_binding(
+        _extract_dtls_fp(pc.localDescription.sdp),
+        _extract_dtls_fp(pc.remoteDescription.sdp),
+    )
+    proof = make_proof(gek, ROLE_CLIENT, group_id, nonce_c, nonce_s, binding)
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
+        "proof": base64.b64encode(proof).decode(),
+    }))
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+
+    if ack.get("type") == MNP.HANDSHAKE_ACK:
+        # The client must authenticate the node too (C3).
+        assert verify_proof(
+            gek, base64.b64decode(ack["proof"]), ROLE_NODE,
+            group_id, nonce_c, nonce_s, binding), "node proof invalid"
+        transcript = handshake_transcript(
+            ROLE_NODE, group_id, nonce_c, nonce_s, binding)
+        Ed25519PublicKey.from_public_bytes(
+            base64.b64decode(ack["node_pk"])
+        ).verify(base64.b64decode(ack["sig"]), transcript)
+    return ack
+
+
+async def _handshake_with_gek_proof(channel, received, sk_hub, gek, groups=None,
+                                    browser_pc=None, group_id=TEST_GROUP):
+    """Send handshake, handle GEK challenge, return handshake_ack."""
+    token = _make_jwt(sk_hub, groups=groups or [group_id])
+    msg = await _do_mnp_handshake(
+        channel, received, token, gek, browser_pc, group_id)
     assert msg["type"] == MNP.HANDSHAKE_ACK
     return msg
 
 
-async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None):
-    """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue)."""
+def _token(sk_hub, jwt_sub, peer_id, group_id, pk_user="test"):
+    """A hub-issued user token, as the browser would present it."""
+    sk_h_pem = sk_hub.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption(),
+    )
+    now = int(time.time())
+    return jwt.encode({
+        "iss": "test-hub", "sub": jwt_sub,
+        "pk_user": pk_user, "hub_id": "test-hub",
+        "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600,
+        "groups": [group_id], "scope": "user",
+    }, sk_h_pem, algorithm="EdDSA")
+
+
+async def _open_channel(transport, peer_id):
+    """
+    Signaling only: a live DataChannel with no MNP handshake performed.
+
+    Separate from `_setup_peer` because someone joining a group for the first time
+    cannot complete the handshake — they have no GEK to prove — and the join has to
+    happen in that window.
+    """
     pc = RTCPeerConnection()
     q = asyncio.Queue()
     buf = bytearray()
@@ -161,6 +239,13 @@ async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_us
     answer_sdp, _ = await transport.handle_offer(pc.localDescription.sdp, peer_id)
     await pc.setRemoteDescription(RTCSessionDescription(sdp=answer_sdp, type="answer"))
     await asyncio.wait_for(ready.wait(), timeout=5.0)
+    return pc, ch, q
+
+
+async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None,
+                      group_id=TEST_GROUP):
+    """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue)."""
+    pc, ch, q = await _open_channel(transport, peer_id)
 
     pk_user = "test"
     if sk_user:
@@ -169,31 +254,9 @@ async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_us
                 serialization.Encoding.Raw, serialization.PublicFormat.Raw)
         ).decode()
 
-    sk_h_pem = sk_hub.private_bytes(
-        serialization.Encoding.PEM,
-        serialization.PrivateFormat.PKCS8,
-        serialization.NoEncryption(),
-    )
-    now = int(time.time())
-    token = jwt.encode({
-        "iss": "test-hub", "sub": jwt_sub,
-        "pk_user": pk_user, "hub_id": "test-hub",
-        "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600,
-        "groups": [],
-    }, sk_h_pem, algorithm="EdDSA")
+    token = _token(sk_hub, jwt_sub, peer_id, group_id, pk_user)
 
-    ch.send(_pack({"type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token}))
-    msg = await asyncio.wait_for(q.get(), timeout=5.0)
-    if msg["type"] == MNP.HANDSHAKE_CHALLENGE:
-        nonce = base64.b64decode(msg["nonce"])
-        offer_fp = _extract_dtls_fp(pc.localDescription.sdp)
-        answer_fp = _extract_dtls_fp(pc.remoteDescription.sdp)
-        proof = hmac.new(gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest()
-        ch.send(_pack({
-            "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
-            "proof": base64.b64encode(proof).decode(),
-        }))
-        msg = await asyncio.wait_for(q.get(), timeout=5.0)
+    msg = await _do_mnp_handshake(ch, q, token, gek, pc, group_id)
     assert msg["type"] == MNP.HANDSHAKE_ACK
     return pc, ch, q
 
@@ -676,6 +739,8 @@ async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir)
     token = _make_jwt(sk_hub)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+        "group_id": TEST_GROUP,
+        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
     }))
 
     challenge = await asyncio.wait_for(received.get(), timeout=5.0)
@@ -734,6 +799,8 @@ async def test_webrtc_dtls_channel_binding_detects_mitm(sk_node, sk_hub, gek, sh
     token = _make_jwt(sk_hub)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+        "group_id": TEST_GROUP,
+        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
     }))
 
     challenge = await asyncio.wait_for(received.get(), timeout=5.0)
@@ -783,13 +850,13 @@ async def test_webrtc_admin_challenge_response(sk_node, sk_hub, gek, shared_dir)
 
     challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
-    assert challenge_msg["file_id"] == entry.id
+    assert challenge_msg["op"] == OP_FILE_DELETE
+    assert challenge_msg["subject"] == entry.id
 
-    challenge = base64.b64decode(challenge_msg["challenge"])
-    signature = sk_admin.sign(challenge)
+    signature = sk_admin.sign(_transcript_from(challenge_msg))
     channel.send(_pack({
         "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
-        "file_id": entry.id,
+        "op_id": challenge_msg["op_id"],
         "signature": base64.b64encode(signature).decode(),
     }))
 
@@ -832,11 +899,10 @@ async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_
     challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
 
-    challenge = base64.b64decode(challenge_msg["challenge"])
-    bad_sig = sk_attacker.sign(challenge)
+    bad_sig = sk_attacker.sign(_transcript_from(challenge_msg))
     channel.send(_pack({
         "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
-        "file_id": entry.id,
+        "op_id": challenge_msg["op_id"],
         "signature": base64.b64encode(bad_sig).decode(),
     }))
 
@@ -914,14 +980,14 @@ async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, s
 
     challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
-    assert challenge_msg["file_id"] == entry.id
+    assert challenge_msg["op"] == OP_FILE_DELETE
+    assert challenge_msg["subject"] == entry.id
 
     # Sign with uploader's Ed25519 key
-    challenge = base64.b64decode(challenge_msg["challenge"])
-    signature = sk_uploader.sign(challenge)
+    signature = sk_uploader.sign(_transcript_from(challenge_msg))
     channel.send(_pack({
         "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
-        "file_id": entry.id,
+        "op_id": challenge_msg["op_id"],
         "signature": base64.b64encode(signature).decode(),
     }))
 
@@ -979,11 +1045,10 @@ async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, share
     assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
 
     # Sign with user B's key (wrong key)
-    challenge = base64.b64decode(challenge_msg["challenge"])
-    bad_sig = sk_user_b.sign(challenge)
+    bad_sig = sk_user_b.sign(_transcript_from(challenge_msg))
     channel.send(_pack({
         "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
-        "file_id": entry.id,
+        "op_id": challenge_msg["op_id"],
         "signature": base64.b64encode(bad_sig).decode(),
     }))
 
@@ -1014,54 +1079,122 @@ def x25519_keypair():
 
 
 @pytest.mark.asyncio
-async def test_gek_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir,
-                                          tmp_path, x25519_keypair):
-    """GEK bundle stored on node via DataChannel, then fetched during handshake."""
+async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_dir,
+                                                 tmp_path, x25519_keypair):
+    """
+    The whole invite flow over a real DataChannel, end to end.
+
+    The operator asks for a code; the invitee — who has never held the group key
+    and therefore cannot complete the GEK proof — redeems it in the pre-proof
+    window and the node wraps the key for the X25519 key they just proved they
+    hold. At no point is a public key fetched from the hub, which is the point:
+    that lookup was H3.
+    """
     hub_pk_pem = _hub_pk_pem(sk_hub)
     indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
-    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
-    await bundle_store.open()
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
         shared_root=shared_dir, index=indexer.index,
         stun_servers=[],
     )
-    transport._ctx["bundle_store"] = bundle_store
+    transport._ctx["roster"] = roster
+    transport._ctx["has_admin_authority"] = True
+    transport._ctx["groups"] = {
+        TEST_GROUP: {"gek": gek, "shared_root": shared_dir, "index": indexer.index},
+    }
+
+    # A paired operator, as `meshbay-node operator pair` would have left it.
+    sk_admin = Ed25519PrivateKey.generate()
+    admin_pk_b64 = pk_to_b64(sk_admin.public_key())
+    await roster.pin_identity("user-001", "grenet", admin_pk_b64, "AA==", "code")
+    await roster.set_member("", "user-001", ROLE_OPERATOR, "active", "local-cli")
 
-    # Connect as admin and store a GEK bundle for user-002
     pc_admin, ch_admin, q_admin = await _setup_peer(
         transport, sk_hub, gek, "peer-admin")
 
-    sk_x_raw, pk_x_raw = x25519_keypair
-    bundle = wrap_gek(gek, pk_x_raw)
+    # 1. The operator asks the node for an invitation code.
+    ch_admin.send(_pack({
+        "type": MNP.INVITE_CREATE, "v": MNP_VERSION,
+        "user_id": "user-002", "group_id": TEST_GROUP, "username": "bob",
+    }))
+    challenge_msg = await asyncio.wait_for(q_admin.get(), timeout=5.0)
+    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
+    assert challenge_msg["op"] == OP_INVITE_CREATE
+    assert challenge_msg["subject"] == "user-002"
 
     ch_admin.send(_pack({
-        "type": MNP.GEK_BUNDLE_STORE,
-        "v": MNP_VERSION,
-        "user_id": "user-002",
-        "group_id": "g",
-        "pk_eph_b64": bundle["pk_eph_b64"],
-        "nonce_b64": bundle["nonce_b64"],
-        "wrapped_b64": bundle["wrapped_b64"],
+        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
+        "op_id": challenge_msg["op_id"],
+        "signature": base64.b64encode(
+            sk_admin.sign(_transcript_from(challenge_msg))).decode(),
     }))
-    ack = await asyncio.wait_for(q_admin.get(), timeout=5.0)
-    assert ack["type"] == "ack"
-    assert ack["detail"] == "gek_bundle_stored"
+    invite = await asyncio.wait_for(q_admin.get(), timeout=5.0)
+    assert invite["type"] == MNP.INVITE_RESULT
+    code = invite["code"]
+    assert code and len(code) == 9        # XXXX-XXXX
 
-    # Verify bundle was persisted
-    stored = await bundle_store.fetch("g", "user-002")
-    assert stored is not None
-    assert stored["pk_eph_b64"] == bundle["pk_eph_b64"]
+    # 2. Bob connects. He cannot prove GEK possession — he has never had it — so
+    #    he redeems the code in the pre-proof window instead.
+    sk_x_raw, pk_x_raw = x25519_keypair
+    sk_bob_ed = Ed25519PrivateKey.generate()
+    pc_bob, ch_bob, q_bob = await _open_channel(transport, "peer-bob")
 
-    # Unwrap to verify it's correct
-    recovered = unwrap_gek(stored, sk_x_raw, pk_x_raw)
-    assert recovered == gek
+    nonce_c = os.urandom(NONCE_LEN)
+    ch_bob.send(_pack({
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
+        "token": _token(sk_hub, "user-002", "peer-bob", TEST_GROUP),
+        "group_id": TEST_GROUP,
+        "nonce": base64.b64encode(nonce_c).decode(),
+    }))
+    challenge = await asyncio.wait_for(q_bob.get(), timeout=5.0)
+    assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE
+    nonce_s = base64.b64decode(challenge["nonce"])
 
-    await bundle_store.close()
+    # Bob signs a transcript naming the node, and he cannot complete the handshake
+    # that would prove its key — he has no GEK yet. So he has to be able to learn
+    # it from the challenge; taking it from the test's own knowledge of sk_node
+    # would hide the fact that a real client cannot.
+    assert challenge["node_pk"] == pk_to_b64(sk_node.public_key()), (
+        "the challenge must announce the node key to a first-time joiner")
+    node_pk_b64 = challenge["node_pk"]
+
+    pk_ed_b64 = pk_to_b64(sk_bob_ed.public_key())
+    pk_x_b64 = base64.b64encode(pk_x_raw).decode()
+    ts = int(time.time())
+    transcript = join_transcript(
+        node_pk_b64=node_pk_b64,
+        group_id=TEST_GROUP, user_id="user-002",
+        pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64,
+        nonce_node=nonce_s, ts=ts,
+    )
+    ch_bob.send(_pack({
+        "type": MNP.JOIN_REQUEST, "v": MNP_VERSION,
+        "group_id": TEST_GROUP,
+        "pk_ed25519": pk_ed_b64, "pk_x25519": pk_x_b64,
+        "code": code, "ts": ts,
+        "sig": base64.b64encode(sk_bob_ed.sign(transcript)).decode(),
+    }))
+
+    result = await asyncio.wait_for(q_bob.get(), timeout=5.0)
+    assert result["type"] == MNP.JOIN_RESULT
+    assert result["ok"] is True
+    assert result["gek"] is True
+    assert result["role"] == ROLE_MEMBER
+
+    # 3. The key really is the group key, and only Bob's secret opens it.
+    assert unwrap_gek_aes(result, sk_x_raw, pk_x_raw) == gek
+
+    # 4. The code is spent.
+    assert await roster.consume_invite(code, "user-002") is None
+
+    await roster.close()
     await pc_admin.close()
+    await pc_bob.close()
     await transport.close_all()
 
 
@@ -1124,8 +1257,10 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di
 
     # Step 1: Send handshake with group_id so _pending_group is set
     token = _make_jwt(sk_hub, groups=["g"])
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
@@ -1140,11 +1275,12 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di
     recovered_gek = unwrap_gek(bundle_resp, sk_x_raw, pk_x_raw)
     assert recovered_gek == gek
 
-    nonce = base64.b64decode(msg["nonce"])
-    offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp)
-    answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp)
-    proof = hmac.new(recovered_gek, nonce + offer_fp + answer_fp,
-                     hashlib.sha256).digest()
+    nonce_s = base64.b64decode(msg["nonce"])
+    binding = webrtc_binding(
+        _extract_dtls_fp(browser_pc.localDescription.sdp),
+        _extract_dtls_fp(browser_pc.remoteDescription.sdp),
+    )
+    proof = make_proof(recovered_gek, ROLE_CLIENT, "g", nonce_c, nonce_s, binding)
 
     # Step 4: Complete handshake
     channel.send(_pack({
@@ -1229,8 +1365,10 @@ async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir,
     await asyncio.wait_for(ready.wait(), timeout=5.0)
 
     token = _make_jwt(sk_hub, groups=["g"])
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
@@ -1296,8 +1434,10 @@ async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir,
     await asyncio.wait_for(ready.wait(), timeout=5.0)
 
     token = _make_jwt(sk_hub, groups=["g"])
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
@@ -1313,9 +1453,18 @@ async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir,
 
 
 @pytest.mark.asyncio
-async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shared_dir,
+async def test_gek_not_auto_activated_on_bundle_store(sk_node, sk_hub, gek, shared_dir,
                                                        tmp_path, x25519_keypair):
-    """Storing the node operator's GEK bundle auto-activates GEK (AES variant)."""
+    """
+    A GEK bundle arriving over MNP must NOT become the node's live key (C5b).
+
+    This test previously asserted the opposite: storing a bundle addressed to the
+    node operator auto-activated it, with no signature required. Because the
+    operator's X25519 public key is public — the node publishes it in handshake_ack
+    — any group member could wrap a key of their own choosing for it and take over
+    the group, locking every legitimate member out. GEK activation now happens only
+    through the node's local admin UI or CLI.
+    """
     hub_pk_pem = _hub_pk_pem(sk_hub)
     indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
@@ -1324,7 +1473,8 @@ async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shar
     bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
     await bundle_store.open()
 
-    new_gek = generate_gek()
+    attacker_gek = generate_gek()
+    assert attacker_gek != gek
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
@@ -1336,14 +1486,18 @@ async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shar
     transport._ctx["sk_x25519_raw"] = sk_x_raw
     transport._ctx["pk_x25519_raw"] = pk_x_raw
     transport._ctx["pk_x25519_b64"] = base64.b64encode(pk_x_raw).decode()
+    transport._ctx["admin_pk_ed25519"] = Ed25519PrivateKey.generate().public_key()
 
     pc_admin, ch_admin, q_admin = await _setup_peer(
         transport, sk_hub, gek, "peer-setup-admin")
 
-    # Store GEK bundle wrapped with AES-GCM (browser-compatible)
-    node_bundle = wrap_gek_aes(new_gek, pk_x_raw)
+    # An ordinary member wraps a key of their choosing for the operator's public
+    # key and offers it to the node. The message that used to carry this no longer
+    # exists (the node wraps the GEK itself now), so it reaches no handler at all —
+    # a stronger outcome than the admin challenge this test used to assert.
+    node_bundle = wrap_gek_aes(attacker_gek, pk_x_raw)
     ch_admin.send(_pack({
-        "type": MNP.GEK_BUNDLE_STORE,
+        "type": "gek_bundle_store",
         "v": MNP_VERSION,
         "user_id": "node-operator",
         "group_id": "g",
@@ -1351,12 +1505,12 @@ async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shar
         "nonce_b64": node_bundle["nonce_b64"],
         "wrapped_b64": node_bundle["wrapped_b64"],
     }))
-    ack = await asyncio.wait_for(q_admin.get(), timeout=5.0)
-    assert ack["type"] == "ack"
 
-    await asyncio.sleep(0.2)
+    await asyncio.sleep(0.5)
+    assert q_admin.empty(), "the retired bundle message still gets a response"
 
-    assert transport._ctx.get("gek") == new_gek
+    assert transport._ctx.get("gek") == gek, "group key was seized over MNP (C5b)"
+    assert await bundle_store.fetch("g", "node-operator") is None
 
     await bundle_store.close()
     await pc_admin.close()
@@ -1398,6 +1552,8 @@ async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir):
     token = _make_jwt(sk_hub)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+        "group_id": TEST_GROUP,
+        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
     }))
 
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
@@ -1457,8 +1613,10 @@ async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_
     await asyncio.wait_for(ready.wait(), timeout=5.0)
 
     token = _make_jwt(sk_hub, groups=["g"])
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == MNP.HANDSHAKE_CHALLENGE