""" node_status, root_add, root_remove over MNP. These test the D5 node management panel's server-side behaviour: the admin identity check on node_status, the list_groups operation, and the root add/remove flows through the MNP handlers. """ import base64 from dataclasses import asdict from pathlib import Path import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from conftest import one_root from meshbay_common.adminop import ( OP_ROOT_ADD, OP_ROOT_REMOVE, OP_GROUP_ATTACH, OP_MEMBER_UNPIN, admin_transcript, ) from meshbay_node.transport.quic_server import Denylist from meshbay_common.crypto import pk_to_b64 from meshbay_common.join import ROLE_OPERATOR from meshbay_common.protocol import MNP from meshbay_node import ops from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roster import open_roster from meshbay_node.roots import RootSet from meshbay_node.transport.webrtc_server import WebRTCPeerSession 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 async def _sign_and_exec(session, op: str, subject: str, sk, exec_fn, group_id: str = GROUP): challenge = _last(session) assert challenge["type"] == "admin_challenge", challenge transcript = admin_transcript( op=op, node_pk_b64=session._node_pk_b64(), group_id=group_id, subject=subject, nonce=base64.b64decode(challenge["nonce"]), ts=challenge["ts"]) pending = session._admin_ops.get(challenge["op_id"]) or { "op": op, "subject": subject} await exec_fn(pending, transcript, sk.sign(transcript)) # ── _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() # ── node_status ───────────────────────────────────────────────────────────── async def test_node_status_returns_groups_for_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") session._spawn(session._do_node_status({})) await _drain(session) msg = _last(session) assert msg["type"] == MNP.NODE_STATUS_ACK assert len(msg["groups"]) == 1 assert msg["groups"][0]["id"] == GROUP async def test_node_status_refused_for_non_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=False, node_user_id="grenet") session._spawn(session._do_node_status({})) await _drain(session) msg = _last(session) assert msg["type"] == "error" assert "operator" in msg["detail"].lower() async def test_node_status_refused_when_user_is_operator_but_ids_mismatch( tmp_path, roster, ): """A paired operator who is not the node owner cannot see node_status.""" session = await _session(tmp_path, roster, operator=True, node_user_id="someone-else") session._spawn(session._do_node_status({})) await _drain(session) msg = _last(session) assert msg["type"] == "error" async def test_node_status_catches_send_failure(tmp_path, roster): """If _send itself throws (e.g. msgpack encoding fails), the error must not silently vanish — it used to, because _send was outside the try block.""" session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") original_send = session._send sent = [] call_count = [0] def _exploding_send(msg): call_count[0] += 1 if msg.get("type") == "node_status_ack": raise TypeError("msgpack cannot encode this") sent.append(msg) session._send = _exploding_send session._spawn(session._do_node_status({})) await _drain(session) # The try/except around _send should catch the error and send an error reply assert any(m.get("type") == "error" for m in sent) # ── 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 NodeConfig, GroupConfig, 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') 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, RootSpec, NodeConfig 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') 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, RootSpec, NodeConfig 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') 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().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, RootSpec, NodeConfig 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') 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() 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().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, RootSpec, NodeConfig 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') 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() 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, RootSpec, NodeConfig 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') 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_add MNP handler ─────────────────────────────────────────────────── async def test_root_add_issues_challenge(tmp_path, roster): session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") session._do_root_add({"group_id": GROUP, "path": "/tmp/test"}) msg = _last(session) assert msg["type"] == "admin_challenge" async def test_root_add_refuses_without_authority(tmp_path, roster): session = await _session(tmp_path, roster, operator=False) session._do_root_add({"group_id": GROUP, "path": "/tmp/test"}) msg = _last(session) assert msg["type"] == "error" assert "authorized" in msg["detail"].lower() async def test_root_add_refuses_missing_fields(tmp_path, roster): session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") session._do_root_add({"group_id": GROUP}) assert _last(session)["type"] == "error" assert "Missing" in _last(session)["detail"] # ── 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"] # ── roster_read MNP handler ────────────────────────────────────────────── async def test_roster_read_returns_members_for_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") session._spawn(session._do_roster_read({"group_id": GROUP})) await _drain(session) msg = _last(session) assert msg["type"] == "roster_read_ack" assert "members" in msg assert "identities" in msg async def test_roster_read_refused_for_non_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=False, node_user_id="grenet") session._spawn(session._do_roster_read({"group_id": GROUP})) await _drain(session) msg = _last(session) assert msg["type"] == "error" async def test_roster_read_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") sk2, 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 sk3, pk3 = _keypair() await roster.pin_identity("real", "real", pk3, pk3, "code") await roster.set_member(GROUP, "real", "member", "active", "local-cli") session._spawn(session._do_roster_read({"group_id": GROUP})) await _drain(session) msg = _last(session) assert msg["type"] == "roster_read_ack" ghost = [m for m in msg["members"] if m["user_id"] == "ghost"] assert len(ghost) == 0, "ghost members must be filtered out" real = [m for m in msg["members"] if m["user_id"] == "real"] assert len(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") # ghost has no identity row session._do_member_unpin({"user_id": "ghost"}) challenge = _last(session) assert challenge["type"] == "admin_challenge" await _sign_and_exec(session, OP_MEMBER_UNPIN, "ghost", session.sk_op, session._admin_exec_member_unpin) msg = _last(session) assert msg["type"] == "error" assert "No such pinned identity" in msg["detail"] async def test_unpin_succeeds_for_real_identity(tmp_path, roster): """Full unpin flow: challenge → sign → exec → identity deleted.""" session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") sk2, pk2 = _keypair() await roster.pin_identity("target", "target", pk2, pk2, "code") await roster.set_member(GROUP, "target", "member", "active", "local-cli") session._do_member_unpin({"user_id": "target"}) challenge = _last(session) assert challenge["type"] == "admin_challenge" await _sign_and_exec(session, OP_MEMBER_UNPIN, "target", session.sk_op, session._admin_exec_member_unpin) msg = _last(session) assert msg["type"] == "member_unpin_ack" assert msg["user_id"] == "target" idents = await roster.list_identities() assert not any(i["user_id"] == "target" for i in idents) # ── denylist_read MNP handler ──────────────────────────────────────────── async def test_denylist_read_returns_entries_for_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") session.denylist.deny_user("bad-user") session._spawn(session._do_denylist_read({})) await _drain(session) msg = _last(session) assert msg["type"] == "denylist_read_ack" assert msg["count"] == 1 assert "bad-user" in msg["users"] async def test_denylist_read_refused_for_non_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=False, node_user_id="grenet") session._spawn(session._do_denylist_read({})) await _drain(session) msg = _last(session) assert msg["type"] == "error" # ── denylist_clear MNP handler ─────────────────────────────────────────── async def test_denylist_clear_removes_entry_for_admin(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") session._spawn(session._do_denylist_clear({"subject": "bad-user"})) await _drain(session) msg = _last(session) assert msg["type"] == "denylist_clear_ack" assert msg["removed"] == 1 assert "bad-user" not in session.denylist.entries()["users"] assert "other-user" in session.denylist.entries()["users"] async def test_denylist_clear_all_for_admin(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") session._spawn(session._do_denylist_clear({"subject": ""})) await _drain(session) msg = _last(session) assert msg["type"] == "denylist_clear_ack" assert msg["removed"] == 3 async def test_denylist_clear_refused_for_non_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=False, node_user_id="grenet") session._spawn(session._do_denylist_clear({"subject": "bad-user"})) await _drain(session) msg = _last(session) assert msg["type"] == "error" # ── group_attach MNP handler ──────────────────────────────────────────── async def test_group_attach_issues_challenge(tmp_path, roster): session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") session._do_group_attach({"name": "test-group", "shared_dir": "/tmp/share"}) msg = _last(session) assert msg["type"] == "admin_challenge" async def test_group_attach_refused_without_authority(tmp_path, roster): session = await _session(tmp_path, roster, operator=False) session._do_group_attach({"name": "test-group", "shared_dir": "/tmp/share"}) msg = _last(session) assert msg["type"] == "error" assert "authorized" in msg["detail"].lower() async def test_group_attach_refuses_missing_fields(tmp_path, roster): session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") session._do_group_attach({"name": "test-group"}) msg = _last(session) assert msg["type"] == "error" assert "Missing" in msg["detail"] # ── node_reload MNP handler ───────────────────────────────────────────── async def test_node_reload_runs_for_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=True, node_user_id="grenet") session._spawn(session._do_node_reload({})) await _drain(session) msg = _last(session) assert msg["type"] == "node_reload_ack" assert msg["status"] == "reloaded" assert len(session.reload_called) == 1 async def test_node_reload_refused_for_non_admin(tmp_path, roster): session = await _session(tmp_path, roster, operator=False, node_user_id="grenet") session._spawn(session._do_node_reload({})) await _drain(session) msg = _last(session) assert msg["type"] == "error"