fix(node): group isolation, upload confinement, GEK seizure, admin challenge

Phase 11.5 — findings H1, C5a, H2, C5b, H5 (see second-review.md).

Batched together because the node-side changes share webrtc_server.py and
cannot be separated into working commits.

H1 — cross-group chat leak. chat_store, the peer registry and the display-name
cache were read from the shared transport context, and daemon.py hoisted the
FIRST group's chat store onto it. On a node hosting several groups every
group's messages went to one database, chat_history served them back to members
of every other group, and chat broadcast reached all peers regardless of group.
All three now resolve through _group_ctx().

C5a — upload confinement. Uploads landed in the shared root under a
client-chosen name and overwrote whatever was there. Any member could destroy
the operator's files, and by becoming the recorded uploader of the replaced
file could then delete it through the uploader path, bypassing the Ed25519
admin challenge. Uploads now go to a per-user quarantine (.uploads/{user_id}/),
refuse to overwrite, and enforce chunk ordering, a filename allowlist and a
size cap.

H2 — stored XSS in the node admin UI. Filenames chosen by any group member were
interpolated raw into the localhost UI, which has no authentication, so script
execution there equals control of the node admin API. Now html.escape()
throughout, textContent in the audit table, plus CSP/nosniff/no-referrer. The
CSP contains exfiltration but cannot stop injected inline script — escaping is
the fix.

C5b — group key seizure. gek_bundle_store wrote whatever any member sent and
auto-activated bundles addressed to the node operator. The operator's X25519
public key is public (the node publishes it in handshake_ack), so any member
could wrap a key of their choosing for it and take over the group, locking
every legitimate member out. Storing now requires an operator signature and
_try_activate_gek is removed: nothing arriving over MNP can set a live GEK.

H5 — unbound signing oracle. The node challenged with 32 raw random bytes and
the client signed them blind, so a signature named no operation, subject, node
or time. New meshbay_common/adminop.py defines a length-prefixed,
domain-separated transcript; both sides build it independently and the client
refuses to sign when the announced op/subject do not match its request.

BREAKING: a group admin who does not operate the node can no longer store GEK
bundles on it. Invites must be performed by the node operator.

Adds tests/test_security_regressions.py. Verified against pre-fix source via
git stash. Three pre-existing tests asserted the vulnerable behaviour as a
feature and were inverted: gek auto-activation, and the transport-wide
chat_store in test_daemon.

Tests: 109 node, 132 hub+common.

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

3 files changed, 453 insertions, 24 deletions

diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py
index e899639..9ebf945 100644
--- a/packages/meshbay-node/tests/test_daemon.py
+++ b/packages/meshbay-node/tests/test_daemon.py
@@ -121,7 +121,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"
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..a480d21
--- /dev/null
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -0,0 +1,372 @@
+"""
+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
+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_gek_bundle_store_requires_admin_challenge(tmp_path):
+    """
+    C5b: gek_bundle_store used to write whatever any authenticated member sent.
+    It must now answer with a challenge and store nothing until a valid
+    node-operator signature arrives.
+    """
+    session = _session(tmp_path, "ordinary-member")
+    session._group_id = None
+    session._admin_ops = {}
+    session._ctx["admin_pk_ed25519"] = Ed25519PrivateKey.generate().public_key()
+
+    stored = []
+
+    class _Store:
+        async def store(self, *args):
+            stored.append(args)
+
+    session._ctx["bundle_store"] = _Store()
+    session._do_gek_bundle_store({
+        "user_id": "victim", "group_id": "g" * 32,
+        "pk_eph_b64": "AA==", "nonce_b64": "AA==", "wrapped_b64": "AA==",
+    })
+
+    assert stored == [], "bundle written without operator authorization (C5b)"
+    assert any(m.get("type") == "admin_challenge" for m in session.sent)
+
+
+def test_gek_bundle_store_refused_without_pinned_admin_key(tmp_path):
+    """C5b: deny by default — no pinned key means no privileged operation."""
+    session = _session(tmp_path, "ordinary-member")
+    session._group_id = None
+    session._admin_ops = {}
+    session._ctx["bundle_store"] = object()
+
+    session._do_gek_bundle_store({
+        "user_id": "victim", "group_id": "g" * 32,
+        "pk_eph_b64": "AA==", "nonce_b64": "AA==", "wrapped_b64": "AA==",
+    })
+    assert any(m.get("type") == "error" for m in session.sent)
+
+
+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", "gek_bundle_store"),
+    ("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_GEK_BUNDLE_STORE
+
+    sk_admin = Ed25519PrivateKey.generate()
+    delete_transcript = _transcript(op=OP_FILE_DELETE)
+    signature = sk_admin.sign(delete_transcript)
+
+    store_transcript = _transcript(op=OP_GEK_BUNDLE_STORE)
+    with pytest.raises(Exception):
+        sk_admin.public_key().verify(signature, store_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_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"
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index 693a68b..d57f4f5 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -32,6 +32,11 @@ from meshbay_common.crypto import (
 )
 from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
 from meshbay_common.protocol import MNP
+from meshbay_common.adminop import (
+    OP_FILE_DELETE,
+    OP_GEK_BUNDLE_STORE,
+    admin_transcript,
+)
 from meshbay_node.bundle_store import BundleStore
 from meshbay_node.indexer import DirectoryIndexer
 from meshbay_node.transport.webrtc_server import WebRTCTransport
@@ -86,6 +91,21 @@ def _make_jwt(sk_hub, groups=None, pk_user="test"):
     }, 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
@@ -783,13 +803,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 +852,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 +933,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 +998,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(),
     }))
 
@@ -1031,7 +1049,11 @@ async def test_gek_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir,
     )
     transport._ctx["bundle_store"] = bundle_store
 
-    # Connect as admin and store a GEK bundle for user-002
+    # Storing a bundle is a node-operator operation (C5b): the node challenges and
+    # only the pinned admin key is accepted.
+    sk_admin = Ed25519PrivateKey.generate()
+    transport._ctx["admin_pk_ed25519"] = sk_admin.public_key()
+
     pc_admin, ch_admin, q_admin = await _setup_peer(
         transport, sk_hub, gek, "peer-admin")
 
@@ -1047,6 +1069,19 @@ async def test_gek_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir,
         "nonce_b64": bundle["nonce_b64"],
         "wrapped_b64": bundle["wrapped_b64"],
     }))
+
+    challenge_msg = await asyncio.wait_for(q_admin.get(), timeout=5.0)
+    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
+    assert challenge_msg["op"] == OP_GEK_BUNDLE_STORE
+    assert challenge_msg["subject"] == "user-002"
+
+    signature = sk_admin.sign(_transcript_from(challenge_msg))
+    ch_admin.send(_pack({
+        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
+        "op_id": challenge_msg["op_id"],
+        "signature": base64.b64encode(signature).decode(),
+    }))
+
     ack = await asyncio.wait_for(q_admin.get(), timeout=5.0)
     assert ack["type"] == "ack"
     assert ack["detail"] == "gek_bundle_stored"
@@ -1313,9 +1348,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 +1368,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,12 +1381,13 @@ 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.
+    node_bundle = wrap_gek_aes(attacker_gek, pk_x_raw)
     ch_admin.send(_pack({
         "type": MNP.GEK_BUNDLE_STORE,
         "v": MNP_VERSION,
@@ -1351,12 +1397,16 @@ 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"
+
+    # The node demands an operator signature instead of storing and adopting it.
+    reply = await asyncio.wait_for(q_admin.get(), timeout=5.0)
+    assert reply["type"] == MNP.ADMIN_CHALLENGE
+    assert reply["op"] == OP_GEK_BUNDLE_STORE
 
     await asyncio.sleep(0.2)
 
-    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()