diff options
Diffstat (limited to 'packages/meshbay-node')
15 files changed, 1012 insertions, 68 deletions
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 + # 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 - 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) + 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} <hr> <small>MeshBay Node v{__version__} — <a href="/api/status">JSON status</a> - — <a href="/api/files">JSON files</a></small> + — <a href="/api/files">JSON files</a> — <a href="/chat">Chat</a></small> +</body> +</html>""" + + # ── 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"""<!DOCTYPE html> +<html lang="en"> +<head> + <meta charset="utf-8"> + <title>MeshBay Chat</title> + <style> + body {{ font-family: monospace; max-width: 700px; margin: 40px auto; padding: 0 20px; }} + #messages {{ border: 1px solid #ccc; padding: 10px; height: 400px; overflow-y: auto; + background: #fafafa; margin-bottom: 10px; }} + .msg {{ margin: 4px 0; }} + .sender {{ font-weight: bold; color: #2563eb; }} + .time {{ color: #9ca3af; font-size: 0.8em; }} + </style> +</head> +<body> + <h1>MeshBay Chat</h1> + <div id="messages"></div> + <p><a href="/">Back to status</a></p> + <script> + const box = document.getElementById('messages'); + const ws = new WebSocket('ws://' + location.host + '/ws/chat'); + ws.onmessage = (e) => {{ + const msg = JSON.parse(e.data); + const div = document.createElement('div'); + div.className = 'msg'; + const t = new Date(msg.timestamp * 1000).toLocaleTimeString(); + div.innerHTML = '<span class="time">' + t + '</span> ' + + '<span class="sender">' + msg.sender_id + '</span>: ' + + '(encrypted message #' + msg.iteration + ')'; + box.appendChild(div); + box.scrollTop = box.scrollHeight; + }}; + </script> </body> </html>""" diff --git a/packages/meshbay-node/tests/test_chat_store.py b/packages/meshbay-node/tests/test_chat_store.py new file mode 100644 index 0000000..d74310d --- /dev/null +++ b/packages/meshbay-node/tests/test_chat_store.py @@ -0,0 +1,91 @@ +""" +Tests for the SQLite-backed chat message store. +""" + +import pytest +import pytest_asyncio +from pathlib import Path + +from meshbay_node.chat.store import ChatStore + + +@pytest_asyncio.fixture +async def store(tmp_path): + s = ChatStore(db_path=tmp_path / "test_chat.db") + await s.open() + yield s + await s.close() + + +@pytest.mark.asyncio +async def test_save_and_retrieve(store): + row_id = await store.save_message( + sender_id="alice", iteration=0, payload=b"hello", + ) + assert row_id == 1 + + msgs = await store.get_messages() + assert len(msgs) == 1 + assert msgs[0].sender_id == "alice" + assert msgs[0].iteration == 0 + assert msgs[0].payload == b"hello" + assert msgs[0].thread_id is None + + +@pytest.mark.asyncio +async def test_message_count(store): + assert await store.message_count() == 0 + await store.save_message("alice", 0, b"msg1") + await store.save_message("bob", 1, b"msg2") + assert await store.message_count() == 2 + + +@pytest.mark.asyncio +async def test_get_messages_since(store): + await store.save_message("alice", 0, b"old") + all_msgs = await store.get_messages() + cutoff = all_msgs[0].timestamp + await store.save_message("bob", 1, b"new") + + msgs = await store.get_messages(since=cutoff) + assert len(msgs) == 1 + assert msgs[0].sender_id == "bob" + + +@pytest.mark.asyncio +async def test_thread_messages(store): + await store.save_message("alice", 0, b"root", thread_id="t1") + await store.save_message("bob", 1, b"reply", thread_id="t1") + await store.save_message("carol", 2, b"other") + + thread = await store.get_thread("t1") + assert len(thread) == 2 + assert thread[0].sender_id == "alice" + assert thread[1].sender_id == "bob" + + +@pytest.mark.asyncio +async def test_message_ordering(store): + for i in range(5): + await store.save_message(f"user-{i}", i, f"msg-{i}".encode()) + + msgs = await store.get_messages() + assert len(msgs) == 5 + for i, m in enumerate(msgs): + assert m.sender_id == f"user-{i}" + + +@pytest.mark.asyncio +async def test_limit(store): + for i in range(10): + await store.save_message("alice", i, f"msg-{i}".encode()) + + msgs = await store.get_messages(limit=3) + assert len(msgs) == 3 + + +@pytest.mark.asyncio +async def test_context_manager(tmp_path): + async with ChatStore(db_path=tmp_path / "ctx_test.db") as store: + await store.save_message("alice", 0, b"test") + assert await store.message_count() == 1 diff --git a/packages/meshbay-node/tests/test_multi_group.py b/packages/meshbay-node/tests/test_multi_group.py new file mode 100644 index 0000000..9be8d47 --- /dev/null +++ b/packages/meshbay-node/tests/test_multi_group.py @@ -0,0 +1,169 @@ +""" +Multi-group isolation test: two groups on one QUIC server. + +Verifies that: + - A user in group-a can fetch files from group-a + - A user in group-a is rejected when requesting group-b + - A user in both groups can access both +""" + +import os +import time +import jwt +import pytest +from pathlib import Path +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives import serialization + +from meshbay_common.crypto import generate_gek, pk_to_b64 +from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from meshbay_node.transport.quic_server import QuicChunkServer +from meshbay_node.transport.quic_client import QuicChunkClient + + +@pytest.fixture +def sk_node(): + return Ed25519PrivateKey.generate() + +@pytest.fixture +def sk_hub(): + return Ed25519PrivateKey.generate() + +@pytest.fixture +def gek_a(): + return generate_gek() + +@pytest.fixture +def gek_b(): + return generate_gek() + +@pytest.fixture +def dir_a(tmp_path): + d = tmp_path / "group_a" + d.mkdir() + (d / "file_a.txt").write_bytes(b"content from group A " * 100) + return d + +@pytest.fixture +def dir_b(tmp_path): + d = tmp_path / "group_b" + d.mkdir() + (d / "file_b.txt").write_bytes(b"content from group B " * 100) + return d + + +def make_jwt(sk_hub, pk_node_b64, groups, user_id="user-001", ttl=3600): + sk_pem = sk_hub.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + now = int(time.time()) + return jwt.encode({ + "iss": "test-hub", "sub": user_id, + "pk_user": pk_node_b64, "hub_id": "test-hub", + "jti": "test-jti", "iat": now, "exp": now + ttl, + "groups": groups, + }, sk_pem, algorithm="EdDSA") + + +@pytest.fixture +async def multi_group_server(sk_node, sk_hub, gek_a, gek_b, dir_a, dir_b, tmp_path): + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer_a = DirectoryIndexer(root=dir_a, group_id="group-a", sk_node=sk_node, gek=gek_a) + await indexer_a.initial_scan() + + indexer_b = DirectoryIndexer(root=dir_b, group_id="group-b", sk_node=sk_node, gek=gek_b) + await indexer_b.initial_scan() + + groups = { + "group-a": {"gek": gek_a, "shared_root": dir_a, "index": indexer_a.index}, + "group-b": {"gek": gek_b, "shared_root": dir_b, "index": indexer_b.index}, + } + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + server = QuicChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, + gek=gek_a, shared_root=dir_a, index=indexer_a.index, + host="127.0.0.1", port=19200, + cert_path=cert_path, key_path=key_path, + groups=groups, + ) + await server.start() + yield server, indexer_a, indexer_b + await server.stop() + + +@pytest.mark.asyncio +async def test_user_can_access_own_group( + multi_group_server, sk_node, sk_hub, gek_a, +): + """User in group-a can fetch index and chunks from group-a.""" + server, indexer_a, _ = multi_group_server + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) + + async with QuicChunkClient( + host="127.0.0.1", port=19200, + jwt_token=token, gek=gek_a, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-a", + ) as client: + wire = await client.fetch_index() + recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek_a) + assert recovered.count == 1 + + entry = recovered.entries[0] + assert entry.name == "file_a.txt" + chunk = await client.fetch_chunk(entry.id, chunk_index=0) + assert chunk == b"content from group A " * 100 + + +@pytest.mark.asyncio +async def test_user_rejected_from_other_group( + multi_group_server, sk_node, sk_hub, gek_b, +): + """User in group-a only is rejected when requesting group-b.""" + server, _, _ = multi_group_server + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) + + with pytest.raises(ConnectionError, match="rejected"): + async with QuicChunkClient( + host="127.0.0.1", port=19200, + jwt_token=token, gek=gek_b, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-b", + ) as client: + await client.fetch_index() + + +@pytest.mark.asyncio +async def test_dual_group_user_accesses_both( + multi_group_server, sk_node, sk_hub, gek_a, gek_b, +): + """User in both groups can access either group's files.""" + server, indexer_a, indexer_b = multi_group_server + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a", "group-b"]) + + async with QuicChunkClient( + host="127.0.0.1", port=19200, + jwt_token=token, gek=gek_a, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-a", + ) as client_a: + wire_a = await client_a.fetch_index() + idx_a = GroupIndex.deserialize(wire_a, sk_node=sk_node, gek=gek_a) + assert idx_a.entries[0].name == "file_a.txt" + + async with QuicChunkClient( + host="127.0.0.1", port=19200, + jwt_token=token, gek=gek_b, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-b", + ) as client_b: + wire_b = await client_b.fetch_index() + idx_b = GroupIndex.deserialize(wire_b, sk_node=sk_node, gek=gek_b) + assert idx_b.entries[0].name == "file_b.txt" diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py index 2abd465..0c1a1cd 100644 --- a/packages/meshbay-node/tests/test_quic_transport.py +++ b/packages/meshbay-node/tests/test_quic_transport.py @@ -14,7 +14,7 @@ from cryptography.hazmat.primitives import serialization from meshbay_common.crypto import generate_gek, pk_to_b64 from meshbay_node.indexer import DirectoryIndexer, GroupIndex -from meshbay_node.transport.quic_server import QuicChunkServer +from meshbay_node.transport.quic_server import QuicChunkServer, Denylist from meshbay_node.transport.quic_client import QuicChunkClient @@ -38,7 +38,7 @@ def shared_dir(tmp_path): (d / "small.txt").write_bytes(b"hello quic " * 100) return d -def make_jwt(sk_hub, pk_node_b64, ttl=3600): +def make_jwt(sk_hub, pk_node_b64, ttl=3600, groups=None): sk_pem = sk_hub.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, @@ -49,6 +49,7 @@ def make_jwt(sk_hub, pk_node_b64, ttl=3600): "iss": "test-hub", "sub": "user-001", "pk_user": pk_node_b64, "hub_id": "test-hub", "jti": "test-jti", "iat": now, "exp": now + ttl, + "groups": groups or [], }, sk_pem, algorithm="EdDSA") @@ -155,3 +156,132 @@ async def test_quic_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p await client.fetch_index() await server.stop() + + +@pytest.mark.asyncio +async def test_quic_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path): + """QUIC server rejects a client whose JWT groups don't include the requested group_id.""" + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + server = QuicChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + host="127.0.0.1", port=19103, + cert_path=cert_path, key_path=key_path, + ) + await server.start() + + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) + + with pytest.raises(ConnectionError, match="rejected"): + async with QuicChunkClient( + host="127.0.0.1", port=19103, + jwt_token=token, gek=gek, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-b", + ) as client: + await client.fetch_index() + + await server.stop() + + +@pytest.mark.asyncio +async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_path): + """QUIC 0-RTT: connect, save session ticket, reconnect with ticket.""" + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + server = QuicChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + host="127.0.0.1", port=19104, + cert_path=cert_path, key_path=key_path, + ) + await server.start() + + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key())) + pk_b64 = pk_to_b64(sk_node.public_key()) + + # First connection — captures session ticket + saved_ticket = None + async with QuicChunkClient( + host="127.0.0.1", port=19104, + jwt_token=token, gek=gek, pk_node_b64=pk_b64, + ) as client: + wire = await client.fetch_index() + assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + saved_ticket = client.session_ticket + + # Allow server to process the close + await asyncio.sleep(0.1) + + # Second connection — reuses session ticket (0-RTT) + async with QuicChunkClient( + host="127.0.0.1", port=19104, + jwt_token=token, gek=gek, pk_node_b64=pk_b64, + session_ticket=saved_ticket, + ) as client: + wire = await client.fetch_index() + assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + + await server.stop() + + +@pytest.mark.asyncio +async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_path): + """QUIC server rejects a connection when the user is on the denylist.""" + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + denylist = Denylist() + server = QuicChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + host="127.0.0.1", port=19105, + cert_path=cert_path, key_path=key_path, + denylist=denylist, + ) + await server.start() + + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key())) + pk_b64 = pk_to_b64(sk_node.public_key()) + + # Connection works before denylisting + async with QuicChunkClient( + host="127.0.0.1", port=19105, + jwt_token=token, gek=gek, pk_node_b64=pk_b64, + ) as client: + wire = await client.fetch_index() + assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + + # Add user to denylist + denylist.deny_user("user-001") + + # Connection now rejected + with pytest.raises(Exception): + async with QuicChunkClient( + host="127.0.0.1", port=19105, + jwt_token=token, gek=gek, pk_node_b64=pk_b64, + ) as client: + await client.fetch_index() + + await server.stop() diff --git a/packages/meshbay-node/tests/test_transport.py b/packages/meshbay-node/tests/test_transport.py index 2064ba5..0e70d72 100644 --- a/packages/meshbay-node/tests/test_transport.py +++ b/packages/meshbay-node/tests/test_transport.py @@ -41,7 +41,7 @@ def shared_dir(tmp_path): (d / "small.txt").write_bytes(b"hello meshbay " * 100) return d -def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600): +def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600, groups=None): sk_pem = sk_hub.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, @@ -53,6 +53,7 @@ def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600): "pk_user": pk_node_b64, "hub_id": "test-hub", "jti": "test-jti", "iat": now, "exp": now + ttl, + "groups": groups or [], }, sk_pem, algorithm="EdDSA") @@ -154,6 +155,42 @@ async def test_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path): @pytest.mark.asyncio +async def test_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path): + """TCP+TLS server rejects a client whose JWT groups don't include the requested group_id.""" + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer = DirectoryIndexer(root=shared_dir, group_id="g", + sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + server = ChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + host="127.0.0.1", port=0, + cert_path=cert_path, key_path=key_path, + ) + await server.start() + port = server._server.sockets[0].getsockname()[1] + + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) + + with pytest.raises(ConnectionError, match="rejected"): + async with ChunkClient( + host="127.0.0.1", port=port, + jwt_token=token, gek=gek, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-b", + ) as client: + pass + + await server.stop() + + +@pytest.mark.asyncio async def test_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path): hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) |