diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 317 |
1 files changed, 288 insertions, 29 deletions
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.") |