path: root/packages/meshbay-node/src/meshbay_node/ops/groups.py

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