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 --- .../meshbay-common/src/meshbay_common/__init__.py | 2 +- .../meshbay-common/src/meshbay_common/adminop.py | 8 + .../meshbay-common/src/meshbay_common/paths.py | 150 +++++++ .../meshbay-common/src/meshbay_common/protocol.py | 12 +- packages/meshbay-common/tests/test_paths.py | 123 ++++++ packages/meshbay-hub/src/meshbay_hub/static/app.js | 27 +- .../src/meshbay_hub/static/locales/de.js | 1 + .../src/meshbay_hub/static/locales/en.js | 1 + .../src/meshbay_hub/static/locales/es.js | 1 + .../src/meshbay_hub/static/locales/fr.js | 1 + .../src/meshbay_hub/static/locales/it.js | 1 + .../src/meshbay_hub/static/locales/ja.js | 1 + .../src/meshbay_hub/static/locales/nl.js | 1 + .../src/meshbay_hub/static/locales/pl.js | 1 + .../src/meshbay_hub/static/locales/pt-BR.js | 1 + .../src/meshbay_hub/static/locales/zh-CN.js | 1 + .../meshbay-hub/src/meshbay_hub/static/style.css | 10 + packages/meshbay-node/src/meshbay_node/config.py | 83 +++- packages/meshbay-node/src/meshbay_node/daemon.py | 317 ++++++++++++-- .../src/meshbay_node/indexer/group_index.py | 10 + .../src/meshbay_node/indexer/indexer.py | 404 ++++++++++++++--- packages/meshbay-node/src/meshbay_node/ops.py | 487 +++++++++++++++++++++ packages/meshbay-node/src/meshbay_node/roots.py | 322 ++++++++++++++ .../src/meshbay_node/transport/quic_server.py | 40 +- .../src/meshbay_node/transport/webrtc_server.py | 301 ++++++++++--- packages/meshbay-node/src/meshbay_node/ui/app.py | 318 +++----------- 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 ++--- 40 files changed, 3491 insertions(+), 523 deletions(-) create mode 100644 packages/meshbay-common/src/meshbay_common/paths.py create mode 100644 packages/meshbay-common/tests/test_paths.py create mode 100644 packages/meshbay-node/src/meshbay_node/ops.py create mode 100644 packages/meshbay-node/src/meshbay_node/roots.py 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') diff --git a/packages/meshbay-common/src/meshbay_common/__init__.py b/packages/meshbay-common/src/meshbay_common/__init__.py index 60a9dc7..814c606 100644 --- a/packages/meshbay-common/src/meshbay_common/__init__.py +++ b/packages/meshbay-common/src/meshbay_common/__init__.py @@ -4,5 +4,5 @@ __version__ = "0.5.0" # 0.2: added PING/PONG, and `before`/`has_more` on chat history. Both are # additive — an 0.1 peer sends no `before` and gets the newest page, which is # what it wanted — so this is a MINOR bump, not a MAJOR one. -MNP_VERSION = "0.2" +MNP_VERSION = "0.3" MHP_VERSION = "0.1" diff --git a/packages/meshbay-common/src/meshbay_common/adminop.py b/packages/meshbay-common/src/meshbay_common/adminop.py index fe4be83..4e28f65 100644 --- a/packages/meshbay-common/src/meshbay_common/adminop.py +++ b/packages/meshbay-common/src/meshbay_common/adminop.py @@ -37,6 +37,14 @@ OP_FILE_DELETE = "file_delete" OP_DIR_DELETE = "dir_delete" OP_INVITE_CREATE = "invite_create" OP_MEMBER_REVOKE = "member_revoke" +# Rotating the group key is what actually takes it away from a revoked member: +# revocation stops the node serving the *next* key, and they still hold the +# current one. The node generates the new key itself with its own CSPRNG, so +# nothing arriving over MNP contributes key material — the C5b rule is about +# key material from outside, not about the instruction. +OP_GEK_ROTATE = "gek_rotate" +# Forgetting a pinned identity, so someone can pair again after losing a device. +OP_MEMBER_UNPIN = "member_unpin" # OP_GEK_BUNDLE_STORE is gone. Members no longer hand the node key material at # all: the node holds the GEK and wraps it itself, for a key the recipient proved # they hold (see `join.py` and docs/invite-pairing-v1.md). The operation existed diff --git a/packages/meshbay-common/src/meshbay_common/paths.py b/packages/meshbay-common/src/meshbay_common/paths.py new file mode 100644 index 0000000..45bf34e --- /dev/null +++ b/packages/meshbay-common/src/meshbay_common/paths.py @@ -0,0 +1,150 @@ +""" +Name rules that have to give the same answer everywhere. + +Most people share from an external drive formatted exFAT or NTFS, on Windows. +Those filesystems are case-insensitive and case-preserving, store no POSIX +permissions and no symlinks, and Windows refuses a set of names outright. So two +names that are plainly different on ext4 can be the same file somewhere else, or +no file at all — and an index built on one machine is read on another. + +"Are these two names the same?" therefore has one answer, and it lives here. + +What this module deliberately does **not** do: rewrite names. A name is stored as +the filesystem gave it, because that is the string that opens the file. Folding +and normalization exist for *comparison*, never for storage. +""" + +from __future__ import annotations + +import os +import re +import unicodedata +from pathlib import Path, PurePath + +# Windows refuses these as a basename, with or without an extension: `AUX.txt` +# is as impossible as `AUX`. A group indexed on Linux can hold them, and a +# Windows client then cannot write the file it just downloaded. +WINDOWS_RESERVED = frozenset({ + "CON", "PRN", "AUX", "NUL", + *(f"COM{i}" for i in range(1, 10)), + *(f"LPT{i}" for i in range(1, 10)), +}) + +# Reserved on Windows; `/` is reserved everywhere. Control characters go too. +_RESERVED_CHARS = set('<>:"/\\|?*') | {chr(c) for c in range(32)} + +# Windows without long-path support. A deep media library reaches this. +MAX_PATH_WINDOWS = 260 + + +def nfc(text: str) -> str: + """ + Canonical form for comparison. + + `Café.mkv` written on macOS (NFD: `e` + combining acute) and on Windows + (NFC: precomposed `é`) are different byte strings that name the same file to + a human, and to most filesystems. For French names this is routine. + """ + return unicodedata.normalize("NFC", text) + + +def fold(text: str) -> str: + """ + The form in which two names are "the same file". + + NFC first, then `casefold` — which is not `lower()`: it handles the cases + `lower()` misses, and those are the ones that show up as a bug report rather + than a test failure. + """ + return nfc(text).casefold() + + +def fold_path(rel: str) -> str: + """`fold` applied per segment, so separators survive.""" + return "/".join(fold(seg) for seg in rel.split("/")) + + +def portable_name_problem(name: str) -> str | None: + """ + Why `name` cannot be written on some supported platform, or None. + + Used to warn, not to refuse: a file already on the operator's disk is a fact, + and the answer is to tell whoever downloads it that it was renamed — not to + pretend it is not there. + """ + if not name: + return "empty name" + if name in (".", ".."): + return "reserved name" + bad = sorted(set(name) & _RESERVED_CHARS) + if bad: + printable = "".join(c if c.isprintable() else "?" for c in bad) + return f"contains reserved characters: {printable}" + # Windows strips these silently, so `file .` and `file` become the same + # thing after a round trip. + if name[-1] in " .": + return "ends with a space or a dot" + stem = name.split(".", 1)[0] + if stem.upper() in WINDOWS_RESERVED: + return f"reserved on Windows ({stem.upper()})" + return None + + +def is_portable_name(name: str) -> bool: + return portable_name_problem(name) is None + + +def long_path(path: Path | str) -> str: + """ + A path string safe to hand to the OS. + + On Windows, prefix with `\\\\?\\` so `MAX_PATH` does not truncate a deep + library. The prefix requires an absolute, already-normalized path, and it is + a no-op everywhere else. + """ + text = str(path) + if os.name != "nt" or text.startswith("\\\\?\\"): + return text + resolved = str(PurePath(text)) + if len(resolved) < MAX_PATH_WINDOWS and not text.startswith("\\\\"): + return text + if text.startswith("\\\\"): # UNC: \\server\share → \\?\UNC\server\share + return "\\\\?\\UNC" + text[1:] + return "\\\\?\\" + text + + +def find_fold_collisions(names: list[str]) -> dict[str, list[str]]: + """ + Names that collide once folded, keyed by the folded form. + + Two entries that fold together cannot both exist on a case-insensitive + filesystem. On ext4 they can, which is how an index becomes unrepresentable + for the person who downloads it — so this is reported to the operator rather + than resolved silently: only they know which file they meant. + """ + seen: dict[str, list[str]] = {} + for name in names: + seen.setdefault(fold(name), []).append(name) + return {k: v for k, v in seen.items() if len(v) > 1} + + +_TRAILING = re.compile(r"[ .]+$") + + +def sanitize_for_download(name: str, *, replacement: str = "_") -> str: + """ + A name that can be written on the running platform, from one that may not be. + + For the client saving a file, never for the node storing one. Returns the + name unchanged when it is already portable, so the common case is identity + and the caller can tell whether it renamed anything by comparing. + """ + if is_portable_name(name): + return name + out = "".join(replacement if c in _RESERVED_CHARS else c for c in name) + out = _TRAILING.sub("", out) + stem, dot, ext = out.partition(".") + if stem.upper() in WINDOWS_RESERVED: + stem = f"{stem}{replacement}" + out = f"{stem}{dot}{ext}" + return out or "unnamed" diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 1aadafe..a1c971f 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -11,8 +11,10 @@ Every message carries a "v" field for protocol version. from dataclasses import dataclass, field from typing import Any -MNP_VERSION = "0.1" -MHP_VERSION = "0.1" +# The wire versions live in meshbay_common/__init__.py — one source, because a +# second copy here said "0.1" while every message on the wire carried "0.2". +# Nothing imported it, which is the only reason it was harmless. +from meshbay_common import MNP_VERSION, MHP_VERSION # noqa: F401 (re-export) # ── MNP message types ───────────────────────────────────────────────────────── @@ -74,6 +76,12 @@ class MNP: INVITE_CREATE = "invite_create" # operator → node: issue a pairing code MEMBER_REVOKE = "member_revoke" # operator → node: stop serving the key MEMBER_REVOKE_ACK = "member_revoke_ack" + MEMBER_UNPIN = "member_unpin" # operator → node: forget an identity + MEMBER_UNPIN_ACK = "member_unpin_ack" + # Rotation is the half of revocation that revocation cannot do: the node + # generates a fresh key itself, so no key material crosses the wire. + GEK_ROTATE = "gek_rotate" # operator → node: new group key + GEK_ROTATE_ACK = "gek_rotate_ack" INVITE_RESULT = "invite_result" # node → operator: the code, once diff --git a/packages/meshbay-common/tests/test_paths.py b/packages/meshbay-common/tests/test_paths.py new file mode 100644 index 0000000..b9052e3 --- /dev/null +++ b/packages/meshbay-common/tests/test_paths.py @@ -0,0 +1,123 @@ +""" +Name rules that have to give the same answer on every platform. + +Most people share from an external drive formatted exFAT or NTFS, on Windows, so +these are not compatibility niceties — they decide whether two index entries are +one file, and whether a member can save what they downloaded. +""" + +import pytest + +from meshbay_common.paths import ( + find_fold_collisions, + fold, + fold_path, + is_portable_name, + nfc, + portable_name_problem, + sanitize_for_download, +) + + +# ── Same file or not ───────────────────────────────────────────────────────── + +def test_case_differences_fold_together(): + assert fold("README.TXT") == fold("readme.txt") + + +def test_folding_is_not_lowercasing(): + """`casefold` handles what `lower()` misses, and those are the pairs that + arrive as a bug report rather than a test failure.""" + assert fold("STRASSE") == fold("straße") + assert "STRASSE".lower() != "straße".lower() + + +def test_the_two_unicode_spellings_of_an_accent_are_one_name(): + """ + `Café.mkv` written on macOS (NFD) and on Windows (NFC) are different byte + strings naming the same file. For French filenames this is routine. + """ + nfc_form = "Café.mkv" # é precomposed + nfd_form = "Café.mkv" # e + combining acute + assert nfc_form != nfd_form + assert nfc(nfd_form) == nfc_form + assert fold(nfd_form) == fold(nfc_form) + + +def test_distinct_names_stay_distinct(): + assert fold("film.mkv") != fold("film2.mkv") + + +def test_folding_a_path_keeps_its_separators(): + assert fold_path("Films/2024/A.MKV") == "films/2024/a.mkv" + + +def test_collisions_are_found_and_grouped(): + found = find_fold_collisions( + ["README.txt", "readme.TXT", "notes.md", "Café.mkv", "Café.mkv"]) + groups = {frozenset(v) for v in found.values()} + assert len(groups) == 2 + assert frozenset({"README.txt", "readme.TXT"}) in groups + # The other pair is the two Unicode spellings of the same accented name, + # which are different byte strings — so compare by size, not by literal. + assert any(len(g) == 2 and all(f.endswith('.mkv') for f in g) + for g in groups) + assert all("notes.md" not in v for v in found.values()) + + +def test_no_collision_means_no_report(): + assert find_fold_collisions(["a.txt", "b.txt"]) == {} + + +# ── What Windows refuses ───────────────────────────────────────────────────── + +@pytest.mark.parametrize("name", ["CON", "PRN", "AUX", "NUL", "COM1", "LPT9", + "aux", "Aux.txt", "COM3.tar.gz"]) +def test_windows_reserved_names_are_flagged(name): + """`AUX.txt` is as impossible as `AUX` — the extension does not help.""" + assert portable_name_problem(name) is not None + + +@pytest.mark.parametrize("name", ['ab', 'a:b', 'a"b', "a/b", "a\\b", + "a|b", "a?b", "a*b", "a\x00b", "a\tb"]) +def test_reserved_characters_are_flagged(name): + assert portable_name_problem(name) is not None + + +@pytest.mark.parametrize("name", ["trailing ", "trailing.", " "]) +def test_a_trailing_space_or_dot_is_flagged(name): + """Windows strips them silently, so `file .` and `file` come back the same.""" + assert portable_name_problem(name) is not None + + +@pytest.mark.parametrize("name", ["film.mkv", "Café.mkv", "rapport (1).pdf", + "été.txt", "COMET.txt", "AUXILIARY.doc", + "日本語.txt"]) +def test_ordinary_names_are_portable(name): + assert is_portable_name(name), portable_name_problem(name) + + +def test_empty_and_dot_names_are_refused(): + assert portable_name_problem("") is not None + assert portable_name_problem(".") is not None + assert portable_name_problem("..") is not None + + +# ── Saving a file whose name the platform refuses ──────────────────────────── + +def test_a_portable_name_is_returned_unchanged(): + """Identity in the common case, so a caller can tell whether it renamed + anything by comparing.""" + assert sanitize_for_download("film.mkv") == "film.mkv" + assert sanitize_for_download("Café (2024).mkv") == "Café (2024).mkv" + + +def test_sanitizing_produces_something_writable(): + for hostile in ["ac.txt", "AUX", "trailing .", "with/slash.txt", "COM1.log"]: + cleaned = sanitize_for_download(hostile) + assert is_portable_name(cleaned), f"{hostile!r} → {cleaned!r} still refused" + + +def test_sanitizing_never_returns_nothing(): + assert sanitize_for_download("...") not in ("", None) + assert sanitize_for_download("???") not in ("", None) diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js index cb7a462..4a4e712 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js @@ -1240,6 +1240,10 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth, // Directories are not index entries, so a new empty one needs a nudge // to appear in the breadcrumb listing. const [nodeDirs, setNodeDirs] = useState([]); + // The group's roots and whether each is readable. A root whose drive is + // unplugged keeps its files listed — they are frozen, not deleted — so this is + // the only thing that lets the UI say which of the two it is. + const [nodeRoots, setNodeRoots] = useState([]); const [isNodeAdmin, setIsNodeAdmin] = useState(false); // Paired ≠ operator account. `is_node_admin` says the hub account owning this @@ -1275,6 +1279,7 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth, const fresh = indexMsg.entries || []; setEntries(fresh); if (indexMsg.dirs) setNodeDirs(indexMsg.dirs); + if (indexMsg.roots) setNodeRoots(indexMsg.roots); cacheGroupIndex(groupId, group ? group.name : groupId, fresh); }, [groupId, group]); @@ -1492,6 +1497,7 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth, const indexMsg = await transport.fetchIndex(); if (indexMsg.entries) setEntries(indexMsg.entries); if (indexMsg.dirs) setNodeDirs(indexMsg.dirs); + if (indexMsg.roots) setNodeRoots(indexMsg.roots); } catch (err) { setError(err.message); } @@ -1671,6 +1677,19 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth, } const subdirs = [...dirs].sort(); + // At the top of a group the folders on screen ARE the roots, so their state + // belongs there. Deeper in, everything shown lives inside one readable root + // and there is nothing to flag. + const rootState = new Map(nodeRoots.map(r => [r.name, r])); + const unavailableHere = currentPath + ? [] + : subdirs.filter(d => rootState.get(d) && rootState.get(d).available === false); + // A member cannot create a folder at the top of a group: that level is the + // set of roots, which is the operator's configuration and not a directory on + // anyone's disk. The node refuses it, so offering it would only produce an + // error nobody can act on. + const canCreateDir = Boolean(currentPath); + const baseLabel = { idle: t('status.idle'), discovering: t('status.discovering'), @@ -1875,9 +1894,11 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth, + ${canCreateDir && html` + `}
@@ -1944,8 +1965,10 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth, onChange=${() => toggle(dirKey(d))} /> `} - \u{1F4C1} - ${d}/ + ${unavailableHere.includes(d) ? '\u{26A0}' : '\u{1F4C1}'} + ${d}/${unavailableHere.includes(d) ? html` + ${t('group.root_unavailable')} + ` : ''} ${inside.length ? formatSize(bytes) : ''} diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js index 15401bf..aaed0d1 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js @@ -92,6 +92,7 @@ export default { 'group.upload': 'Hochladen', 'group.mkdir': 'Neuer Ordner', 'group.mkdir_prompt': 'Name des neuen Ordners:', + 'group.root_unavailable': '(nicht verfügbar — Laufwerk getrennt)', 'group.view': 'Ansehen', 'group.delete': 'Löschen', 'group.delete_confirm': '{name} löschen?', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js index a98eb1a..02532d6 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js @@ -92,6 +92,7 @@ export default { 'group.upload': 'Upload', 'group.mkdir': 'New folder', 'group.mkdir_prompt': 'Name of the new folder:', + 'group.root_unavailable': '(unavailable — the drive is disconnected)', 'group.view': 'View', 'group.delete': 'Delete', 'group.delete_confirm': 'Delete {name}?', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js index 596d952..49bfc1c 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js @@ -90,6 +90,7 @@ export default { 'group.upload': 'Subir', 'group.mkdir': 'Nueva carpeta', 'group.mkdir_prompt': 'Nombre de la nueva carpeta:', + 'group.root_unavailable': '(no disponible — la unidad está desconectada)', 'group.view': 'Ver', 'group.delete': 'Eliminar', 'group.delete_confirm': '¿Eliminar {name}?', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js index 61dc99e..1afc5b1 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js @@ -91,6 +91,7 @@ export default { 'group.upload': 'Envoyer', 'group.mkdir': 'Nouveau dossier', 'group.mkdir_prompt': 'Nom du nouveau dossier :', + 'group.root_unavailable': '(indisponible — le disque est déconnecté)', 'group.view': 'Afficher', 'group.delete': 'Supprimer', 'group.delete_confirm': 'Supprimer {name} ?', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js index 6693115..44b4353 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js @@ -91,6 +91,7 @@ export default { 'group.upload': 'Carica', 'group.mkdir': 'Nuova cartella', 'group.mkdir_prompt': 'Nome della nuova cartella:', + 'group.root_unavailable': '(non disponibile — l’unità è scollegata)', 'group.view': 'Visualizza', 'group.delete': 'Elimina', 'group.delete_confirm': 'Eliminare {name}?', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js index 8e18a19..33fdee3 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js @@ -89,6 +89,7 @@ export default { 'group.upload': 'アップロード', 'group.mkdir': '新しいフォルダー', 'group.mkdir_prompt': '新しいフォルダーの名前：', + 'group.root_unavailable': '(利用不可 — ドライブが切断されています)', 'group.view': '表示', 'group.delete': '削除', 'group.delete_confirm': '{name} を削除しますか？', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js index 6fd9220..aa0385e 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js @@ -92,6 +92,7 @@ export default { 'group.upload': 'Uploaden', 'group.mkdir': 'Nieuwe map', 'group.mkdir_prompt': 'Naam van de nieuwe map:', + 'group.root_unavailable': '(niet beschikbaar — de schijf is losgekoppeld)', 'group.view': 'Bekijken', 'group.delete': 'Verwijderen', 'group.delete_confirm': '{name} verwijderen?', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js index 81f0139..e66dd12 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js @@ -95,6 +95,7 @@ export default { 'group.upload': 'Wyślij', 'group.mkdir': 'Nowy folder', 'group.mkdir_prompt': 'Nazwa nowego folderu:', + 'group.root_unavailable': '(niedostępne — dysk jest odłączony)', 'group.view': 'Podgląd', 'group.delete': 'Usuń', 'group.delete_confirm': 'Usunąć {name}?', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js index a2f41af..03de040 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js @@ -92,6 +92,7 @@ export default { 'group.upload': 'Enviar', 'group.mkdir': 'Nova pasta', 'group.mkdir_prompt': 'Nome da nova pasta:', + 'group.root_unavailable': '(indisponível — a unidade está desconectada)', 'group.view': 'Visualizar', 'group.delete': 'Excluir', 'group.delete_confirm': 'Excluir {name}?', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js index 86c7580..72995dd 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js @@ -88,6 +88,7 @@ export default { 'group.upload': '上传', 'group.mkdir': '新建文件夹', 'group.mkdir_prompt': '新文件夹的名称：', + 'group.root_unavailable': '(不可用 — 驱动器已断开连接)', 'group.view': '查看', 'group.delete': '删除', 'group.delete_confirm': '删除 {name}？', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/style.css b/packages/meshbay-hub/src/meshbay_hub/static/style.css index 1ba0ef0..6b0b6fe 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/style.css +++ b/packages/meshbay-hub/src/meshbay_hub/static/style.css @@ -1894,3 +1894,13 @@ button:disabled { opacity: 0.5; cursor: not-allowed; } } .btn-primary:hover:not(:disabled) { background: var(--accent-hover); } .btn-primary:disabled { opacity: 0.5; cursor: not-allowed; } + +/* A root whose volume is absent. Its files stay listed — they are frozen, not + deleted — so the row has to carry the difference. */ +.root-offline { + font-size: 0.85em; + font-weight: 500; + color: var(--warn, #d97706); + margin-left: 6px; +} +.dir-row .root-offline { white-space: nowrap; } diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index 3de2473..42ebcfd 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -47,13 +47,29 @@ max_concurrent_streams = 8 # Browser and native clients reach this node over WebRTC DataChannel via hub # signaling — no inbound port to open. QUIC is the optional direct path. -# Multiple groups — each with its own directory +# Multiple groups — each with one or more named directories ("roots"). +# +# A root's name is the directory's basename, and it becomes the first segment of +# every path members see: /home/user/Media appears to everyone as "Media/". +# Two roots cannot share a name (compared without regard to case), and no root +# may sit inside another. Exactly one root receives uploads. [[groups]] id = "" # set after joining name = "My Media" -shared_dir = "/home/user/Media" quic_port = 19010 + [[groups.roots]] + path = "/home/user/Media" + upload = true + + [[groups.roots]] + path = "/run/media/user/USB/Musique" # an external drive is fine: if it is + kind = "audio" # unplugged the root goes unavailable + # and its files stay in the index, + # rather than looking deleted + +# The single-directory form still works and means the same thing — one root, +# named after the directory, receiving uploads. [[groups]] id = "" name = "Public Archive" @@ -96,11 +112,25 @@ class NodeConfig: max_concurrent_streams: int = 8 +@dataclass +class RootSpec: + """One named directory inside a group. See `meshbay_node/roots.py`.""" + path: str = "" + name: str = "" # empty → the directory's basename, derived at load + kind: str = "generic" # generic|video|audio|photo — a view hint, unused for now + upload: bool = False # exactly one root per group receives uploads + + @dataclass class GroupConfig: id: str = "" name: str = "" - shared_dir: str = "" + # A group's content is several named roots. `shared_dir` is the single-root + # form and is still read: it becomes one root named after its basename, which + # is why every path gained a segment. See roots.py for why there is no + # unprefixed shape. + roots: list[RootSpec] = field(default_factory=list) + shared_dir: str = "" # legacy single-root form, migrated at load visibility: str = "private" # public|private — discoverability, not admission # Admission. "invite" (default) means a newcomer needs a one-time pairing code # before the node wraps the group key for them; "open" means the node pins @@ -112,6 +142,19 @@ class GroupConfig: join_policy: str = "invite" # invite|open quic_port: int = 19010 # QUIC MNP port + def __post_init__(self) -> None: + """ + The single-directory form becomes one root, whoever built this. + + On the dataclass rather than in the TOML reader, because a GroupConfig is + also built by the CLI, by `group attach` and by tests. Putting the + migration in the parser alone left every one of those paths with a group + that had no directory at all — and it presented as "skipping group", + which reads like configuration rather than a bug. + """ + if not self.roots and self.shared_dir.strip(): + self.roots = [RootSpec(path=self.shared_dir.strip(), upload=True)] + @dataclass class KeystoreConfig: @@ -157,6 +200,35 @@ def _positive(value: object, default: int, name: str) -> int: return n +def _read_roots(group: dict) -> list[RootSpec]: + """ + A group's roots, from `[[groups.roots]]` or from the legacy `shared_dir`. + + Both forms are accepted and `shared_dir` is not deprecated for a single + directory — it is the same thing said shorter. Naming both is refused rather + than merged: which one receives uploads would be a guess, and a wrong guess + is discovered weeks later. + """ + specs = [ + RootSpec( + path=str(r.get("path", "")), + name=str(r.get("name", "")), + kind=str(r.get("kind", "generic")), + upload=bool(r.get("upload", False)), + ) + for r in group.get("roots", []) or [] + ] + legacy = str(group.get("shared_dir", "") or "").strip() + if specs and legacy: + log.warning( + "group %r declares both shared_dir and [[groups.roots]] — using " + "roots and ignoring shared_dir = %s", + group.get("name", ""), legacy) + # A bare shared_dir needs no handling here: GroupConfig.__post_init__ turns + # it into one root for every construction path, not just this one. + return specs + + def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: """ Load config from TOML file. Supports both single [group] and @@ -190,7 +262,10 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.groups.append(GroupConfig( id=g.get("id", ""), name=g.get("name", ""), - shared_dir=g.get("shared_dir", ""), + roots=_read_roots(g), + # Ignored when roots are given explicitly (warned about in + # _read_roots); otherwise __post_init__ migrates it. + shared_dir="" if _read_roots(g) else g.get("shared_dir", ""), visibility=g.get("visibility", "private"), join_policy=g.get("join_policy", "invite"), quic_port=g.get("quic_port", cfg.node.quic_port), diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 8b4a1d7..21331f2 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -26,6 +26,7 @@ Usage: """ import asyncio +from dataclasses import asdict import base64 import json import logging @@ -42,6 +43,7 @@ from meshbay_node.audit import AuditStore from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config +from meshbay_node.roots import RootSet, RootError from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer from meshbay_node.keystore import load_or_create_keystore @@ -197,17 +199,30 @@ class NodeDaemon: # 4. Build per-group contexts groups_ctx: dict[str, dict] = {} for group_cfg in self._config.groups: - if not group_cfg.id or not group_cfg.shared_dir: - log.warning("Group %r missing id or shared_dir — skipping", - group_cfg.name) + if not group_cfg.id or not group_cfg.roots: + log.warning("Group %r has no id or no shared directory — " + "skipping", group_cfg.name) continue - shared_root = Path(group_cfg.shared_dir).expanduser().resolve() - if not shared_root.exists(): - log.warning("Shared dir not found: %s — skipping group %s", - shared_root, group_cfg.name) + try: + roots = RootSet.build([asdict(r) for r in group_cfg.roots]) + except RootError as e: + # Configuration the operator has to fix; guessing would put + # a member's file on the wrong disk or index one twice. + log.error("Group %r: %s — skipping", group_cfg.name, e) continue + roots.refresh_availability() + if not any(r.available for r in roots): + # Not skipped for being empty: a group whose only drive is + # unplugged still exists, and its index is frozen rather + # than lost. But there is nothing to serve until it returns. + log.warning( + "Group %r: none of its %d root(s) are readable right now " + "(%s) — serving nothing until one returns", + group_cfg.name, len(roots), + ", ".join(str(r.path) for r in roots)) + gek = None if group_cfg.visibility == "private": gek = await self._load_gek( @@ -219,7 +234,7 @@ class NodeDaemon: group_cfg.name) indexer = DirectoryIndexer( - root=shared_root, + roots=roots, group_id=group_cfg.id, sk_node=keys.sk_ed25519, gek=gek, @@ -229,11 +244,13 @@ class NodeDaemon: self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index log.info("Indexing group %s: %s (%d files)", - group_cfg.name, shared_root, indexer.index.count) + group_cfg.name, + ", ".join(f"{r.name}={r.path}" for r in roots), + indexer.index.count) groups_ctx[group_cfg.id] = { "gek": gek, - "shared_root": shared_root, + "roots": roots, "index": indexer.index, "visibility": group_cfg.visibility, # Admission policy comes from node.toml, never from the hub: @@ -270,7 +287,7 @@ class NodeDaemon: sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, gek=first["gek"], - shared_root=first["shared_root"], + roots=first["roots"], index=first["index"], groups=groups_ctx, denylist=denylist, @@ -290,6 +307,11 @@ class NodeDaemon: self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64 self._webrtc._ctx["roster"] = self._roster + # The MNP adapter calls the same operations as the loopback API + # (meshbay_node.ops), and those take the daemon's state. Handing + # the transport a second set of lookups is how two paths to one + # operation start disagreeing — the shape of C1 and C6. + self._webrtc._ctx["daemon_state"] = self._state self._webrtc._ctx["invite_ttl"] = ( self._config.node.invite_ttl_hours * 3600) paired = await self._roster.has_operator() if self._roster else False @@ -310,7 +332,7 @@ class NodeDaemon: sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, gek=first["gek"], - shared_root=first["shared_root"], + roots=first["roots"], index=first["index"], host="::", port=self._config.node.quic_port, @@ -383,7 +405,13 @@ class NodeDaemon: self._state["roster"] = self._roster self._state["node_user_id"] = session.user_id self._state["webrtc"] = self._webrtc + self._state["quic_server"] = self._quic_server self._state["hub"] = hub + # Rotating a key has to reach every transport holding a copy of it, + # and clearing the denylist has to reach the one the handshake + # consults — so both are published rather than reachable only + # through the object that happens to own them. + self._state["denylist"] = self._denylist self._state["pk_x25519_raw"] = pk_x_raw self._state["status"] = "running" @@ -410,10 +438,78 @@ class NodeDaemon: loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, stop_event.set) + # Milestone 14.8: re-read node.toml without dropping connections. + try: + loop.add_signal_handler( + signal.SIGHUP, + lambda: asyncio.ensure_future(self._reload_config())) + except (NotImplementedError, AttributeError): + pass # no SIGHUP on Windows; `reload` says so there await stop_event.wait() await self._shutdown() + async def _reload_config(self) -> None: + """ + Re-read node.toml on SIGHUP. + + Deliberately narrow: it picks up **root changes on groups already + hosted**, which is what an operator adjusts day to day, and reports + anything else as needing a restart. Adding or removing a whole group + means new indexers, chat stores, GEK loads and transport contexts, and + doing that under a live daemon is how a half-built group ends up serving + content. Saying "restart for that" is honest and costs one restart. + + Nothing here touches connections: a member watching a film keeps + watching it. + """ + log.info("SIGHUP — re-reading %s", self._config_path) + try: + fresh = load_config(self._config_path) + except Exception as e: + log.error("Reload failed, keeping the running config: %s", e) + return + + groups_ctx = self._state.get("groups_ctx") or {} + hosted = set(groups_ctx) + incoming = {g.id for g in fresh.groups if g.id} + if incoming != hosted: + added = ", ".join(sorted(incoming - hosted)) or "none" + removed = ", ".join(sorted(hosted - incoming)) or "none" + log.warning("Group set changed (added: %s, removed: %s) — restart the " + "daemon for that; roots of existing groups reloaded anyway", + added, removed) + + changed = 0 + for group_cfg in fresh.groups: + ctx = groups_ctx.get(group_cfg.id) + if not ctx: + continue + try: + roots = RootSet.build([asdict(r) for r in group_cfg.roots]) + except RootError as e: + log.error("Group %r: %s — keeping the roots already loaded", + group_cfg.name, e) + continue + before = {(r.name, str(r.path)) for r in ctx["roots"]} + after = {(r.name, str(r.path)) for r in roots} + if before == after: + continue + roots.refresh_availability() + indexer = next((i for i in self._indexers + if i.group_id == group_cfg.id), None) + if indexer is None: + continue + log.info("Group %r roots changed: %s", group_cfg.name, + ", ".join(f"{r.name}={r.path}" for r in roots)) + await indexer.retarget(roots) + ctx["roots"] = roots + changed += 1 + + self._config = fresh + self._state["config"] = fresh + log.info("Reload complete — %d group(s) re-rooted", changed) + async def _login_with_retry(self, hub: HubClient): """Login to hub, retrying if the node key hasn't been linked yet.""" import httpx as _httpx @@ -675,21 +771,27 @@ def main() -> None: parser = argparse.ArgumentParser(description="MeshBay Node daemon") parser.add_argument("command", nargs="?", - choices=["init", "status", "ui", "gek-init", "operator", - "member", "group", "calibrate-argon2"], + choices=["init", "status", "ui", "gek-init", "gek", + "operator", "member", "group", "file", + "denylist", "reload", "calibrate-argon2"], help="init: write example config | status: node state and keys " "| ui: print the admin UI URL | operator pair: pair a " "browser with this node | member list|invite|revoke|unpin " - "| group add --dir : host another of your " - "groups | calibrate-argon2: benchmark") + "| group list|add | gek init|rotate | file list|rm " + "| denylist show|clear | reload: re-read node.toml " + "| calibrate-argon2: benchmark") parser.add_argument("subcommand", nargs="?", help="'pair' for operator; list|invite|revoke|unpin for " - "member; 'add' for group") + "member; list|add for group; init|rotate for gek; " + "list|rm for file; show|clear for denylist") parser.add_argument("target", nargs="?", - help="username for member invite|revoke|unpin; " - "group name for group add") + help="username for member invite|revoke|unpin; group name " + "for group add; file id for file rm; identifier for " + "denylist clear") parser.add_argument("--dir", default=None, help="shared directory, for group add") + parser.add_argument("--yes", action="store_true", + help="skip the confirmation for destructive commands") parser.add_argument("--config", type=Path, default=None, help="Config file path") parser.add_argument("--group", default=None, @@ -699,8 +801,8 @@ def main() -> None: args = parser.parse_args() # Query commands print a report; library logging would interleave with it. - quiet = args.command in ("status", "ui", "gek-init", "operator", "member", - "group") + quiet = args.command in ("status", "ui", "gek-init", "gek", "operator", + "member", "group", "file", "denylist", "reload") logging.basicConfig( level=logging.ERROR if quiet else getattr(logging, args.log_level), format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", @@ -755,11 +857,17 @@ def main() -> None: print(f"config {DEFAULT_CONFIG_PATH}") if not cfg.groups: print("groups none configured — create a group on the hub, then add") - print(" a [[groups]] entry with its id and shared_dir") + print(" a [[groups]] entry with its id and a directory") else: for g in cfg.groups: print(f" group {g.name} [{g.visibility}] {g.id or ''}") - print(f" {g.shared_dir or ''}") + if not g.roots: + print(" ") + for r in g.roots: + label = r.name or Path(r.path).name + flag = " (uploads)" if r.upload else "" + live = "" if Path(r.path).expanduser().is_dir() else " [UNAVAILABLE]" + print(f" {label} → {r.path}{flag}{live}") # Node authority: the roster is the source of truth, node.toml the legacy # form. Read the DB directly so this reports correctly while the daemon is # stopped — the state an operator is most often in when checking. @@ -874,22 +982,172 @@ def main() -> None: print("usage: meshbay-node member list|invite|revoke|unpin") sys.exit(1) - if args.command == "gek-init": + if args.command in ("gek-init", "gek"): + # `gek-init` is the original spelling and still works. `gek rotate` is + # the one that matters after a revocation: the ex-member holds the + # current key and nothing else takes it from them. + sub = "init" if args.command == "gek-init" else (args.subcommand or "init") + if sub not in ("init", "rotate"): + print("usage: meshbay-node gek init|rotate [--group NAME]") + sys.exit(1) + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) group_id = _resolve_group(cfg, args.group) - out = _daemon_api(cfg, f"/api/groups/{group_id}/gek", + + if sub == "rotate" and not args.yes: + print("Rotating replaces this group's key.") + print(" · every member re-receives it automatically on their next connect") + print(" · anyone revoked keeps the OLD key and loses access to new content") + print(" · content already downloaded stays readable to whoever has it") + if input("rotate now? [y/N] ").strip().lower() not in ("y", "yes"): + print("cancelled") + return + + out = _daemon_api(cfg, f"/api/groups/{group_id}/gek" + f"{'?rotate=true' if sub == 'rotate' else ''}", method="POST", timeout=60) - print(f"GEK ready for {group_id}") + verb = "rotated" if out.get("rotated") else "ready" + print(f"GEK {verb} for {group_id}") print(f" {out.get('authorized_members', 0)} authorized member(s) — each " f"receives the key on connect") for err in out.get("errors") or []: print(f" ! {err}") return + if args.command == "reload": + # Milestone 14.8. The daemon re-reads node.toml; groups that appeared or + # whose roots changed are picked up without dropping live connections. + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + import os as _os + import signal as _signal + import subprocess as _subprocess + # `--` is pgrep's own end-of-options marker and must be its own argument; + # folded into the pattern it searches for a process literally called + # "-- -m …". Anchored to the end of the command line so it matches the + # daemon and never a shell that merely mentions it — the same trap + # deploy-node.sh documents, where an unanchored pattern kills the script. + pid_out = _subprocess.run( + ["pgrep", "-f", "--", r"-m meshbay_node\.daemon$"], + capture_output=True, text=True) + pids = [int(x) for x in pid_out.stdout.split()] + if not pids: + print("Node is not running — start it with: meshbay-node") + sys.exit(1) + for pid in pids: + _os.kill(pid, _signal.SIGHUP) + print(f"sent SIGHUP to {len(pids)} daemon process(es)") + print("watch the result: tail -f /tmp/meshbay-node.log") + return + + if args.command == "denylist": + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + sub = args.subcommand or "show" + + if sub == "show": + out = _daemon_api(cfg, "/api/denylist") + total = out.get("count", 0) + if not total: + print("denylist empty — nothing is being refused") + return + for kind in ("users", "groups", "jtis"): + for entry in out.get(kind, []): + print(f" {kind[:-1]:<6} {entry}") + print(f"\n{total} entr(y/ies). These survive a restart (finding H4).") + return + + if sub == "clear": + if not args.yes: + what = args.target or "EVERY entry" + print(f"Clearing the denylist re-admits {what}.") + print("A revocation the hub sent will not come back on its own.") + if input("clear now? [y/N] ").strip().lower() not in ("y", "yes"): + print("cancelled") + return + out = _daemon_api(cfg, f"/api/denylist/clear?subject={args.target or ''}", + method="POST") + print(f"removed {out['removed']} entr(y/ies) ({out['subject']})") + return + + print("usage: meshbay-node denylist show|clear [identifier] [--yes]") + sys.exit(1) + + if args.command == "file": + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + sub = args.subcommand or "list" + group_id = _resolve_group(cfg, args.group) + + if sub == "list": + out = _daemon_api(cfg, f"/api/groups/{group_id}/files") + files = sorted(out.get("files", []), key=lambda f: (f["path"], f["name"])) + if not files: + print("no files indexed") + return + for f in files: + print(f" {f['id'][:12]} {f['size']:>12} {f['path']}/{f['name']}") + print(f"\n{len(files)} file(s). Remove one with: " + f"meshbay-node file rm ") + return + + if sub == "rm": + # Milestone 14.11 — the last operator action that needed a browser. + if not args.target: + print("usage: meshbay-node file rm [--group NAME]") + sys.exit(1) + out = _daemon_api(cfg, f"/api/groups/{group_id}/files",) + matches = [f for f in out.get("files", []) + if f["id"].startswith(args.target)] + if not matches: + print(f"no file whose id starts with {args.target!r}") + sys.exit(1) + if len(matches) > 1: + print(f"{args.target!r} matches {len(matches)} files — be more specific:") + for f in matches[:10]: + print(f" {f['id'][:16]} {f['path']}/{f['name']}") + sys.exit(1) + target = matches[0] + if not args.yes: + print(f"Delete {target['path']}/{target['name']} " + f"({target['size']} bytes) from disk?") + print("This removes the file itself, not just the listing.") + if input("delete? [y/N] ").strip().lower() not in ("y", "yes"): + print("cancelled") + return + _daemon_api(cfg, f"/api/groups/{group_id}/files/{target['id']}", + method="DELETE") + print(f"deleted {target['path']}/{target['name']}") + return + + print("usage: meshbay-node file list|rm [--group NAME] [--yes]") + sys.exit(1) + if args.command == "group": + if args.subcommand in (None, "list"): + # Milestone 14.2. + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + out = _daemon_api(cfg, "/api/groups") + groups = out.get("groups", []) + if not groups: + print("no groups hosted — add one with: " + "meshbay-node group add --dir ") + return + for g in groups: + key = "GEK" if g.get("has_gek") else "NO KEY" + print(f" {g['name']} [{g['visibility']}/{g.get('join_policy')}] " + f"{key} {g['file_count']} file(s) " + f"{g.get('peers', 0)} peer(s)") + print(f" {g['id']}") + for r in g.get("roots", []): + flags = " (uploads)" if r.get("upload") else "" + live = "" if r.get("available", True) else " [UNAVAILABLE]" + print(f" root {r['name']}{flags}{live}") + if not g.get("has_gek"): + print(f" give it a key: meshbay-node gek init " + f"--group {g['name']}") + return + if args.subcommand != "add": - print("usage: meshbay-node group add --dir ") + print("usage: meshbay-node group list|add --dir ") sys.exit(1) if not args.target or not args.dir: print("usage: meshbay-node group add --dir ") @@ -904,8 +1162,9 @@ def main() -> None: print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}") print(f" shared_dir {out['shared_dir']}") print() - print("Restart the daemon so it picks the group up, then give it a key:") - print(f" meshbay-node gek-init --group {out['name']}") + print("Tell the daemon to re-read its config, then give the group a key:") + print(" meshbay-node reload") + print(f" meshbay-node gek init --group {out['name']}") print() print("The key is this group's own — members of your other groups cannot") print("read it, and joining one says nothing about the other.") diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py index a69429c..5dbdc5d 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py @@ -59,6 +59,12 @@ class GroupIndex: sk_node: Ed25519PrivateKey gek: bytes | None = None # None → public group (no encryption) version: int = 1 + # The group's roots and whether each is readable right now. Travels inside + # the encrypted payload because it names the operator's directories, and a + # member needs it to tell "temporarily unavailable" from "deleted" — a + # distinction the entries alone cannot carry, since an unavailable root's + # files are still listed. Absent in an index written before roots existed. + roots: list = field(default_factory=list) _entries: dict = field(default_factory=dict, repr=False) # id → IndexEntry # ── Entry management ────────────────────────────────────────────────────── @@ -90,6 +96,7 @@ class GroupIndex: payload = msgpack.packb({ "group_id": self.group_id, "version": self.version, + "roots": list(self.roots), "entries": [asdict(e) for e in self.entries], }, use_bin_type=True) @@ -170,6 +177,9 @@ class GroupIndex: sk_node=sk_node, gek=gek, version=payload["version"], + # Absent from an index written before roots existed; an empty list + # reads as "nothing known about availability", not "no roots". + roots=payload.get("roots") or [], ) for e in payload["entries"]: idx.add_entry(IndexEntry(**e)) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 60dc04b..d9b1d2c 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -1,15 +1,29 @@ """ -Directory indexer — watches a directory and maintains a GroupIndex. - -Uses watchdog for filesystem events. On any change (create/modify/delete/move), -the affected file is re-scanned and the GroupIndex is updated. -File metadata (blake3 hash, size, type, duration) is computed on first scan. -Heavy operations (hashing large files) run in a thread pool to avoid blocking. +Directory indexer — watches a group's roots and maintains a GroupIndex. + +Uses watchdog for filesystem events. On any change the affected file is +re-scanned and the GroupIndex is updated. File metadata (blake3 hash, size, +type) is computed on first scan; hashing runs in a thread pool. + +Two properties are worth stating because they are what the code is shaped +around, not incidental: + +**A root that goes away freezes; it never empties.** Unmounting a volume either +makes watchdog emit a deletion for every file under it or presents an empty +directory to the next scan. Both would propagate deletions for a whole library +as though the owner had erased it, to every member. So a deletion is acted on +only once the root it belongs to has been confirmed still readable, and a root +that is not is marked unavailable with its entries left exactly where they are. + +**Events are not trusted to be complete.** `ReadDirectoryChangesW` drops events +when its buffer overflows under a burst, and inotify on a FUSE mount misses +changes made outside it. Most users are on Windows sharing from exFAT, so both +apply. A periodic reconciliation scan is therefore not a belt-and-braces extra; +it is the only thing that recovers a missed event. """ import asyncio import logging -import mimetypes import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path @@ -20,8 +34,10 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from watchdog.events import FileSystemEvent, FileSystemEventHandler from watchdog.observers import Observer +from meshbay_common.paths import fold, find_fold_collisions, long_path from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roots import Root, RootSet log = logging.getLogger(__name__) @@ -59,7 +75,19 @@ def _is_indexable(path: Path) -> bool: _HASH_CHUNK = 8 * 1024 * 1024 # 8 MB streaming hash chunks -def _scan_file(root: Path, file_path: Path) -> IndexEntry | None: +def _virtual_dir(root: Root, file_path: Path) -> str: + """ + The directory a file appears in, as members see it: `"Films/2024"`. + + The root name is the first segment for every root, including the only one of + a single-root group — one path shape has to be got right once, two have to + be kept right forever. + """ + rel = file_path.parent.relative_to(root.path) + return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}" + + +def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: """Compute IndexEntry for a file. Blocking — run in executor. Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.) don't require loading the whole file into memory.""" @@ -68,33 +96,33 @@ def _scan_file(root: Path, file_path: Path) -> IndexEntry | None: try: stat = file_path.stat() hasher = blake3.blake3() - with open(file_path, "rb") as f: + # long_path is a no-op off Windows; there it is what lets a deep media + # library past MAX_PATH. + with open(long_path(file_path), "rb") as f: while chunk := f.read(_HASH_CHUNK): hasher.update(chunk) - file_id = hasher.hexdigest() - rel_path = str(file_path.parent.relative_to(root)) - if rel_path == ".": - rel_path = "" return IndexEntry( - id=file_id, + id=hasher.hexdigest(), + # Stored exactly as the filesystem gave it: this is the string that + # opens the file. Normalization is for comparison only. name=file_path.name, - path=rel_path, + path=_virtual_dir(root, file_path), size=stat.st_size, type=_detect_type(file_path), added_at=int(stat.st_mtime), ) - except (OSError, PermissionError) as e: + except (OSError, PermissionError, ValueError) as e: log.warning("Cannot index %s: %s", file_path, e) return None class DirectoryIndexer: """ - Watches a directory and keeps a GroupIndex up to date. + Watches a group's roots and keeps a GroupIndex up to date. Usage: indexer = DirectoryIndexer( - root=Path("/home/user/shared"), + roots=RootSet.build([{"path": "/home/user/shared", "upload": True}]), group_id="my-group", sk_node=sk, gek=gek_bytes, @@ -105,61 +133,147 @@ class DirectoryIndexer: await indexer.stop() """ + # How often to re-check which roots are readable and reconcile the index + # against what is actually on disk. Not a poll for changes — a backstop for + # the events the OS did not deliver, and the way a re-plugged drive is + # noticed. + RECONCILE_SECS = 60.0 + def __init__( self, - root: Path, + roots: RootSet, group_id: str, sk_node: Ed25519PrivateKey, gek: bytes | None, on_change: Callable[["DirectoryIndexer"], Awaitable[None]] | None = None, ): - self.root = root.resolve() + self.roots = roots self.group_id = group_id self.sk_node = sk_node self.gek = gek self.on_change = on_change self._index = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek) + self._index.roots = roots.describe() self._executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="indexer") self._observer: Observer | None = None self._loop: asyncio.AbstractEventLoop | None = None + self._reconciler: asyncio.Task | None = None + self._pending_timers: dict[str, asyncio.TimerHandle] = {} @property def index(self) -> GroupIndex: return self._index + # ── Root lookup ─────────────────────────────────────────────────────────── + + def _root_for(self, file_path: Path) -> Root | None: + """Which root a real path belongs to, longest match first.""" + try: + resolved = file_path.resolve() + except OSError: + resolved = file_path + best: Root | None = None + for root in self.roots: + try: + resolved.relative_to(root.path) + except ValueError: + continue + if best is None or len(root.path.parts) > len(best.path.parts): + best = root + return best + # ── Initial scan ────────────────────────────────────────────────────────── async def initial_scan(self) -> None: - """Scan the entire directory tree. Run once at startup.""" - log.info("Scanning %s ...", self.root) + """Scan every available root. Run once at startup.""" + self.roots.refresh_availability() + total = 0 + for root in self.roots: + if not root.available: + log.warning("Root %r is not readable at startup (%s) — its files " + "are not indexed yet and will appear when it returns", + root.name, root.path) + continue + total += await self._scan_root(root) + self._index.version = int(time.time()) + self._index.roots = self.roots.describe() + self._report_collisions() + log.info("Initial scan complete: %d files across %d root(s)", + total, len(self.roots)) + + async def _scan_root(self, root: Root) -> int: + log.info("Scanning %s (root %r) ...", root.path, root.name) loop = asyncio.get_event_loop() - files = [p for p in self.root.rglob("*") if p.is_file()] count = 0 + try: + files = [p for p in root.path.rglob("*") if p.is_file()] + except OSError as e: + log.warning("Cannot scan root %r: %s", root.name, e) + return 0 for file_path in files: entry = await loop.run_in_executor( - self._executor, _scan_file, self.root, file_path) + self._executor, _scan_file, root, file_path) if entry: self._index.add_entry(entry) count += 1 - self._index.version = int(time.time()) - log.info("Initial scan complete: %d files indexed", count) + return count + + def _report_collisions(self) -> None: + """ + Names that are the same file on a case-insensitive filesystem. + + Reported, never resolved: on ext4 both files exist and only the operator + knows which was meant. Left silent, the pair reaches somebody on Windows + who can save one of them. + """ + by_dir: dict[str, list[str]] = {} + for entry in self._index.entries: + by_dir.setdefault(fold(entry.path), []).append(entry.name) + for folded_dir, names in by_dir.items(): + for _, clashing in find_fold_collisions(names).items(): + log.warning( + "Names that differ only by case or accent form in %s: %s — " + "these are one file on NTFS or exFAT, and a member on Windows " + "can only keep one of them", + folded_dir or "/", ", ".join(sorted(clashing))) # ── Watchdog integration ────────────────────────────────────────────────── async def start(self) -> None: - """Start initial scan + filesystem watcher.""" + """Start initial scan + filesystem watcher + reconciler.""" self._loop = asyncio.get_event_loop() await self.initial_scan() + self._start_observer() + self._reconciler = asyncio.create_task(self._reconcile_loop()) + def _start_observer(self) -> None: handler = _WatchdogHandler(self) self._observer = Observer() - self._observer.schedule(handler, str(self.root), recursive=True) + watched = 0 + for root in self.roots: + if not root.available: + continue + try: + self._observer.schedule(handler, str(root.path), recursive=True) + watched += 1 + except OSError as e: + log.warning("Cannot watch root %r: %s", root.name, e) self._observer.start() - log.info("Watching %s for changes", self.root) + log.info("Watching %d root(s) for changes", watched) async def stop(self) -> None: - """Stop the filesystem watcher.""" + """Stop the filesystem watcher and the reconciler.""" + if self._reconciler: + self._reconciler.cancel() + try: + await self._reconciler + except asyncio.CancelledError: + pass + self._reconciler = None + for handle in self._pending_timers.values(): + handle.cancel() + self._pending_timers.clear() if self._observer: self._observer.stop() self._observer.join() @@ -167,6 +281,171 @@ class DirectoryIndexer: self._executor.shutdown(wait=False) log.info("Indexer stopped") + async def retarget(self, roots: RootSet) -> None: + """ + Point this indexer at a new set of roots, without a restart (14.8). + + Entries under a root that is gone from the config are dropped — the + operator removed it deliberately, which is not the same event as a + volume disappearing, and conflating the two is what §6.9 exists to + prevent. Roots that survive keep their entries; new ones are scanned. + """ + old_names = {r.folded for r in self.roots} + new_names = {r.folded for r in roots} + + for root in self.roots: + if root.folded not in new_names: + dropped = self._entries_under(root) + log.info("Root %r removed from the config — dropping %d entries", + root.name, len(dropped)) + for entry in dropped: + self._index.remove_entry(entry.id) + + self.roots = roots + roots.refresh_availability() + for root in roots: + if root.folded not in old_names and root.available: + await self._scan_root(root) + + self._index.roots = roots.describe() + self._index.version = int(time.time()) + self._restart_observer() + if self.on_change: + await self.on_change(self) + + # ── Reconciliation ──────────────────────────────────────────────────────── + + async def _reconcile_loop(self) -> None: + while True: + try: + await asyncio.sleep(self.RECONCILE_SECS) + await self.reconcile() + except asyncio.CancelledError: + raise + except Exception: + log.exception("Reconcile failed — continuing") + + async def reconcile(self) -> None: + """ + Re-check availability, and rescan roots that came back. + + The only place a root's entries are dropped: when the root is readable + and the files are genuinely gone. A root that is not readable is left + untouched, which is the whole point. + """ + changed = self.roots.refresh_availability() + touched = False + + for root, available in changed: + if available: + log.info("Root %r is back — rescanning", root.name) + self._drop_root_entries(root) + await self._scan_root(root) + touched = True + else: + # Frozen: entries stay, marked unavailable to members through + # the roots table in the index payload. + log.warning("Root %r went away — %d entries frozen, not deleted", + root.name, len(self._entries_under(root))) + touched = True + + if changed: + self._restart_observer() + + if await self._sweep_available_roots(): + touched = True + + if touched: + self._index.roots = self.roots.describe() + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + + async def _sweep_available_roots(self) -> bool: + """ + Catch what the watcher missed: files gone, and files never announced. + + Only touches roots that are readable right now — a root whose volume is + absent has nothing to compare against, and comparing anyway is exactly + the mistake this module exists to avoid. + """ + loop = asyncio.get_event_loop() + changed = False + for root in self.roots: + if not root.available: + continue + try: + on_disk = {p.resolve() for p in root.path.rglob("*") + if _is_indexable(p)} + except OSError as e: + log.warning("Cannot reconcile root %r: %s", root.name, e) + continue + + known: dict[Path, str] = {} + for entry in self._entries_under(root): + abs_path = self._entry_path(root, entry) + if abs_path: + known[abs_path] = entry.id + + for missing in set(known) - on_disk: + # Duplicate content is handled without a special case here: the + # entry goes, and the add loop below re-indexes the surviving + # copy under its own path, because the id is then absent. A + # dedicated "find the survivor" lookup was written first and + # deleted — it rehashed every file under the root on any single + # deletion, and a test proved it changed nothing. + self._index.remove_entry(known[missing]) + log.info("Reconcile: %s is gone", missing) + changed = True + + for added in on_disk - set(known): + entry = await loop.run_in_executor( + self._executor, _scan_file, root, added) + if not entry: + continue + # The index is keyed by **content hash**, so two identical files + # at two paths are one entry and the path comparison above + # cannot see the second. Adding it anyway rewrites that entry's + # path every cycle, bumps the version, and pushes an index + # update to every connected peer once a minute — for ever. + # Measured on a live node: `clip.mp4` present at the root and in + # uploads/ with the same bytes. + if self._index.get_entry(entry.id) is not None: + log.debug("Reconcile: %s duplicates content already indexed " + "as %s — leaving the index alone", + added, entry.id[:8]) + continue + self._index.add_entry(entry) + log.info("Reconcile: %s appeared (missed event)", added) + changed = True + return changed + + def _entries_under(self, root: Root) -> list[IndexEntry]: + prefix = fold(root.name) + return [e for e in self._index.entries + if fold(e.path).split("/", 1)[0] == prefix] + + def _drop_root_entries(self, root: Root) -> None: + for entry in self._entries_under(root): + self._index.remove_entry(entry.id) + + @staticmethod + def _entry_path(root: Root, entry: IndexEntry) -> Path | None: + _, _, tail = entry.path.partition("/") + try: + return (root.path / tail / entry.name).resolve() if tail else \ + (root.path / entry.name).resolve() + except OSError: + return None + + def _restart_observer(self) -> None: + """Re-schedule watches after roots appeared or disappeared.""" + if self._observer: + self._observer.stop() + self._observer.join() + self._observer = None + self._start_observer() + # ── Internal update ─────────────────────────────────────────────────────── _DEBOUNCE_SECS = 2.0 @@ -175,39 +454,60 @@ class DirectoryIndexer: """Called from watchdog thread — schedule debounced async update.""" if not self._loop: return - key = str(file_path.resolve()) - self._loop.call_soon_threadsafe( - self._debounce, key, file_path, deleted) + key = str(file_path) + self._loop.call_soon_threadsafe(self._debounce, key, file_path, deleted) def _debounce(self, key: str, file_path: Path, deleted: bool) -> None: - if not hasattr(self, "_pending_timers"): - self._pending_timers: dict[str, asyncio.TimerHandle] = {} old = self._pending_timers.pop(key, None) if old: old.cancel() - handle = self._loop.call_later( - self._DEBOUNCE_SECS, - lambda: asyncio.ensure_future(self._update_entry(file_path, deleted)), - ) - self._pending_timers[key] = handle - - def _remove_by_path(self, file_path: Path) -> None: - """Remove any existing entries that match this file's path + name.""" - resolved = file_path.resolve() - to_remove = [ - e.id for e in self._index.entries - if (self.root / e.path / e.name).resolve() == resolved - ] - for fid in to_remove: - self._index.remove_entry(fid) + + def fire() -> None: + self._pending_timers.pop(key, None) + asyncio.ensure_future(self._update_entry(file_path, deleted)) + + self._pending_timers[key] = self._loop.call_later(self._DEBOUNCE_SECS, fire) + + def _remove_by_path(self, root: Root, file_path: Path) -> None: + """Remove any existing entries that point at this file.""" + try: + resolved = file_path.resolve() + except OSError: + resolved = file_path + for entry in self._entries_under(root): + if self._entry_path(root, entry) == resolved: + self._index.remove_entry(entry.id) async def _update_entry(self, file_path: Path, deleted: bool) -> None: - self._remove_by_path(file_path) + root = self._root_for(file_path) + if root is None: + return + + if deleted and not root.is_live(): + # The volume went away rather than the file. Freeze: mark the root + # and touch nothing. Every other event for this root will arrive + # here too and be dropped the same way, which is the intent — one + # unplugged drive must not empty a library. + if root.available: + root.available = False + self._index.roots = self.roots.describe() + log.warning("Root %r disappeared — ignoring deletion events and " + "freezing %d entries", root.name, + len(self._entries_under(root))) + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + return + + if not root.available: + return + + self._remove_by_path(root, file_path) if not deleted: loop = asyncio.get_event_loop() entry = await loop.run_in_executor( - self._executor, _scan_file, self.root, file_path) + self._executor, _scan_file, root, file_path) if entry: self._index.add_entry(entry) log.debug("Indexed: %s (%s, %d bytes)", diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py new file mode 100644 index 0000000..c111581 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -0,0 +1,487 @@ +""" +Operator operations — one implementation, several front doors. + +Three things ask this node to act: the CLI (over the loopback admin API), the +local admin UI, and — from Stage B3 — signed MNP messages from a paired client. +They must agree, and the way to make them agree is not to write the operation +three times and hope. + +**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 +the same shape one size down. So each operation lives here once, takes the +daemon's `state`, and knows nothing about HTTP, argv or MNP. The adapters +translate: `ui/app.py` turns `OpError` into a JSON response, the CLI prints it, +the MNP handler sends an error frame. + +**Authorization is not here.** Reaching this module already means the caller got +past its adapter's check — the loopback session token (11.5.3) for the API, an +Ed25519 signature verified against the roster for MNP. These functions do what +they are told; deciding who may tell them is the adapter's job and stays visible +in the adapter. +""" + +from __future__ import annotations + +import logging +from dataclasses import asdict +from pathlib import Path +from typing import Any + +from meshbay_common.crypto import generate_gek, wrap_gek_aes +from meshbay_node.config import DEFAULT_CONFIG_PATH +from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR +from meshbay_node.roots import RootError, RootSet + +log = logging.getLogger(__name__) + + +class OpError(Exception): + """ + An operation refused, with enough for any adapter to report it. + + `status` is an HTTP code because one adapter needs one; the others ignore it. + `extra` carries the "here is what would have worked" payload — a bare "no + such group" leaves an operator guessing at a UUID. + """ + + def __init__(self, message: str, *, status: int = 400, + extra: dict[str, Any] | None = None): + super().__init__(message) + self.message = message + self.status = status + self.extra = extra or {} + + def as_dict(self) -> dict: + return {"error": self.message, **self.extra} + + +# ── Shared lookups ─────────────────────────────────────────────────────────── + +def _roster(state: dict): + roster = state.get("roster") + if not roster: + raise OpError("Roster not available", status=503) + return roster + + +def _hub(state: dict): + hub = state.get("hub") + if not hub or not hub._session: + raise OpError("Hub not connected", status=503) + return hub + + +def _group_ctx(state: dict, group_id: str) -> dict: + groups_ctx = state.get("groups_ctx", {}) + if group_id not in groups_ctx: + raise OpError("Group not hosted on this node", status=404, + extra={"available": [ + {"id": gid} for gid in groups_ctx]}) + return groups_ctx[group_id] + + +def _config(state: dict): + config = state.get("config") + if not config: + raise OpError("No config loaded", status=503) + return config + + +# ── Roster ─────────────────────────────────────────────────────────────────── + +async def read_roster(state: dict, group_id: str = "") -> dict: + roster = state.get("roster") + if not roster: + return {"identities": [], "members": [], "invites": []} + return { + "identities": await roster.list_identities(), + "members": await roster.list_members(group_id or None), + "invites": await roster.list_invites(), + } + + +async def resolve_user(state: dict, username: str) -> dict: + """ + Map a username to an account id. + + The roster answers first — it is the node's own record. The hub is the + fallback for identities pinned before invitations carried a name, and for + people admitted through an open-join group. Only an account id comes back; + no key is ever taken from there. + """ + roster = state.get("roster") + if roster: + for ident in await roster.list_identities(): + if ident["username"] == username: + return {"user_id": ident["user_id"], "source": "roster"} + hub = state.get("hub") + if hub and hub._session: + try: + account = await hub.get_user_pubkeys(username) + return {"user_id": account["user_id"], "source": "hub"} + except Exception: + pass + raise OpError(f"Unknown user {username!r}", status=404) + + +async def pair_operator(state: dict) -> dict: + """ + Issue a one-time code that pairs a browser as this node's operator. + + The code is the whole point: it binds the operator's browser identity key to + their account without asking the hub, which is what stops a hub from naming + itself node administrator (M3, and the same substitution as H3). Returned + once and stored only as a hash. + """ + roster = _roster(state) + user_id = state.get("node_user_id") + if not user_id: + raise OpError("Node not connected to hub yet", status=503) + + config = state.get("config") + ttl = (config.node.pair_ttl_hours if config else 24) * 3600 + code = await roster.create_invite( + group_id="", # operator authority is node-wide + user_id=user_id, + role=ROLE_OPERATOR, + created_by="local-cli", + ttl=ttl, + username=(config.hub.username if config else ""), + ) + invites = await roster.list_invites() + expires = next((i["expires_at"] for i in invites + if i["user_id"] == user_id and i["role"] == ROLE_OPERATOR), "") + return {"code": code, "expires_at": expires, "user_id": user_id} + + +async def create_invite(state: dict, group_id: str, username: str) -> dict: + """ + Issue an invitation code. + + The hub is asked for the account id and nothing else — never for a key. A hub + that answered with the wrong account would produce an invite whose code it + never learns, since the code goes to a human out of band. + """ + roster = _roster(state) + _group_ctx(state, group_id) + hub = _hub(state) + try: + account = await hub.get_user_pubkeys(username) + except Exception as e: + raise OpError(f"Unknown user {username!r}: {e}", status=404) from e + + config = state.get("config") + ttl = (config.node.invite_ttl_hours if config else 168) * 3600 + code = await roster.create_invite( + group_id=group_id, + user_id=account["user_id"], + role=ROLE_MEMBER, + created_by="local-cli", + ttl=ttl, + username=username, + ) + invites = await roster.list_invites() + expires = next((i["expires_at"] for i in invites + if i["user_id"] == account["user_id"] + and i["group_id"] == group_id), "") + return {"code": code, "expires_at": expires, + "username": username, "user_id": account["user_id"]} + + +async def revoke_member(state: dict, user_id: str, group_id: str) -> dict: + """ + Stop serving the group key to someone. + + Takes effect on their next connection: the key is wrapped on demand, so there + is no stored bundle left behind that would outlive this. Rotating the group + key is still required — they hold the current one. + """ + roster = _roster(state) + if not await roster.set_status(group_id, user_id, "revoked"): + raise OpError("No such member in that group", status=404) + log.info("Member revoked: user=%s group=%s", user_id[:8], group_id[:8]) + return {"status": "revoked", "user_id": user_id, "group_id": group_id, + "reminder": "rotate the group key: meshbay-node gek rotate"} + + +async def unpin_member(state: dict, user_id: str) -> dict: + """Forget a pinned identity, so the person can pair again with a new key.""" + roster = _roster(state) + if not await roster.unpin(user_id): + raise OpError("No such pinned identity", status=404) + log.info("Identity unpinned: user=%s", user_id[:8]) + return {"status": "unpinned", "user_id": user_id} + + +# ── Group keys ─────────────────────────────────────────────────────────────── + +async def set_gek(state: dict, group_id: str, *, rotate: bool = False) -> dict: + """ + Generate the group key and activate it, or rotate an existing one. + + Nothing is pre-wrapped for members. Each member's copy is produced when they + connect, for a key they proved they hold (`join_request`) — pre-wrapping used + to fetch public keys from the hub, which is H3 with the node as the victim + instead of the inviter. Only the node's own copy is stored, so the daemon can + reload the key across restarts without the operator's browser. + + **`rotate` generates a fresh key even when one exists.** That is the point of + it: after a revocation the ex-member still holds the current key, and nothing + else takes it away from them. Without `rotate` an existing key is kept, so + running this twice is not destructive by accident. + """ + ctx = _group_ctx(state, group_id) + hub = _hub(state) + + bundle_store = state.get("bundle_store") + if not bundle_store: + raise OpError("Bundle store not available", status=503) + + existing = ctx.get("gek") + gek = generate_gek() if (rotate or not existing) else existing + rotated = bool(existing) and gek is not existing + errors: list[str] = [] + + roster = state.get("roster") + authorized = len(await roster.list_members(group_id)) if roster else 0 + + node_user_id = hub._session.user_id if hub._session else None + pk_x_node_raw = state.get("pk_x25519_raw") + if pk_x_node_raw and node_user_id: + try: + node_bundle = wrap_gek_aes(gek, pk_x_node_raw) + await bundle_store.store( + group_id, f"_node_{node_user_id}", + node_bundle["pk_eph_b64"], node_bundle["nonce_b64"], + node_bundle["wrapped_b64"], + ) + log.info("GEK wrapped for node keystore (daemon reload)") + except Exception as e: + errors.append(f"node keystore: {e}") + log.warning("Failed to wrap GEK for node keystore: %s", e) + + ctx["gek"] = gek + log.info("GEK %s for group %s — %d authorized member(s) will receive it " + "on connect", "rotated" if rotated else "initialized", + group_id[:8], authorized) + + # The transport holds its own view of the group; a rotation that did not + # reach it would keep serving the old key until the daemon restarted. + for transport_key in ("webrtc", "quic_server"): + transport = state.get(transport_key) + groups = getattr(transport, "_ctx", {}).get("groups") if transport else None + if groups and group_id in groups: + groups[group_id]["gek"] = gek + + indexes = state.get("indexes") or {} + index = indexes.get(group_id) + if index is not None: + # The index is encrypted under the GEK; leaving the old key on it would + # serve members a listing they cannot open. + index.gek = gek + + return { + "status": "rotated" if rotated else "ok", + "group_id": group_id, + "rotated": rotated, + "authorized_members": authorized, + "errors": errors, + } + + +# ── Groups and roots ───────────────────────────────────────────────────────── + +async def list_groups(state: dict) -> dict: + """What this node hosts, with live status. Milestone 14.2.""" + config = state.get("config") + groups_ctx = state.get("groups_ctx", {}) + peers = state.get("peers") or {} + out = [] + for gid, ctx in groups_ctx.items(): + cfg = next((g for g in config.groups if g.id == gid), None) if config else None + idx = ctx.get("index") + roots = ctx.get("roots") + out.append({ + "id": gid, + "name": cfg.name if cfg else gid[:8], + "visibility": cfg.visibility if cfg else "private", + "join_policy": cfg.join_policy if cfg else "invite", + "has_gek": bool(ctx.get("gek")), + "file_count": idx.count if idx else 0, + "index_version": idx.version if idx else 0, + "roots": roots.describe() if roots else [], + "peers": sum(1 for p in peers.values() if p.get("group_id") == gid), + }) + return {"groups": out} + + +async def attach_group(state: dict, name: str, shared_dir: str) -> dict: + """ + Write a new [[groups]] block into node.toml. + + The name-to-id lookup happens here because this process is the one logged + into the hub. Nothing is created on the hub: the group already exists, this + only tells the node to host it. + """ + if not name or not shared_dir: + raise OpError("name and shared_dir are required") + config = _config(state) + hub = _hub(state) + try: + mine = await hub.list_my_groups() + except Exception as e: + raise OpError(f"Could not list groups: {e}", status=502) from e + + match = [g for g in mine if g["id"] == name or g["name"] == name] + if not match: + raise OpError(f"No group of yours is called {name!r}", status=404, + extra={"available": [{"name": g["name"], "id": g["id"]} + for g in mine]}) + if len(match) > 1: + raise OpError(f"Several of your groups are called {name!r} — use the id", + status=409, + extra={"available": [{"name": g["name"], "id": g["id"]} + for g in match]}) + group = match[0] + + if any(g.id == group["id"] for g in config.groups): + raise OpError(f"{group['name']!r} is already hosted by this node", status=409) + + path = Path(shared_dir).expanduser() + try: + path.mkdir(parents=True, exist_ok=True) + except OSError as e: + raise OpError(f"Cannot create {path}: {e}") from e + + conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH) + # Appended as text rather than re-serialised: node.toml is hand-written and + # full of comments explaining decisions, and a round trip through a TOML + # writer would throw all of that away. + block = (f'\n[[groups]]\n' + f'id = "{group["id"]}"\n' + f'name = "{group["name"]}"\n' + f'visibility = "{group.get("visibility", "private")}"\n' + f'\n [[groups.roots]]\n' + f' path = "{path}"\n' + f' upload = true\n') + try: + with conf_path.open("a") as f: + f.write(block) + except OSError as e: + raise OpError(f"Cannot write {conf_path}: {e}", status=500) from e + + return {"group_id": group["id"], "name": group["name"], + "shared_dir": str(path), "config": str(conf_path), + "note": "restart the node to pick it up"} + + +async def add_root(state: dict, group_id: str, path: str, *, + name: str = "", kind: str = "generic", + upload: bool = False) -> dict: + """ + Add a directory to a group, refusing anything ambiguous. + + Validated against the group's existing roots *before* being written, so a + config that would be refused at startup is refused here instead — where the + operator is watching and can fix it. + """ + config = _config(state) + cfg = next((g for g in config.groups if g.id == group_id), None) + if cfg is None: + raise OpError("Group not configured on this node", status=404) + + specs = [asdict(r) for r in cfg.roots] + specs.append({"path": path, "name": name, "kind": kind, "upload": upload}) + try: + built = RootSet.build(specs) + except RootError as e: + raise OpError(str(e)) from e + + added = built.roots[-1] + conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH) + raise OpError( + # Writing into the middle of a hand-written TOML file means finding the + # right [[groups]] block and appending inside it, which a text append + # cannot do. Until that is written, say so plainly rather than appending + # to the wrong group. + f"Add this to {conf_path} under the [[groups]] block for " + f"{cfg.name!r}, then restart the node:\n\n" + f' [[groups.roots]]\n' + f' path = "{added.path}"\n' + + (f' name = "{added.name}"\n' if name else "") + + (f' kind = "{added.kind}"\n' if kind != "generic" else "") + + (f' upload = true\n' if upload else ""), + status=501, + extra={"validated": True, "name": added.name, "path": str(added.path)}, + ) + + +# ── Files ──────────────────────────────────────────────────────────────────── + +async def delete_file(state: dict, group_id: str, file_id: str) -> dict: + """ + Remove a file from a group. Milestone 14.11 — the last operator action that + needed a browser. + + Authorization happened in the adapter. On the loopback path that is the + session token, which means physical or SSH access to the machine hosting the + files — an operator who can run this can also `rm` the file, so the check is + not weaker than the alternative. + """ + ctx = _group_ctx(state, group_id) + index = ctx.get("index") + roots = ctx.get("roots") + if not index or not roots: + raise OpError("Group has no index", status=503) + + entry = index.get_entry(file_id) + if not entry: + raise OpError("No such file in this group", status=404) + + from meshbay_node.roots import entry_abs_path + path = entry_abs_path(roots, entry) + if path is None: + raise OpError( + f"{entry.name!r} is in root {entry.path.split('/')[0]!r}, which is " + f"not readable right now — the file is frozen, not gone", status=409) + + try: + path.unlink() + except FileNotFoundError: + # Already gone from disk; drop the stale entry rather than refusing. + log.warning("Index named a file that is not on disk: %s", path) + except OSError as e: + raise OpError(f"Cannot delete {entry.name!r}: {e}", status=500) from e + + index.remove_entry(file_id) + log.info("File deleted by operator: %s/%s", entry.path, entry.name) + return {"status": "deleted", "name": entry.name, "path": entry.path, + "group_id": group_id} + + +# ── Revocation denylist ────────────────────────────────────────────────────── + +async def read_denylist(state: dict) -> dict: + """Milestone 14.10 — what the node is currently refusing.""" + denylist = state.get("denylist") + if not denylist: + return {"users": [], "groups": [], "jtis": [], "count": 0} + entries = denylist.entries() + return {**entries, "count": sum(len(v) for v in entries.values())} + + +async def clear_denylist(state: dict, *, subject: str = "") -> dict: + """ + Drop denylist entries — all of them, or one identifier. + + Deliberately not silent: a cleared denylist re-admits whoever it was keeping + out, and the count is what tells the operator whether they undid one + revocation or all of them. + """ + denylist = state.get("denylist") + if not denylist: + raise OpError("No denylist in this process", status=503) + removed = denylist.clear(subject) + log.warning("Denylist cleared (%s): %d entr(y/ies) removed", + subject or "all", removed) + return {"status": "cleared", "removed": removed, "subject": subject or "all"} diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py new file mode 100644 index 0000000..8f999d7 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -0,0 +1,322 @@ +""" +A group's content is several named roots, not one directory. + + / (the group's virtual root) + ├── Films/ → D:\\Media\\Films + ├── Musique/ → E:\\Audio (external drive) + └── Documents/ → C:\\Users\\me\\Partage + +Every index path carries the root name as its first segment, uniformly — a group +with one root is not a special case, because two path shapes would have to be +kept right forever and one shape only has to be got right once. + +Three rules, and each of them is load-bearing rather than tidy: + + * **The name is the directory's basename, derived once and stored.** Never + recomputed from the path, or renaming a folder on disk silently re-identifies + every file under it. + * **No root may contain another.** Otherwise the same bytes are indexed twice + under two identities, and deleting one leaves the other pointing at nothing. + * **Availability is per root.** A root whose volume goes away *freezes*: its + entries stay in the index, marked unavailable. Emptying it would propagate + deletions for a whole library as though the owner had erased it. + +Comparison of names is case-insensitive and NFC-normalized (`meshbay_common.paths`), +because most of these directories live on exFAT or NTFS, where `Films` and `films` +are one directory. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from pathlib import Path + +from meshbay_common.paths import fold, portable_name_problem + +log = logging.getLogger(__name__) + +VALID_KINDS = ("generic", "video", "audio", "photo") + + +class RootError(ValueError): + """A root set that cannot be built. The message is shown to the operator.""" + + +@dataclass +class Root: + """One named directory inside a group.""" + + name: str + path: Path + kind: str = "generic" + upload: bool = False + # Runtime, not configuration: set by the indexer when the directory can no + # longer be read, and cleared when it comes back. + available: bool = True + + @property + def folded(self) -> str: + return fold(self.name) + + def is_live(self) -> bool: + """Readable right now. The question `available` caches.""" + try: + return self.path.is_dir() + except OSError: + return False + + +def derive_name(path: Path) -> str: + """ + The name a directory gets when it is added: its basename. + + A path that has no usable basename — a drive root such as `E:\\`, or `/` — + has nothing to derive from, and the operator has to supply a name. + """ + name = path.name or "" + if not name: + raise RootError( + f"{path} has no directory name to use — give the root an explicit " + f"name (a drive or filesystem root cannot supply one)") + return name + + +@dataclass +class RootSet: + """ + The roots of one group, and the only place a virtual path is resolved. + + `resolve()` is the single entry point for turning something that arrived + over the wire into a path on disk. Callers must not join paths themselves — + that is how a traversal gets in through the one site nobody reviewed. + """ + + roots: list[Root] = field(default_factory=list) + + # ── Construction ───────────────────────────────────────────────────────── + + @classmethod + def build(cls, specs: list[dict]) -> "RootSet": + """ + Build from configuration, refusing anything ambiguous. + + `specs` are dicts with `path`, and optionally `name`, `kind`, `upload`. + Raises RootError with a message meant for an operator reading a log. + """ + roots: list[Root] = [] + by_folded: dict[str, Root] = {} + + for spec in specs: + raw = str(spec.get("path", "")).strip() + if not raw: + raise RootError("a root has no path") + path = Path(raw).expanduser() + try: + path = path.resolve() + except OSError as e: + raise RootError(f"{raw}: {e}") from e + + name = str(spec.get("name") or "").strip() or derive_name(path) + + problem = portable_name_problem(name) + if problem: + raise RootError( + f"root name {name!r} ({problem}) — every member sees this as a " + f"folder name, including on Windows. Give the root an explicit " + f"name in the config") + + clash = by_folded.get(fold(name)) + if clash: + raise RootError( + f"two roots would both be called {name!r}: {clash.path} and " + f"{path}. Names are compared without regard to case. Give one " + f"of them an explicit name") + + kind = str(spec.get("kind") or "generic").strip().lower() + if kind not in VALID_KINDS: + log.warning("root %r: unknown kind %r — using 'generic'", name, kind) + kind = "generic" + + root = Root(name=name, path=path, kind=kind, + upload=bool(spec.get("upload", False))) + _refuse_nesting(root, roots) + roots.append(root) + by_folded[root.folded] = root + + cls._settle_upload_root(roots) + return cls(roots=roots) + + @staticmethod + def _settle_upload_root(roots: list[Root]) -> None: + """ + Exactly one root receives uploads, and the operator picks it. + + Not guessed when several are marked, because "uploads went somewhere + else" is discovered weeks later. With none marked and a single root, the + answer is not ambiguous, so it is taken. + """ + marked = [r for r in roots if r.upload] + if len(marked) > 1: + names = ", ".join(r.name for r in marked) + raise RootError( + f"several roots are marked upload = true ({names}) — exactly one " + f"receives uploads") + if not marked and len(roots) == 1: + roots[0].upload = True + + # ── Lookup ─────────────────────────────────────────────────────────────── + + def by_name(self, name: str) -> Root | None: + target = fold(name) + for root in self.roots: + if root.folded == target: + return root + return None + + @property + def upload_root(self) -> Root | None: + for root in self.roots: + if root.upload: + return root + return None + + @property + def names(self) -> list[str]: + return [r.name for r in self.roots] + + def __bool__(self) -> bool: + return bool(self.roots) + + def __len__(self) -> int: + return len(self.roots) + + def __iter__(self): + return iter(self.roots) + + # ── Resolution ─────────────────────────────────────────────────────────── + + def split(self, virtual: str) -> tuple[Root, str] | None: + """ + `"Films/2024"` → (the Films root, `"2024"`). None if no such root. + + Does not touch the filesystem, so it is safe to call on a root whose + volume is absent. + """ + rel = (virtual or "").strip().strip("/") + if not rel: + return None + head, _, tail = rel.partition("/") + root = self.by_name(head) + if root is None: + return None + return root, tail + + def resolve(self, virtual: str, *, require_available: bool = True) -> Path | None: + """ + A virtual path from the wire → a real path inside its root, or None. + + Refuses `..`, absolute segments, and anything whose resolved form escapes + the root — symlinks included, which is why this resolves before + comparing rather than checking the string. + """ + found = self.split(virtual) + if found is None: + return None + root, tail = found + if require_available and not root.available: + return None + + parts = [seg for seg in tail.split("/") if seg not in ("", ".")] + if any(seg == ".." for seg in parts): + return None + try: + target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve() + base = root.path.resolve() + except OSError: + return None + if target != base and base not in target.parents: + return None + return target + + def virtual_of(self, absolute: Path) -> str | None: + """A real path anywhere in this group → `"Films/2024"`, or None.""" + for root in self.roots: + virtual = self.virtual_path(root, absolute) + if virtual is not None: + return virtual + return None + + def virtual_path(self, root: Root, absolute: Path) -> str | None: + """The inverse of `resolve`: a real path → `"Films/2024"`.""" + try: + rel = absolute.resolve().relative_to(root.path.resolve()) + except (ValueError, OSError): + return None + return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}" + + # ── Availability ───────────────────────────────────────────────────────── + + def refresh_availability(self) -> list[tuple[Root, bool]]: + """ + Re-read which roots are readable. Returns the ones that changed. + + Called periodically and after a filesystem event that looks like a + disappearance. A change here never edits the index: a root going away + freezes its entries, and a root coming back triggers a rescan. + """ + changed: list[tuple[Root, bool]] = [] + for root in self.roots: + live = root.is_live() + if live != root.available: + root.available = live + changed.append((root, live)) + log.warning("Root %r is now %s (%s)", root.name, + "available" if live else "UNAVAILABLE", root.path) + return changed + + def describe(self) -> list[dict]: + """Per-root state for the index payload and the admin UI.""" + return [ + {"name": r.name, "kind": r.kind, "available": r.available, + "upload": r.upload} + for r in self.roots + ] + + +def entry_abs_path(roots: RootSet, entry) -> Path | None: + """ + Where an index entry actually lives, or None if its root is gone. + + The one place an entry becomes a path. `entry.path` starts with a root name, + so a group whose drive is unplugged answers None here rather than opening + something unrelated that happens to share a relative path with another root. + """ + parent = roots.resolve(entry.path) + return (parent / entry.name) if parent else None + + +def _refuse_nesting(new: Root, existing: list[Root]) -> None: + """ + No root may contain another, compared case-insensitively. + + `D:\\Media` and `D:\\Media\\Films` together would index the same bytes twice + under two identities. On NTFS and exFAT `d:\\media` is the same directory as + `D:\\Media`, so a string comparison that respects case would miss it. + """ + new_parts = [fold(p) for p in new.path.parts] + for other in existing: + other_parts = [fold(p) for p in other.path.parts] + if new_parts == other_parts: + raise RootError( + f"roots {new.name!r} and {other.name!r} are the same directory " + f"({new.path})") + shorter, longer, inner, outer = ( + (other_parts, new_parts, new, other) + if len(other_parts) < len(new_parts) + else (new_parts, other_parts, other, new)) + if longer[:len(shorter)] == shorter: + raise RootError( + f"root {inner.name!r} ({inner.path}) is inside root " + f"{outer.name!r} ({outer.path}) — its files would be indexed " + f"twice under two names") diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 73e668c..ce6fe17 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -35,6 +35,7 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived, StreamReset from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION +from meshbay_node.roots import RootSet, entry_abs_path from meshbay_common.handshake import ( NONCE_LEN, ROLE_CLIENT, @@ -98,6 +99,37 @@ class Denylist: log.info("Denied jti: %s", jti[:8]) self._save() + def entries(self) -> dict[str, list[str]]: + """What is currently refused, for the operator to inspect (14.10).""" + return { + "users": sorted(self.user_ids), + "groups": sorted(self.group_ids), + "jtis": sorted(self.jtis), + } + + def clear(self, subject: str = "") -> int: + """ + Drop everything, or one identifier. Returns how many entries went. + + Not silent by design: clearing re-admits whoever it was keeping out, and + the count is what tells the operator whether they undid one revocation + or all of them. + """ + before = len(self.user_ids) + len(self.group_ids) + len(self.jtis) + if subject: + self.user_ids.discard(subject) + self.group_ids.discard(subject) + self.jtis.discard(subject) + else: + self.user_ids.clear() + self.group_ids.clear() + self.jtis.clear() + after = len(self.user_ids) + len(self.group_ids) + len(self.jtis) + removed = before - after + if removed: + self._save() + return removed + def _load(self) -> None: if not self._path or not self._path.exists(): return @@ -352,7 +384,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return @@ -379,7 +411,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return @@ -506,7 +538,7 @@ class QuicChunkServer: sk_node: Ed25519PrivateKey, hub_pk_pem: bytes, gek: bytes, - shared_root: Path, + roots: RootSet, index: GroupIndex, host: str = "::", # listen IPv4 + IPv6 (dual-stack Linux) port: int = 19000, @@ -519,7 +551,7 @@ class QuicChunkServer: "sk_node": sk_node, "hub_pk_pem": hub_pk_pem, "gek": gek, - "shared_root": shared_root, + "roots": roots, "index": index, } if groups: diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 6b87e31..4ec841f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -60,6 +60,8 @@ from meshbay_common.adminop import ( OP_FILE_DELETE, OP_INVITE_CREATE, OP_MEMBER_REVOKE, + OP_GEK_ROTATE, + OP_MEMBER_UNPIN, admin_transcript, ) from meshbay_common.crypto import pk_to_b64, wrap_gek_aes @@ -72,6 +74,8 @@ from meshbay_common.join import ( from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex +from meshbay_node import ops +from meshbay_node.roots import RootSet, entry_abs_path from meshbay_node.roster import DEFAULT_INVITE_TTL log = logging.getLogger(__name__) @@ -156,37 +160,47 @@ def _free_name(directory: Path, filename: str) -> str: raise FileExistsError(filename) -def safe_subdir(shared_root: Path, rel: str) -> Path | None: +def safe_subdir(roots: RootSet, rel: str) -> Path | None: """ - Resolve a client-supplied directory under the shared root, or refuse. + Resolve a client-supplied directory inside one of the group's roots, or refuse. - Uploads land where the member is looking now rather than in a per-user - quarantine, so the path arrives from the wire and every part of it has to be - checked: each segment against the same allowlist as filenames, and the - resolved result against the root. `..`, absolute paths, symlinks pointing - out, and anything with a separator in a segment are all refused here rather - than in the caller, so there is one place to get it right. + The path arrives from the wire, so every part is checked: the first segment + must name a root that is readable right now, each later segment against the + same allowlist as filenames, and the resolved result against that root's + directory. `..`, absolute paths, symlinks pointing out, and anything with a + separator in a segment are all refused here rather than in the caller, so + there is one place to get it right. + + The virtual root itself — `""` — is deliberately **not** resolvable. It is + not a directory on anyone's disk: a file cannot be written there and a + directory cannot be created there, because it belongs to no volume. Callers + that used to receive the shared root for an empty path now receive None, + which is the honest answer. The quarantine was the fix for C5a; what actually mattered in it — no overwrite, a name allowlist, and confinement — is kept by this plus the caller's existing checks. """ - rel = (rel or "").strip().strip("/") - if not rel: - return shared_root - parts = [seg for seg in rel.split("/") if seg not in ("", ".")] + found = roots.split(rel or "") + if found is None: + return None + root, tail = found + if not root.available: + return None + parts = [seg for seg in tail.split("/") if seg not in ("", ".")] if any(seg == ".." or not SAFE_UPLOAD_NAME.match(seg) for seg in parts): return None try: - target = (shared_root / Path(*parts)).resolve() - root = shared_root.resolve() + target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve() + base = root.path.resolve() except OSError: return None - if target != root and root not in target.parents: + if target != base and base not in target.parents: return None return target + def _extract_dtls_fingerprint(sdp: str) -> bytes: """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes.""" for line in sdp.splitlines(): @@ -439,6 +453,10 @@ class WebRTCPeerSession: self._do_invite_create(msg) elif mtype == MNP.MEMBER_REVOKE: self._do_member_revoke(msg) + elif mtype == MNP.MEMBER_UNPIN: + self._do_member_unpin(msg) + elif mtype == MNP.GEK_ROTATE: + self._do_gek_rotate(msg) elif mtype == MNP.KEYPAIR_BUNDLE_STORE: self._spawn(self._do_keypair_bundle_store(msg)) elif mtype == MNP.KEYPAIR_BUNDLE_DELETE: @@ -1023,10 +1041,9 @@ class WebRTCPeerSession: but it writes to the operator's disk, so it is audited like one. """ ctx = self._group_ctx() - shared_root = ctx.get("shared_root") - if not shared_root: - self._send({"type": "error", "detail": "No shared directory", - "filename": filename}) + roots: RootSet | None = ctx.get("roots") + if not roots: + self._send({"type": "error", "detail": "No shared directory"}) return name = str(msg.get("name", "")).strip() @@ -1034,12 +1051,21 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "Invalid directory name"}) return - parent = safe_subdir(shared_root, msg.get("dir") or "") + # The virtual root is not a directory on anyone's disk, so a member + # cannot create one there — that would be adding a root, which is the + # operator's configuration and not a file operation. + parent_rel = (msg.get("dir") or "").strip("/") + if not parent_rel: + self._send({"type": "error", + "detail": "Choose a folder to create this in"}) + return + + parent = safe_subdir(roots, parent_rel) if parent is None or not parent.is_dir(): self._send({"type": "error", "detail": "Invalid directory"}) return - target = safe_subdir(shared_root, f"{(msg.get('dir') or '').strip('/')}/{name}") + target = safe_subdir(roots, f"{parent_rel}/{name}") if target is None: self._send({"type": "error", "detail": "Invalid directory"}) return @@ -1048,14 +1074,20 @@ class WebRTCPeerSession: return target.mkdir(parents=False) - log.info("Directory created by %s: %s", self._user_id[:8], - target.relative_to(shared_root)) - self._audit("dir_create", str(target.relative_to(shared_root))) + virtual = roots.virtual_of(target) or f"{parent_rel}/{name}" + log.info("Directory created by %s: %s", self._user_id[:8], virtual) + self._audit("dir_create", virtual) self._send({ "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION, - "dir": str(target.relative_to(shared_root)), + "dir": virtual, }) + @staticmethod + def _names_a_root(roots: RootSet, rel: str) -> bool: + """True when `rel` is a bare root name rather than something inside one.""" + found = roots.split(rel or "") + return found is not None and not found[1] + def _do_dir_delete(self, msg: dict) -> None: """ Remove an empty directory, for the node operator. @@ -1068,14 +1100,17 @@ class WebRTCPeerSession: operator deletes the files first and sees what they are losing. """ ctx = self._group_ctx() - shared_root = ctx.get("shared_root") - if not shared_root: - self._send({"type": "error", "detail": "No shared directory", - "filename": filename}) + roots: RootSet | None = ctx.get("roots") + if not roots: + self._send({"type": "error", "detail": "No shared directory"}) return - target = safe_subdir(shared_root, msg.get("dir") or "") - if target is None or target == shared_root: + rel = (msg.get("dir") or "").strip("/") + target = safe_subdir(roots, rel) + # A root itself is not deletable here: removing one is a configuration + # change, and doing it through a file operation would leave the group + # config naming a directory nobody can reach. + if target is None or self._names_a_root(roots, rel): self._send({"type": "error", "detail": "Invalid directory"}) return if not target.is_dir(): @@ -1089,16 +1124,17 @@ class WebRTCPeerSession: return self._issue_admin_challenge( - OP_DIR_DELETE, str(target.relative_to(shared_root))) + OP_DIR_DELETE, roots.virtual_of(target) or rel) async def _admin_exec_dir_delete( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: rel = pending["subject"] ctx = self._group_ctx() - shared_root = ctx.get("shared_root") - target = safe_subdir(shared_root, rel) if shared_root else None - if target is None or target == shared_root or not target.is_dir(): + roots: RootSet | None = ctx.get("roots") + target = safe_subdir(roots, rel) if roots else None + if (target is None or self._names_a_root(roots, rel) + or not target.is_dir()): self._send({"type": "error", "detail": "Not a directory"}) return @@ -1145,6 +1181,95 @@ class WebRTCPeerSession: return self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id) + def _do_gek_rotate(self, msg: dict) -> None: + """ + Ask for a new group key. Operator only, and signed. + + This is what actually removes a revoked member's access: revocation + stops the node serving the *next* key, and they still hold the current + one. The node generates the replacement itself — nothing arriving here + contributes key material, which is what the C5b rule is about. + """ + if not self._group_id: + self._send({"type": "error", "detail": "No group on this connection"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_GEK_ROTATE, self._group_id) + + async def _admin_exec_gek_rotate( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"gek_rotate:{pending['subject'][:8]}") + return + try: + result = await self._run_op( + ops.set_gek, pending["subject"], rotate=True) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("gek_rotate", pending["subject"]) + self._send({ + "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION, + "group_id": pending["subject"], + "authorized_members": result.get("authorized_members", 0), + # Said plainly, because rotating is the step people skip: content + # already downloaded stays readable to whoever holds it. + "note": "members re-receive the key on their next connect; content " + "already downloaded is unaffected", + }) + + def _do_member_unpin(self, msg: dict) -> None: + """Forget a pinned identity, so someone can pair again with a new key.""" + user_id = str(msg.get("user_id", "")).strip() + if not user_id: + self._send({"type": "error", "detail": "Missing user_id"}) + return + if user_id == self._user_id: + # Unpinning yourself over the connection your pin authorizes would + # end that connection's authority mid-operation. + self._send({"type": "error", "detail": "Cannot unpin yourself"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_MEMBER_UNPIN, user_id) + + async def _admin_exec_member_unpin( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + user_id = pending["subject"] + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"member_unpin:{user_id[:8]}") + return + try: + await self._run_op(ops.unpin_member, user_id) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("member_unpin", user_id) + self._send({"type": MNP.MEMBER_UNPIN_ACK, "v": MNP_VERSION, + "user_id": user_id}) + + async def _run_op(self, fn, *args, **kwargs): + """ + Call an operation from `meshbay_node.ops` with the daemon's own view. + + The transport carries its own context and the loopback API carries the + daemon state; they overlap but are not the same dict. Handing the MNP + path a *second* set of lookups is exactly how two implementations of one + operation start disagreeing — C1 and C6 one size down — so the daemon + publishes its state here and both adapters call the same function. + """ + state = self._ctx.get("daemon_state") + if state is None: + raise ops.OpError("Node state not available", status=503) + return await fn(state, *args, **kwargs) + async def _admin_exec_member_revoke( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: @@ -1285,24 +1410,39 @@ class WebRTCPeerSession: # Directories are not index entries, so the client used to infer them # from file paths — which means a folder someone just created, or one # they emptied, simply did not exist as far as the UI was concerned. - "dirs": self._list_dirs(ctx.get("shared_root")), + "dirs": self._list_dirs(ctx.get("roots")), + # Which top-level folders are roots, and whether each is readable. + # A frozen root's files stay listed, so without this a member cannot + # tell "the drive is unplugged" from "it is all still there". + "roots": ctx["roots"].describe() if ctx.get("roots") else [], }) @staticmethod - def _list_dirs(shared_root: Path | None) -> list[str]: - """Directories under the shared root, relative and sorted.""" - if not shared_root: - return [] - out = [] - try: - for path in sorted(shared_root.rglob("*")): - if path.is_dir() and not path.name.startswith("."): - rel = path.relative_to(shared_root) - if not any(part.startswith(".") for part in rel.parts): - out.append(str(rel)) - except OSError: + def _list_dirs(roots: RootSet | None) -> list[str]: + """ + Every directory in the group, as members address them, sorted. + + Each root appears as a directory in its own right, so a root holding no + files yet is still somewhere a member can navigate to and upload into. + An unavailable root is listed too — its content is frozen, not gone, and + hiding it would look exactly like deletion. + """ + if not roots: return [] - return out[:2000] + out: list[str] = [] + for root in roots: + out.append(root.name) + if not root.available: + continue + try: + for path in sorted(root.path.rglob("*")): + if path.is_dir() and not path.name.startswith("."): + rel = path.relative_to(root.path) + if not any(part.startswith(".") for part in rel.parts): + out.append(f"{root.name}/{rel.as_posix()}") + except OSError: + continue + return sorted(out)[:2000] async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() @@ -1314,7 +1454,7 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return @@ -1372,7 +1512,7 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return @@ -1542,20 +1682,41 @@ class WebRTCPeerSession: "filename": filename}) return - shared_root = ctx.get("shared_root") - if not shared_root: - self._send({"type": "error", "detail": "No shared directory", + roots: RootSet | None = ctx.get("roots") + upload_root = roots.upload_root if roots else None + if upload_root is None: + # Refused, never guessed. With several roots, picking one would send + # a member's file to a disk the operator did not intend, and that is + # discovered weeks later. + self._send({"type": "error", + "detail": "No upload folder is configured for this group", + "filename": filename}) + return + if not upload_root.available: + # The designated root's volume is absent. Falling back to another + # root would scatter uploads across disks depending on what happened + # to be plugged in. + self._send({"type": "error", + "detail": f"The upload folder ({upload_root.name}) is " + f"currently unavailable", "filename": filename}) return - # One destination, chosen here and not by the client: uploads/ at the root - # of the shared directory. C5a is still honoured — the name passed the - # allowlist above, and an existing file is never replaced, which was the - # real defect (overwriting a file also made the attacker its recorded - # uploader, and therefore able to delete it). - rel_dir = UPLOAD_DIR_NAME - target_dir = shared_root / UPLOAD_DIR_NAME - target_dir.mkdir(parents=True, exist_ok=True) + # One destination, chosen by the operator and not by the client: + # uploads/ inside the group's designated root. C5a is still honoured — + # the name passed the allowlist above, and an existing file is never + # replaced, which was the real defect (overwriting a file also made the + # attacker its recorded uploader, and therefore able to delete it). + rel_dir = f"{upload_root.name}/{UPLOAD_DIR_NAME}" + target_dir = upload_root.path / UPLOAD_DIR_NAME + try: + target_dir.mkdir(parents=True, exist_ok=True) + except OSError as e: + log.warning("Cannot create upload folder in root %r: %s", + upload_root.name, e) + self._send({"type": "error", "detail": "Upload folder unavailable", + "filename": filename}) + return upload_key = f"{rel_dir}/{filename}" state = self._uploads.get(upload_key) @@ -1789,6 +1950,12 @@ class WebRTCPeerSession: elif pending["op"] == OP_INVITE_CREATE: self._spawn( self._admin_exec_invite_create(pending, transcript, sig_bytes)) + elif pending["op"] == OP_GEK_ROTATE: + self._spawn( + self._admin_exec_gek_rotate(pending, transcript, sig_bytes)) + elif pending["op"] == OP_MEMBER_UNPIN: + self._spawn( + self._admin_exec_member_unpin(pending, transcript, sig_bytes)) else: self._send({"type": "error", "detail": "Unknown admin operation"}) @@ -1865,7 +2032,7 @@ class WebRTCPeerSession: }) def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if file_path.exists(): file_path.unlink() log.info("File deleted: %s", entry.name) @@ -2049,7 +2216,7 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = ctx["shared_root"] / entry.path / entry.name + file_path = entry_abs_path(ctx["roots"], entry) if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return @@ -2265,7 +2432,7 @@ class WebRTCTransport: Manages WebRTC peer connections for browser clients. Usage: - transport = WebRTCTransport(sk_node, hub_pk_pem, gek, shared_root, index) + transport = WebRTCTransport(sk_node, hub_pk_pem, gek, roots, index) answer_sdp = await transport.handle_offer(offer_sdp, peer_id) # Return answer_sdp to the browser via hub signaling """ @@ -2275,7 +2442,7 @@ class WebRTCTransport: sk_node: Ed25519PrivateKey, hub_pk_pem: bytes, gek: bytes, - shared_root: Path, + roots: RootSet, index: GroupIndex, groups: dict[str, dict] | None = None, denylist: Any | None = None, @@ -2286,7 +2453,7 @@ class WebRTCTransport: "sk_node": sk_node, "hub_pk_pem": hub_pk_pem, "gek": gek, - "shared_root": shared_root, + "roots": roots, "index": index, "_peers": {}, # None means "the operator said nothing" — the default applies. It diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index b78f78d..050829e 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -23,6 +23,7 @@ from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query from fastapi.responses import HTMLResponse, JSONResponse from meshbay_node import __version__ +from meshbay_node import ops from meshbay_node.config import DEFAULT_CONFIG_PATH from meshbay_common.crypto import generate_gek, wrap_gek_aes from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR @@ -30,6 +31,24 @@ from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR log = logging.getLogger(__name__) +def _op(coro): + """ + Run an operation and translate its refusal into a JSON response. + + The operations live in `meshbay_node.ops` and know nothing about HTTP. This + is the whole of the HTTP adapter: without it each handler would carry its + own status codes, and the MNP handler in Stage B3 would carry a second set + that slowly stopped agreeing. + """ + async def run(): + try: + return await coro() + except ops.OpError as e: + return JSONResponse(e.as_dict(), e.status) + return run() + + + def create_ui_app(state: dict) -> FastAPI: app = FastAPI( title="MeshBay Node Admin", @@ -110,90 +129,27 @@ def create_ui_app(state: dict) -> FastAPI: @app.get("/api/groups") async def api_groups(): - groups_ctx = state.get("groups_ctx", {}) - config = state.get("config") - result = [] - for gid, ctx in groups_ctx.items(): - cfg = None - if config: - cfg = next((g for g in config.groups if g.id == gid), None) - idx = ctx.get("index") - result.append({ - "id": gid, - "name": cfg.name if cfg else gid[:8], - "shared_dir": str(ctx.get("shared_root", "")), - "visibility": cfg.visibility if cfg else "private", - "file_count": idx.count if idx else 0, - "index_version": idx.version if idx else 0, - }) - return {"groups": result} - + return await _op(lambda: ops.list_groups(state)) @app.post("/api/groups/attach") async def attach_group(payload: dict): - """ - Write a new [[groups]] block into node.toml. + return await _op(lambda: ops.attach_group( + state, + (payload.get("name") or "").strip(), + (payload.get("shared_dir") or "").strip(), + )) - The name-to-id lookup happens here because this process is the one logged - into the hub. Nothing is created on the hub: the group already exists, - this only tells the node to host it. - """ - name = (payload.get("name") or "").strip() - shared_dir = (payload.get("shared_dir") or "").strip() - if not name or not shared_dir: - return JSONResponse({"error": "name and shared_dir are required"}, 400) + @app.delete("/api/groups/{group_id}/files/{file_id}") + async def delete_file(group_id: str, file_id: str): + """Milestone 14.11 — the last operator action that needed a browser.""" + return await _op(lambda: ops.delete_file(state, group_id, file_id)) - config = state.get("config") - if not config: - return JSONResponse({"error": "No config loaded"}, 503) + @app.get("/api/denylist") + async def api_denylist(): + return await _op(lambda: ops.read_denylist(state)) - hub = state.get("hub") - if not hub or not hub._session: - return JSONResponse({"error": "Hub not connected"}, 503) - try: - mine = await hub.list_my_groups() - except Exception as e: - return JSONResponse({"error": f"Could not list groups: {e}"}, 502) - - match = [g for g in mine if g["id"] == name or g["name"] == name] - if not match: - return JSONResponse({ - "error": f"No group of yours is called {name!r}", - "available": [{"name": g["name"], "id": g["id"]} for g in mine], - }, 404) - if len(match) > 1: - return JSONResponse({ - "error": f"Several of your groups are called {name!r} — use the id", - "available": [{"name": g["name"], "id": g["id"]} for g in match], - }, 409) - group = match[0] - - if any(g.id == group["id"] for g in config.groups): - return JSONResponse( - {"error": f"{group['name']!r} is already hosted by this node"}, 409) - - path = Path(shared_dir).expanduser() - try: - path.mkdir(parents=True, exist_ok=True) - except OSError as e: - return JSONResponse({"error": f"Cannot create {path}: {e}"}, 400) - - # Appended as text rather than re-serialised: node.toml is hand-written - # and full of comments explaining decisions, and a round trip through a - # TOML writer would throw all of that away. - conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH) - block = (f'\n[[groups]]\n' - f'id = "{group["id"]}"\n' - f'name = "{group["name"]}"\n' - f'shared_dir = "{path}"\n' - f'visibility = "{group.get("visibility", "private")}"\n') - try: - with conf_path.open("a") as f: - f.write(block) - except OSError as e: - return JSONResponse({"error": f"Cannot write {conf_path}: {e}"}, 500) - - return {"group_id": group["id"], "name": group["name"], - "shared_dir": str(path), "config": str(conf_path)} + @app.post("/api/denylist/clear") + async def api_denylist_clear(subject: str = ""): + return await _op(lambda: ops.clear_denylist(state, subject=subject)) @app.get("/api/groups/{group_id}/files") async def api_group_files(group_id: str): @@ -279,7 +235,11 @@ def create_ui_app(state: dict) -> FastAPI: { "id": g.id, "name": g.name, - "shared_dir": g.shared_dir, + "roots": [ + {"path": r.path, "name": r.name, "kind": r.kind, + "upload": r.upload} + for r in g.roots + ], "visibility": g.visibility, } for g in config.groups @@ -290,205 +250,33 @@ def create_ui_app(state: dict) -> FastAPI: @app.post("/api/operator/pair") async def operator_pair(): - """ - Issue a one-time code that pairs a browser as this node's operator. - - The code is the whole point: it binds the operator's browser identity key - to their account without asking the hub, which is what stops a hub from - naming itself node administrator (M3, and the same substitution as H3). - It is returned once and stored only as a hash. - """ - roster = state.get("roster") - user_id = state.get("node_user_id") - if not roster or not user_id: - return JSONResponse({"error": "Node not connected to hub yet"}, 503) - - config = state.get("config") - ttl = (config.node.pair_ttl_hours if config else 24) * 3600 - code = await roster.create_invite( - group_id="", # operator authority is node-wide - user_id=user_id, - role=ROLE_OPERATOR, - created_by="local-cli", - ttl=ttl, - username=(config.hub.username if config else ""), - ) - invites = await roster.list_invites() - expires = next((i["expires_at"] for i in invites - if i["user_id"] == user_id and i["role"] == ROLE_OPERATOR), "") - return {"code": code, "expires_at": expires, "user_id": user_id} + return await _op(lambda: ops.pair_operator(state)) @app.get("/api/roster") async def api_roster(group_id: str = ""): - roster = state.get("roster") - if not roster: - return {"identities": [], "members": [], "invites": []} - return { - "identities": await roster.list_identities(), - "members": await roster.list_members(group_id or None), - "invites": await roster.list_invites(), - } + return await _op(lambda: ops.read_roster(state, group_id)) @app.post("/api/groups/{group_id}/invites") async def create_invite(group_id: str, username: str): - """ - Issue an invitation code from the CLI, without a browser. - - The hub is asked for the account id and nothing else — never for a key. - A hub that answered with the wrong account would produce an invite whose - code it never learns, since the code goes to a human out of band. - """ - roster = state.get("roster") - groups_ctx = state.get("groups_ctx", {}) - if not roster: - return JSONResponse({"error": "Roster not available"}, 503) - if group_id not in groups_ctx: - return JSONResponse({"error": "Group not hosted on this node"}, 404) - - hub = state.get("hub") - if not hub or not hub._session: - return JSONResponse({"error": "Hub not connected"}, 503) - try: - account = await hub.get_user_pubkeys(username) - except Exception as e: - return JSONResponse({"error": f"Unknown user {username!r}: {e}"}, 404) - - config = state.get("config") - ttl = (config.node.invite_ttl_hours if config else 168) * 3600 - code = await roster.create_invite( - group_id=group_id, - user_id=account["user_id"], - role=ROLE_MEMBER, - created_by="local-cli", - ttl=ttl, - username=username, - ) - invites = await roster.list_invites() - expires = next((i["expires_at"] for i in invites - if i["user_id"] == account["user_id"] - and i["group_id"] == group_id), "") - return {"code": code, "expires_at": expires, - "username": username, "user_id": account["user_id"]} + return await _op(lambda: ops.create_invite(state, group_id, username)) @app.get("/api/resolve") async def resolve_user(username: str): - """ - Map a username to an account id for the CLI. - - The roster answers first — it is the node's own record. The hub is the - fallback for identities pinned before invitations carried a name, and for - people admitted through an open-join group. Only an account id comes back; - no key is ever taken from here. - """ - roster = state.get("roster") - if roster: - for ident in await roster.list_identities(): - if ident["username"] == username: - return {"user_id": ident["user_id"], "source": "roster"} - hub = state.get("hub") - if hub and hub._session: - try: - account = await hub.get_user_pubkeys(username) - return {"user_id": account["user_id"], "source": "hub"} - except Exception: - pass - return JSONResponse({"error": f"Unknown user {username!r}"}, 404) + return await _op(lambda: ops.resolve_user(state, username)) @app.post("/api/members/{user_id}/revoke") async def revoke_member(user_id: str, group_id: str): - """ - Stop serving the group key to someone. - - Takes effect on their next connection: the key is wrapped on demand, so - there is no stored bundle left behind that would outlive this. Rotating - the group key is still required — they hold the current one. - """ - roster = state.get("roster") - if not roster: - return JSONResponse({"error": "Roster not available"}, 503) - if not await roster.set_status(group_id, user_id, "revoked"): - return JSONResponse({"error": "No such member in that group"}, 404) - log.info("Member revoked: user=%s group=%s", user_id[:8], group_id[:8]) - return {"status": "revoked", "user_id": user_id, "group_id": group_id, - "reminder": "rotate the group key: meshbay-node gek-init"} + return await _op(lambda: ops.revoke_member(state, user_id, group_id)) @app.post("/api/members/{user_id}/unpin") async def unpin_member(user_id: str): - """Forget a pinned identity, so the person can pair again with a new key.""" - roster = state.get("roster") - if not roster: - return JSONResponse({"error": "Roster not available"}, 503) - if not await roster.unpin(user_id): - return JSONResponse({"error": "No such pinned identity"}, 404) - log.info("Identity unpinned: user=%s", user_id[:8]) - return {"status": "unpinned", "user_id": user_id} + return await _op(lambda: ops.unpin_member(state, user_id)) # ── GEK initialization (operator only, localhost) ────────────────────── @app.post("/api/groups/{group_id}/gek") - async def init_gek(group_id: str): - """ - Generate the group key and activate it. - - It used to be wrapped here for every member, using public keys fetched from - the hub — which is H3 with the node as the victim instead of the inviter: a - hub answering with its own key was handed the group key by the node itself. - - Nothing is pre-wrapped for members now. Each member's copy is produced when - they connect, for a key they proved they hold (`join_request`). Only the - node's own copy is stored, so the daemon can reload the key across restarts - without the operator's browser. - """ - groups_ctx = state.get("groups_ctx", {}) - if group_id not in groups_ctx: - return JSONResponse({"error": "Group not hosted on this node"}, 404) - - hub = state.get("hub") - if not hub or not hub._session: - return JSONResponse({"error": "Hub not connected"}, 503) - - bundle_store = state.get("bundle_store") - if not bundle_store: - return JSONResponse({"error": "Bundle store not available"}, 503) - - existing_gek = groups_ctx[group_id].get("gek") - gek = existing_gek or generate_gek() - errors: list[str] = [] - - roster = state.get("roster") - authorized = len(await roster.list_members(group_id)) if roster else 0 - - # Store a copy wrapped for the node keystore X25519 key so the daemon can - # reload the GEK on restart without the operator's browser keys. - node_user_id = hub._session.user_id if hub._session else None - pk_x_node_raw = state.get("pk_x25519_raw") - if pk_x_node_raw and node_user_id: - try: - node_bundle = wrap_gek_aes(gek, pk_x_node_raw) - await bundle_store.store( - group_id, f"_node_{node_user_id}", - node_bundle["pk_eph_b64"], node_bundle["nonce_b64"], - node_bundle["wrapped_b64"], - ) - log.info("GEK wrapped for node keystore (daemon reload)") - except Exception as e: - errors.append(f"node keystore: {e}") - log.warning("Failed to wrap GEK for node keystore: %s", e) - - groups_ctx[group_id]["gek"] = gek - log.info("GEK initialized for group %s — %d authorized member(s) will " - "receive it on connect", group_id[:8], authorized) - - webrtc = state.get("webrtc") - if webrtc and "groups" in webrtc._ctx and group_id in webrtc._ctx["groups"]: - webrtc._ctx["groups"][group_id]["gek"] = gek - - return { - "status": "ok", - "group_id": group_id, - "authorized_members": authorized, - "errors": errors, - } + async def init_gek(group_id: str, rotate: bool = False): + return await _op(lambda: ops.set_gek(state, group_id, rotate=rotate)) # ── Chat endpoints ─────────────────────────────────────────────────────── @@ -656,7 +444,15 @@ def _render_page(state: dict, roster_view: dict | None = None) -> str: cfg = next((g for g in config.groups if g.id == gid), None) idx = ctx.get("index") name = cfg.name if cfg else gid[:8] - shared = ctx.get("shared_root", "") + roots = ctx.get("roots") + # An unavailable root is shown as such rather than hidden: its files are + # still listed and still in the index, and hiding the root would make a + # frozen library look deleted — the exact confusion this is meant to + # prevent. + shared = ", ".join( + f"{r.name} → {r.path}" + ("" if r.available else " [UNAVAILABLE]") + for r in roots + ) if roots else "" vis = cfg.visibility if cfg else "private" fcount = idx.count if idx else 0 total_size = sum(e.size for e in idx.entries) if idx else 0 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