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, 487 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
new file mode 100644
index 0000000..c111581
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -0,0 +1,487 @@
+"""
+Operator operations — one implementation, several front doors.
+
+Three things ask this node to act: the CLI (over the loopback admin API), the
+local admin UI, and — from Stage B3 — signed MNP messages from a paired client.
+They must agree, and the way to make them agree is not to write the operation
+three times and hope.
+
+**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
+the same shape one size down. So each operation lives here once, takes the
+daemon's `state`, and knows nothing about HTTP, argv or MNP. The adapters
+translate: `ui/app.py` turns `OpError` into a JSON response, the CLI prints it,
+the MNP handler sends an error frame.
+
+**Authorization is not here.** Reaching this module already means the caller got
+past its adapter's check — the loopback session token (11.5.3) for the API, an
+Ed25519 signature verified against the roster for MNP. These functions do what
+they are told; deciding who may tell them is the adapter's job and stays visible
+in the adapter.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any
+
+from meshbay_common.crypto import generate_gek, wrap_gek_aes
+from meshbay_node.config import DEFAULT_CONFIG_PATH
+from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
+from meshbay_node.roots import RootError, RootSet
+
+log = logging.getLogger(__name__)
+
+
+class OpError(Exception):
+    """
+    An operation refused, with enough for any adapter to report it.
+
+    `status` is an HTTP code because one adapter needs one; the others ignore it.
+    `extra` carries the "here is what would have worked" payload — a bare "no
+    such group" leaves an operator guessing at a UUID.
+    """
+
+    def __init__(self, message: str, *, status: int = 400,
+                 extra: dict[str, Any] | None = None):
+        super().__init__(message)
+        self.message = message
+        self.status = status
+        self.extra = extra or {}
+
+    def as_dict(self) -> dict:
+        return {"error": self.message, **self.extra}
+
+
+# ── Shared lookups ───────────────────────────────────────────────────────────
+
+def _roster(state: dict):
+    roster = state.get("roster")
+    if not roster:
+        raise OpError("Roster not available", status=503)
+    return roster
+
+
+def _hub(state: dict):
+    hub = state.get("hub")
+    if not hub or not hub._session:
+        raise OpError("Hub not connected", status=503)
+    return hub
+
+
+def _group_ctx(state: dict, group_id: str) -> dict:
+    groups_ctx = state.get("groups_ctx", {})
+    if group_id not in groups_ctx:
+        raise OpError("Group not hosted on this node", status=404,
+                      extra={"available": [
+                          {"id": gid} for gid in groups_ctx]})
+    return groups_ctx[group_id]
+
+
+def _config(state: dict):
+    config = state.get("config")
+    if not config:
+        raise OpError("No config loaded", status=503)
+    return config
+
+
+# ── Roster ───────────────────────────────────────────────────────────────────
+
+async def read_roster(state: dict, group_id: str = "") -> dict:
+    roster = state.get("roster")
+    if not roster:
+        return {"identities": [], "members": [], "invites": []}
+    return {
+        "identities": await roster.list_identities(),
+        "members": await roster.list_members(group_id or None),
+        "invites": await roster.list_invites(),
+    }
+
+
+async def resolve_user(state: dict, username: str) -> dict:
+    """
+    Map a username to an account id.
+
+    The roster answers first — it is the node's own record. The hub is the
+    fallback for identities pinned before invitations carried a name, and for
+    people admitted through an open-join group. Only an account id comes back;
+    no key is ever taken from there.
+    """
+    roster = state.get("roster")
+    if roster:
+        for ident in await roster.list_identities():
+            if ident["username"] == username:
+                return {"user_id": ident["user_id"], "source": "roster"}
+    hub = state.get("hub")
+    if hub and hub._session:
+        try:
+            account = await hub.get_user_pubkeys(username)
+            return {"user_id": account["user_id"], "source": "hub"}
+        except Exception:
+            pass
+    raise OpError(f"Unknown user {username!r}", status=404)
+
+
+async def pair_operator(state: dict) -> dict:
+    """
+    Issue a one-time code that pairs a browser as this node's operator.
+
+    The code is the whole point: it binds the operator's browser identity key to
+    their account without asking the hub, which is what stops a hub from naming
+    itself node administrator (M3, and the same substitution as H3). Returned
+    once and stored only as a hash.
+    """
+    roster = _roster(state)
+    user_id = state.get("node_user_id")
+    if not user_id:
+        raise OpError("Node not connected to hub yet", status=503)
+
+    config = state.get("config")
+    ttl = (config.node.pair_ttl_hours if config else 24) * 3600
+    code = await roster.create_invite(
+        group_id="",                 # operator authority is node-wide
+        user_id=user_id,
+        role=ROLE_OPERATOR,
+        created_by="local-cli",
+        ttl=ttl,
+        username=(config.hub.username if config else ""),
+    )
+    invites = await roster.list_invites()
+    expires = next((i["expires_at"] for i in invites
+                    if i["user_id"] == user_id and i["role"] == ROLE_OPERATOR), "")
+    return {"code": code, "expires_at": expires, "user_id": user_id}
+
+
+async def create_invite(state: dict, group_id: str, username: str) -> dict:
+    """
+    Issue an invitation code.
+
+    The hub is asked for the account id and nothing else — never for a key. A hub
+    that answered with the wrong account would produce an invite whose code it
+    never learns, since the code goes to a human out of band.
+    """
+    roster = _roster(state)
+    _group_ctx(state, group_id)
+    hub = _hub(state)
+    try:
+        account = await hub.get_user_pubkeys(username)
+    except Exception as e:
+        raise OpError(f"Unknown user {username!r}: {e}", status=404) from e
+
+    config = state.get("config")
+    ttl = (config.node.invite_ttl_hours if config else 168) * 3600
+    code = await roster.create_invite(
+        group_id=group_id,
+        user_id=account["user_id"],
+        role=ROLE_MEMBER,
+        created_by="local-cli",
+        ttl=ttl,
+        username=username,
+    )
+    invites = await roster.list_invites()
+    expires = next((i["expires_at"] for i in invites
+                    if i["user_id"] == account["user_id"]
+                    and i["group_id"] == group_id), "")
+    return {"code": code, "expires_at": expires,
+            "username": username, "user_id": account["user_id"]}
+
+
+async def revoke_member(state: dict, user_id: str, group_id: str) -> dict:
+    """
+    Stop serving the group key to someone.
+
+    Takes effect on their next connection: the key is wrapped on demand, so there
+    is no stored bundle left behind that would outlive this. Rotating the group
+    key is still required — they hold the current one.
+    """
+    roster = _roster(state)
+    if not await roster.set_status(group_id, user_id, "revoked"):
+        raise OpError("No such member in that group", status=404)
+    log.info("Member revoked: user=%s group=%s", user_id[:8], group_id[:8])
+    return {"status": "revoked", "user_id": user_id, "group_id": group_id,
+            "reminder": "rotate the group key: meshbay-node gek rotate"}
+
+
+async def unpin_member(state: dict, user_id: str) -> dict:
+    """Forget a pinned identity, so the person can pair again with a new key."""
+    roster = _roster(state)
+    if not await roster.unpin(user_id):
+        raise OpError("No such pinned identity", status=404)
+    log.info("Identity unpinned: user=%s", user_id[:8])
+    return {"status": "unpinned", "user_id": user_id}
+
+
+# ── Group keys ───────────────────────────────────────────────────────────────
+
+async def set_gek(state: dict, group_id: str, *, rotate: bool = False) -> dict:
+    """
+    Generate the group key and activate it, or rotate an existing one.
+
+    Nothing is pre-wrapped for members. Each member's copy is produced when they
+    connect, for a key they proved they hold (`join_request`) — pre-wrapping used
+    to fetch public keys from the hub, which is H3 with the node as the victim
+    instead of the inviter. Only the node's own copy is stored, so the daemon can
+    reload the key across restarts without the operator's browser.
+
+    **`rotate` generates a fresh key even when one exists.** That is the point of
+    it: after a revocation the ex-member still holds the current key, and nothing
+    else takes it away from them. Without `rotate` an existing key is kept, so
+    running this twice is not destructive by accident.
+    """
+    ctx = _group_ctx(state, group_id)
+    hub = _hub(state)
+
+    bundle_store = state.get("bundle_store")
+    if not bundle_store:
+        raise OpError("Bundle store not available", status=503)
+
+    existing = ctx.get("gek")
+    gek = generate_gek() if (rotate or not existing) else existing
+    rotated = bool(existing) and gek is not existing
+    errors: list[str] = []
+
+    roster = state.get("roster")
+    authorized = len(await roster.list_members(group_id)) if roster else 0
+
+    node_user_id = hub._session.user_id if hub._session else None
+    pk_x_node_raw = state.get("pk_x25519_raw")
+    if pk_x_node_raw and node_user_id:
+        try:
+            node_bundle = wrap_gek_aes(gek, pk_x_node_raw)
+            await bundle_store.store(
+                group_id, f"_node_{node_user_id}",
+                node_bundle["pk_eph_b64"], node_bundle["nonce_b64"],
+                node_bundle["wrapped_b64"],
+            )
+            log.info("GEK wrapped for node keystore (daemon reload)")
+        except Exception as e:
+            errors.append(f"node keystore: {e}")
+            log.warning("Failed to wrap GEK for node keystore: %s", e)
+
+    ctx["gek"] = gek
+    log.info("GEK %s for group %s — %d authorized member(s) will receive it "
+             "on connect", "rotated" if rotated else "initialized",
+             group_id[:8], authorized)
+
+    # The transport holds its own view of the group; a rotation that did not
+    # reach it would keep serving the old key until the daemon restarted.
+    for transport_key in ("webrtc", "quic_server"):
+        transport = state.get(transport_key)
+        groups = getattr(transport, "_ctx", {}).get("groups") if transport else None
+        if groups and group_id in groups:
+            groups[group_id]["gek"] = gek
+
+    indexes = state.get("indexes") or {}
+    index = indexes.get(group_id)
+    if index is not None:
+        # The index is encrypted under the GEK; leaving the old key on it would
+        # serve members a listing they cannot open.
+        index.gek = gek
+
+    return {
+        "status": "rotated" if rotated else "ok",
+        "group_id": group_id,
+        "rotated": rotated,
+        "authorized_members": authorized,
+        "errors": errors,
+    }
+
+
+# ── Groups and roots ─────────────────────────────────────────────────────────
+
+async def list_groups(state: dict) -> dict:
+    """What this node hosts, with live status. Milestone 14.2."""
+    config = state.get("config")
+    groups_ctx = state.get("groups_ctx", {})
+    peers = state.get("peers") or {}
+    out = []
+    for gid, ctx in groups_ctx.items():
+        cfg = next((g for g in config.groups if g.id == gid), None) if config else None
+        idx = ctx.get("index")
+        roots = ctx.get("roots")
+        out.append({
+            "id": gid,
+            "name": cfg.name if cfg else gid[:8],
+            "visibility": cfg.visibility if cfg else "private",
+            "join_policy": cfg.join_policy if cfg else "invite",
+            "has_gek": bool(ctx.get("gek")),
+            "file_count": idx.count if idx else 0,
+            "index_version": idx.version if idx else 0,
+            "roots": roots.describe() if roots else [],
+            "peers": sum(1 for p in peers.values() if p.get("group_id") == gid),
+        })
+    return {"groups": out}
+
+
+async def attach_group(state: dict, name: str, shared_dir: str) -> dict:
+    """
+    Write a new [[groups]] block into node.toml.
+
+    The name-to-id lookup happens here because this process is the one logged
+    into the hub. Nothing is created on the hub: the group already exists, this
+    only tells the node to host it.
+    """
+    if not name or not shared_dir:
+        raise OpError("name and shared_dir are required")
+    config = _config(state)
+    hub = _hub(state)
+    try:
+        mine = await hub.list_my_groups()
+    except Exception as e:
+        raise OpError(f"Could not list groups: {e}", status=502) from e
+
+    match = [g for g in mine if g["id"] == name or g["name"] == name]
+    if not match:
+        raise OpError(f"No group of yours is called {name!r}", status=404,
+                      extra={"available": [{"name": g["name"], "id": g["id"]}
+                                           for g in mine]})
+    if len(match) > 1:
+        raise OpError(f"Several of your groups are called {name!r} — use the id",
+                      status=409,
+                      extra={"available": [{"name": g["name"], "id": g["id"]}
+                                           for g in match]})
+    group = match[0]
+
+    if any(g.id == group["id"] for g in config.groups):
+        raise OpError(f"{group['name']!r} is already hosted by this node", status=409)
+
+    path = Path(shared_dir).expanduser()
+    try:
+        path.mkdir(parents=True, exist_ok=True)
+    except OSError as e:
+        raise OpError(f"Cannot create {path}: {e}") from e
+
+    conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
+    # Appended as text rather than re-serialised: node.toml is hand-written and
+    # full of comments explaining decisions, and a round trip through a TOML
+    # writer would throw all of that away.
+    block = (f'\n[[groups]]\n'
+             f'id          = "{group["id"]}"\n'
+             f'name        = "{group["name"]}"\n'
+             f'visibility  = "{group.get("visibility", "private")}"\n'
+             f'\n  [[groups.roots]]\n'
+             f'  path   = "{path}"\n'
+             f'  upload = true\n')
+    try:
+        with conf_path.open("a") as f:
+            f.write(block)
+    except OSError as e:
+        raise OpError(f"Cannot write {conf_path}: {e}", status=500) from e
+
+    return {"group_id": group["id"], "name": group["name"],
+            "shared_dir": str(path), "config": str(conf_path),
+            "note": "restart the node to pick it up"}
+
+
+async def add_root(state: dict, group_id: str, path: str, *,
+                   name: str = "", kind: str = "generic",
+                   upload: bool = False) -> dict:
+    """
+    Add a directory to a group, refusing anything ambiguous.
+
+    Validated against the group's existing roots *before* being written, so a
+    config that would be refused at startup is refused here instead — where the
+    operator is watching and can fix it.
+    """
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    specs = [asdict(r) for r in cfg.roots]
+    specs.append({"path": path, "name": name, "kind": kind, "upload": upload})
+    try:
+        built = RootSet.build(specs)
+    except RootError as e:
+        raise OpError(str(e)) from e
+
+    added = built.roots[-1]
+    conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
+    raise OpError(
+        # Writing into the middle of a hand-written TOML file means finding the
+        # right [[groups]] block and appending inside it, which a text append
+        # cannot do. Until that is written, say so plainly rather than appending
+        # to the wrong group.
+        f"Add this to {conf_path} under the [[groups]] block for "
+        f"{cfg.name!r}, then restart the node:\n\n"
+        f'  [[groups.roots]]\n'
+        f'  path   = "{added.path}"\n'
+        + (f'  name   = "{added.name}"\n' if name else "")
+        + (f'  kind   = "{added.kind}"\n' if kind != "generic" else "")
+        + (f'  upload = true\n' if upload else ""),
+        status=501,
+        extra={"validated": True, "name": added.name, "path": str(added.path)},
+    )
+
+
+# ── Files ────────────────────────────────────────────────────────────────────
+
+async def delete_file(state: dict, group_id: str, file_id: str) -> dict:
+    """
+    Remove a file from a group. Milestone 14.11 — the last operator action that
+    needed a browser.
+
+    Authorization happened in the adapter. On the loopback path that is the
+    session token, which means physical or SSH access to the machine hosting the
+    files — an operator who can run this can also `rm` the file, so the check is
+    not weaker than the alternative.
+    """
+    ctx = _group_ctx(state, group_id)
+    index = ctx.get("index")
+    roots = ctx.get("roots")
+    if not index or not roots:
+        raise OpError("Group has no index", status=503)
+
+    entry = index.get_entry(file_id)
+    if not entry:
+        raise OpError("No such file in this group", status=404)
+
+    from meshbay_node.roots import entry_abs_path
+    path = entry_abs_path(roots, entry)
+    if path is None:
+        raise OpError(
+            f"{entry.name!r} is in root {entry.path.split('/')[0]!r}, which is "
+            f"not readable right now — the file is frozen, not gone", status=409)
+
+    try:
+        path.unlink()
+    except FileNotFoundError:
+        # Already gone from disk; drop the stale entry rather than refusing.
+        log.warning("Index named a file that is not on disk: %s", path)
+    except OSError as e:
+        raise OpError(f"Cannot delete {entry.name!r}: {e}", status=500) from e
+
+    index.remove_entry(file_id)
+    log.info("File deleted by operator: %s/%s", entry.path, entry.name)
+    return {"status": "deleted", "name": entry.name, "path": entry.path,
+            "group_id": group_id}
+
+
+# ── Revocation denylist ──────────────────────────────────────────────────────
+
+async def read_denylist(state: dict) -> dict:
+    """Milestone 14.10 — what the node is currently refusing."""
+    denylist = state.get("denylist")
+    if not denylist:
+        return {"users": [], "groups": [], "jtis": [], "count": 0}
+    entries = denylist.entries()
+    return {**entries, "count": sum(len(v) for v in entries.values())}
+
+
+async def clear_denylist(state: dict, *, subject: str = "") -> dict:
+    """
+    Drop denylist entries — all of them, or one identifier.
+
+    Deliberately not silent: a cleared denylist re-admits whoever it was keeping
+    out, and the count is what tells the operator whether they undid one
+    revocation or all of them.
+    """
+    denylist = state.get("denylist")
+    if not denylist:
+        raise OpError("No denylist in this process", status=503)
+    removed = denylist.clear(subject)
+    log.warning("Denylist cleared (%s): %d entr(y/ies) removed",
+                subject or "all", removed)
+    return {"status": "cleared", "removed": removed, "subject": subject or "all"}