diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 219 |
1 files changed, 217 insertions, 2 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 83c0ed7..bdac6b8 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -1,8 +1,223 @@ -"""Entry point for the meshbay-node systemd service.""" +""" +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: - raise NotImplementedError("Node daemon not yet implemented — see Phase 2") + 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__": |