feat(node): several named roots per group, and one implementation per operation

Stage A — a group's content is a set of named roots
---------------------------------------------------
`shared_dir` becomes a list of {name, path, kind}. The name is the directory's
basename, derived once at add time and *stored*: recomputing it would
re-identify a whole library the day someone renames a folder on disk. Duplicate
names are refused case-insensitively and no root may contain another — both
compared with NFC folding, because most of these directories live on exFAT or
NTFS where `Films` and `films` are one directory.

Every index path carries its root name, in a one-root group as much as in a
five-root one. One path shape has to be got right once; two have to be kept
right for ever.

**A root that goes away freezes; it never empties.** Unmounting a volume makes
watchdog report every file under it as deleted, or presents an empty directory
to the next scan. Acting on either propagates deletions for a whole library to
every member, as though the owner had erased it. So a deletion is acted on only
once its root is confirmed readable, and availability is tracked per root — one
unplugged drive leaves the others serving. 12 tests, verified to fail against an
indexer without the check.

Events are not trusted to be complete either: ReadDirectoryChangesW drops them
under load and inotify on a FUSE mount misses changes made outside it. A
periodic reconciliation sweep is the only thing that recovers a missed event.

MNP 0.2 → 0.3 (additive). The hub needs no change: SwarmSource carries a content
hash, a node id and an endpoint — no paths, no filenames — and private groups
register nothing (H7).

Stage B — one implementation behind every front door
----------------------------------------------------
C1 and C6 were both "a second path into the node with its own weaker
handshake". Two implementations of `revoke` with two authorization checks is
that shape one size down. `meshbay_node/ops.py` holds each operation once,
takes the daemon state, and knows nothing about HTTP, argv or MNP. The loopback
API is one `_op(...)` line per endpoint; the MNP handlers call the same
functions. test_ops.py asserts the shape rather than trusting it.

Phase 14 is finished on top of it — `group list`, `gek init|rotate`, `reload`
(SIGHUP), `denylist show|clear`, `file list|rm`. **No operator action requires a
browser any more.** Plus `gek_rotate` and `member_unpin` as operator-signed MNP
operations: rotation is the half of revocation that revocation cannot do, since
the ex-member holds the current key, and the node generates the replacement
with its own CSPRNG — no key material crosses the wire, which is what the C5b
rule is actually about.

Two bugs found by running it rather than by testing it
------------------------------------------------------
GroupIndex is keyed by **content hash**, so the same bytes at two paths are one
entry — which is also why a scan reports ten files and indexes nine.
Reconciliation compared paths, so it decided the second path was a missed event
every 60 s, rewrote the entry and pushed an index update to every connected
peer. Seen in a live node's log.

`meshbay-node reload` crashed on first use with `subprocess` unimported: the
module compiles fine, which is the "syntax, not names" trap already recorded for
the SPA. test_cli_dispatch.py now walks every verb and refuses to let one be
added to the parser without an entry there.

Also corrected: protocol.py declared a second MNP_VERSION of "0.1" while the
wire carried "0.2" — harmless only because nothing imported it. And
_do_dir_create/_do_dir_delete referenced an undefined `filename` on their error
path.

740 tests pass; QE/deploy/e2e.py passes end to end against the live deployment.

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

1 files changed, 290 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_admin_ops_mnp.py b/packages/meshbay-node/tests/test_admin_ops_mnp.py
new file mode 100644
index 0000000..1bf8365
--- /dev/null
+++ b/packages/meshbay-node/tests/test_admin_ops_mnp.py
@@ -0,0 +1,290 @@
+"""
+`gek_rotate` and `member_unpin` over MNP.
+
+Both are destructive and both are new, so the tests are negative assertions:
+nobody without the operator's pinned key can reach them, a signature over the
+wrong transcript does not count, and the operation cannot be triggered by the
+request message alone.
+
+The rule these live under is worth restating, because it is easy to read
+draft-v5 §5.1 as forbidding them: **"nothing arriving over MNP can activate a
+GEK" is about key material arriving from outside** (C5b — a member handing the
+node a key of their choosing). An operator-signed instruction where the node
+generates the key with its own CSPRNG is a different shape, and it is the only
+thing that finishes a revocation: the ex-member still holds the current key.
+"""
+
+import base64
+import time
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from conftest import one_root
+from meshbay_common.adminop import (
+    OP_GEK_ROTATE,
+    OP_MEMBER_UNPIN,
+    admin_transcript,
+)
+from meshbay_common.crypto import pk_to_b64
+from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import open_roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+GROUP = "g" * 32
+
+
+@pytest.fixture
+async def roster(tmp_path):
+    r = await open_roster(tmp_path)
+    yield r
+    await r.close()
+
+
+def _keypair():
+    sk = Ed25519PrivateKey.generate()
+    return sk, pk_to_b64(sk.public_key())
+
+
+async def _session(tmp_path: Path, roster, *, operator: bool) -> WebRTCPeerSession:
+    """A peer session with an operator pinned, or deliberately without one."""
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    roots = one_root(shared)
+
+    sk_op, pk_op = _keypair()
+    if operator:
+        await roster.pin_identity("grenet", "grenet", pk_op, pk_op, "code")
+        await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
+
+    group_ctx = {"gek": b"\x01" * 32, "roots": roots, "index": index,
+                 "join_policy": "invite"}
+    state = {
+        "groups_ctx": {GROUP: group_ctx},
+        "roster": roster,
+        "indexes": {GROUP: index},
+        "bundle_store": _FakeBundleStore(),
+        "pk_x25519_raw": b"\x02" * 32,
+        "hub": _FakeHub(),
+        "node_user_id": "node-user",
+    }
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {
+        "roots": roots, "index": index, "sk_node": index.sk_node,
+        "roster": roster, "groups": {GROUP: group_ctx},
+        "has_admin_authority": operator,
+        "daemon_state": state,
+    }
+    session._group_id = GROUP
+    session._user_id = "grenet" if operator else "mallory"
+    session._username = session._user_id
+    session._pk_user = ""
+    session._uploads = {}
+    session._admin_ops = {}
+    session._remote_ip = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    session.state = state
+    session.sk_op = sk_op
+    session.spawned = []
+    session._spawn = session.spawned.append
+    return session
+
+
+class _FakeBundleStore:
+    def __init__(self):
+        self.stored = []
+
+    async def store(self, *args):
+        self.stored.append(args)
+
+
+class _FakeHub:
+    class _S:
+        user_id = "node-user"
+    _session = _S()
+
+
+def _last(session):
+    return session.sent[-1] if session.sent else {}
+
+
+async def _drain(session):
+    """Await whatever `_spawn` started. The real session holds its tasks; a
+    hand-built one collects them here so the assertion sees the result."""
+    for coro in session.spawned:
+        await coro
+    session.spawned.clear()
+
+
+async def _sign_and_exec(session, op: str, subject: str, sk, exec_fn):
+    challenge = _last(session)
+    assert challenge["type"] == "admin_challenge", challenge
+    transcript = admin_transcript(
+        op=op, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
+        subject=subject, nonce=base64.b64decode(challenge["nonce"]),
+        ts=challenge["ts"])
+    pending = session._admin_ops.get(challenge["op_id"]) or {
+        "op": op, "subject": subject}
+    await exec_fn(pending, transcript, sk.sign(transcript))
+
+
+# ── gek_rotate ───────────────────────────────────────────────────────────────
+
+async def test_rotation_needs_an_operator(tmp_path, roster):
+    """Without a paired operator there is nobody who could sign, so the node
+    fails closed and says why rather than issuing a challenge nobody can meet."""
+    session = await _session(tmp_path, roster, operator=False)
+
+    session._do_gek_rotate({})
+
+    assert _last(session)["type"] == "error"
+    assert "authorized key" in _last(session)["detail"]
+    assert not session._admin_ops
+
+
+async def test_the_request_alone_rotates_nothing(tmp_path, roster):
+    """The message asks; only a signature acts. A node that rotated here would
+    let any member lock the group out."""
+    session = await _session(tmp_path, roster, operator=True)
+    before = session._ctx["groups"][GROUP]["gek"]
+
+    session._do_gek_rotate({})
+
+    assert _last(session)["type"] == "admin_challenge"
+    assert session._ctx["groups"][GROUP]["gek"] == before
+
+
+async def test_a_members_signature_does_not_rotate(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    sk_mallory, pk_mallory = _keypair()
+    await roster.pin_identity("mallory", "mallory", pk_mallory, pk_mallory, "code")
+    await roster.set_member(GROUP, "mallory", ROLE_MEMBER, "active", "grenet")
+    before = session._ctx["groups"][GROUP]["gek"]
+
+    session._do_gek_rotate({})
+    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, sk_mallory,
+                         session._admin_exec_gek_rotate)
+
+    assert _last(session)["type"] == "error"
+    assert session._ctx["groups"][GROUP]["gek"] == before
+
+
+async def test_a_signature_over_another_operation_does_not_count(tmp_path, roster):
+    """
+    H5's rule: the node rebuilds the transcript from the operation it is holding
+    and verifies against *that*, so a signature collected for one act cannot be
+    presented as another.
+
+    Driven through `_do_admin_response`, deliberately. Handing a transcript
+    straight to `_admin_exec_*` would skip the reconstruction that is the
+    control, and the test would pass while proving nothing.
+    """
+    session = await _session(tmp_path, roster, operator=True)
+    before = session._ctx["groups"][GROUP]["gek"]
+
+    session._do_gek_rotate({})
+    challenge = _last(session)
+
+    # Signed over member_unpin, presented against the pending gek_rotate.
+    wrong = admin_transcript(
+        op=OP_MEMBER_UNPIN, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
+        subject=GROUP, nonce=base64.b64decode(challenge["nonce"]),
+        ts=challenge["ts"])
+    session._do_admin_response({
+        "op_id": challenge["op_id"],
+        "signature": base64.b64encode(session.sk_op.sign(wrong)).decode(),
+    })
+    await _drain(session)
+
+    assert _last(session)["type"] == "error"
+    assert session.state["groups_ctx"][GROUP]["gek"] == before
+
+
+async def test_the_operator_rotates_and_the_node_makes_the_key(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    before = session._ctx["groups"][GROUP]["gek"]
+
+    session._do_gek_rotate({})
+    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op,
+                         session._admin_exec_gek_rotate)
+
+    ack = _last(session)
+    assert ack["type"] == MNP.GEK_ROTATE_ACK, ack
+    after = session.state["groups_ctx"][GROUP]["gek"]
+    assert after != before, "the key did not change"
+    assert len(after) == 32
+    # Produced here, not received: no key material crossed the wire (C5b).
+    assert session.state["bundle_store"].stored, (
+        "the node's own copy was not stored — the daemon could not reload it")
+
+
+async def test_rotation_reaches_the_index(tmp_path, roster):
+    """The index is encrypted under the GEK. Leaving the old key on it would
+    serve members a listing they cannot open."""
+    session = await _session(tmp_path, roster, operator=True)
+
+    session._do_gek_rotate({})
+    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op,
+                         session._admin_exec_gek_rotate)
+
+    assert session.state["indexes"][GROUP].gek == \
+        session.state["groups_ctx"][GROUP]["gek"]
+
+
+# ── member_unpin ─────────────────────────────────────────────────────────────
+
+async def test_unpinning_needs_an_operator(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=False)
+    session._do_member_unpin({"user_id": "bob"})
+    assert _last(session)["type"] == "error"
+
+
+async def test_unpinning_yourself_is_refused(tmp_path, roster):
+    """It would end the authority of the connection performing the operation,
+    halfway through it."""
+    session = await _session(tmp_path, roster, operator=True)
+    session._do_member_unpin({"user_id": "grenet"})
+    assert _last(session)["detail"] == "Cannot unpin yourself"
+
+
+async def test_a_members_signature_does_not_unpin(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    sk_bob, pk_bob = _keypair()
+    await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")
+
+    session._do_member_unpin({"user_id": "bob"})
+    await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", sk_bob,
+                         session._admin_exec_member_unpin)
+
+    assert _last(session)["type"] == "error"
+    assert await roster.get_identity("bob") is not None, (
+        "a member removed their own pin — only the operator may")
+
+
+async def test_the_operator_unpins(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    _, pk_bob = _keypair()
+    await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")
+
+    session._do_member_unpin({"user_id": "bob"})
+    await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", session.sk_op,
+                         session._admin_exec_member_unpin)
+
+    assert _last(session)["type"] == MNP.MEMBER_UNPIN_ACK
+    assert await roster.get_identity("bob") is None
+
+
+async def test_unpinning_someone_unknown_says_so(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    session._do_member_unpin({"user_id": "nobody"})
+    await _sign_and_exec(session, OP_MEMBER_UNPIN, "nobody", session.sk_op,
+                         session._admin_exec_member_unpin)
+    assert _last(session)["type"] == "error"
+    assert "No such pinned identity" in _last(session)["detail"]