diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/ops/groups.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ops/groups.py | 248 |
1 files changed, 248 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/ops/groups.py b/packages/meshbay-node/src/meshbay_node/ops/groups.py new file mode 100644 index 0000000..905aa44 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/ops/groups.py @@ -0,0 +1,248 @@ +"""The groups this node hosts and their keys.""" + +from __future__ import annotations + +import logging +from pathlib import Path + +from meshbay_common.crypto import generate_gek, wrap_gek_aes + +from meshbay_node.config import DEFAULT_CONFIG_PATH +from meshbay_node.ops.core import OpError, _config, _group_ctx, _hub +from meshbay_node.ops.node_toml import _find_group_range + +log = logging.getLogger("meshbay_node.ops") + + +# ── 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) + + if rotate and ctx.get("visibility") == "public": + raise OpError( + "Key rotation is not available for public groups", status=400) + + 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, + # With paths: this answers the loopback API, which is the + # operator's own channel. `meshbay-node root list` printed "?" for + # every directory without it — it was reading a field the member + # form of this deliberately omits. + "roots": roots.describe(with_paths=True) if roots else [], + "peers": sum(1 for p in peers.values() if p.get("group_id") == gid), + }) + roster = state.get("roster") + has_operator = False + if roster: + members = await roster.list_members() + has_operator = any(m["role"] == "operator" and m["status"] == "active" + for m in members) + from meshbay_node.config import node_settings_defaults + # No config (a test, an unconfigured node) falls back to NodeConfig()'s own + # values rather than to numbers repeated here, which is the copy this used + # to be: it was missing three settings and reported them as null. + defaults = node_settings_defaults(config.node if config else None) + if roster: + settings = await roster.node_settings(defaults) + else: + settings = defaults + return {"groups": out, "operator_paired": has_operator, "settings": settings} + + +async def attach_group(state: dict, name: str, shared_dir: str, + writable: bool = True) -> 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) + join_policy = group.get("join_policy", "invite") + block = (f'\n[[groups]]\n' + f'id = "{group["id"]}"\n' + f'name = "{group["name"]}"\n' + f'visibility = "{group.get("visibility", "private")}"\n' + f'join_policy = "{join_policy}"\n') + # No `upload_dir` here. `GroupConfig.__post_init__` still *reads* it, so an + # existing node.toml keeps working — but what it does on read is force every + # other root read-only and append that path as the one writable one, which + # is the model this refactor replaced. Writing it into a group created + # today would mean two mechanisms deciding the same thing, one of them + # invisible: `group add --dir X --writable --upload-dir Y` silently made X + # read-only. A second writable directory is `root add <path> --writable`. + block += (f'\n [[groups.roots]]\n' + # Forward slashes: a Windows path in a TOML basic string is a + # parse error (`\U`, `\a`, ... are escapes). pathlib reads `/`. + f' path = "{path.as_posix()}"\n' + f' writable = {"true" if writable else "false"}\n') + try: + with conf_path.open("a", encoding="utf-8", newline="\n") as f: + f.write(block) + except OSError as e: + raise OpError(f"Cannot write {conf_path}: {e}", status=500) from e + + result = {"group_id": group["id"], "name": group["name"], + "shared_dir": str(path), "config": str(conf_path), + "writable": writable, + "note": "restart the node to pick it up"} + return result + + +async def detach_group(state: dict, name: str) -> dict: + """ + Remove a [[groups]] block from node.toml. + + Does not touch the hub — only stops this node from hosting the group + after the next reload or restart. + """ + if not name: + raise OpError("group name or id is required") + config = _config(state) + + match = [g for g in config.groups if g.id == name or g.name == name] + if not match: + raise OpError(f"No hosted group matches {name!r}", status=404, + extra={"available": [{"name": g.name, "id": g.id} + for g in config.groups]}) + group = match[0] + + conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH) + text = conf_path.read_text(encoding="utf-8") + lines = text.split("\n") + + rng = _find_group_range(lines, group.id) + if rng is None: + raise OpError(f"Group {group.id[:8]} not found in {conf_path}") + + start, end = rng + while end < len(lines) and lines[end].strip() == "": + end += 1 + + new_lines = lines[:start] + lines[end:] + conf_path.write_text("\n".join(new_lines), encoding="utf-8", newline="\n") + log.info("Group detached: %s (%s) removed from %s", group.name, group.id[:8], conf_path) + + return {"group_id": group.id, "name": group.name, "config": str(conf_path), + "note": "restart the node to stop hosting it"} |