From e9d5e979fdab9a1cc3c729d602e6f27207b9480c Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 18 Aug 2026 02:15:02 +0200 Subject: feat(node): several named roots per group, and one implementation per operation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage A — a group's content is a set of named roots --------------------------------------------------- `shared_dir` becomes a list of {name, path, kind}. The name is the directory's basename, derived once at add time and *stored*: recomputing it would re-identify a whole library the day someone renames a folder on disk. Duplicate names are refused case-insensitively and no root may contain another — both compared with NFC folding, because most of these directories live on exFAT or NTFS where `Films` and `films` are one directory. Every index path carries its root name, in a one-root group as much as in a five-root one. One path shape has to be got right once; two have to be kept right for ever. **A root that goes away freezes; it never empties.** Unmounting a volume makes watchdog report every file under it as deleted, or presents an empty directory to the next scan. Acting on either propagates deletions for a whole library to every member, as though the owner had erased it. So a deletion is acted on only once its root is confirmed readable, and availability is tracked per root — one unplugged drive leaves the others serving. 12 tests, verified to fail against an indexer without the check. Events are not trusted to be complete either: ReadDirectoryChangesW drops them under load and inotify on a FUSE mount misses changes made outside it. A periodic reconciliation sweep is the only thing that recovers a missed event. MNP 0.2 → 0.3 (additive). The hub needs no change: SwarmSource carries a content hash, a node id and an endpoint — no paths, no filenames — and private groups register nothing (H7). Stage B — one implementation behind every front door ---------------------------------------------------- C1 and C6 were both "a second path into the node with its own weaker handshake". Two implementations of `revoke` with two authorization checks is that shape one size down. `meshbay_node/ops.py` holds each operation once, takes the daemon state, and knows nothing about HTTP, argv or MNP. The loopback API is one `_op(...)` line per endpoint; the MNP handlers call the same functions. test_ops.py asserts the shape rather than trusting it. Phase 14 is finished on top of it — `group list`, `gek init|rotate`, `reload` (SIGHUP), `denylist show|clear`, `file list|rm`. **No operator action requires a browser any more.** Plus `gek_rotate` and `member_unpin` as operator-signed MNP operations: rotation is the half of revocation that revocation cannot do, since the ex-member holds the current key, and the node generates the replacement with its own CSPRNG — no key material crosses the wire, which is what the C5b rule is actually about. Two bugs found by running it rather than by testing it ------------------------------------------------------ GroupIndex is keyed by **content hash**, so the same bytes at two paths are one entry — which is also why a scan reports ten files and indexes nine. Reconciliation compared paths, so it decided the second path was a missed event every 60 s, rewrote the entry and pushed an index update to every connected peer. Seen in a live node's log. `meshbay-node reload` crashed on first use with `subprocess` unimported: the module compiles fine, which is the "syntax, not names" trap already recorded for the SPA. test_cli_dispatch.py now walks every verb and refuses to let one be added to the parser without an entry there. Also corrected: protocol.py declared a second MNP_VERSION of "0.1" while the wire carried "0.2" — harmless only because nothing imported it. And _do_dir_create/_do_dir_delete referenced an undefined `filename` on their error path. 740 tests pass; QE/deploy/e2e.py passes end to end against the live deployment. Co-Authored-By: Claude Opus 5 --- packages/meshbay-node/tests/conftest.py | 18 ++ packages/meshbay-node/tests/test_admin_ops_mnp.py | 290 ++++++++++++++++++++ packages/meshbay-node/tests/test_cli_dispatch.py | 129 +++++++++ packages/meshbay-node/tests/test_daemon.py | 7 +- packages/meshbay-node/tests/test_indexer.py | 11 +- packages/meshbay-node/tests/test_multi_group.py | 11 +- packages/meshbay-node/tests/test_ops.py | 179 ++++++++++++ packages/meshbay-node/tests/test_quic_transport.py | 25 +- .../meshbay-node/tests/test_root_availability.py | 301 +++++++++++++++++++++ packages/meshbay-node/tests/test_roots.py | 242 +++++++++++++++++ packages/meshbay-node/tests/test_roster_pairing.py | 41 ++- .../tests/test_security_regressions.py | 32 ++- .../tests/test_stream_capacity_config.py | 5 +- .../meshbay-node/tests/test_webrtc_transport.py | 99 +++---- 14 files changed, 1289 insertions(+), 101 deletions(-) create mode 100644 packages/meshbay-node/tests/conftest.py create mode 100644 packages/meshbay-node/tests/test_admin_ops_mnp.py create mode 100644 packages/meshbay-node/tests/test_cli_dispatch.py create mode 100644 packages/meshbay-node/tests/test_ops.py create mode 100644 packages/meshbay-node/tests/test_root_availability.py create mode 100644 packages/meshbay-node/tests/test_roots.py (limited to 'packages/meshbay-node/tests') diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py new file mode 100644 index 0000000..03dfb42 --- /dev/null +++ b/packages/meshbay-node/tests/conftest.py @@ -0,0 +1,18 @@ +"""Shared fixtures and helpers for node tests.""" + +from pathlib import Path + +from meshbay_node.roots import RootSet + + +def one_root(path: Path, *, name: str = "", kind: str = "generic") -> RootSet: + """ + A RootSet with a single root over `path`, receiving uploads. + + The equivalent of the old `shared_dir`. Note what it implies for assertions: + a file directly in `path` now has `entry.path == `, not + `""` — every index path carries its root name, in a group with one root as + much as in a group with five. + """ + return RootSet.build([{"path": str(path), "name": name, "kind": kind, + "upload": True}]) diff --git a/packages/meshbay-node/tests/test_admin_ops_mnp.py b/packages/meshbay-node/tests/test_admin_ops_mnp.py new file mode 100644 index 0000000..1bf8365 --- /dev/null +++ b/packages/meshbay-node/tests/test_admin_ops_mnp.py @@ -0,0 +1,290 @@ +""" +`gek_rotate` and `member_unpin` over MNP. + +Both are destructive and both are new, so the tests are negative assertions: +nobody without the operator's pinned key can reach them, a signature over the +wrong transcript does not count, and the operation cannot be triggered by the +request message alone. + +The rule these live under is worth restating, because it is easy to read +draft-v5 §5.1 as forbidding them: **"nothing arriving over MNP can activate a +GEK" is about key material arriving from outside** (C5b — a member handing the +node a key of their choosing). An operator-signed instruction where the node +generates the key with its own CSPRNG is a different shape, and it is the only +thing that finishes a revocation: the ex-member still holds the current key. +""" + +import base64 +import time +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_GEK_ROTATE, + OP_MEMBER_UNPIN, + admin_transcript, +) +from meshbay_common.crypto import pk_to_b64 +from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR +from meshbay_common.protocol import MNP +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import open_roster +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()) + + +async def _session(tmp_path: Path, roster, *, operator: bool) -> WebRTCPeerSession: + """A peer session with an operator pinned, or deliberately without one.""" + 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"} + 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", + } + + 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, + } + session._group_id = GROUP + session._user_id = "grenet" if operator else "mallory" + session._username = session._user_id + 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.spawned = [] + session._spawn = session.spawned.append + return session + + +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): + """Await whatever `_spawn` started. The real session holds its tasks; a + hand-built one collects them here so the assertion sees the result.""" + for coro in session.spawned: + await coro + session.spawned.clear() + + +async def _sign_and_exec(session, op: str, subject: str, sk, exec_fn): + challenge = _last(session) + assert challenge["type"] == "admin_challenge", challenge + transcript = admin_transcript( + op=op, node_pk_b64=session._node_pk_b64(), group_id=GROUP, + 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)) + + +# ── gek_rotate ─────────────────────────────────────────────────────────────── + +async def test_rotation_needs_an_operator(tmp_path, roster): + """Without a paired operator there is nobody who could sign, so the node + fails closed and says why rather than issuing a challenge nobody can meet.""" + session = await _session(tmp_path, roster, operator=False) + + session._do_gek_rotate({}) + + assert _last(session)["type"] == "error" + assert "authorized key" in _last(session)["detail"] + assert not session._admin_ops + + +async def test_the_request_alone_rotates_nothing(tmp_path, roster): + """The message asks; only a signature acts. A node that rotated here would + let any member lock the group out.""" + session = await _session(tmp_path, roster, operator=True) + before = session._ctx["groups"][GROUP]["gek"] + + session._do_gek_rotate({}) + + assert _last(session)["type"] == "admin_challenge" + assert session._ctx["groups"][GROUP]["gek"] == before + + +async def test_a_members_signature_does_not_rotate(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True) + sk_mallory, pk_mallory = _keypair() + await roster.pin_identity("mallory", "mallory", pk_mallory, pk_mallory, "code") + await roster.set_member(GROUP, "mallory", ROLE_MEMBER, "active", "grenet") + before = session._ctx["groups"][GROUP]["gek"] + + session._do_gek_rotate({}) + await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, sk_mallory, + session._admin_exec_gek_rotate) + + assert _last(session)["type"] == "error" + assert session._ctx["groups"][GROUP]["gek"] == before + + +async def test_a_signature_over_another_operation_does_not_count(tmp_path, roster): + """ + H5's rule: the node rebuilds the transcript from the operation it is holding + and verifies against *that*, so a signature collected for one act cannot be + presented as another. + + Driven through `_do_admin_response`, deliberately. Handing a transcript + straight to `_admin_exec_*` would skip the reconstruction that is the + control, and the test would pass while proving nothing. + """ + session = await _session(tmp_path, roster, operator=True) + before = session._ctx["groups"][GROUP]["gek"] + + session._do_gek_rotate({}) + challenge = _last(session) + + # Signed over member_unpin, presented against the pending gek_rotate. + wrong = admin_transcript( + op=OP_MEMBER_UNPIN, node_pk_b64=session._node_pk_b64(), group_id=GROUP, + subject=GROUP, nonce=base64.b64decode(challenge["nonce"]), + ts=challenge["ts"]) + session._do_admin_response({ + "op_id": challenge["op_id"], + "signature": base64.b64encode(session.sk_op.sign(wrong)).decode(), + }) + await _drain(session) + + assert _last(session)["type"] == "error" + assert session.state["groups_ctx"][GROUP]["gek"] == before + + +async def test_the_operator_rotates_and_the_node_makes_the_key(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True) + before = session._ctx["groups"][GROUP]["gek"] + + session._do_gek_rotate({}) + await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op, + session._admin_exec_gek_rotate) + + ack = _last(session) + assert ack["type"] == MNP.GEK_ROTATE_ACK, ack + after = session.state["groups_ctx"][GROUP]["gek"] + assert after != before, "the key did not change" + assert len(after) == 32 + # Produced here, not received: no key material crossed the wire (C5b). + assert session.state["bundle_store"].stored, ( + "the node's own copy was not stored — the daemon could not reload it") + + +async def test_rotation_reaches_the_index(tmp_path, roster): + """The index is encrypted under the GEK. Leaving the old key on it would + serve members a listing they cannot open.""" + session = await _session(tmp_path, roster, operator=True) + + session._do_gek_rotate({}) + await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op, + session._admin_exec_gek_rotate) + + assert session.state["indexes"][GROUP].gek == \ + session.state["groups_ctx"][GROUP]["gek"] + + +# ── member_unpin ───────────────────────────────────────────────────────────── + +async def test_unpinning_needs_an_operator(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False) + session._do_member_unpin({"user_id": "bob"}) + assert _last(session)["type"] == "error" + + +async def test_unpinning_yourself_is_refused(tmp_path, roster): + """It would end the authority of the connection performing the operation, + halfway through it.""" + session = await _session(tmp_path, roster, operator=True) + session._do_member_unpin({"user_id": "grenet"}) + assert _last(session)["detail"] == "Cannot unpin yourself" + + +async def test_a_members_signature_does_not_unpin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True) + sk_bob, pk_bob = _keypair() + await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code") + + session._do_member_unpin({"user_id": "bob"}) + await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", sk_bob, + session._admin_exec_member_unpin) + + assert _last(session)["type"] == "error" + assert await roster.get_identity("bob") is not None, ( + "a member removed their own pin — only the operator may") + + +async def test_the_operator_unpins(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True) + _, pk_bob = _keypair() + await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code") + + session._do_member_unpin({"user_id": "bob"}) + await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", session.sk_op, + session._admin_exec_member_unpin) + + assert _last(session)["type"] == MNP.MEMBER_UNPIN_ACK + assert await roster.get_identity("bob") is None + + +async def test_unpinning_someone_unknown_says_so(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True) + session._do_member_unpin({"user_id": "nobody"}) + await _sign_and_exec(session, OP_MEMBER_UNPIN, "nobody", session.sk_op, + session._admin_exec_member_unpin) + assert _last(session)["type"] == "error" + assert "No such pinned identity" in _last(session)["detail"] diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py new file mode 100644 index 0000000..faca0ce --- /dev/null +++ b/packages/meshbay-node/tests/test_cli_dispatch.py @@ -0,0 +1,129 @@ +""" +Every CLI verb reaches its own code without falling over on a name. + +`meshbay-node reload` shipped with `subprocess` unimported and crashed with a +NameError the first time it was typed. Python compiles that file happily — +`node --check validates syntax, not names` is already in CLAUDE.md about the +SPA, and it is the same class here: nothing in the module is wrong until the +branch runs. + +So this walks every documented verb with the daemon stubbed out, and asserts +that the branch executes. It is deliberately shallow — what each command *does* +is tested in `test_ops.py` and `test_roster_pairing.py`. What this catches is a +branch nobody ever ran. +""" + +import sys +from pathlib import Path + +import pytest + +from meshbay_node import daemon as daemon_mod + +# Each verb, with the arguments that reach its branch. `--yes` where the command +# would otherwise stop for a confirmation nobody can type in a test. +VERBS = [ + ["status"], + ["ui"], + ["group", "list"], + ["group", "add"], # missing --dir: usage, then exit + ["gek", "init"], + ["gek", "rotate", "--yes"], + ["gek-init"], + ["member", "list"], + ["member", "invite", "bob"], + ["member", "revoke", "bob"], + ["member", "unpin", "bob"], + ["operator", "pair"], + ["file", "list"], + ["file", "rm", "abc", "--yes"], + ["denylist", "show"], + ["denylist", "clear", "--yes"], + ["reload"], +] + + +@pytest.fixture +def stub_daemon(monkeypatch, tmp_path): + """ + Answer every loopback call with an empty-ish payload. + + The point is to reach the branch, not to exercise the daemon: a command that + only crashes when the node is running is still a command that crashes. + """ + calls: list[tuple] = [] + + def fake_api(cfg, path, method="GET", timeout=30, body=None): + calls.append((method, path)) + return { + "groups": [], "files": [], "identities": [], "members": [], + "invites": [], "users": [], "jtis": [], "count": 0, "removed": 0, + "subject": "all", "code": "TEST-CODE", "expires_at": "", + "user_id": "u", "authorized_members": 0, "errors": [], + "name": "g", "group_id": "g", "shared_dir": str(tmp_path), + "config": str(tmp_path / "node.toml"), + } + + monkeypatch.setattr(daemon_mod, "_daemon_api", fake_api) + monkeypatch.setattr(daemon_mod, "_resolve_group", lambda cfg, g: "g" * 32) + + conf = tmp_path / "node.toml" + conf.write_text('[hub]\nurl = "https://example.invalid"\n') + monkeypatch.setattr(daemon_mod, "DEFAULT_CONFIG_PATH", conf) + + # No interactive prompt left to hang on. + monkeypatch.setattr("builtins.input", lambda *a: "n") + + # `reload` looks for a real daemon and signals it. Without this the test + # SIGHUPs whatever node happens to be running on the machine — which it did, + # once, before this was added. A test must not reach outside itself. + import os + import subprocess + + signalled: list[int] = [] + monkeypatch.setattr( + subprocess, "run", + lambda *a, **k: subprocess.CompletedProcess(a[0], 0, stdout="4242\n", + stderr="")) + monkeypatch.setattr(os, "kill", lambda pid, sig: signalled.append(pid)) + calls.append(("_signalled", signalled)) + return calls + + +@pytest.mark.parametrize("argv", VERBS, ids=lambda a: "-".join(a)) +def test_every_verb_reaches_its_branch(argv, stub_daemon, monkeypatch, capsys): + monkeypatch.setattr(sys, "argv", ["meshbay-node", *argv]) + try: + daemon_mod.main() + except SystemExit: + # A usage message and exit(1) is a branch that ran, which is what this + # asserts. A NameError or AttributeError is not. + pass + except (NameError, AttributeError) as e: # pragma: no cover + pytest.fail(f"{' '.join(argv)} crashed on a name: {e}") + + out = capsys.readouterr() + assert out.out or out.err, f"{' '.join(argv)} printed nothing at all" + + +def test_the_verb_list_here_matches_the_parser(): + """ + A verb added to the parser and not to this file would go untested, which is + exactly how `reload` shipped broken. + """ + import argparse + import inspect + + source = inspect.getsource(daemon_mod.main) + start = source.index('choices=[') + len('choices=[') + end = source.index(']', start) + declared = {c.strip().strip('"\'') for c in source[start:end].split(',') + if c.strip()} + + exercised = {argv[0] for argv in VERBS} + # `init` writes a config file and `calibrate-argon2` burns CPU for seconds; + # both are excluded on purpose rather than by omission. + untested = declared - exercised - {"init", "calibrate-argon2"} + assert not untested, ( + f"CLI verbs with no dispatch test: {sorted(untested)} — add them to " + f"VERBS above") diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py index 60ef11f..7974be5 100644 --- a/packages/meshbay-node/tests/test_daemon.py +++ b/packages/meshbay-node/tests/test_daemon.py @@ -18,6 +18,7 @@ from unittest.mock import AsyncMock, MagicMock, patch from meshbay_common.crypto import generate_gek from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from conftest import one_root from meshbay_node.daemon import NodeDaemon from meshbay_node.indexer import DirectoryIndexer @@ -222,7 +223,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu sk_node = Ed25519PrivateKey.generate() indexer = DirectoryIndexer( - root=shared_dir, group_id="a" * 32, + roots=one_root(shared_dir), group_id="a" * 32, sk_node=sk_node, gek=gek) await indexer.initial_scan() @@ -272,7 +273,7 @@ async def test_daemon_index_change_registers_swarm_for_public_group( daemon._state["endpoint_hint"] = "node123" indexer = DirectoryIndexer( - root=shared_dir, group_id="a" * 32, + roots=one_root(shared_dir), group_id="a" * 32, sk_node=Ed25519PrivateKey.generate(), gek=gek) await indexer.initial_scan() @@ -302,7 +303,7 @@ async def test_daemon_index_change_skips_other_group_peers( sk_node = Ed25519PrivateKey.generate() indexer = DirectoryIndexer( - root=shared_dir, group_id="a" * 32, + roots=one_root(shared_dir), group_id="a" * 32, sk_node=sk_node, gek=gek) await indexer.initial_scan() diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index 380d752..c304361 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -10,6 +10,7 @@ from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from conftest import one_root from meshbay_node.keystore import NodeKeys @@ -115,7 +116,7 @@ def test_group_index_diff(sk_node, gek): @pytest.mark.asyncio async def test_initial_scan(shared_dir, sk_node, gek): indexer = DirectoryIndexer( - root=shared_dir, + roots=one_root(shared_dir), group_id="scan-test", sk_node=sk_node, gek=gek, @@ -134,7 +135,7 @@ async def test_initial_scan(shared_dir, sk_node, gek): @pytest.mark.asyncio async def test_type_detection(shared_dir, sk_node, gek): - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() by_name = {e.name: e.type for e in indexer.index.entries} @@ -152,7 +153,7 @@ async def test_hidden_files_excluded(tmp_path, sk_node, gek): (d / "visible.txt").write_bytes(b"visible") (d / "file.tmp").write_bytes(b"tmp") - indexer = DirectoryIndexer(root=d, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() names = {e.name for e in indexer.index.entries} @@ -169,7 +170,7 @@ async def test_on_change_callback(shared_dir, sk_node, gek): changes.append(idx.index.count) indexer = DirectoryIndexer( - root=shared_dir, group_id="g", sk_node=sk_node, gek=gek, + roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek, on_change=on_change) await indexer.start() await asyncio.sleep(0.1) @@ -183,7 +184,7 @@ async def test_on_change_callback(shared_dir, sk_node, gek): @pytest.mark.asyncio async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek): - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() wire = indexer.index.serialize() diff --git a/packages/meshbay-node/tests/test_multi_group.py b/packages/meshbay-node/tests/test_multi_group.py index 9be8d47..2828d08 100644 --- a/packages/meshbay-node/tests/test_multi_group.py +++ b/packages/meshbay-node/tests/test_multi_group.py @@ -17,6 +17,7 @@ from cryptography.hazmat.primitives import serialization from meshbay_common.crypto import generate_gek, pk_to_b64 from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from conftest import one_root from meshbay_node.transport.quic_server import QuicChunkServer from meshbay_node.transport.quic_client import QuicChunkClient @@ -72,15 +73,15 @@ async def multi_group_server(sk_node, sk_hub, gek_a, gek_b, dir_a, dir_b, tmp_pa hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - indexer_a = DirectoryIndexer(root=dir_a, group_id="group-a", sk_node=sk_node, gek=gek_a) + indexer_a = DirectoryIndexer(roots=one_root(dir_a), group_id="group-a", sk_node=sk_node, gek=gek_a) await indexer_a.initial_scan() - indexer_b = DirectoryIndexer(root=dir_b, group_id="group-b", sk_node=sk_node, gek=gek_b) + indexer_b = DirectoryIndexer(roots=one_root(dir_b), group_id="group-b", sk_node=sk_node, gek=gek_b) await indexer_b.initial_scan() groups = { - "group-a": {"gek": gek_a, "shared_root": dir_a, "index": indexer_a.index}, - "group-b": {"gek": gek_b, "shared_root": dir_b, "index": indexer_b.index}, + "group-a": {"gek": gek_a, "roots": dir_a, "index": indexer_a.index}, + "group-b": {"gek": gek_b, "roots": dir_b, "index": indexer_b.index}, } cert_path = tmp_path / "node.crt" @@ -88,7 +89,7 @@ async def multi_group_server(sk_node, sk_hub, gek_a, gek_b, dir_a, dir_b, tmp_pa server = QuicChunkServer( sk_node=sk_node, hub_pk_pem=hub_pk_pem, - gek=gek_a, shared_root=dir_a, index=indexer_a.index, + gek=gek_a, roots=one_root(dir_a), index=indexer_a.index, host="127.0.0.1", port=19200, cert_path=cert_path, key_path=key_path, groups=groups, diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py new file mode 100644 index 0000000..f7fd259 --- /dev/null +++ b/packages/meshbay-node/tests/test_ops.py @@ -0,0 +1,179 @@ +""" +One implementation, several front doors. + +The point of `meshbay_node.ops` is not tidiness. C1 and C6 were both "a second +path into the node with its own weaker handshake", and two implementations of +`revoke` with two authorization checks is that shape one size down. So the tests +that matter here are the ones that would fail if a second implementation +appeared: the adapters must be thin, and the operations must not decide who may +call them. +""" + +import inspect +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from conftest import one_root +from meshbay_node import ops +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.quic_server import Denylist + + + +def _state(tmp_path: Path) -> dict: + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + return { + "groups_ctx": {"g" * 32: {"index": index, "roots": one_root(shared), + "gek": None}}, + "denylist": Denylist(path=tmp_path / "denylist.json"), + "indexes": {"g" * 32: index}, + } + + +# ── The adapters stay thin ─────────────────────────────────────────────────── + +def test_operations_take_state_and_nothing_web_shaped(): + """ + An operation that knew about HTTP could not be called from MNP without a + second copy. Every public operation therefore takes `state` first and + returns plain data. + """ + public = [(n, f) for n, f in vars(ops).items() + if inspect.iscoroutinefunction(f) and not n.startswith("_")] + assert public, "no operations found — did the module move?" + for name, fn in public: + params = list(inspect.signature(fn).parameters) + assert params and params[0] == "state", ( + f"{name} does not take state first — an adapter would have to " + f"assemble something for it, which is where a second implementation " + f"begins") + + +def test_the_http_adapter_adds_no_logic(): + """ + Each loopback handler should be a call into `ops` and nothing else. A + handler that grew a check of its own would be a rule the MNP path does not + have. + """ + import meshbay_node.ui.app as ui + source = inspect.getsource(ui) + # Every endpoint that performs an operation routes through _op(...). + for endpoint in ("operator_pair", "create_invite", "revoke_member", + "unpin_member", "init_gek", "attach_group", "delete_file"): + start = source.index(f"async def {endpoint}(") + body = source[start:start + 700] + assert "_op(" in body.split("\n\n")[0] + body, ( + f"{endpoint} does not go through the shared adapter") + assert "roster.set_status" not in body and "generate_gek" not in body, ( + f"{endpoint} performs the operation itself instead of calling ops") + + +def test_op_errors_carry_a_status_without_importing_http(): + source = inspect.getsource(ops) + for forbidden in ("JSONResponse", "fastapi", "starlette", "HTTPException"): + assert forbidden not in source, ( + f"ops imports {forbidden} — it must not know which adapter called it") + + +# ── Denylist (14.10) ───────────────────────────────────────────────────────── + +async def test_denylist_reports_what_it_refuses(tmp_path): + state = _state(tmp_path) + state["denylist"].deny_user("alice") + state["denylist"].deny_group("g" * 32) + + out = await ops.read_denylist(state) + + assert out["users"] == ["alice"] + assert out["groups"] == ["g" * 32] + assert out["count"] == 2 + + +async def test_clearing_one_subject_leaves_the_rest(tmp_path): + state = _state(tmp_path) + state["denylist"].deny_user("alice") + state["denylist"].deny_user("bob") + + out = await ops.clear_denylist(state, subject="alice") + + assert out["removed"] == 1 + assert (await ops.read_denylist(state))["users"] == ["bob"] + + +async def test_clearing_everything_says_how_much(tmp_path): + """The count is the point: it tells the operator whether they undid one + revocation or all of them.""" + state = _state(tmp_path) + for name in ("a", "b", "c"): + state["denylist"].deny_user(name) + + out = await ops.clear_denylist(state) + + assert out["removed"] == 3 + assert (await ops.read_denylist(state))["count"] == 0 + + +async def test_denylist_survives_a_restart(tmp_path): + """Finding H4: revocations used to live only in memory, so a restart + silently un-revoked everyone.""" + state = _state(tmp_path) + state["denylist"].deny_user("alice") + + reopened = Denylist(path=tmp_path / "denylist.json") + assert reopened.is_denied("alice", jti="", group_id="") + + +# ── File deletion (14.11) ──────────────────────────────────────────────────── + +async def test_deleting_a_file_removes_it_from_disk_and_index(tmp_path): + state = _state(tmp_path) + ctx = state["groups_ctx"]["g" * 32] + target = ctx["roots"].roots[0].path / "gone.txt" + target.write_text("x") + from meshbay_common.protocol import IndexEntry + ctx["index"].add_entry(IndexEntry(id="a" * 64, name="gone.txt", path="shared", + size=1, type="other", added_at=0)) + + out = await ops.delete_file(state, "g" * 32, "a" * 64) + + assert out["status"] == "deleted" + assert not target.exists() + assert ctx["index"].get_entry("a" * 64) is None + + +async def test_deleting_from_an_unavailable_root_is_refused(tmp_path): + """ + A frozen root's files are still listed. Deleting one would either fail + obscurely or — worse, once the drive returns — leave the index and the disk + disagreeing. + """ + state = _state(tmp_path) + ctx = state["groups_ctx"]["g" * 32] + from meshbay_common.protocol import IndexEntry + ctx["index"].add_entry(IndexEntry(id="a" * 64, name="frozen.txt", path="shared", + size=1, type="other", added_at=0)) + ctx["roots"].roots[0].available = False + + with pytest.raises(ops.OpError, match="frozen, not gone"): + await ops.delete_file(state, "g" * 32, "a" * 64) + + assert ctx["index"].get_entry("a" * 64) is not None + + +async def test_deleting_an_unknown_file_says_so(tmp_path): + state = _state(tmp_path) + with pytest.raises(ops.OpError, match="No such file"): + await ops.delete_file(state, "g" * 32, "f" * 64) + + +async def test_an_unhosted_group_offers_what_it_does_host(tmp_path): + """A bare "no such group" leaves an operator guessing at a UUID.""" + state = _state(tmp_path) + with pytest.raises(ops.OpError) as exc: + await ops.delete_file(state, "z" * 32, "a" * 64) + assert exc.value.status == 404 + assert exc.value.extra.get("available") diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py index 93ab1b0..5720043 100644 --- a/packages/meshbay-node/tests/test_quic_transport.py +++ b/packages/meshbay-node/tests/test_quic_transport.py @@ -14,6 +14,7 @@ from cryptography.hazmat.primitives import serialization from meshbay_common.crypto import generate_gek, pk_to_b64 from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from conftest import one_root from meshbay_node.transport.quic_server import QuicChunkServer, Denylist from meshbay_node.transport.quic_client import QuicChunkClient @@ -60,7 +61,7 @@ async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path): hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() cert_path = tmp_path / "node.crt" @@ -68,7 +69,7 @@ async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path): server = QuicChunkServer( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, host="127.0.0.1", port=19100, cert_path=cert_path, key_path=key_path, ) @@ -98,7 +99,7 @@ async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path): hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() cert_path = tmp_path / "node.crt" @@ -106,7 +107,7 @@ async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path): server = QuicChunkServer( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, host="127.0.0.1", port=19101, cert_path=cert_path, key_path=key_path, ) @@ -133,7 +134,7 @@ async def test_quic_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() cert_path = tmp_path / "node.crt" @@ -141,7 +142,7 @@ async def test_quic_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p server = QuicChunkServer( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, host="127.0.0.1", port=19102, cert_path=cert_path, key_path=key_path, ) @@ -167,7 +168,7 @@ async def test_quic_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() cert_path = tmp_path / "node.crt" @@ -175,7 +176,7 @@ async def test_quic_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p server = QuicChunkServer( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, host="127.0.0.1", port=19103, cert_path=cert_path, key_path=key_path, ) @@ -201,7 +202,7 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() cert_path = tmp_path / "node.crt" @@ -209,7 +210,7 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat server = QuicChunkServer( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, host="127.0.0.1", port=19104, cert_path=cert_path, key_path=key_path, ) @@ -255,7 +256,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() cert_path = tmp_path / "node.crt" @@ -264,7 +265,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p denylist = Denylist() server = QuicChunkServer( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, host="127.0.0.1", port=19105, cert_path=cert_path, key_path=key_path, denylist=denylist, diff --git a/packages/meshbay-node/tests/test_root_availability.py b/packages/meshbay-node/tests/test_root_availability.py new file mode 100644 index 0000000..028c308 --- /dev/null +++ b/packages/meshbay-node/tests/test_root_availability.py @@ -0,0 +1,301 @@ +""" +A root that goes away freezes; it never empties. + +This is the property the whole per-root availability design exists for. Unplug a +drive while the node is running and the filesystem watcher either reports every +file under it as deleted, or the next scan sees an empty directory. Acting on +either propagates deletions for a whole library, to every member, as though the +owner had erased it — and the index is what the node serves, so the loss is not +local. + +Every test here is written as "the entries are still there". They fail against +an indexer that treats a vanished root as a set of deletions, which is what the +straightforward implementation does. +""" + +import asyncio +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet + +pytestmark = pytest.mark.asyncio + + +def _roots(*paths: Path) -> RootSet: + specs = [{"path": str(p)} for p in paths] + specs[0]["upload"] = True + return RootSet.build(specs) + + +async def _indexer(roots: RootSet) -> DirectoryIndexer: + idx = DirectoryIndexer(roots=roots, group_id="g" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=None) + await idx.initial_scan() + return idx + + +def _names(idx: DirectoryIndexer) -> set[str]: + return {e.name for e in idx.index.entries} + + +# ── The freeze ─────────────────────────────────────────────────────────────── + +async def test_a_vanished_root_does_not_empty_the_index(tmp_path): + films = tmp_path / "Films" + films.mkdir() + (films / "a.mkv").write_bytes(b"a") + (films / "b.mkv").write_bytes(b"b") + + idx = await _indexer(_roots(films)) + assert _names(idx) == {"a.mkv", "b.mkv"} + + # The volume goes away. Watchdog would now report both files as deleted. + for f in films.iterdir(): + f.unlink() + films.rmdir() + + for f in ("a.mkv", "b.mkv"): + await idx._update_entry(films / f, deleted=True) + + assert _names(idx) == {"a.mkv", "b.mkv"}, ( + "an unplugged drive emptied the index — every member would see the " + "library as deleted") + assert idx.roots.roots[0].available is False + + +async def test_reconcile_does_not_delete_from_an_unavailable_root(tmp_path): + """The sweep must skip roots it cannot read: there is nothing to compare + against, and comparing anyway deletes everything.""" + films = tmp_path / "Films" + films.mkdir() + (films / "a.mkv").write_bytes(b"a") + + idx = await _indexer(_roots(films)) + (films / "a.mkv").unlink() + films.rmdir() + + await idx.reconcile() + + assert _names(idx) == {"a.mkv"} + assert idx.roots.roots[0].available is False + + +async def test_one_root_going_away_leaves_the_others_alone(tmp_path): + films = tmp_path / "Films" + music = tmp_path / "Music" + films.mkdir() + music.mkdir() + (films / "a.mkv").write_bytes(b"a") + (music / "b.mp3").write_bytes(b"b") + + idx = await _indexer(_roots(films, music)) + assert _names(idx) == {"a.mkv", "b.mp3"} + + (music / "b.mp3").unlink() + music.rmdir() + await idx.reconcile() + + assert _names(idx) == {"a.mkv", "b.mp3"} + by_name = {r.name: r.available for r in idx.roots} + assert by_name == {"Films": True, "Music": False} + + +async def test_members_are_told_which_roots_are_unavailable(tmp_path): + """Frozen entries stay listed, so without this a member cannot tell + "temporarily unavailable" from "still there".""" + films = tmp_path / "Films" + films.mkdir() + (films / "a.mkv").write_bytes(b"a") + + idx = await _indexer(_roots(films)) + assert idx.index.roots == [ + {"name": "Films", "kind": "generic", "available": True, "upload": True}] + + (films / "a.mkv").unlink() + films.rmdir() + await idx.reconcile() + + assert idx.index.roots[0]["available"] is False + + +# ── The counter-property ───────────────────────────────────────────────────── + +async def test_a_file_deleted_from_a_live_root_is_removed(tmp_path): + """ + The freeze must not become "deletions never happen". A root that is readable + and a file that is genuinely gone is an ordinary deletion. + """ + films = tmp_path / "Films" + films.mkdir() + (films / "a.mkv").write_bytes(b"a") + (films / "b.mkv").write_bytes(b"b") + + idx = await _indexer(_roots(films)) + (films / "a.mkv").unlink() + await idx._update_entry(films / "a.mkv", deleted=True) + + assert _names(idx) == {"b.mkv"} + + +async def test_reconcile_removes_what_is_genuinely_gone(tmp_path): + films = tmp_path / "Films" + films.mkdir() + (films / "a.mkv").write_bytes(b"a") + (films / "b.mkv").write_bytes(b"b") + + idx = await _indexer(_roots(films)) + (films / "a.mkv").unlink() + await idx.reconcile() + + assert _names(idx) == {"b.mkv"} + + +async def test_reconcile_picks_up_a_file_the_watcher_missed(tmp_path): + """ + `ReadDirectoryChangesW` drops events under load and inotify on a FUSE mount + misses changes made outside it. Both are the common case here, so the sweep + is the only thing that recovers. + """ + films = tmp_path / "Films" + films.mkdir() + idx = await _indexer(_roots(films)) + + (films / "late.mkv").write_bytes(b"x") # no event delivered + await idx.reconcile() + + assert _names(idx) == {"late.mkv"} + + +async def test_a_returning_root_is_rescanned(tmp_path): + films = tmp_path / "Films" + films.mkdir() + (films / "a.mkv").write_bytes(b"a") + + idx = await _indexer(_roots(films)) + (films / "a.mkv").unlink() + films.rmdir() + await idx.reconcile() + assert _names(idx) == {"a.mkv"} # frozen + + films.mkdir() + (films / "a.mkv").write_bytes(b"a") + (films / "c.mkv").write_bytes(b"c") + await idx.reconcile() + + assert _names(idx) == {"a.mkv", "c.mkv"} + assert idx.roots.roots[0].available is True + + +async def test_a_root_absent_at_startup_is_not_an_error(tmp_path): + """ + Someone starts the node with the drive unplugged. The group still exists and + the other roots still serve; this one fills in when it returns. + """ + films = tmp_path / "Films" + music = tmp_path / "Music" + films.mkdir() + music.mkdir() + (films / "a.mkv").write_bytes(b"a") + roots = _roots(films, music) + music.rmdir() + + idx = await _indexer(roots) + + assert _names(idx) == {"a.mkv"} + assert {r.name: r.available for r in idx.roots} == {"Films": True, "Music": False} + + +# ── Paths carry their root ─────────────────────────────────────────────────── + +async def test_every_path_starts_with_its_root_name(tmp_path): + films = tmp_path / "Films" + (films / "2024").mkdir(parents=True) + (films / "top.mkv").write_bytes(b"t") + (films / "2024" / "deep.mkv").write_bytes(b"d") + + idx = await _indexer(_roots(films)) + by_name = {e.name: e.path for e in idx.index.entries} + + assert by_name == {"top.mkv": "Films", "deep.mkv": "Films/2024"} + + +async def test_a_single_root_group_is_not_a_special_case(tmp_path): + """One path shape has to be got right once; two have to be kept right + forever. A lone root prefixes exactly like any other.""" + only = tmp_path / "Shared" + only.mkdir() + (only / "x.txt").write_bytes(b"x") + + idx = await _indexer(_roots(only)) + assert [e.path for e in idx.index.entries] == ["Shared"] + + +async def test_same_relative_path_in_two_roots_stays_distinct(tmp_path): + films = tmp_path / "Films" + music = tmp_path / "Music" + (films / "2024").mkdir(parents=True) + (music / "2024").mkdir(parents=True) + (films / "2024" / "same.dat").write_bytes(b"film") + (music / "2024" / "same.dat").write_bytes(b"music") + + idx = await _indexer(_roots(films, music)) + paths = sorted(e.path for e in idx.index.entries) + + assert paths == ["Films/2024", "Music/2024"] + assert len(idx.index.entries) == 2 + + +# ── Duplicate content ──────────────────────────────────────────────────────── + +async def test_identical_files_do_not_churn_the_index(tmp_path): + """ + The index is keyed by content hash, so the same bytes at two paths are one + entry. Reconciliation compares paths, so without care it decides the second + path is a missed event **every cycle** — rewriting that entry, bumping the + version, and pushing an index update to every connected peer once a minute. + + Found on a live node: `clip.mp4` sat at the root of a shared directory and + in `uploads/` with identical bytes. + """ + films = tmp_path / "Films" + (films / "uploads").mkdir(parents=True) + (films / "clip.mp4").write_bytes(b"same bytes") + (films / "uploads" / "clip.mp4").write_bytes(b"same bytes") + + idx = await _indexer(_roots(films)) + assert len(idx.index.entries) == 1, "content-addressed index, so one entry" + + await idx.reconcile() + first = (idx.index.version, idx.index.entries[0].path) + await idx.reconcile() + second = (idx.index.version, idx.index.entries[0].path) + + assert first == second, ( + "reconciliation rewrote the entry for a duplicate it cannot represent — " + "every peer would receive an index update every cycle") + + +async def test_deleting_one_copy_keeps_the_other_listed(tmp_path): + """ + The mirror case: the recorded path goes, identical content stays. Dropping + the entry would delist a file that is still on disk and still servable. + """ + films = tmp_path / "Films" + (films / "uploads").mkdir(parents=True) + (films / "clip.mp4").write_bytes(b"same bytes") + (films / "uploads" / "clip.mp4").write_bytes(b"same bytes") + + idx = await _indexer(_roots(films)) + recorded = idx.index.entries[0].path + survivor = "Films/uploads" if recorded == "Films" else "Films" + + (films / "clip.mp4").unlink() if recorded == "Films" else \ + (films / "uploads" / "clip.mp4").unlink() + await idx.reconcile() + + assert len(idx.index.entries) == 1, "the surviving copy was delisted" + assert idx.index.entries[0].path == survivor diff --git a/packages/meshbay-node/tests/test_roots.py b/packages/meshbay-node/tests/test_roots.py new file mode 100644 index 0000000..ea4ba6a --- /dev/null +++ b/packages/meshbay-node/tests/test_roots.py @@ -0,0 +1,242 @@ +""" +Several named roots per group. + +Most of these are negative assertions — a root set that would be ambiguous is +refused rather than resolved, because every ambiguity here ends as either "my +file went to the wrong disk" or "the same film is listed twice and deleting one +copy breaks the other". +""" + +from pathlib import Path + +import pytest + +from meshbay_node.roots import Root, RootError, RootSet, entry_abs_path +from meshbay_common.protocol import IndexEntry + + +def _spec(path, **kw): + return {"path": str(path), **kw} + + +def _entry(path: str, name: str) -> IndexEntry: + return IndexEntry(id="f" * 64, name=name, path=path, size=1, + type="other", added_at=0) + + +# ── Naming ─────────────────────────────────────────────────────────────────── + +def test_the_name_is_the_directory_basename(tmp_path): + (tmp_path / "Films").mkdir() + roots = RootSet.build([_spec(tmp_path / "Films")]) + assert roots.names == ["Films"] + + +def test_an_explicit_name_wins_over_the_basename(tmp_path): + (tmp_path / "Films").mkdir() + roots = RootSet.build([_spec(tmp_path / "Films", name="Cinema")]) + assert roots.names == ["Cinema"] + + +def test_two_roots_cannot_share_a_name(tmp_path): + for parent in ("a", "b"): + (tmp_path / parent / "Films").mkdir(parents=True) + with pytest.raises(RootError, match="both be called"): + RootSet.build([_spec(tmp_path / "a" / "Films"), + _spec(tmp_path / "b" / "Films")]) + + +def test_names_clash_without_regard_to_case(tmp_path): + """ + `Films` and `films` are one directory on NTFS and exFAT, which is where most + of these live. A comparison that respected case would let the pair through + and produce two roots a Windows member cannot tell apart. + """ + (tmp_path / "a" / "Films").mkdir(parents=True) + (tmp_path / "b" / "films").mkdir(parents=True) + with pytest.raises(RootError, match="both be called"): + RootSet.build([_spec(tmp_path / "a" / "Films"), + _spec(tmp_path / "b" / "films")]) + + +def test_a_name_windows_cannot_write_is_refused(tmp_path): + """ + The root name is a folder every member sees, including on Windows, where + `AUX` cannot be created at all. + """ + (tmp_path / "AUX").mkdir() + with pytest.raises(RootError, match="reserved on Windows"): + RootSet.build([_spec(tmp_path / "AUX")]) + + +# ── Nesting ────────────────────────────────────────────────────────────────── + +def test_a_root_inside_another_is_refused(tmp_path): + """ + Both roots would index the same bytes under two identities, and deleting + through one would leave the other pointing at nothing. + """ + (tmp_path / "Media" / "Films").mkdir(parents=True) + with pytest.raises(RootError, match="is inside root"): + RootSet.build([_spec(tmp_path / "Media"), + _spec(tmp_path / "Media" / "Films")]) + + +def test_nesting_is_refused_in_either_order(tmp_path): + (tmp_path / "Media" / "Films").mkdir(parents=True) + with pytest.raises(RootError, match="is inside root"): + RootSet.build([_spec(tmp_path / "Media" / "Films"), + _spec(tmp_path / "Media")]) + + +def test_the_same_directory_twice_is_refused(tmp_path): + (tmp_path / "Media").mkdir() + with pytest.raises(RootError, match="same directory"): + RootSet.build([_spec(tmp_path / "Media"), + _spec(tmp_path / "Media", name="Other")]) + + +def test_a_sibling_with_a_shared_prefix_is_fine(tmp_path): + """`/data/Media` and `/data/Media2` are unrelated — a string prefix test + would wrongly call the second nested inside the first.""" + (tmp_path / "Media").mkdir() + (tmp_path / "Media2").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media"), _spec(tmp_path / "Media2")]) + assert roots.names == ["Media", "Media2"] + + +# ── Uploads ────────────────────────────────────────────────────────────────── + +def test_a_single_root_receives_uploads_without_being_asked(tmp_path): + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.upload_root is roots.roots[0] + + +def test_several_roots_and_no_designation_means_no_uploads(tmp_path): + """ + Refused, never guessed: picking one would send a member's file to a disk the + operator did not intend, and that is discovered weeks later. + """ + (tmp_path / "A").mkdir() + (tmp_path / "B").mkdir() + roots = RootSet.build([_spec(tmp_path / "A"), _spec(tmp_path / "B")]) + assert roots.upload_root is None + + +def test_two_upload_roots_are_refused(tmp_path): + (tmp_path / "A").mkdir() + (tmp_path / "B").mkdir() + with pytest.raises(RootError, match="exactly one"): + RootSet.build([_spec(tmp_path / "A", upload=True), + _spec(tmp_path / "B", upload=True)]) + + +# ── Resolution ─────────────────────────────────────────────────────────────── + +def test_resolution_finds_a_path_inside_its_root(tmp_path): + (tmp_path / "Media" / "2024").mkdir(parents=True) + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.resolve("Media/2024") == (tmp_path / "Media" / "2024").resolve() + + +def test_the_virtual_root_resolves_to_nothing(tmp_path): + """ + It is not a directory on anyone's disk — it belongs to no volume — so a file + cannot be written there and a directory cannot be created there. + """ + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + for attempt in ("", "/", ".", " "): + assert roots.resolve(attempt) is None, f"{attempt!r} resolved" + + +@pytest.mark.parametrize("attempt", [ + "Media/../..", "Media/../../etc", "Media/sub/../../../etc", + "../Media", "..", "Unknown/x", +]) +def test_escaping_a_root_is_refused(tmp_path, attempt): + (tmp_path / "Media" / "sub").mkdir(parents=True) + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.resolve(attempt) is None, f"{attempt!r} escaped its root" + + +def test_a_symlink_out_of_the_root_is_refused(tmp_path): + """Resolved before comparing, so a link is followed and then rejected — + checking the string would have accepted it.""" + (tmp_path / "Media").mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (tmp_path / "Media" / "escape").symlink_to(outside) + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.resolve("Media/escape") is None + + +def test_resolution_is_case_insensitive_on_the_root_name(tmp_path): + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + assert roots.resolve("media") == (tmp_path / "Media").resolve() + + +def test_an_unavailable_root_resolves_to_nothing(tmp_path): + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + roots.roots[0].available = False + assert roots.resolve("Media") is None + # …but the mapping is still known, so entries can still be listed as frozen + # rather than vanishing from the index. + assert roots.split("Media")[0].name == "Media" + + +def test_an_entry_under_a_missing_root_has_no_path(tmp_path): + """ + An unplugged drive must answer "nowhere", not open a file that happens to + share a relative path with another root. + """ + (tmp_path / "Media").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media")]) + entry = _entry("Media", "film.mkv") + assert entry_abs_path(roots, entry) == (tmp_path / "Media" / "film.mkv").resolve() + roots.roots[0].available = False + assert entry_abs_path(roots, entry) is None + + +def test_virtual_path_round_trips(tmp_path): + (tmp_path / "Media" / "2024").mkdir(parents=True) + roots = RootSet.build([_spec(tmp_path / "Media")]) + real = roots.resolve("Media/2024") + assert roots.virtual_of(real) == "Media/2024" + assert roots.virtual_of(roots.resolve("Media")) == "Media" + + +# ── Availability ───────────────────────────────────────────────────────────── + +def test_availability_follows_the_directory(tmp_path): + target = tmp_path / "Media" + target.mkdir() + roots = RootSet.build([_spec(target)]) + assert roots.refresh_availability() == [] + + target.rmdir() # stands in for an unmounted volume + changed = roots.refresh_availability() + assert [(r.name, live) for r, live in changed] == [("Media", False)] + assert roots.roots[0].available is False + + target.mkdir() + changed = roots.refresh_availability() + assert [(r.name, live) for r, live in changed] == [("Media", True)] + + +def test_describe_reports_what_a_member_needs(tmp_path): + (tmp_path / "Media").mkdir() + (tmp_path / "Music").mkdir() + roots = RootSet.build([_spec(tmp_path / "Media", upload=True), + _spec(tmp_path / "Music", kind="audio")]) + described = roots.describe() + assert described == [ + {"name": "Media", "kind": "generic", "available": True, "upload": True}, + {"name": "Music", "kind": "audio", "available": True, "upload": False}, + ] + # Deliberately no paths: a member is told what exists and whether it is + # readable, not where on the operator's disk it lives. + assert not any("path" in d for d in described) diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py index 435cc76..9e45dbc 100644 --- a/packages/meshbay-node/tests/test_roster_pairing.py +++ b/packages/meshbay-node/tests/test_roster_pairing.py @@ -22,6 +22,7 @@ from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import generate_gek, pk_to_b64, unwrap_gek_aes from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript from meshbay_node.indexer.group_index import GroupIndex +from conftest import one_root from meshbay_node.roster import Roster, hash_code, normalize_code from meshbay_node.transport.webrtc_server import WebRTCPeerSession @@ -65,7 +66,7 @@ def _session(tmp_path: Path, roster, user_id: str = "grenet", session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = { - "shared_root": shared_root, + "roots": one_root(shared_root), "index": index, "sk_node": index.sk_node, "roster": roster, @@ -74,7 +75,7 @@ def _session(tmp_path: Path, roster, user_id: str = "grenet", session._ctx["groups"] = { group_id: { "gek": gek, - "shared_root": shared_root, + "roots": one_root(shared_root), "index": index, "join_policy": join_policy, }, @@ -599,7 +600,7 @@ async def test_revoke_endpoint_stops_authorization(tmp_path, roster): resp = client.post(f"/api/members/bob/revoke?group_id={GROUP}&t=tok") assert resp.status_code == 200 - assert "gek-init" in resp.json()["reminder"], ( + assert "gek rotate" in resp.json()["reminder"], ( "revocation must remind the operator to rotate the key they still hold") assert not await roster.is_authorized(GROUP, "bob") @@ -823,7 +824,7 @@ async def test_a_directory_with_anything_in_it_is_refused(tmp_path, roster): full.mkdir() (full / "keep.txt").write_text("still here") - session._do_dir_delete({"dir": "full"}) + session._do_dir_delete({"dir": "shared/full"}) assert _last(session).get("detail") == "Directory is not empty" assert full.exists() and (full / "keep.txt").exists() @@ -835,15 +836,23 @@ async def test_no_challenge_is_issued_without_an_operator(tmp_path, roster): session._ctx["has_admin_authority"] = False (tmp_path / "shared" / "empty").mkdir() - session._do_dir_delete({"dir": "empty"}) + session._do_dir_delete({"dir": "shared/empty"}) assert _last(session).get("detail") == "No authorized key for deletion" assert (tmp_path / "shared" / "empty").exists() -async def test_the_shared_root_itself_is_not_a_target(tmp_path, roster): +async def test_a_root_itself_is_not_a_target(tmp_path, roster): + """ + Neither the virtual root nor a root directory can be removed this way. + + Removing a root is a configuration change: doing it through a file operation + would leave the group config naming a directory nobody can reach. And the + virtual root is not a directory on anyone's disk at all — it belongs to no + volume. + """ session = await _dir_session(tmp_path, roster) - for attempt in ("", ".", "/", "../shared"): + for attempt in ("", ".", "/", "../shared", "shared", "shared/", "SHARED"): session._do_dir_delete({"dir": attempt}) assert _last(session).get("type") == "error", f"{attempt!r} was accepted" assert (tmp_path / "shared").is_dir() @@ -854,7 +863,11 @@ async def test_escaping_the_shared_root_is_refused(tmp_path, roster): outside = tmp_path / "outside" outside.mkdir() - for attempt in ("../outside", "../../outside", "sub/../../outside"): + # Both shapes: a path that names no root at all, and one that starts inside + # a real root and then climbs out of it. + for attempt in ("../outside", "../../outside", "sub/../../outside", + "shared/../outside", "shared/../../outside", + "shared/sub/../../outside"): session._do_dir_delete({"dir": attempt}) assert _last(session).get("type") == "error", f"{attempt!r} was accepted" assert outside.is_dir(), "a path leaving the shared root removed a directory" @@ -871,19 +884,19 @@ async def test_an_empty_directory_needs_a_signature_and_then_goes(tmp_path, rost session = await _dir_session(tmp_path, roster) (tmp_path / "shared" / "gone").mkdir() - session._do_dir_delete({"dir": "gone"}) + session._do_dir_delete({"dir": "shared/gone"}) challenge = _last(session) assert challenge["type"] == "admin_challenge" assert challenge["op"] == "dir_delete" - assert challenge["subject"] == "gone" + assert challenge["subject"] == "shared/gone" transcript = admin_transcript( op="dir_delete", node_pk_b64=session._node_pk_b64(), group_id="g1", - subject="gone", nonce=base64.b64decode(challenge["nonce"]), + subject="shared/gone", nonce=base64.b64decode(challenge["nonce"]), ts=challenge["ts"]) await session._admin_exec_dir_delete( session._admin_ops.pop(challenge["op_id"]) if session._admin_ops - else {"op": "dir_delete", "subject": "gone"}, + else {"op": "dir_delete", "subject": "shared/gone"}, transcript, sk_ed.sign(transcript)) assert _last(session)["type"] == "dir_delete_ack" @@ -906,9 +919,9 @@ async def test_someone_elses_signature_does_not_remove_it(tmp_path, roster): transcript = admin_transcript( op="dir_delete", node_pk_b64=session._node_pk_b64(), group_id="g1", - subject="theirs", nonce=b"\x22" * 32, ts=int(time.time())) + subject="shared/theirs", nonce=b"\x22" * 32, ts=int(time.time())) await session._admin_exec_dir_delete( - {"op": "dir_delete", "subject": "theirs"}, + {"op": "dir_delete", "subject": "shared/theirs"}, transcript, sk_member.sign(transcript)) assert _last(session).get("detail") == "Signature verification failed" diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index e13dec0..78a631a 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -18,6 +18,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.group_index import GroupIndex +from conftest import one_root from meshbay_node.transport.webrtc_server import WebRTCPeerSession @@ -129,12 +130,24 @@ def test_the_node_never_generates_a_name_it_would_refuse(tmp_path): f"the node picked {chosen!r} and would then reject it on the next upload") +def _uploads_dir(session) -> Path: + """ + Where this session's uploads land: uploads/ inside the group's upload root. + + Asked of the root set rather than assembled by hand, so a test cannot pass + while agreeing with a wrong answer the code also produced. + """ + root = session._ctx["roots"].upload_root + assert root is not None, "the fixture must designate an upload root" + return root.path / "uploads" + + def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: """A peer session wired to a real shared root, with sending stubbed out.""" shared_root = tmp_path / "shared" shared_root.mkdir(exist_ok=True) index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) - ctx = {"shared_root": shared_root, "index": index, "sk_node": index.sk_node} + ctx = {"roots": one_root(shared_root), "index": index, "sk_node": index.sk_node} session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = ctx @@ -161,9 +174,7 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): this test now asserts — an existing file is never replaced. """ victim = _session(tmp_path, "victim-user") - shared_root = victim._ctx["shared_root"] - - uploads = shared_root / "uploads" + uploads = _uploads_dir(victim) uploads.mkdir() original = uploads / "important.mp4" original.write_bytes(b"operator's original content") @@ -190,7 +201,7 @@ def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): session.sent.clear() session._do_file_upload(dict(payload)) - uploads = session._ctx["shared_root"] / "uploads" + uploads = _uploads_dir(session) assert (uploads / "movie.mp4").read_bytes() == b"first", ( "the first upload was replaced") assert (uploads / "movie (2).mp4").read_bytes() == b"first" @@ -221,7 +232,6 @@ def test_upload_ignores_any_directory_the_client_asks_for(tmp_path): client-chosen destination would open does not exist on this path. """ session = _session(tmp_path, "user-1") - shared_root = session._ctx["shared_root"] session._do_file_upload({ "filename": "note.txt", "dir": "../../etc", @@ -229,7 +239,7 @@ def test_upload_ignores_any_directory_the_client_asks_for(tmp_path): "data": base64.b64encode(b"x").decode(), }) - assert (shared_root / "uploads" / "note.txt").read_bytes() == b"x" + assert (_uploads_dir(session) / "note.txt").read_bytes() == b"x" assert not (tmp_path / "etc").exists() @@ -249,7 +259,7 @@ def test_two_members_can_send_the_same_filename(tmp_path): "data": base64.b64encode(b"second").decode(), }) - uploads = first._ctx["shared_root"] / "uploads" + uploads = _uploads_dir(first) assert (uploads / "IMG_1234.jpg").read_bytes() == b"first" assert (uploads / "IMG_1234 (2).jpg").read_bytes() == b"second" @@ -270,8 +280,8 @@ def test_chat_store_and_peers_are_per_group(tmp_path): index_a = GroupIndex(group_id="a" * 32, sk_node=Ed25519PrivateKey.generate()) index_b = GroupIndex(group_id="b" * 32, sk_node=Ed25519PrivateKey.generate()) groups = { - "a" * 32: {"chat_store": "STORE_A", "index": index_a, "shared_root": tmp_path}, - "b" * 32: {"chat_store": "STORE_B", "index": index_b, "shared_root": tmp_path}, + "a" * 32: {"chat_store": "STORE_A", "index": index_a, "roots": one_root(tmp_path / "a")}, + "b" * 32: {"chat_store": "STORE_B", "index": index_b, "roots": one_root(tmp_path / "b")}, } ctx = {"groups": groups} @@ -663,7 +673,7 @@ def test_admin_ui_escapes_filenames(tmp_path): html = _render_page({ "status": "running", - "groups_ctx": {"g" * 32: {"index": index, "shared_root": tmp_path}}, + "groups_ctx": {"g" * 32: {"index": index, "roots": one_root(tmp_path)}}, "indexes": {"g" * 32: index}, }) diff --git a/packages/meshbay-node/tests/test_stream_capacity_config.py b/packages/meshbay-node/tests/test_stream_capacity_config.py index 7c33a2f..cf90e22 100644 --- a/packages/meshbay-node/tests/test_stream_capacity_config.py +++ b/packages/meshbay-node/tests/test_stream_capacity_config.py @@ -25,6 +25,7 @@ from pathlib import Path import pytest from meshbay_node.config import load_config +from meshbay_node.roots import RootSet from meshbay_node.transport.webrtc_server import ( MAX_CONCURRENT_TRANSCODES, WebRTCPeerSession, @@ -95,7 +96,7 @@ class _FakePC: def _semaphore_size(n): t = WebRTCTransport( sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32, - shared_root=Path("/tmp"), index=None, max_concurrent_streams=n) + roots=RootSet(), index=None, max_concurrent_streams=n) s = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="p") return s._transcode_semaphore()._value @@ -117,7 +118,7 @@ def test_the_budget_is_shared_between_peers(): """ t = WebRTCTransport( sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32, - shared_root=Path("/tmp"), index=None, max_concurrent_streams=2) + roots=RootSet(), index=None, max_concurrent_streams=2) async def go(): a = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="a") diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index cc0c6a2..5727ef9 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -48,6 +48,7 @@ from meshbay_common.adminop import ( ) from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript from meshbay_node.bundle_store import BundleStore +from conftest import one_root from meshbay_node.roster import Roster from meshbay_node.indexer import DirectoryIndexer from meshbay_node.transport.webrtc_server import WebRTCTransport @@ -265,12 +266,12 @@ async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_us async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: browser sends MNP handshake, node responds with handshake_ack.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -309,12 +310,12 @@ async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir): async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: full file transfer — handshake, index, fetch chunk, decrypt.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -396,12 +397,12 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir async def test_webrtc_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: invalid JWT is rejected with error.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -444,12 +445,12 @@ async def test_webrtc_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir): async def test_webrtc_request_before_handshake_rejected(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: request without handshake is rejected.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -490,7 +491,7 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm from meshbay_node.chat.store import ChatStore hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() chat_store = ChatStore(db_path=tmp_path / "chat_test.db") @@ -498,7 +499,7 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["chat_store"] = chat_store @@ -537,12 +538,12 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm async def test_webrtc_chat_history_no_store(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: chat history without chat_store returns empty list.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -566,7 +567,7 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path) from meshbay_node.chat.store import ChatStore hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() chat_store = ChatStore(db_path=tmp_path / "chat_bc.db") @@ -574,7 +575,7 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path) transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["chat_store"] = chat_store @@ -604,12 +605,12 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path) async def test_webrtc_group_membership_enforced(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: JWT without matching group claim is rejected.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -651,12 +652,12 @@ async def test_webrtc_group_membership_enforced(sk_node, sk_hub, gek, shared_dir async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: peer removed from _peers dict on session close.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -678,12 +679,12 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir): async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: stream_segment for non-existent file returns error.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -708,12 +709,12 @@ async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_d async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: wrong GEK proof is rejected — hub admin can't fake membership.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -768,12 +769,12 @@ async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir) async def test_webrtc_dtls_channel_binding_detects_mitm(sk_node, sk_hub, gek, shared_dir): """WebRTC: DTLS channel binding detects fingerprint substitution (simulated MitM).""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -850,14 +851,14 @@ async def test_webrtc_admin_challenge_response(sk_node, sk_hub, gek, shared_dir, tmp_path): """WebRTC DataChannel: admin file delete requires Ed25519 challenge-response.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() sk_admin = Ed25519PrivateKey.generate() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["roster"] = await _paired_operator_roster(tmp_path, sk_admin) @@ -899,7 +900,7 @@ async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_ tmp_path): """WebRTC DataChannel: wrong Ed25519 signature is rejected — hub can't fake admin.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() sk_admin = Ed25519PrivateKey.generate() @@ -907,7 +908,7 @@ async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_ transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["roster"] = await _paired_operator_roster(tmp_path, sk_admin) @@ -946,12 +947,12 @@ async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_ async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: stream_request for non-existent file returns error.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -975,7 +976,7 @@ async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_d async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, shared_dir): """Uploader must prove Ed25519 key ownership to delete — no uploader shortcut.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() sk_uploader = Ed25519PrivateKey.generate() @@ -986,7 +987,7 @@ async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, s transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) # No admin_pk configured — only uploader_pk should authorize deletion @@ -1032,7 +1033,7 @@ async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, s async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, shared_dir): """Hub-forged JWT with same sub cannot delete — wrong Ed25519 key is rejected.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() # User A uploaded the file @@ -1047,7 +1048,7 @@ async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, share transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) # No admin_pk — only uploader_pk matters @@ -1117,7 +1118,7 @@ async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_di that lookup was H3. """ hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() roster = Roster(db_path=tmp_path / "roster.db") @@ -1125,13 +1126,13 @@ async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_di transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["roster"] = roster transport._ctx["has_admin_authority"] = True transport._ctx["groups"] = { - TEST_GROUP: {"gek": gek, "shared_root": shared_dir, "index": indexer.index}, + TEST_GROUP: {"gek": gek, "roots": shared_dir, "index": indexer.index}, } # A paired operator, as `meshbay-node operator pair` would have left it. @@ -1229,7 +1230,7 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di tmp_path, x25519_keypair): """Browser fetches GEK bundle from node during the handshake challenge window.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() sk_x_raw, pk_x_raw = x25519_keypair @@ -1244,7 +1245,7 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["bundle_store"] = bundle_store @@ -1328,7 +1329,7 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, tmp_path): """Keypair bundle stored on node, then fetched during handshake window.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() bundle_store = BundleStore(db_path=tmp_path / "bundles.db") @@ -1336,7 +1337,7 @@ async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["bundle_store"] = bundle_store @@ -1415,7 +1416,7 @@ async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path): """Keypair bundle fetch returns found=false when no bundle exists.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() bundle_store = BundleStore(db_path=tmp_path / "bundles.db") @@ -1423,7 +1424,7 @@ async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["bundle_store"] = bundle_store @@ -1492,7 +1493,7 @@ async def test_gek_not_auto_activated_on_bundle_store(sk_node, sk_hub, gek, shar through the node's local admin UI or CLI. """ hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() sk_x_raw, pk_x_raw = x25519_keypair @@ -1504,7 +1505,7 @@ async def test_gek_not_auto_activated_on_bundle_store(sk_node, sk_hub, gek, shar transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["bundle_store"] = bundle_store @@ -1547,12 +1548,12 @@ async def test_gek_not_auto_activated_on_bundle_store(sk_node, sk_hub, gek, shar async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir): """WebRTC DataChannel: connection refused when GEK is not initialized.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=None) await indexer.initial_scan() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=None, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) @@ -1594,7 +1595,7 @@ async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir): async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path): """GEK bundle fetch returns found=false when no bundle exists.""" hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() bundle_store = BundleStore(db_path=tmp_path / "bundles.db") @@ -1602,7 +1603,7 @@ async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_ transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, + roots=one_root(shared_dir), index=indexer.index, stun_servers=[], ) transport._ctx["bundle_store"] = bundle_store -- cgit v1.2.3