diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 160 |
1 files changed, 132 insertions, 28 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 5851b34..fe12909 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -23,6 +23,7 @@ Usage: """ import asyncio +import base64 import json import logging import signal @@ -30,12 +31,14 @@ import sys from pathlib import Path import uvicorn +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from meshbay_common import MNP_VERSION from meshbay_common.protocol import MNP 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, load_config, write_example_config +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 @@ -111,6 +114,7 @@ class NodeDaemon: self._denylist = Denylist() if Denylist else None self._chat_stores: dict[str, ChatStore] = {} self._audit_store: AuditStore | None = None + self._bundle_store: BundleStore | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] self._hub: HubClient | None = None @@ -126,18 +130,46 @@ class NodeDaemon: ) log.info("Keys loaded: %s", keys.pk_ed25519_b64[:16]) - # 2. Hub connection + # 2. Start admin UI early (so operator can copy node key before hub login) + self._state["pk_node_ed25519"] = keys.pk_ed25519_b64 + self._state["config"] = self._config + from meshbay_node.ui import create_ui_app + ui_app = create_ui_app(self._state) + ui_cfg = uvicorn.Config( + ui_app, + host="127.0.0.1", + port=self._config.node.ui_port, + log_level="warning", + ) + ui_server = uvicorn.Server(ui_cfg) + self._tasks.append(asyncio.create_task(ui_server.serve())) + log.info("Admin UI at http://localhost:%d", self._config.node.ui_port) + + # 3. Hub connection (Ed25519 auth — retries until node key is linked) hub_cfg = HubConfig( hub_url=self._config.hub.url, username=self._config.hub.username, - password=self._config.hub.password, ) async with HubClient(hub_cfg, keys) as hub: self._hub = hub - session = await hub.startup(endpoint_hint=None) + session = await self._login_with_retry(hub) self._state["endpoint_hint"] = session.node_id - # 3. Build per-group contexts + # 4. Bundle store (P2P GEK bundles) + data_dir = self._config.data_dir + data_dir.mkdir(parents=True, exist_ok=True) + self._bundle_store = BundleStore(db_path=data_dir / "bundles.db") + await self._bundle_store.open() + log.info("Bundle store opened: %s", data_dir / "bundles.db") + + # X25519 key material for GEK unwrapping + from cryptography.hazmat.primitives import serialization + sk_x_raw = keys.sk_x25519.private_bytes( + serialization.Encoding.Raw, serialization.PrivateFormat.Raw, + serialization.NoEncryption()) + pk_x_raw = base64.b64decode(keys.pk_x25519_b64) + + # 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: @@ -153,12 +185,13 @@ class NodeDaemon: gek = None if group_cfg.visibility == "private": - try: - gek = await hub.fetch_gek(group_cfg.id) + gek = await self._load_gek( + group_cfg.id, session.user_id, sk_x_raw, pk_x_raw) + if gek: log.info("GEK loaded for group %s", group_cfg.id[:8]) - except LookupError: - log.warning("No GEK for group %s — skipping", group_cfg.name) - continue + else: + log.info("No GEK yet for group %s — will accept first setup", + group_cfg.name) indexer = DirectoryIndexer( root=shared_root, @@ -183,9 +216,7 @@ class NodeDaemon: log.error("No valid groups configured — exiting") return - # 4. Chat stores (one SQLite DB per group) - data_dir = self._config.data_dir - data_dir.mkdir(parents=True, exist_ok=True) + # 5. Chat stores (one SQLite DB per group) for gid in groups_ctx: chat_db = data_dir / gid[:16] / "chat.db" store = ChatStore(db_path=chat_db) @@ -194,7 +225,7 @@ class NodeDaemon: groups_ctx[gid]["chat_store"] = store log.info("Chat stores opened: %d groups", len(self._chat_stores)) - # 4b. Audit store (legal compliance — IP + action logging) + # 6. Audit store (legal compliance — IP + action logging) audit_db = data_dir / "audit.db" self._audit_store = AuditStore(db_path=audit_db) await self._audit_store.open() @@ -219,6 +250,17 @@ class NodeDaemon: self._webrtc._ctx["hub_ws"] = _WsSender(hub) self._webrtc._ctx["node_user_id"] = session.user_id self._webrtc._ctx["audit_store"] = self._audit_store + self._webrtc._ctx["bundle_store"] = self._bundle_store + self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw + 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) + if admin_pk: + self._webrtc._ctx["admin_pk_ed25519"] = admin_pk + log.info("Admin Ed25519 key pinned for node sovereignty") + else: + log.warning("No admin_pk_ed25519 — admin operations disabled") log.info("WebRTC transport ready") else: log.warning("WebRTC not available (aiortc not installed)") @@ -323,23 +365,13 @@ class NodeDaemon: log.info("HTTP API on port %d for group %s", group_cfg.http_port, group_cfg.name) - # 10. Local web UI + # 10. Update admin UI state (UI already running from step 2) self._state["groups_ctx"] = groups_ctx - self._state["config"] = self._config self._state["audit_store"] = self._audit_store + self._state["bundle_store"] = self._bundle_store self._state["webrtc"] = self._webrtc self._state["hub"] = hub - from meshbay_node.ui import create_ui_app - ui_app = create_ui_app(self._state) - ui_cfg = uvicorn.Config( - ui_app, - host="127.0.0.1", - port=self._config.node.ui_port, - log_level="warning", - ) - ui_server = uvicorn.Server(ui_cfg) - self._tasks.append(asyncio.create_task(ui_server.serve())) - log.info("Local UI at http://localhost:%d", self._config.node.ui_port) + self._state["pk_x25519_raw"] = pk_x_raw self._state["status"] = "running" log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s", @@ -363,6 +395,75 @@ class NodeDaemon: await self._shutdown() + 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 + while True: + try: + return await hub.startup(endpoint_hint=None) + except _httpx.HTTPStatusError as e: + body = e.response.text if hasattr(e.response, 'text') else '' + if e.response.status_code == 401 and "No node key" in body: + self._state["status"] = "waiting_for_node_key" + log.warning( + "Node key not linked — open admin UI at " + "http://localhost:%d, copy the key, and paste it in " + "Settings > Link Node on the hub. Retrying in 30s...", + self._config.node.ui_port, + ) + await asyncio.sleep(30) + else: + raise + except Exception as e: + log.warning("Hub login failed: %s — retrying in 10s", e) + await asyncio.sleep(10) + + async def _load_gek( + self, + group_id: str, + node_user_id: str, + sk_x_raw: bytes, + pk_x_raw: bytes, + ) -> bytes | None: + """Load GEK from local bundle store (node-only, hub never touches crypto).""" + from meshbay_common.crypto import unwrap_gek_aes + + if not self._bundle_store: + return None + + # Try node-specific bundle first (stored by init_gek for daemon reload), + # then fall back to operator's user bundle (legacy / pre-dual-key) + for user_key in [f"_node_{node_user_id}", node_user_id]: + bundle = await self._bundle_store.fetch(group_id, user_key) + if not bundle: + continue + try: + gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw) + log.info("GEK loaded from local bundle store for group %s (key=%s)", + group_id[:8], user_key[:16]) + return gek + except Exception as e: + log.debug("Failed to unwrap GEK bundle (key=%s): %s", user_key[:16], e) + + 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 + + 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 + async def _on_index_change(self, indexer: DirectoryIndexer) -> None: """Called when a DirectoryIndexer detects file changes.""" group_id = indexer.group_id @@ -429,6 +530,9 @@ class NodeDaemon: if self._audit_store: await self._audit_store.close() + if self._bundle_store: + await self._bundle_store.close() + for store in self._chat_stores.values(): await store.close() @@ -475,7 +579,7 @@ def main() -> None: calibrate_argon2() return - cfg = load_config(args.config) + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) if not cfg.hub.username: print("Error: hub.username not set in config. Run: meshbay-node init") sys.exit(1) |