""" MeshBay Node daemon — main process. Startup sequence: 1. Load config (~/.config/meshbay/node.toml) 2. Load or create keystore (Argon2id unlock) 3. Connect to hub: register → login → announce node 4. Fetch GEK bundle from hub (if group configured) 5. Start directory indexer (watchdog) 6. Start TCP+TLS chunk server on node.port 7. Start local web UI on node.ui_port (localhost only) 8. Run until SIGINT/SIGTERM Usage: meshbay-node # interactive password prompt meshbay-node --config /path # custom config meshbay-node init # write example config + create keystore meshbay-node --calibrate-argon2 # benchmark Argon2id, suggest parameters """ import asyncio import logging import signal import sys from pathlib import Path import uvicorn from meshbay_node.config import Config, load_config, write_example_config from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer from meshbay_node.keystore import load_or_create_keystore from meshbay_node.transport import ChunkServer from meshbay_node.ui import create_ui_app log = logging.getLogger(__name__) # ── Argon2id calibration ────────────────────────────────────────────────────── def calibrate_argon2(target_ms: int = 500) -> None: """Benchmark Argon2id and suggest parameters targeting ~target_ms.""" import time import os from meshbay_common.crypto import derive_keystore_key print(f"Calibrating Argon2id (target: {target_ms}ms) ...") salt = os.urandom(16) for mem in [65536, 131072, 262144, 524288]: from cryptography.hazmat.primitives.kdf.argon2 import Argon2id t0 = time.perf_counter() Argon2id(salt=salt, length=32, iterations=3, lanes=4, memory_cost=mem).derive(b"benchmark") elapsed_ms = (time.perf_counter() - t0) * 1000 print(f" memory_cost={mem:>7} ({mem//1024:>4}MB): {elapsed_ms:.0f}ms", end="") if abs(elapsed_ms - target_ms) < target_ms * 0.3: print(" ← recommended") else: print() print("Set memory_cost in meshbay_common/crypto.py: ARGON2_MEMORY_COST") # ── Daemon ──────────────────────────────────────────────────────────────────── class NodeDaemon: def __init__(self, config: Config): self._config = config self._state = { "status": "starting", "hub_url": config.hub.url, "username": config.hub.username, "group_id": config.group.id, "group_name": config.group.name, "node_port": config.node.port, "endpoint_hint": None, "index": None, } self._server: ChunkServer | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] async def run(self) -> None: log.info("MeshBay Node starting up") # 1. Keystore keys = load_or_create_keystore( path=self._config.keystore.path, unlock_file=self._config.keystore.unlock_file, ) log.info("Keys loaded: %s", keys.pk_ed25519_b64[:16]) # 2. Hub connection 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: session = await hub.startup(endpoint_hint=None) self._state["endpoint_hint"] = session.node_id # 3. Fetch GEK if group configured if self._config.group.id: try: gek = await hub.fetch_gek(self._config.group.id) keys.gek = gek log.info("GEK loaded for group %s", self._config.group.id[:8]) except LookupError: log.warning("No GEK bundle found for group %s — " "wait for admin to add you", self._config.group.id[:8]) # 4. Directory indexers async def on_index_change(indexer: DirectoryIndexer) -> None: self._state["index"] = indexer.index for shared_dir in self._config.node.shared_dirs: d = Path(shared_dir).expanduser().resolve() if not d.exists(): log.warning("Shared directory not found: %s — skipping", d) continue indexer = DirectoryIndexer( root=d, group_id=self._config.group.id, sk_node=keys.sk_ed25519, gek=keys.gek, on_change=on_index_change, ) await indexer.start() self._indexers.append(indexer) self._state["index"] = indexer.index log.info("Indexing: %s (%d files)", d, indexer.index.count) # 5. Chunk server if self._indexers and keys.gek: self._server = ChunkServer( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, gek=keys.gek, shared_root=Path(self._config.node.shared_dirs[0]).expanduser(), index=self._indexers[0].index, host="0.0.0.0", port=self._config.node.port, ) await self._server.start() log.info("Chunk server on port %d", self._config.node.port) # 6. Local web UI 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["status"] = "running" log.info("Node ready") # 7. Wait for shutdown stop_event = asyncio.Event() loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, stop_event.set) await stop_event.wait() await self._shutdown() async def _shutdown(self) -> None: log.info("Shutting down...") self._state["status"] = "stopping" for task in self._tasks: task.cancel() for indexer in self._indexers: await indexer.stop() if self._server: await self._server.stop() log.info("Node stopped") # ── Entry point ─────────────────────────────────────────────────────────────── def main() -> None: import argparse parser = argparse.ArgumentParser(description="MeshBay Node daemon") parser.add_argument("command", nargs="?", choices=["init", "calibrate-argon2"], help="init: write example config | calibrate-argon2: benchmark") parser.add_argument("--config", type=Path, default=None, help="Config file path") parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) args = parser.parse_args() logging.basicConfig( level=getattr(logging, args.log_level), format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", ) if args.command == "init": write_example_config() print(f"Example config written. Edit it and run: meshbay-node") return if args.command == "calibrate-argon2": calibrate_argon2() return cfg = load_config(args.config) if not cfg.hub.username: print("Error: hub.username not set in config. Run: meshbay-node init") sys.exit(1) daemon = NodeDaemon(cfg) asyncio.run(daemon.run()) if __name__ == "__main__": main()