diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 229 |
1 files changed, 173 insertions, 56 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 46a1716..7f84abe 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -45,7 +45,8 @@ from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer -from meshbay_node.keystore import NodeKeys, load_or_create_keystore +from meshbay_node.keystore import load_or_create_keystore +from meshbay_node.roster import Roster from meshbay_node.transport import ( Denylist, QUIC_AVAILABLE, @@ -117,6 +118,7 @@ class NodeDaemon: self._chat_stores: dict[str, ChatStore] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None + self._roster: Roster | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] self._hub: HubClient | None = None @@ -174,6 +176,14 @@ class NodeDaemon: await self._bundle_store.open() log.info("Bundle store opened: %s", data_dir / "bundles.db") + # 4b. Roster — who this node recognises and which keys are theirs. + # Node authority is established here, locally, and never learned from + # the hub: a hub that could name the operator's key could install + # itself as node administrator. + self._roster = Roster(db_path=data_dir / "roster.db") + await self._roster.open() + await self._roster.purge_expired() + # X25519 key material for GEK unwrapping from cryptography.hazmat.primitives import serialization sk_x_raw = keys.sk_x25519.private_bytes( @@ -223,6 +233,9 @@ class NodeDaemon: "shared_root": shared_root, "index": indexer.index, "visibility": group_cfg.visibility, + # Admission policy comes from node.toml, never from the hub: + # a hub that could declare a group open would be handed its key. + "join_policy": group_cfg.join_policy, } if not groups_ctx: @@ -272,12 +285,20 @@ class NodeDaemon: self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64 - admin_pk = self._resolve_admin_pk(keys) + self._webrtc._ctx["roster"] = self._roster + admin_pk = self._legacy_admin_pk() + paired = await self._roster.has_operator() if self._roster else False if admin_pk: self._webrtc._ctx["admin_pk_ed25519"] = admin_pk - log.info("Admin Ed25519 key pinned for node sovereignty") + self._webrtc._ctx["has_admin_authority"] = paired + if paired or admin_pk: + sources = ([] if not paired else ["paired operator"]) + \ + ([] if not admin_pk else ["node.toml admin_pk"]) + log.info("Node authority: %s", " + ".join(sources)) else: - log.warning("No admin_pk_ed25519 — admin operations disabled") + log.warning( + "No operator paired — invites and file deletion are " + "refused. Run: meshbay-node operator pair") log.info("WebRTC transport ready") else: log.warning("WebRTC not available (aiortc not installed)") @@ -358,6 +379,8 @@ class NodeDaemon: self._state["groups_ctx"] = groups_ctx self._state["audit_store"] = self._audit_store self._state["bundle_store"] = self._bundle_store + self._state["roster"] = self._roster + self._state["node_user_id"] = session.user_id self._state["webrtc"] = self._webrtc self._state["hub"] = hub self._state["pk_x25519_raw"] = pk_x_raw @@ -457,21 +480,25 @@ class NodeDaemon: log.warning("No unwrappable GEK bundle found for group %s", group_id[:8]) return None - def _resolve_admin_pk(self, keys: NodeKeys) -> Ed25519PublicKey | None: - """Resolve the admin Ed25519 public key: config → auto-pin from node keystore.""" - if self._config.admin_pk_ed25519: - try: - raw = base64.b64decode(self._config.admin_pk_ed25519) - return Ed25519PublicKey.from_public_bytes(raw) - except Exception as e: - log.error("Invalid admin_pk_ed25519 in config: %s", e) - return None + def _legacy_admin_pk(self) -> Ed25519PublicKey | None: + """ + The pre-roster way of naming the operator: `admin_pk_ed25519` in node.toml. - pk = keys.sk_ed25519.public_key() - from meshbay_common.crypto import pk_to_b64 - pk_b64 = pk_to_b64(pk) - log.info("Auto-pinning admin key from node keystore: %s", pk_b64[:16]) - return pk + Still honoured so a deployment configured that way keeps working, but no + longer the only path — and the auto-pin that used to stand in for it is + gone. It pinned the node's *keystore* key while the browser signed with the + user's *identity* key, so admin operations failed closed with a signature + error that looked like a bug elsewhere (finding M3). An operator now pairs + a browser with `meshbay-node operator pair`. + """ + if not self._config.admin_pk_ed25519: + return None + try: + raw = base64.b64decode(self._config.admin_pk_ed25519) + return Ed25519PublicKey.from_public_bytes(raw) + except Exception as e: + log.error("Invalid admin_pk_ed25519 in config: %s", e) + return None async def _on_index_change(self, indexer: DirectoryIndexer) -> None: """Called when a DirectoryIndexer detects file changes.""" @@ -554,6 +581,9 @@ class NodeDaemon: if self._bundle_store: await self._bundle_store.close() + if self._roster: + await self._roster.close() + for store in self._chat_stores.values(): await store.close() @@ -570,6 +600,60 @@ class NodeDaemon: log.info("Node stopped") +# ── CLI helpers ─────────────────────────────────────────────────────────────── + +def _daemon_api(cfg: Config, path: str, method: str = "GET", + timeout: int = 30) -> dict: + """ + Call the daemon's loopback API. + + The daemon owns the roster, the hub session and the live group contexts, so + the CLI asks it to act rather than opening its databases behind its back. It + also means every operator action goes through the same authorization as the + admin UI (the per-run session token, 11.5.3). + """ + import json as _json + import urllib.error + import urllib.parse + import urllib.request + + token_file = cfg.data_dir / "ui-token" + if not token_file.exists(): + print("Node is not running — start it with: meshbay-node") + sys.exit(1) + + sep = "&" if "?" in path else "?" + url = (f"http://127.0.0.1:{cfg.node.ui_port}{path}" + f"{sep}t={token_file.read_text().strip()}") + try: + req = urllib.request.Request(url, method=method) + with urllib.request.urlopen(req, timeout=timeout) as r: + return _json.loads(r.read()) + except urllib.error.HTTPError as e: + body = e.read().decode()[:300] + try: + detail = _json.loads(body).get("error", body) + except Exception: + detail = body + print(f"failed: {detail}") + sys.exit(1) + except Exception as e: + print(f"failed: {e}") + sys.exit(1) + + +def _resolve_group(cfg: Config, group: str | None) -> str: + """The group argument, or the only configured one.""" + if group: + return group + configured = [g.id for g in cfg.groups if g.id] + if len(configured) == 1: + return configured[0] + print("--group is required (several groups configured)" + if configured else "no group configured in node.toml") + sys.exit(1) + + # ── Entry point ─────────────────────────────────────────────────────────────── def main() -> None: @@ -577,19 +661,23 @@ def main() -> None: parser = argparse.ArgumentParser(description="MeshBay Node daemon") parser.add_argument("command", nargs="?", - choices=["init", "status", "ui", "gek-init", "calibrate-argon2"], + choices=["init", "status", "ui", "gek-init", "operator", + "calibrate-argon2"], help="init: write example config | status: node state and keys " - "| ui: print the admin UI URL | calibrate-argon2: benchmark") + "| ui: print the admin UI URL | operator pair: pair a " + "browser with this node | calibrate-argon2: benchmark") + parser.add_argument("subcommand", nargs="?", + help="'pair' for the operator command") parser.add_argument("--config", type=Path, default=None, help="Config file path") parser.add_argument("--group", default=None, - help="group id for gek-init (optional if only one is configured)") + help="group id (optional if only one is configured)") parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) args = parser.parse_args() # Query commands print a report; library logging would interleave with it. - quiet = args.command in ("status", "ui", "gek-init") + quiet = args.command in ("status", "ui", "gek-init", "operator") logging.basicConfig( level=logging.ERROR if quiet else getattr(logging, args.log_level), format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", @@ -649,50 +737,79 @@ def main() -> None: 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 cfg.admin_pk_ed25519: - print("admin key NOT pinned — file deletion and member invites will be") - print(" refused (node.toml: admin_pk_ed25519)") - return - - if args.command == "gek-init": - import json as _json - import urllib.error - import urllib.request + # 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. + import asyncio as _asyncio - cfg = load_config(args.config or DEFAULT_CONFIG_PATH) - group_id = args.group - if not group_id: - if len(cfg.groups) == 1: - group_id = cfg.groups[0].id - else: - print("--group is required (several groups configured)") - sys.exit(1) + from meshbay_node.roster import Roster as _Roster - token_file = cfg.data_dir / "ui-token" - if not token_file.exists(): - print("Node is not running — start it with: meshbay-node") - sys.exit(1) + async def _read_roster() -> tuple[list, int]: + r = _Roster(db_path=cfg.data_dir / "roster.db") + await r.open() + try: + return (await r.list_members()), len(await r.list_invites()) + finally: + await r.close() - # The daemon holds the hub session and the live group contexts, so the CLI - # asks it to do the work rather than duplicating it. Same operation as the - # admin UI button — an operator on a headless host should never need a - # browser on that host to initialise a group key. - url = (f"http://127.0.0.1:{cfg.node.ui_port}/api/groups/{group_id}/gek" - f"?t={token_file.read_text().strip()}") try: - req = urllib.request.Request(url, method="POST") - with urllib.request.urlopen(req, timeout=60) as r: - out = _json.loads(r.read()) - except urllib.error.HTTPError as e: - print(f"failed: {e.code} {e.read().decode()[:300]}") - sys.exit(1) + members, pending = _asyncio.run(_read_roster()) + except Exception as e: + members, pending = [], 0 + print(f"roster <unreadable: {e}>") + + operators = [m for m in members if m["role"] == "operator" + and m["status"] == "active"] + if operators: + for op in operators: + print(f"operator {op.get('username') or op['user_id'][:8]}" + f" key {(op.get('pk_ed25519') or '')[:16]}…" + f" paired {op.get('pinned_at', '?')}") + elif cfg.admin_pk_ed25519: + print("operator node.toml admin_pk_ed25519 (legacy)") + print(" run `meshbay-node operator pair` to replace it") + else: + print("operator NONE PAIRED — file deletion and member invites are") + print(" refused. Run: meshbay-node operator pair") + if pending: + print(f"invites {pending} pending code(s)") + return + + if args.command == "gek-init": + 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", + method="POST", timeout=60) print(f"GEK ready for {group_id}") - print(f" wrapped for {out.get('wrapped_count')}/{out.get('total_members')} members") + 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 == "operator": + if args.subcommand != "pair": + print("usage: meshbay-node operator pair") + sys.exit(1) + + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + out = _daemon_api(cfg, "/api/operator/pair", method="POST") + + from meshbay_node.roster import write_code_file + path = write_code_file(cfg.data_dir, out["code"], out.get("expires_at", "")) + + print(f"PAIRING CODE {out['code']}") + print(f"valid until {out.get('expires_at', '?')}") + print() + print("Sign in to the web app as this node's operator, open one of your") + print("groups, go to the Members tab and enter the code there.") + print("It works once, for that account only, and authorizes invites and") + print("file deletion from that browser.") + print() + print(f"also written to {path}") + return + if args.command == "ui": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) token_file = cfg.data_dir / "ui-token" |