diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/config.py | 4 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 214 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/hub_client.py | 16 |
3 files changed, 205 insertions, 29 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index c28105f..9e6a391 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -93,6 +93,7 @@ class Config: node: NodeConfig = field(default_factory=NodeConfig) groups: list[GroupConfig] = field(default_factory=list) keystore: KeystoreConfig = field(default_factory=KeystoreConfig) + data_dir: Path = field(default_factory=lambda: Path.home() / ".local" / "share" / "meshbay") # Back-compat: single-group access @property @@ -140,6 +141,9 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: name=grp.get("name", ""), )) + if "data_dir" in raw: + cfg.data_dir = Path(raw["data_dir"]).expanduser().resolve() + ks = raw.get("keystore", {}) if "path" in ks: cfg.keystore.path = Path(ks["path"]).expanduser() diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 93ba3c4..6a8bbc9 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -7,9 +7,13 @@ Startup sequence: 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 + 6. Create chat stores (one SQLite DB per group) + 7. Create WebRTC transport (browser clients via DataChannel) + 8. Start QUIC+TCP chunk servers (native clients) + 9. Start HTTP file API (public content) + 10. Start hub WebSocket (signaling, revocations, WebRTC offers) + 11. Start local web UI on node.ui_port (localhost only) + 12. Run until SIGINT/SIGTERM Usage: meshbay-node # interactive password prompt @@ -19,6 +23,7 @@ Usage: """ import asyncio +import json import logging import signal import sys @@ -26,13 +31,23 @@ from pathlib import Path import uvicorn +from meshbay_node.chat.store import ChatStore 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.transport.quic_server import QuicChunkServer -from meshbay_node.ui import create_ui_app +from meshbay_node.keystore import NodeKeys, load_or_create_keystore +from meshbay_node.transport import ( + ChunkServer, + Denylist, + QUIC_AVAILABLE, + WEBRTC_AVAILABLE, + create_http_app, +) + +if QUIC_AVAILABLE: + from meshbay_node.transport import QuicChunkServer +if WEBRTC_AVAILABLE: + from meshbay_node.transport import WebRTCTransport log = logging.getLogger(__name__) @@ -43,7 +58,6 @@ 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) @@ -61,12 +75,24 @@ def calibrate_argon2(target_ms: int = 500) -> None: print("Set memory_cost in meshbay_common/crypto.py: ARGON2_MEMORY_COST") +# ── Hub WS sender bridge ───────────────────────────────────────────────────── + +class _WsSender: + """Thin bridge so WebRTC context can call hub_ws.send() for chat_notify.""" + + def __init__(self, hub_client: HubClient): + self._hub = hub_client + + async def send(self, data: str) -> None: + await self._hub.send_ws(data) + + # ── Daemon ──────────────────────────────────────────────────────────────────── class NodeDaemon: def __init__(self, config: Config): - self._config = config - self._state = { + self._config = config + self._state: dict = { "status": "starting", "hub_url": config.hub.url, "username": config.hub.username, @@ -77,9 +103,14 @@ class NodeDaemon: "indexes": {}, } self._tcp_server: ChunkServer | None = None - self._quic_server: QuicChunkServer | None = None + self._quic_server = None + self._webrtc = None + self._denylist = Denylist() if Denylist else None + self._chat_stores: dict[str, ChatStore] = {} self._indexers: list[DirectoryIndexer] = [] - self._tasks: list[asyncio.Task] = [] + self._tasks: list[asyncio.Task] = [] + self._hub: HubClient | None = None + self._http_servers: list[uvicorn.Server] = [] async def run(self) -> None: log.info("MeshBay Node starting up") @@ -98,6 +129,7 @@ class NodeDaemon: password=self._config.hub.password, ) async with HubClient(hub_cfg, keys) as hub: + self._hub = hub session = await hub.startup(endpoint_hint=None) self._state["endpoint_hint"] = session.node_id @@ -142,9 +174,45 @@ class NodeDaemon: "index": indexer.index, } - # 4. QUIC chunk server (primary transport, all groups on one port) - if groups_ctx: - first = next(iter(groups_ctx.values())) + if not groups_ctx: + 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) + for gid in groups_ctx: + chat_db = data_dir / gid[:16] / "chat.db" + store = ChatStore(db_path=chat_db) + await store.open() + self._chat_stores[gid] = store + groups_ctx[gid]["chat_store"] = store + log.info("Chat stores opened: %d groups", len(self._chat_stores)) + + # 5. Denylist + denylist = self._denylist + + # 6. WebRTC transport (browser clients) + first = next(iter(groups_ctx.values())) + if WEBRTC_AVAILABLE: + self._webrtc = WebRTCTransport( + sk_node=keys.sk_ed25519, + hub_pk_pem=session.hub_pk_pem, + gek=first["gek"], + shared_root=first["shared_root"], + index=first["index"], + groups=groups_ctx, + denylist=denylist, + ) + self._webrtc._ctx["chat_store"] = first.get("chat_store") + self._webrtc._ctx["hub_ws"] = _WsSender(hub) + self._webrtc._ctx["node_user_id"] = session.user_id + log.info("WebRTC transport ready") + else: + log.warning("WebRTC not available (aiortc not installed)") + + # 7. QUIC + TCP chunk servers + if QUIC_AVAILABLE: self._quic_server = QuicChunkServer( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, @@ -154,26 +222,97 @@ class NodeDaemon: host="::", port=self._config.node.quic_port, groups=groups_ctx, + denylist=denylist, ) await self._quic_server.start() log.info("QUIC server on port %d (%d groups)", self._config.node.quic_port, len(groups_ctx)) - # TCP+TLS server (fallback transport, same groups) - self._tcp_server = ChunkServer( + self._tcp_server = ChunkServer( + sk_node=keys.sk_ed25519, + hub_pk_pem=session.hub_pk_pem, + gek=first["gek"], + shared_root=first["shared_root"], + index=first["index"], + host="0.0.0.0", + port=self._config.node.port, + groups=groups_ctx, + ) + await self._tcp_server.start() + log.info("TCP+TLS server on port %d", self._config.node.port) + + # 8. Hub WebSocket (signaling + revocations + WebRTC offers) + async def on_webrtc_offer(sdp, peer_id, ice_candidates): + if not self._webrtc: + return None + try: + answer_sdp, answer_ice = await self._webrtc.handle_offer( + sdp, peer_id) + log.info("WebRTC answer for peer=%s (%d peers)", + peer_id, self._webrtc.active_peers) + return (answer_sdp, answer_ice) + except Exception as e: + log.error("WebRTC offer failed: %s", e) + return None + + async def on_incoming(peer_ip, peer_port): + if self._quic_server: + self._quic_server.punch_nat(peer_ip, peer_port) + + def on_revocation(token): + if denylist and token: + import jwt as _jwt + try: + payload = _jwt.decode( + token, session.hub_pk_pem, algorithms=["EdDSA"], + options={"verify_exp": False}) + target = payload.get("target") + tid = payload.get("target_id", "") + if target == "user": + denylist.deny_user(tid) + elif target == "jti": + denylist.deny_jti(tid) + except Exception as e: + log.warning("Invalid revocation token: %s", e) + + ws_task = asyncio.create_task(hub.maintain_ws( + on_incoming=on_incoming, + on_revocation=on_revocation, + on_webrtc_offer=on_webrtc_offer, + group_ids=list(groups_ctx.keys()), + )) + self._tasks.append(ws_task) + log.info("Hub WS task started") + + # 9. HTTP file API (one per group) + for gid, gctx in groups_ctx.items(): + group_cfg = next( + (g for g in self._config.groups if g.id == gid), None) + if not group_cfg: + continue + http_app = create_http_app( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, - gek=first["gek"], - shared_root=first["shared_root"], - index=first["index"], + shared_root=gctx["shared_root"], + index=gctx["index"], + group_id=gid, + group_name=group_cfg.name, + gek=gctx.get("gek"), + ) + http_cfg = uvicorn.Config( + http_app, host="0.0.0.0", - port=self._config.node.port, - groups=groups_ctx, + port=group_cfg.http_port, + log_level="warning", ) - await self._tcp_server.start() - log.info("TCP+TLS server on port %d", self._config.node.port) + http_server = uvicorn.Server(http_cfg) + self._http_servers.append(http_server) + self._tasks.append(asyncio.create_task(http_server.serve())) + log.info("HTTP API on port %d for group %s", + group_cfg.http_port, group_cfg.name) - # 5. Local web UI + # 10. Local web UI + from meshbay_node.ui import create_ui_app ui_app = create_ui_app(self._state) ui_cfg = uvicorn.Config( ui_app, @@ -186,9 +325,12 @@ class NodeDaemon: log.info("Local UI at http://localhost:%d", self._config.node.ui_port) self._state["status"] = "running" - log.info("Node ready — %d groups", len(groups_ctx)) + log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s", + len(groups_ctx), + "yes" if self._webrtc else "no", + "yes" if self._quic_server else "no") - # 6. Wait for shutdown + # 11. Wait for shutdown stop_event = asyncio.Event() loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): @@ -203,13 +345,29 @@ class NodeDaemon: for task in self._tasks: task.cancel() + for task in self._tasks: + try: + await task + except (asyncio.CancelledError, Exception): + pass + + if self._webrtc: + await self._webrtc.close_all() + + for store in self._chat_stores.values(): + await store.close() + for indexer in self._indexers: await indexer.stop() + if self._quic_server: await self._quic_server.stop() if self._tcp_server: await self._tcp_server.stop() + for server in self._http_servers: + server.should_exit = True + log.info("Node stopped") @@ -235,7 +393,7 @@ def main() -> None: if args.command == "init": write_example_config() - print(f"Example config written. Edit it and run: meshbay-node") + print("Example config written. Edit it and run: meshbay-node") return if args.command == "calibrate-argon2": diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index a9a1d6c..ca0e776 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -81,6 +81,7 @@ class HubClient: self._keys = keys self._http = httpx.AsyncClient(timeout=15, base_url=config.hub_url) self._session: HubSession | None = None + self._ws: Any = None async def __aenter__(self): return self @@ -246,6 +247,15 @@ class HubClient: # ── Persistent WebSocket (signaling + revocations) ────────────────────── + async def send_ws(self, data: str) -> None: + """Send a message on the hub WebSocket (if connected). Best-effort.""" + ws = self._ws + if ws: + try: + await ws.send(data) + except Exception: + pass + async def maintain_ws( self, on_incoming: Any = None, @@ -282,6 +292,7 @@ class HubClient: log.error("WS auth failed: %s", auth_resp) return + self._ws = ws log.info("Hub WS connected") async for raw in ws: @@ -310,10 +321,13 @@ class HubClient: elif mtype == "pong": pass + except asyncio.CancelledError: + raise except Exception as e: log.warning("Hub WS disconnected: %s — reconnecting in 5s", e) - import asyncio await asyncio.sleep(5) + finally: + self._ws = None # ── Convenience: full startup sequence ─────────────────────────────────── |