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"}