diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/config.py | 83 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 317 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/group_index.py | 10 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/indexer/indexer.py | 400 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ops.py | 487 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/roots.py | 322 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/quic_server.py | 40 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 301 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ui/app.py | 318 |
9 files changed, 1863 insertions, 415 deletions
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" @@ -97,10 +113,24 @@ class NodeConfig: @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 <name> --dir <path>: 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 '<no id>'}") - print(f" {g.shared_dir or '<no shared_dir>'}") + if not g.roots: + print(" <no directory configured>") + 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 <id>") + 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 <file-id> [--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 <id> [--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 <name> --dir <path>") + 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 <name> --dir <path>") + print("usage: meshbay-node group list|add <name> --dir <path>") sys.exit(1) if not args.target or not args.dir: print("usage: meshbay-node group add <name> --dir <path>") @@ -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. +Directory indexer — watches a group's roots 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. +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 |