path: root/packages/meshbay-node/tests/test_node_status.py

"""
The node management panel's server-side behaviour.

The `_is_node_admin` hint, root removal over MNP, and the operations the Node
page and the CLI reach over loopback: listing groups, adding and updating
roots, the roster, the denylist and unpinning. Their own MNP messages
(`node_status`, `root_add`, `roster_read`, …) are gone since MNP 6.0.
"""

from dataclasses import asdict
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import pk_to_b64
from meshbay_common.join import ROLE_OPERATOR
from meshbay_node import ops
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import RootSet
from meshbay_node.roster import open_roster
from meshbay_node.transport.quic_server import Denylist
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root

GROUP = "g" * 32


@pytest.fixture
async def roster(tmp_path):
    r = await open_roster(tmp_path)
    yield r
    await r.close()


def _keypair():
    sk = Ed25519PrivateKey.generate()
    return sk, pk_to_b64(sk.public_key())


class _FakeBundleStore:
    def __init__(self):
        self.stored = []

    async def store(self, *args):
        self.stored.append(args)


class _FakeHub:
    class _S:
        user_id = "node-user"
    _session = _S()


def _last(session):
    return session.sent[-1] if session.sent else {}


async def _drain(session):
    for coro in session.spawned:
        await coro
    session.spawned.clear()


async def _session(
    tmp_path: Path, roster, *, operator: bool, node_user_id: str = "node-user",
) -> WebRTCPeerSession:
    shared = tmp_path / "shared"
    shared.mkdir(exist_ok=True)
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)

    sk_op, pk_op = _keypair()
    if operator:
        await roster.pin_identity("grenet", "grenet", pk_op, pk_op, "code")
        await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")

    group_ctx = {"gek": b"\x01" * 32, "roots": roots, "index": index,
                 "join_policy": "invite"}
    denylist = Denylist()
    reload_called = []
    async def _reload():
        reload_called.append(True)

    state = {
        "groups_ctx": {GROUP: group_ctx},
        "roster": roster,
        "indexes": {GROUP: index},
        "bundle_store": _FakeBundleStore(),
        "pk_x25519_raw": b"\x02" * 32,
        "hub": _FakeHub(),
        "node_user_id": node_user_id,
        "denylist": denylist,
        "reload_fn": _reload,
    }

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {
        "roots": roots, "index": index, "sk_node": index.sk_node,
        "roster": roster, "groups": {GROUP: group_ctx},
        "has_admin_authority": operator,
        "daemon_state": state,
        "node_user_id": node_user_id,
    }
    session._group_id = GROUP
    session._user_id = "grenet" if operator else "mallory"
    session._username = session._user_id
    # The device on the connection, as `device_hello` leaves it. Node-wide
    # controls are gated on a key the node pinned as an operator and *proved*
    # on this connection, not on the account id in the token — a hub can choose
    # the second and cannot produce the first (NS4, M3). A fixture that sets
    # only the account tests a check that is no longer the check.
    session._pinned_pk = pk_op if operator else ""
    session._device_confirmed = bool(operator)
    session._pk_user = ""
    session._uploads = {}
    session._admin_ops = {}
    session._remote_ip = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    session.state = state
    session.sk_op = sk_op
    session.denylist = denylist
    session.reload_called = reload_called
    session.spawned = []
    session._spawn = session.spawned.append
    return session


# ── _is_node_admin ──────────────────────────────────────────────────────────

async def test_is_node_admin_matches_user_id(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    assert session._is_node_admin()


async def test_is_node_admin_rejects_different_user(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="someone-else")
    assert not session._is_node_admin()


async def test_is_node_admin_rejects_missing_node_user_id(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    del session._ctx["node_user_id"]
    assert not session._is_node_admin()


# ── ops.list_groups ─────────────────────────────────────────────────────────

async def test_list_groups_returns_group_metadata(tmp_path):
    shared = tmp_path / "shared"
    shared.mkdir()
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)
    state = {
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
        "peers": {"p1": {"group_id": GROUP}, "p2": {"group_id": GROUP},
                  "p3": {"group_id": "other"}},
        "config": None,
    }
    result = await ops.list_groups(state)
    groups = result["groups"]
    assert len(groups) == 1
    g = groups[0]
    assert g["id"] == GROUP
    assert g["has_gek"] is True
    assert g["peers"] == 2
    assert isinstance(g["roots"], list)


async def test_list_groups_empty(tmp_path):
    state = {"groups_ctx": {}, "peers": {}, "config": None}
    result = await ops.list_groups(state)
    assert result["groups"] == []


# ── ops.add_root ────────────────────────────────────────────────────────────

async def test_add_root_creates_directory_and_returns_info(tmp_path):
    shared = tmp_path / "shared"
    shared.mkdir()
    new_dir = tmp_path / "new_root"

    from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(shared), name="shared", kind="generic", writable=True),
    ])

    conf = tmp_path / "node.toml"
    conf.write_text(f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
                    f'  [[groups.roots]]\n  path = "{shared}"\n', encoding="utf-8")
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }

    result = await ops.add_root(state, GROUP, str(new_dir))
    assert result["status"] == "added"
    assert new_dir.is_dir()
    assert len(result["roots"]) == 2


async def test_add_root_rejects_unknown_group(tmp_path):
    from meshbay_node.config import NodeConfig
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = []
    state = {"config": node_cfg, "groups_ctx": {}}
    with pytest.raises(ops.OpError, match="not configured"):
        await ops.add_root(state, "nonexistent", "/tmp/nope")


# ── ops.remove_root ─────────────────────────────────────────────────────────

async def test_remove_root_requires_at_least_one_remaining(tmp_path):
    shared = tmp_path / "shared"
    shared.mkdir()

    from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(shared), name="shared", kind="generic", writable=True),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
                    f'  [[groups.roots]]\n  path = "{shared}"\n', encoding="utf-8")
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }
    with pytest.raises(ops.OpError, match="only root"):
        await ops.remove_root(state, GROUP, "shared")


async def test_removing_a_writable_root_is_allowed(tmp_path):
    """
    It used to be refused: with one designated upload root, removing it left
    the group with nowhere to put an upload and no way to say so. Several roots
    can be writable now, and a group with none is a valid read-only group — so
    the refusal would be protecting a state that is no longer special.
    """
    d1 = tmp_path / "incoming"
    d2 = tmp_path / "shared"
    d1.mkdir()
    d2.mkdir()

    from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(d1), name="incoming", kind="generic", writable=True),
        RootSpec(path=str(d2), name="shared", kind="generic", writable=False),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "incoming"\n  writable = true\n\n'
        f'  [[groups.roots]]\n  path = "{d2}"\n  name = "shared"\n', encoding="utf-8")
    roots = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }
    result = await ops.remove_root(state, GROUP, "incoming")
    assert result["status"] == "removed"
    assert [r["name"] for r in result["roots"]] == ["shared"]
    assert conf.read_text(encoding="utf-8").count("[[groups.roots]]") == 1


async def test_update_root_rewrites_the_flags_in_node_toml(tmp_path):
    """
    The flags live in the operator's config file, so they survive a restart —
    and the file is hand-written and full of comments, so the change is a line
    edit rather than a round trip through a TOML writer that would discard
    every one of them.
    """
    d1 = tmp_path / "media"
    d1.mkdir()

    from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(d1), name="media", kind="generic", writable=False),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
        f'  [[groups.roots]]\n'
        f'  # the operator explained this one to themselves\n'
        f'  path = "{d1}"\n  name = "media"\n', encoding="utf-8")
    roots = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }

    result = await ops.update_root(state, GROUP, "media",
                                   writable=True, removable=True)
    assert result["status"] == "updated"
    text = conf.read_text(encoding="utf-8")
    assert "writable = true" in text
    assert "removable = true" in text
    assert "the operator explained this one to themselves" in text, (
        "the config file was rewritten instead of edited")

    # And the live root set agrees immediately, without waiting for a reload:
    # the loopback API reads it, and an operator who toggles a switch and sees
    # it snap back assumes the change did not take.
    assert roots.roots[0].writable is True
    assert roots.roots[0].removable is True

    # A second call that changes nothing must not append a duplicate line.
    await ops.update_root(state, GROUP, "media", writable=True, removable=True)
    assert conf.read_text(encoding="utf-8").count("writable =") == 1


async def test_update_root_replaces_a_legacy_upload_line(tmp_path):
    """
    A config written before the refactor says `upload = true`. Leaving it in
    place next to a new `writable` line would give the file two answers, and
    `RootSet.build` prefers `writable` — so the stale one would sit there
    contradicting the running node for as long as anyone read it.
    """
    d1 = tmp_path / "media"
    d1.mkdir()

    from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(d1), name="media", kind="generic", writable=True),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "media"\n'
        f'  upload = true\n', encoding="utf-8")
    roots = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }

    await ops.update_root(state, GROUP, "media", writable=False)
    text = conf.read_text(encoding="utf-8")
    assert "upload = true" not in text
    assert "writable = false" in text


async def test_remove_root_succeeds_with_two_roots(tmp_path):
    d1 = tmp_path / "dir1"
    d2 = tmp_path / "dir2"
    d1.mkdir()
    d2.mkdir()

    from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(d1), name="dir1", kind="generic", writable=True),
        RootSpec(path=str(d2), name="dir2", kind="generic", writable=False),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "dir1"\n\n'
        f'  [[groups.roots]]\n  path = "{d2}"\n  name = "dir2"\n', encoding="utf-8")
    roots = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }

    result = await ops.remove_root(state, GROUP, "dir2")
    assert result["status"] == "removed"
    assert len(cfg.roots) == 1
    assert cfg.roots[0].name == "dir1"


# ── root_remove MNP handler ────────────────────────────────────────────────

async def test_root_remove_issues_challenge(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._do_root_remove({"group_id": GROUP, "root_name": "shared"})
    msg = _last(session)
    assert msg["type"] == "admin_challenge"


async def test_root_remove_refuses_without_authority(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False)
    session._do_root_remove({"group_id": GROUP, "root_name": "shared"})
    msg = _last(session)
    assert msg["type"] == "error"


async def test_root_remove_refuses_missing_fields(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._do_root_remove({"group_id": GROUP})
    assert _last(session)["type"] == "error"
    assert "Missing" in _last(session)["detail"]


# ── The roster, the denylist and unpinning (loopback, CLI) ──────────────────
#
# Their MNP messages are gone (MNP 6.0, no client sent them); the operations are
# what the Node page and the CLI call.

async def test_read_roster_filters_ghost_members(tmp_path, roster):
    """Members whose identity was deleted (revoked then unpinned) are filtered
    out by read_roster — the LEFT JOIN returns them with pk_ed25519 = NULL but
    they should never reach the UI."""
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    _, pk2 = _keypair()
    await roster.pin_identity("ghost", "ghost", pk2, pk2, "code")
    await roster.set_member(GROUP, "ghost", "member", "revoked", "local-cli")
    await roster._db.execute("DELETE FROM identities WHERE user_id = 'ghost'")
    await roster._db.commit()
    # Also add a real member so the roster isn't empty
    _, pk3 = _keypair()
    await roster.pin_identity("real", "real", pk3, pk3, "code")
    await roster.set_member(GROUP, "real", "member", "active", "local-cli")

    result = await ops.read_roster(session.state, GROUP)
    assert "identities" in result
    assert not [m for m in result["members"] if m["user_id"] == "ghost"], (
        "ghost members must be filtered out")
    assert len([m for m in result["members"] if m["user_id"] == "real"]) == 1


async def test_unpin_fails_for_ghost_member(tmp_path, roster):
    """Unpinning a member with no identity gives a clear error, not a crash."""
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    await roster.set_member(GROUP, "ghost", "member", "revoked", "local-cli")
    with pytest.raises(ops.OpError, match="No such pinned identity"):
        await ops.unpin_member(session.state, "ghost")


async def test_read_denylist_lists_entries(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session.denylist.deny_user("bad-user")
    result = await ops.read_denylist(session.state)
    assert result["count"] == 1
    assert "bad-user" in result["users"]


async def test_clear_denylist_removes_one_entry(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session.denylist.deny_user("bad-user")
    session.denylist.deny_user("other-user")
    result = await ops.clear_denylist(session.state, subject="bad-user")
    assert result["removed"] == 1
    assert "bad-user" not in session.denylist.entries()["users"]
    assert "other-user" in session.denylist.entries()["users"]


async def test_clear_denylist_all(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session.denylist.deny_user("a")
    session.denylist.deny_user("b")
    session.denylist.deny_jti("j")
    result = await ops.clear_denylist(session.state, subject="")
    assert result["removed"] == 3