From 4b3e8c3b8b9d10c8ac333dd8db614a7569052472 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Mon, 10 Aug 2026 03:07:56 +0200 Subject: feat: Phase 7 — Node v2 (multi-group, Sender Keys, 0-RTT, chat, denylist) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements all 8 milestones (7.0-7.7): - 7.0: JWT carries `groups` claim; node verifies group membership at MNP handshake (QUIC + TCP+TLS). Resolves security review C2. - 7.1: QUIC 0-RTT session resumption via stored session tickets (17-21ms reconnect vs 47ms cold). - 7.2: Hub→node WebSocket signaling for NAT punch coordination (`client_incoming`/`punch_ready`) + jti denylist push. Denylist class blocks revoked users/jtis at handshake. - 7.3: Multi-group daemon — one QUIC port serves N groups with per-group GEK, shared_root, and index routing. - 7.4: HLS streaming via QUIC (STREAM_SEGMENT message type, ffmpeg segment extraction). - 7.5: Sender Keys protocol for group chat (Signal Groups approach). Each member has own sending chain key, HKDF chain ratchet, AES-256-GCM encryption, Ed25519 signing. Resolves security review C1. - 7.6: Chat store (SQLite via aiosqlite), CHAT_MESSAGE MNP wire type with peer broadcast, web UI with WebSocket push. - 7.7: Argon2id calibration CLI. First security review included (first-review.md). 109 tests, demo-v3 validated against meshbay.org production hub. Co-Authored-By: Claude Opus 4.6 --- .../meshbay-node/src/meshbay_node/chat/__init__.py | 4 + .../meshbay-node/src/meshbay_node/chat/store.py | 119 ++++++++++++++ packages/meshbay-node/src/meshbay_node/config.py | 4 +- packages/meshbay-node/src/meshbay_node/daemon.py | 113 ++++++++----- .../meshbay-node/src/meshbay_node/hub_client.py | 54 +++++++ .../src/meshbay_node/transport/__init__.py | 5 +- .../src/meshbay_node/transport/client.py | 10 +- .../src/meshbay_node/transport/quic_client.py | 45 +++++- .../src/meshbay_node/transport/quic_server.py | 176 +++++++++++++++++++-- .../src/meshbay_node/transport/server.py | 24 ++- packages/meshbay-node/src/meshbay_node/ui/app.py | 95 ++++++++++- 11 files changed, 583 insertions(+), 66 deletions(-) create mode 100644 packages/meshbay-node/src/meshbay_node/chat/__init__.py create mode 100644 packages/meshbay-node/src/meshbay_node/chat/store.py (limited to 'packages/meshbay-node/src/meshbay_node') diff --git a/packages/meshbay-node/src/meshbay_node/chat/__init__.py b/packages/meshbay-node/src/meshbay_node/chat/__init__.py new file mode 100644 index 0000000..f647e19 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/chat/__init__.py @@ -0,0 +1,4 @@ +"""MeshBay Node — chat module (Sender Keys encrypted group messaging).""" +from .store import ChatStore + +__all__ = ["ChatStore"] diff --git a/packages/meshbay-node/src/meshbay_node/chat/store.py b/packages/meshbay-node/src/meshbay_node/chat/store.py new file mode 100644 index 0000000..1dbcc2b --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/chat/store.py @@ -0,0 +1,119 @@ +""" +MeshBay Node — SQLite-backed chat message store. + +One database per group. Stores encrypted Sender Keys messages for offline +retrieval and history. Messages are stored as received (ciphertext) — +decryption happens on the client side. +""" + +import logging +import time +from dataclasses import dataclass +from pathlib import Path + +import aiosqlite + +log = logging.getLogger(__name__) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sender_id TEXT NOT NULL, + iteration INTEGER NOT NULL, + payload BLOB NOT NULL, + timestamp REAL NOT NULL, + thread_id TEXT DEFAULT NULL +); +CREATE INDEX IF NOT EXISTS idx_messages_ts ON messages(timestamp); +CREATE INDEX IF NOT EXISTS idx_messages_thread ON messages(thread_id); +""" + + +@dataclass +class StoredMessage: + id: int + sender_id: str + iteration: int + payload: bytes + timestamp: float + thread_id: str | None + + +class ChatStore: + """Async SQLite chat store for one group.""" + + def __init__(self, db_path: Path): + self._db_path = db_path + self._db: aiosqlite.Connection | None = None + + async def open(self) -> None: + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._db = await aiosqlite.connect(str(self._db_path)) + await self._db.executescript(_SCHEMA) + await self._db.commit() + + async def close(self) -> None: + if self._db: + await self._db.close() + self._db = None + + async def __aenter__(self): + await self.open() + return self + + async def __aexit__(self, *_): + await self.close() + + async def save_message( + self, + sender_id: str, + iteration: int, + payload: bytes, + thread_id: str | None = None, + ) -> int: + """Store a message. Returns the row id.""" + ts = time.time() + cursor = await self._db.execute( + "INSERT INTO messages (sender_id, iteration, payload, timestamp, thread_id) " + "VALUES (?, ?, ?, ?, ?)", + (sender_id, iteration, payload, ts, thread_id), + ) + await self._db.commit() + return cursor.lastrowid + + async def get_messages( + self, + since: float = 0, + limit: int = 100, + ) -> list[StoredMessage]: + """Get messages after a timestamp, most recent last.""" + cursor = await self._db.execute( + "SELECT id, sender_id, iteration, payload, timestamp, thread_id " + "FROM messages WHERE timestamp > ? ORDER BY timestamp ASC LIMIT ?", + (since, limit), + ) + rows = await cursor.fetchall() + return [ + StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], + payload=r[3], timestamp=r[4], thread_id=r[5]) + for r in rows + ] + + async def get_thread(self, thread_id: str, limit: int = 100) -> list[StoredMessage]: + """Get messages in a thread.""" + cursor = await self._db.execute( + "SELECT id, sender_id, iteration, payload, timestamp, thread_id " + "FROM messages WHERE thread_id = ? ORDER BY timestamp ASC LIMIT ?", + (thread_id, limit), + ) + rows = await cursor.fetchall() + return [ + StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], + payload=r[3], timestamp=r[4], thread_id=r[5]) + for r in rows + ] + + async def message_count(self) -> int: + cursor = await self._db.execute("SELECT COUNT(*) FROM messages") + row = await cursor.fetchone() + return row[0] diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index b420013..c28105f 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -73,11 +73,12 @@ class NodeConfig: @dataclass class GroupConfig: id: str = "" + name: str = "" + shared_dir: str = "" visibility: str = "private" # public|private port: int = 19000 # TCP+TLS MNP port for this group quic_port: int = 19010 # QUIC MNP port http_port: int = 19001 # HTTP file API port - name: str = "" @dataclass @@ -125,6 +126,7 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.groups.append(GroupConfig( id=g.get("id", ""), name=g.get("name", ""), + shared_dir=g.get("shared_dir", ""), visibility=g.get("visibility", "private"), port=g.get("port", cfg.node.port), quic_port=g.get("quic_port", cfg.node.quic_port), diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index bdac6b8..93ba3c4 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -31,6 +31,7 @@ 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 log = logging.getLogger(__name__) @@ -69,13 +70,14 @@ class NodeDaemon: "status": "starting", "hub_url": config.hub.url, "username": config.hub.username, - "group_id": config.group.id, - "group_name": config.group.name, + "groups": [g.name for g in config.groups], "node_port": config.node.port, + "quic_port": config.node.quic_port, "endpoint_hint": None, - "index": None, + "indexes": {}, } - self._server: ChunkServer | None = None + self._tcp_server: ChunkServer | None = None + self._quic_server: QuicChunkServer | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] @@ -99,52 +101,79 @@ class NodeDaemon: 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) + # 3. 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: + log.warning("Group %r missing id or shared_dir — skipping", + group_cfg.name) continue + + shared_root = Path(group_cfg.shared_dir).expanduser().resolve() + if not shared_root.exists(): + log.warning("Shared dir not found: %s — skipping group %s", + shared_root, group_cfg.name) + continue + + gek = None + if group_cfg.visibility == "private": + try: + gek = await hub.fetch_gek(group_cfg.id) + 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 + indexer = DirectoryIndexer( - root=d, - group_id=self._config.group.id, + root=shared_root, + group_id=group_cfg.id, sk_node=keys.sk_ed25519, - gek=keys.gek, - on_change=on_index_change, + gek=gek, ) await indexer.start() self._indexers.append(indexer) - self._state["index"] = indexer.index - log.info("Indexing: %s (%d files)", d, indexer.index.count) + self._state["indexes"][group_cfg.id] = indexer.index + log.info("Indexing group %s: %s (%d files)", + group_cfg.name, shared_root, indexer.index.count) + + groups_ctx[group_cfg.id] = { + "gek": gek, + "shared_root": shared_root, + "index": indexer.index, + } + + # 4. QUIC chunk server (primary transport, all groups on one port) + if groups_ctx: + first = next(iter(groups_ctx.values())) + self._quic_server = QuicChunkServer( + sk_node=keys.sk_ed25519, + hub_pk_pem=session.hub_pk_pem, + gek=first["gek"], + shared_root=first["shared_root"], + index=first["index"], + host="::", + port=self._config.node.quic_port, + groups=groups_ctx, + ) + await self._quic_server.start() + log.info("QUIC server on port %d (%d groups)", + self._config.node.quic_port, len(groups_ctx)) - # 5. Chunk server - if self._indexers and keys.gek: - self._server = ChunkServer( + # TCP+TLS server (fallback transport, same groups) + self._tcp_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, + 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._server.start() - log.info("Chunk server on port %d", self._config.node.port) + await self._tcp_server.start() + log.info("TCP+TLS server on port %d", self._config.node.port) - # 6. Local web UI + # 5. Local web UI ui_app = create_ui_app(self._state) ui_cfg = uvicorn.Config( ui_app, @@ -157,9 +186,9 @@ class NodeDaemon: log.info("Local UI at http://localhost:%d", self._config.node.ui_port) self._state["status"] = "running" - log.info("Node ready") + log.info("Node ready — %d groups", len(groups_ctx)) - # 7. Wait for shutdown + # 6. Wait for shutdown stop_event = asyncio.Event() loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): @@ -176,8 +205,10 @@ class NodeDaemon: task.cancel() for indexer in self._indexers: await indexer.stop() - if self._server: - await self._server.stop() + if self._quic_server: + await self._quic_server.stop() + if self._tcp_server: + await self._tcp_server.stop() log.info("Node stopped") diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index d91b945..74851c1 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -19,6 +19,7 @@ import logging import time from dataclasses import dataclass, field from pathlib import Path +from typing import Any, Callable import httpx import jwt @@ -243,6 +244,59 @@ class HubClient: r.raise_for_status() return r.json() + # ── Persistent WebSocket (signaling + revocations) ────────────────────── + + async def maintain_ws( + self, + on_incoming: Any = None, + on_revocation: Any = None, + ) -> None: + """ + Maintain a persistent WebSocket connection to the hub. + Receives NAT punch requests and revocation tokens. + Runs until cancelled. + """ + import websockets + + if self._session is None: + raise RuntimeError("Not logged in") + + hub_url = self._session.hub_url.replace("https://", "wss://").replace("http://", "ws://") + ws_url = f"{hub_url}/v1/nodes/ws" + + while True: + try: + async with websockets.connect(ws_url) as ws: + await ws.send(json.dumps({ + "type": "auth", + "token": self._session.access_token, + })) + auth_resp = json.loads(await ws.recv()) + if auth_resp.get("type") != "auth_ok": + log.error("WS auth failed: %s", auth_resp) + return + + log.info("Hub WS connected") + + async for raw in ws: + msg = json.loads(raw) + mtype = msg.get("type") + + if mtype == "client_incoming" and on_incoming: + await on_incoming(msg["peer_ip"], msg["peer_port"]) + await ws.send(json.dumps({"type": "punch_ready"})) + + elif mtype == "revocation" and on_revocation: + on_revocation(msg.get("token", "")) + + elif mtype == "pong": + pass + + except Exception as e: + log.warning("Hub WS disconnected: %s — reconnecting in 5s", e) + import asyncio + await asyncio.sleep(5) + # ── Convenience: full startup sequence ─────────────────────────────────── async def startup(self, endpoint_hint: str | None = None) -> HubSession: diff --git a/packages/meshbay-node/src/meshbay_node/transport/__init__.py b/packages/meshbay-node/src/meshbay_node/transport/__init__.py index b3144a6..5a1b8d7 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/transport/__init__.py @@ -6,15 +6,16 @@ from .http_server import create_http_app # QUIC transport (MNP v2) — requires aioquic>=1.0 # Falls back gracefully if not installed; node still works via TCP+TLS and HTTP. try: - from .quic_server import QuicChunkServer + from .quic_server import QuicChunkServer, Denylist from .quic_client import QuicChunkClient QUIC_AVAILABLE = True except ImportError: QuicChunkServer = None # type: ignore[assignment,misc] QuicChunkClient = None # type: ignore[assignment,misc] + Denylist = None # type: ignore[assignment,misc] QUIC_AVAILABLE = False __all__ = [ "ChunkServer", "ChunkClient", "create_http_app", - "QuicChunkServer", "QuicChunkClient", "QUIC_AVAILABLE", + "QuicChunkServer", "QuicChunkClient", "Denylist", "QUIC_AVAILABLE", ] diff --git a/packages/meshbay-node/src/meshbay_node/transport/client.py b/packages/meshbay-node/src/meshbay_node/transport/client.py index 3430365..63d50af 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/client.py @@ -58,11 +58,13 @@ class ChunkClient: jwt_token: str, gek: bytes, pk_node_b64: str, # node's Ed25519 PK from hub — used for sig verification + group_id: str = "", ): self._host = host self._port = port self._jwt_token = jwt_token self._gek = gek + self._group_id = group_id self._pk_node = Ed25519PublicKey.from_public_bytes( base64.b64decode(pk_node_b64)) self._reader: asyncio.StreamReader | None = None @@ -80,12 +82,14 @@ class ChunkClient: self._reader, self._writer = await asyncio.open_connection( self._host, self._port, ssl=ssl_ctx) - # MNP handshake - await _send(self._writer, { + handshake_msg = { "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": self._jwt_token, - }) + } + if self._group_id: + handshake_msg["group_id"] = self._group_id + await _send(self._writer, handshake_msg) ack = await _recv(self._reader) if ack.get("type") != MNP.HANDSHAKE_ACK: raise ConnectionError(f"Handshake rejected: {ack}") diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py index a2220ff..83b729e 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py @@ -100,18 +100,22 @@ class QuicChunkClient: jwt_token: str, gek: bytes, pk_node_b64: str, - local_port: int = 0, # 0 = OS choisit; spécifier pour hole punching Port-Restricted + local_port: int = 0, # 0 = OS picks; set for hole punching (Port-Restricted) + group_id: str = "", + session_ticket: object | None = None, ): self._host = host self._port = port self._jwt_token = jwt_token self._gek = gek self._local_port = local_port + self._group_id = group_id self._pk_node = Ed25519PublicKey.from_public_bytes( base64.b64decode(pk_node_b64)) self._proto: _MNPClientProtocol | None = None self._cm = None self._ctrl_stream = 0 + self._session_ticket = session_ticket async def __aenter__(self): await self.connect() @@ -120,6 +124,13 @@ class QuicChunkClient: async def __aexit__(self, *_): await self.close() + @property + def session_ticket(self) -> object | None: + return self._session_ticket + + def _save_ticket(self, ticket: object) -> None: + self._session_ticket = ticket + async def connect(self) -> None: import ssl config = QuicConfiguration( @@ -127,20 +138,25 @@ class QuicChunkClient: alpn_protocols=ALPN, verify_mode=ssl.CERT_NONE, # identity verified via Ed25519 at MNP layer ) + if self._session_ticket: + config.session_ticket = self._session_ticket self._cm = connect( self._host, self._port, configuration=config, create_protocol=_MNPClientProtocol, - local_port=self._local_port, # 0 = aléatoire; local_port=X pour hole punching + local_port=self._local_port, + session_ticket_handler=self._save_ticket, ) self._proto = await self._cm.__aenter__() - # MNP handshake on stream 0 - self._proto._send(self._ctrl_stream, { + handshake_msg = { "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": self._jwt_token, - }) + } + if self._group_id: + handshake_msg["group_id"] = self._group_id + self._proto._send(self._ctrl_stream, handshake_msg) ack = await self._proto._recv(self._ctrl_stream) if ack.get("type") != MNP.HANDSHAKE_ACK: raise ConnectionError(f"QUIC handshake rejected: {ack}") @@ -198,3 +214,22 @@ class QuicChunkClient: raise ValueError("Plaintext hash mismatch after decryption") return plaintext + + async def fetch_stream_segment( + self, file_id: str, segment_index: int, segment_duration: int = 4, + ) -> bytes: + """Fetch one HLS segment (MPEG-TS bytes) over QUIC.""" + sid = self._new_stream() + self._proto._send(sid, { + "type": MNP.STREAM_SEGMENT, + "v": MNP_VERSION, + "file_id": file_id, + "segment_index": segment_index, + "segment_duration": segment_duration, + }) + msg = await self._proto._recv(sid, timeout=30.0) + + if msg.get("type") == "error": + raise LookupError(msg.get("detail", "Unknown error")) + + return base64.b64decode(msg["data_b64"]) diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 9cb3bd8..43c1026 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -21,6 +21,7 @@ import asyncio import base64 import logging import struct +import subprocess from pathlib import Path from typing import Any, Callable @@ -50,6 +51,25 @@ MAX_MSG = 64 * 1024 * 1024 ALPN = ["meshbay-mnp"] +class Denylist: + """Shared denylist for revoked users and invalidated JWTs.""" + + def __init__(self): + self.user_ids: set[str] = set() + self.jtis: set[str] = set() + + def is_denied(self, user_id: str, jti: str) -> bool: + return user_id in self.user_ids or jti in self.jtis + + def deny_user(self, user_id: str) -> None: + self.user_ids.add(user_id) + log.info("Denied user: %s", user_id[:8]) + + def deny_jti(self, jti: str) -> None: + self.jtis.add(jti) + log.info("Denied jti: %s", jti[:8]) + + # ── Wire helpers ────────────────────────────────────────────────────────────── def _pack(obj: dict) -> bytes: @@ -90,6 +110,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): super().__init__(*args, **kwargs) self._ctx = node_ctx # shared server context (keys, index, etc.) self._user_id: str | None = None + self._group_id: str | None = None self._buffers: dict[int, _StreamBuffer] = {} def quic_event_received(self, event: QuicEvent) -> None: @@ -116,6 +137,10 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._do_index_sync_sync(stream_id) elif mtype == MNP.FILE_REQUEST: self._do_file_request_sync(stream_id, msg) + elif mtype == MNP.STREAM_SEGMENT: + self._do_stream_segment_sync(stream_id, msg) + elif mtype == MNP.CHAT_MESSAGE: + self._do_chat_message_sync(stream_id, msg) else: log.warning("Unknown MNP message type: %s", mtype) except Exception as e: @@ -123,8 +148,8 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "error", "detail": str(e)}) def _do_handshake_sync(self, stream_id: int, msg: dict) -> None: - import time token = msg.get("token", "") + group_id = msg.get("group_id", "") try: decoded = jwt.decode(token, self._ctx["hub_pk_pem"], algorithms=["EdDSA"]) except Exception as e: @@ -132,21 +157,45 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._quic.close() return - if decoded.get("exp", 0) < int(time.time()): - self._send(stream_id, {"type": "error", "detail": "JWT expired"}) + denylist = self._ctx.get("denylist") + if denylist and denylist.is_denied(decoded.get("sub", ""), decoded.get("jti", "")): + self._send(stream_id, {"type": "error", "detail": "Token revoked"}) + self._quic.close() + return + + if group_id and group_id not in decoded.get("groups", []): + self._send(stream_id, {"type": "error", "detail": "Not a member of this group"}) + self._quic.close() + return + + if group_id and "groups" in self._ctx and group_id not in self._ctx["groups"]: + self._send(stream_id, {"type": "error", "detail": "Group not hosted on this node"}) self._quic.close() return self._user_id = decoded["sub"] - log.info("QUIC handshake OK — user=%s", self._user_id[:8]) + self._group_id = group_id + + peers = self._ctx.get("_peers") + if peers is not None: + peers[self._user_id] = self + + log.info("QUIC handshake OK — user=%s group=%s", self._user_id[:8], group_id[:8] if group_id else "none") self._send(stream_id, { "type": MNP.HANDSHAKE_ACK, "v": MNP_VERSION, "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()), }) + def _group_ctx(self) -> dict: + """Resolve the active group context (multi-group or legacy single-group).""" + if "groups" in self._ctx and self._group_id: + return self._ctx["groups"][self._group_id] + return self._ctx + def _do_index_sync_sync(self, stream_id: int) -> None: - wire = self._ctx["index"].serialize() + ctx = self._group_ctx() + wire = ctx["index"].serialize() self._send(stream_id, { "type": MNP.INDEX_SYNC, "v": MNP_VERSION, @@ -155,26 +204,96 @@ class _MNPServerProtocol(QuicConnectionProtocol): def _do_file_request_sync(self, stream_id: int, msg: dict) -> None: """Serve file chunk synchronously (blocking I/O — acceptable for test sizes).""" + ctx = self._group_ctx() file_id = msg["file_id"] chunk_index = msg["chunk_index"] - entry = self._ctx["index"].get_entry(file_id) + entry = ctx["index"].get_entry(file_id) if not entry: self._send(stream_id, {"type": "error", "detail": "File not found"}) return - file_path = self._ctx["shared_root"] / entry.path / entry.name + file_path = ctx["shared_root"] / entry.path / entry.name if not file_path.exists(): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return chunk_data = _read_and_encrypt( self._ctx["sk_node"], - self._ctx["gek"], + ctx["gek"], file_path, chunk_index, ) self._send(stream_id, chunk_data) + def _do_stream_segment_sync(self, stream_id: int, msg: dict) -> None: + """Extract and serve one HLS segment via ffmpeg.""" + ctx = self._group_ctx() + file_id = msg["file_id"] + segment_index = msg["segment_index"] + segment_duration = msg.get("segment_duration", 4) + + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send(stream_id, {"type": "error", "detail": "File not found"}) + return + + file_path = ctx["shared_root"] / entry.path / entry.name + if not file_path.exists(): + self._send(stream_id, {"type": "error", "detail": "File not on disk"}) + return + + start_time = segment_index * segment_duration + segment_data = _extract_segment(file_path, start_time, segment_duration) + if segment_data is None: + self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) + return + + self._send(stream_id, { + "type": MNP.STREAM_SEGMENT, + "v": MNP_VERSION, + "file_id": file_id, + "segment_index": segment_index, + "data_b64": base64.b64encode(segment_data).decode(), + "size": len(segment_data), + }) + + def _do_chat_message_sync(self, stream_id: int, msg: dict) -> None: + """Receive a chat message, store it, and broadcast to other connected peers.""" + chat_store = self._ctx.get("chat_store") + if chat_store: + import asyncio + asyncio.ensure_future(chat_store.save_message( + sender_id=msg.get("sender_id", self._user_id), + iteration=msg.get("iteration", 0), + payload=msg.get("payload", b"").encode() if isinstance(msg.get("payload"), str) else msg.get("payload", b""), + thread_id=msg.get("thread_id"), + )) + + peers = self._ctx.get("_peers", {}) + broadcast = { + "type": MNP.CHAT_MESSAGE, + "v": MNP_VERSION, + "sender_id": msg.get("sender_id", self._user_id), + "iteration": msg.get("iteration", 0), + "payload": msg.get("payload", ""), + "thread_id": msg.get("thread_id"), + "group_id": self._group_id or "", + } + for uid, proto in peers.items(): + if uid != self._user_id and proto is not self: + try: + proto._send(0, broadcast) + except Exception: + pass + + self._send(stream_id, {"type": "ack", "v": MNP_VERSION}) + + def connection_lost(self, exc) -> None: + peers = self._ctx.get("_peers") + if peers and self._user_id: + peers.pop(self._user_id, None) + super().connection_lost(exc) + def _send(self, stream_id: int, obj: dict) -> None: self._quic.send_stream_data(stream_id, _pack(obj)) self.transmit() @@ -213,6 +332,25 @@ def _read_and_encrypt( } +def _extract_segment(file_path: Path, start_time: float, duration: float) -> bytes | None: + """Extract one HLS segment via ffmpeg. Returns MPEG-TS bytes or None on failure.""" + try: + result = subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", + "-ss", str(start_time), + "-i", str(file_path), + "-t", str(duration), + "-c:v", "copy", "-c:a", "copy", + "-f", "mpegts", "pipe:1"], + capture_output=True, timeout=30, + ) + if result.returncode == 0 and result.stdout: + return result.stdout + return None + except Exception: + return None + + # ── QuicChunkServer ──────────────────────────────────────────────────────────── class QuicChunkServer: @@ -228,10 +366,12 @@ class QuicChunkServer: gek: bytes, shared_root: Path, index: GroupIndex, - host: str = "::", # écoute IPv4 + IPv6 (dual-stack Linux) + host: str = "::", # listen IPv4 + IPv6 (dual-stack Linux) port: int = 19000, cert_path: Path | None = None, key_path: Path | None = None, + groups: dict[str, dict] | None = None, + denylist: Denylist | None = None, ): self._ctx = { "sk_node": sk_node, @@ -240,17 +380,27 @@ class QuicChunkServer: "shared_root": shared_root, "index": index, } + if groups: + self._ctx["groups"] = groups + self._denylist = denylist or Denylist() + self._ctx["denylist"] = self._denylist + self._ctx["_peers"] = {} self._host = host self._port = port self._cert_path = cert_path or Path.home() / ".config/meshbay/node_tls.crt" self._key_path = key_path or Path.home() / ".config/meshbay/node_tls.key" self._server = None self._task = None + self._session_tickets: dict[bytes, Any] = {} @property def port(self) -> int: return self._port + @property + def denylist(self) -> Denylist: + return self._denylist + def _make_config(self) -> QuicConfiguration: from meshbay_node.transport.tls_cert import generate_self_signed_cert if not self._cert_path.exists(): @@ -259,6 +409,12 @@ class QuicChunkServer: config.load_cert_chain(str(self._cert_path), str(self._key_path)) return config + def _store_ticket(self, ticket: Any) -> None: + self._session_tickets[ticket.ticket] = ticket + + def _fetch_ticket(self, label: bytes) -> Any | None: + return self._session_tickets.pop(label, None) + async def start(self) -> None: config = self._make_config() ctx = self._ctx @@ -270,6 +426,8 @@ class QuicChunkServer: self._host, self._port, configuration=config, create_protocol=protocol_factory, + session_ticket_handler=self._store_ticket, + session_ticket_fetcher=self._fetch_ticket, ) log.info("QuicChunkServer listening on %s:%d (QUIC/UDP)", self._host, self._port) diff --git a/packages/meshbay-node/src/meshbay_node/transport/server.py b/packages/meshbay-node/src/meshbay_node/transport/server.py index 6a1b05b..76ac13a 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/server.py @@ -104,6 +104,7 @@ class _ConnectionHandler: gek: bytes, shared_root: Path, index: GroupIndex, + groups: dict[str, dict] | None = None, ): self._reader = reader self._writer = writer @@ -112,8 +113,10 @@ class _ConnectionHandler: self._gek = gek self._shared_root = shared_root self._index = index + self._groups = groups self._peer = writer.get_extra_info("peername") self._user_id: str | None = None + self._group_id: str | None = None async def handle(self) -> None: try: @@ -133,16 +136,28 @@ class _ConnectionHandler: raise ValueError(f"Expected handshake, got {msg.get('type')!r}") token = msg.get("token", "") + group_id = msg.get("group_id", "") try: decoded = jwt.decode(token, self._hub_pk_pem, algorithms=["EdDSA"]) except Exception as e: raise PermissionError(f"Invalid JWT: {e}") from e - if decoded.get("exp", 0) < int(time.time()): - raise PermissionError("JWT expired") + if group_id and group_id not in decoded.get("groups", []): + raise PermissionError("Not a member of this group") + + if group_id and self._groups and group_id not in self._groups: + raise PermissionError("Group not hosted on this node") self._user_id = decoded["sub"] - log.info("[%s] Handshake OK — user=%s", self._peer, self._user_id[:8]) + self._group_id = group_id + + if group_id and self._groups and group_id in self._groups: + ctx = self._groups[group_id] + self._gek = ctx["gek"] + self._shared_root = ctx["shared_root"] + self._index = ctx["index"] + + log.info("[%s] Handshake OK — user=%s group=%s", self._peer, self._user_id[:8], group_id[:8] if group_id else "none") await _send(self._writer, { "type": MNP.HANDSHAKE_ACK, @@ -221,6 +236,7 @@ class ChunkServer: port: int = 19000, cert_path: Path | None = None, key_path: Path | None = None, + groups: dict[str, dict] | None = None, ): self._sk_node = sk_node self._hub_pk_pem = hub_pk_pem @@ -231,6 +247,7 @@ class ChunkServer: self._port = port self._cert_path = cert_path self._key_path = key_path + self._groups = groups self._server: asyncio.Server | None = None @property @@ -264,5 +281,6 @@ class ChunkServer: reader, writer, self._sk_node, self._hub_pk_pem, self._gek, self._shared_root, self._index, + groups=self._groups, ) await handler.handle() diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index f8978d1..a63e28f 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -10,15 +10,18 @@ Minimal FastAPI app providing: Served only on 127.0.0.1 — not exposed to the network. """ +import asyncio +import json import logging from typing import TYPE_CHECKING -from fastapi import FastAPI +from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from meshbay_node import __version__ if TYPE_CHECKING: + from meshbay_node.chat.store import ChatStore from meshbay_node.indexer import GroupIndex log = logging.getLogger(__name__) @@ -127,7 +130,95 @@ def create_ui_app(state: dict) -> FastAPI: {files_html}
MeshBay Node v{__version__} — JSON status - — JSON files + — JSON files — Chat + +""" + + # ── Chat endpoints ─────────────────────────────────────────────────────── + + _chat_subscribers: list[WebSocket] = [] + + @app.get("/api/chat/history") + async def chat_history(since: float = 0, limit: int = 100): + chat_store = state.get("chat_store") + if not chat_store: + return {"messages": []} + msgs = await chat_store.get_messages(since=since, limit=limit) + return { + "messages": [ + { + "id": m.id, + "sender_id": m.sender_id, + "iteration": m.iteration, + "timestamp": m.timestamp, + "thread_id": m.thread_id, + } + for m in msgs + ] + } + + @app.websocket("/ws/chat") + async def chat_websocket(ws: WebSocket): + """WebSocket for real-time chat push to the local UI.""" + await ws.accept() + _chat_subscribers.append(ws) + try: + while True: + await ws.receive_text() + except WebSocketDisconnect: + pass + finally: + _chat_subscribers.remove(ws) + + async def broadcast_chat_to_ui(msg: dict) -> None: + """Push a chat message to all connected UI WebSocket clients.""" + payload = json.dumps(msg) + dead = [] + for ws in _chat_subscribers: + try: + await ws.send_text(payload) + except Exception: + dead.append(ws) + for ws in dead: + _chat_subscribers.remove(ws) + + app.broadcast_chat = broadcast_chat_to_ui + + @app.get("/chat", response_class=HTMLResponse) + async def chat_page(): + return f""" + + + + MeshBay Chat + + + +

MeshBay Chat

+
+

Back to status

+ """ -- cgit v1.2.3