From e9d5e979fdab9a1cc3c729d602e6f27207b9480c Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 18 Aug 2026 02:15:02 +0200 Subject: feat(node): several named roots per group, and one implementation per operation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- packages/meshbay-node/src/meshbay_node/ops.py | 487 ++++++++++++++++++++++++++ 1 file changed, 487 insertions(+) create mode 100644 packages/meshbay-node/src/meshbay_node/ops.py (limited to 'packages/meshbay-node/src/meshbay_node/ops.py') 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"} -- cgit v1.2.3