diff options
Diffstat (limited to 'packages')
| -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 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_daemon.py | 171 |
4 files changed, 376 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 ─────────────────────────────────── diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py new file mode 100644 index 0000000..e74f7fc --- /dev/null +++ b/packages/meshbay-node/tests/test_daemon.py @@ -0,0 +1,171 @@ +""" +Integration test: Node daemon wires all components correctly. + +Phase 11 — verifies that NodeDaemon creates chat stores, WebRTC transport, +and shuts down cleanly. Hub interaction is mocked. +""" + +import asyncio +import os + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from unittest.mock import AsyncMock, MagicMock, patch + +from meshbay_common.crypto import generate_gek +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon + + +@pytest.fixture +def sk_hub(): + return Ed25519PrivateKey.generate() + + +@pytest.fixture +def hub_pk_pem(sk_hub): + return sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + +@pytest.fixture +def gek(): + return generate_gek() + + +@pytest.fixture +def shared_dir(tmp_path): + d = tmp_path / "shared" + d.mkdir() + (d / "test.bin").write_bytes(os.urandom(2048)) + (d / "hello.txt").write_bytes(b"hello daemon test " * 50) + return d + + +@pytest.fixture +def node_config(tmp_path, shared_dir): + return Config( + hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"), + node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000), + groups=[GroupConfig( + id="g" * 32, + name="test-group", + shared_dir=str(shared_dir), + visibility="private", + port=29000, + quic_port=29010, + http_port=29001, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + + +@pytest.mark.asyncio +async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem): + """Daemon creates ChatStore for each group and shuts down cleanly.""" + daemon = NodeDaemon(node_config) + + sk_node = Ed25519PrivateKey.generate() + mock_keys = MagicMock() + mock_keys.sk_ed25519 = sk_node + mock_keys.pk_ed25519_b64 = "test" + mock_keys.pk_x25519_b64 = "test" + + mock_session = MagicMock() + mock_session.node_id = "node123" + mock_session.user_id = "user123" + mock_session.hub_pk_pem = hub_pk_pem + + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ + patch("meshbay_node.daemon.HubClient") as MockHub: + + hub_instance = AsyncMock() + hub_instance.startup = AsyncMock(return_value=mock_session) + hub_instance.fetch_gek = AsyncMock(return_value=gek) + hub_instance.maintain_ws = AsyncMock() + hub_instance.send_ws = AsyncMock() + hub_instance._ws = None + hub_instance.close = AsyncMock() + hub_instance.__aenter__ = AsyncMock(return_value=hub_instance) + hub_instance.__aexit__ = AsyncMock(return_value=False) + MockHub.return_value = hub_instance + + shutdown_event = asyncio.Event() + + async def mock_maintain_ws(**kwargs): + await shutdown_event.wait() + + hub_instance.maintain_ws = mock_maintain_ws + + async def run_daemon(): + with patch("signal.SIGINT", 2), \ + patch("signal.SIGTERM", 15): + try: + await asyncio.wait_for(daemon.run(), timeout=5) + except (asyncio.TimeoutError, Exception): + pass + + task = asyncio.create_task(run_daemon()) + await asyncio.sleep(1) + + assert daemon._state["status"] == "running" + group_id = "g" * 32 + assert group_id in daemon._chat_stores + assert daemon._chat_stores[group_id]._db is not None + + if daemon._webrtc: + assert "chat_store" in daemon._webrtc._ctx + assert "hub_ws" in daemon._webrtc._ctx + assert "node_user_id" in daemon._webrtc._ctx + assert daemon._webrtc._ctx["node_user_id"] == "user123" + + shutdown_event.set() + await daemon._shutdown() + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + + for store in daemon._chat_stores.values(): + assert store._db is None + + +@pytest.mark.asyncio +async def test_daemon_no_groups_exits(tmp_path): + """Daemon with no valid groups exits cleanly.""" + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"), + node=NodeConfig(), + groups=[GroupConfig(id="", name="empty", shared_dir="")], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + + sk_node = Ed25519PrivateKey.generate() + mock_keys = MagicMock() + mock_keys.sk_ed25519 = sk_node + mock_keys.pk_ed25519_b64 = "test" + + mock_session = MagicMock() + mock_session.node_id = "node123" + mock_session.user_id = "user123" + mock_session.hub_pk_pem = b"pem" + + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ + patch("meshbay_node.daemon.HubClient") as MockHub: + + hub_instance = AsyncMock() + hub_instance.startup = AsyncMock(return_value=mock_session) + hub_instance.close = AsyncMock() + hub_instance.__aenter__ = AsyncMock(return_value=hub_instance) + hub_instance.__aexit__ = AsyncMock(return_value=False) + MockHub.return_value = hub_instance + + await daemon.run() + + assert daemon._state["status"] == "starting" + assert len(daemon._chat_stores) == 0 |