From e23e33adeaf8ee7439187d4451c856b37816a51f Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 11 Aug 2026 04:13:53 +0200 Subject: feat: Phase 9 — Web client SPA with WebRTC P2P transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Complete browser-based client: Preact SPA with login, group file browser, encrypted download, video playback, group chat, i18n, and dark/light theme. Browser connects P2P to nodes behind residential NAT via WebRTC DataChannel (aiortc). Hub handles signaling only — all data flows E2E. Performance: pipelined downloads (8-chunk sliding window), binary msgpack wire format (no base64), redundant I/O elimination. Large file downloads stream to disk via File System Access API (showSaveFilePicker). Validated on SFR + Orange residential NATs, Chrome + Firefox, IPv4/IPv6. 132 tests passing. Deployed to meshbay.org + Orange node. Co-Authored-By: Claude Opus 4.6 --- .../meshbay-node/src/meshbay_node/hub_client.py | 8 +- .../src/meshbay_node/indexer/group_index.py | 8 +- .../src/meshbay_node/indexer/indexer.py | 51 ++++++--- .../src/meshbay_node/transport/http_server.py | 5 +- .../src/meshbay_node/transport/quic_client.py | 7 +- .../src/meshbay_node/transport/quic_server.py | 7 +- .../src/meshbay_node/transport/server.py | 2 +- .../src/meshbay_node/transport/webrtc_server.py | 125 +++++++++++++++++---- 8 files changed, 157 insertions(+), 56 deletions(-) (limited to 'packages/meshbay-node/src/meshbay_node') diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index 3e77ed0..a9a1d6c 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -251,6 +251,7 @@ class HubClient: on_incoming: Any = None, on_revocation: Any = None, on_webrtc_offer: Any = None, + group_ids: list[str] | None = None, ) -> None: """ Maintain a persistent WebSocket connection to the hub. @@ -268,11 +269,14 @@ class HubClient: while True: try: async with websockets.connect(ws_url) as ws: - await ws.send(json.dumps({ + auth_msg = { "type": "auth", "token": self._session.access_token, "node_id": self._session.node_id, - })) + } + if group_ids: + auth_msg["group_ids"] = group_ids + await ws.send(json.dumps(auth_msg)) auth_resp = json.loads(await ws.recv()) if auth_resp.get("type") != "auth_ok": log.error("WS auth failed: %s", auth_resp) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py index 4edeeab..a69429c 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py @@ -25,14 +25,16 @@ import zstandard as zstd from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import ( - chunk_key as derive_chunk_key, - encrypt_chunk, - decrypt_chunk, sign_chunk, verify_chunk_signature, pk_to_b64, generate_gek, ) +from meshbay_common.webcrypto import ( + chunk_key_aes as derive_chunk_key, + encrypt_chunk_aes as encrypt_chunk, + decrypt_chunk_aes as decrypt_chunk, +) from meshbay_common.protocol import IndexEntry, IndexDelta log = logging.getLogger(__name__) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 4bb1527..60dc04b 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -169,30 +169,49 @@ class DirectoryIndexer: # ── Internal update ─────────────────────────────────────────────────────── + _DEBOUNCE_SECS = 2.0 + def _schedule_update(self, file_path: Path, deleted: bool = False) -> None: - """Called from watchdog thread — schedule async update on the event loop.""" - if self._loop: - self._loop.call_soon_threadsafe( - lambda: asyncio.ensure_future( - self._update_entry(file_path, deleted))) + """Called from watchdog thread — schedule debounced async update.""" + if not self._loop: + return + key = str(file_path.resolve()) + self._loop.call_soon_threadsafe( + self._debounce, key, file_path, deleted) + + def _debounce(self, key: str, file_path: Path, deleted: bool) -> None: + if not hasattr(self, "_pending_timers"): + self._pending_timers: dict[str, asyncio.TimerHandle] = {} + old = self._pending_timers.pop(key, None) + if old: + old.cancel() + handle = self._loop.call_later( + self._DEBOUNCE_SECS, + lambda: asyncio.ensure_future(self._update_entry(file_path, deleted)), + ) + self._pending_timers[key] = handle + + def _remove_by_path(self, file_path: Path) -> None: + """Remove any existing entries that match this file's path + name.""" + resolved = file_path.resolve() + to_remove = [ + e.id for e in self._index.entries + if (self.root / e.path / e.name).resolve() == resolved + ] + for fid in to_remove: + self._index.remove_entry(fid) async def _update_entry(self, file_path: Path, deleted: bool) -> None: - if deleted: - # Remove by matching path (hash not available after deletion) - to_remove = [ - e.id for e in self._index.entries - if (self.root / e.path / e.name).resolve() == file_path.resolve() - ] - for fid in to_remove: - self._index.remove_entry(fid) - log.debug("Removed from index: %s", file_path.name) - else: + self._remove_by_path(file_path) + + if not deleted: loop = asyncio.get_event_loop() entry = await loop.run_in_executor( self._executor, _scan_file, self.root, file_path) if entry: self._index.add_entry(entry) - log.debug("Indexed: %s (%s)", file_path.name, entry.id[:8]) + log.debug("Indexed: %s (%s, %d bytes)", + file_path.name, entry.id[:8], entry.size) self._index.version = int(time.time()) if self.on_change: diff --git a/packages/meshbay-node/src/meshbay_node/transport/http_server.py b/packages/meshbay-node/src/meshbay_node/transport/http_server.py index 7554086..151c2e8 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/http_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/http_server.py @@ -37,7 +37,8 @@ from fastapi.responses import FileResponse, JSONResponse, StreamingResponse from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION -from meshbay_common.crypto import chunk_key as derive_chunk_key, encrypt_chunk, sign_chunk, pk_to_b64 +from meshbay_common.crypto import sign_chunk, pk_to_b64 +from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk from meshbay_node import __version__ from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.group_index import GroupIndex @@ -189,7 +190,7 @@ def create_http_app( if not plaintext: raise HTTPException(status_code=416, detail="Chunk out of range") - file_hash = blake3.blake3(file_path.read_bytes()).digest() + file_hash = bytes.fromhex(entry.id) pt_hash = blake3.blake3(plaintext).digest() if gek: 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 83b729e..288465f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py @@ -23,11 +23,8 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from meshbay_common import MNP_VERSION -from meshbay_common.crypto import ( - chunk_key as derive_chunk_key, - decrypt_chunk, - verify_chunk_signature, -) +from meshbay_common.crypto import verify_chunk_signature +from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk from meshbay_common.protocol import MNP log = logging.getLogger(__name__) 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 43c1026..f439e62 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -35,11 +35,10 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION from meshbay_common.crypto import ( - chunk_key as derive_chunk_key, - encrypt_chunk, sign_chunk, pk_to_b64, ) +from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex from meshbay_node.transport.tls_cert import server_ssl_context @@ -217,11 +216,13 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return + file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( self._ctx["sk_node"], ctx["gek"], file_path, chunk_index, + file_hash, ) self._send(stream_id, chunk_data) @@ -304,13 +305,13 @@ def _read_and_encrypt( gek: bytes, file_path: Path, chunk_index: int, + file_hash: bytes, ) -> dict: """Read and encrypt one chunk (blocking — runs in executor).""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) - file_hash = blake3.blake3(file_path.read_bytes()).digest() pt_hash = blake3.blake3(plaintext).digest() ckey = derive_chunk_key(gek, file_hash, chunk_index) nonce, ct = encrypt_chunk(ckey, plaintext) diff --git a/packages/meshbay-node/src/meshbay_node/transport/server.py b/packages/meshbay-node/src/meshbay_node/transport/server.py index 76ac13a..b77f1f2 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/server.py @@ -197,7 +197,7 @@ class _ConnectionHandler: "type": "error", "detail": "File not on disk"}) continue - file_hash = blake3.blake3(file_path.read_bytes()).digest() + file_hash = bytes.fromhex(entry.id) chunk = await loop.run_in_executor( None, _serve_chunk, self._sk_node, self._gek, file_path, file_hash, chunk_index) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 89391b9..7da6623 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -29,19 +29,14 @@ import struct from pathlib import Path from typing import Any -import blake3 import jwt import msgpack from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION -from meshbay_common.crypto import ( - chunk_key as derive_chunk_key, - encrypt_chunk, - sign_chunk, - pk_to_b64, -) +from meshbay_common.crypto import pk_to_b64 +from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex @@ -101,6 +96,7 @@ class WebRTCPeerSession: def _handle_message(self, msg: dict) -> None: mtype = msg.get("type") + log.debug("WebRTC recv: %s", mtype) try: if mtype == MNP.HANDSHAKE: self._do_handshake(msg) @@ -112,8 +108,12 @@ class WebRTCPeerSession: self._do_file_request(msg) elif mtype == MNP.STREAM_SEGMENT: self._do_stream_segment(msg) + elif mtype == MNP.GEK_REQUEST: + self._do_gek_request() elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message(msg) + elif mtype == MNP.CHAT_HISTORY: + self._do_chat_history(msg) else: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: @@ -145,6 +145,10 @@ class WebRTCPeerSession: self._user_id = decoded["sub"] self._group_id = group_id + peers = self._ctx.get("_peers") + if peers is not None: + peers[self._user_id] = self + log.info("WebRTC handshake OK — user=%s group=%s", self._user_id[:8], group_id[:8] if group_id else "none") self._send({ @@ -160,11 +164,32 @@ class WebRTCPeerSession: def _do_index_sync(self) -> None: ctx = self._group_ctx() - wire = ctx["index"].serialize() + idx = ctx["index"] + entries = [ + { + "id": e.id, "name": e.name, "path": e.path, + "size": e.size, "type": e.type, "added_at": e.added_at, + } + for e in idx.entries + ] self._send({ "type": MNP.INDEX_SYNC, "v": MNP_VERSION, - "index_b64": base64.b64encode(wire).decode(), + "group_id": idx.group_id, + "version": idx.version, + "entries": entries, + }) + + def _do_gek_request(self) -> None: + ctx = self._group_ctx() + gek = ctx.get("gek") + if not gek: + self._send({"type": "error", "detail": "No GEK available"}) + return + self._send({ + "type": MNP.GEK_RESPONSE, + "v": MNP_VERSION, + "gek_b64": base64.b64encode(gek).decode(), }) def _do_file_request(self, msg: dict) -> None: @@ -173,6 +198,7 @@ class WebRTCPeerSession: chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) if not entry: + log.warning("File not found: %s", file_id[:16]) self._send({"type": "error", "detail": "File not found"}) return @@ -181,11 +207,13 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not on disk"}) return + file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( self._ctx["sk_node"], ctx["gek"], file_path, chunk_index, + file_hash, ) self._send(chunk_data) @@ -235,21 +263,77 @@ class WebRTCPeerSession: def _do_chat_message(self, msg: dict) -> None: chat_store = self._ctx.get("chat_store") + payload = msg.get("payload", "") if chat_store: + raw = payload.encode() if isinstance(payload, str) else payload 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""), + payload=raw, 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), + "payload": payload, + "thread_id": msg.get("thread_id"), + "timestamp": __import__("time").time(), + } + for uid, session in peers.items(): + if uid != self._user_id and session is not self: + try: + session._send(broadcast) + except Exception: + pass + self._send({"type": "ack", "v": MNP_VERSION}) + def _do_chat_history(self, msg: dict) -> None: + chat_store = self._ctx.get("chat_store") + if not chat_store: + self._send({ + "type": MNP.CHAT_HISTORY_RESPONSE, + "v": MNP_VERSION, + "messages": [], + }) + return + + since = msg.get("since", 0) + limit = msg.get("limit", 100) + asyncio.ensure_future(self._send_chat_history(chat_store, since, limit)) + + async def _send_chat_history(self, chat_store, since: float, limit: int) -> None: + msgs = await chat_store.get_messages(since=since, limit=limit) + self._send({ + "type": MNP.CHAT_HISTORY_RESPONSE, + "v": MNP_VERSION, + "messages": [ + { + "id": m.id, + "sender_id": m.sender_id, + "payload": m.payload.decode("utf-8", errors="replace") + if isinstance(m.payload, bytes) else m.payload, + "timestamp": m.timestamp, + "thread_id": m.thread_id, + } + for m in msgs + ], + }) + def _send(self, obj: dict) -> None: if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) + else: + log.warning("WebRTC send skipped: channel=%s", + self._channel.readyState if self._channel else "none") async def close(self) -> None: + peers = self._ctx.get("_peers") + if peers and self._user_id: + peers.pop(self._user_id, None) await self._pc.close() @@ -258,30 +342,22 @@ def _read_and_encrypt( gek: bytes, file_path: Path, chunk_index: int, + file_hash: bytes, ) -> dict: with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) - file_hash = blake3.blake3(file_path.read_bytes()).digest() - pt_hash = blake3.blake3(plaintext).digest() - ckey = derive_chunk_key(gek, file_hash, chunk_index) - nonce, ct = encrypt_chunk(ckey, plaintext) - ct_hash = blake3.blake3(ct).digest() - sig = sign_chunk(sk_node, chunk_index, nonce, ct_hash) + ckey = chunk_key_aes(gek, file_hash, chunk_index) + nonce, ct = encrypt_chunk_aes(ckey, plaintext) return { "type": MNP.FILE_CHUNK, "v": MNP_VERSION, "chunk_index": chunk_index, "plaintext_size": len(plaintext), - "nonce_b64": base64.b64encode(nonce).decode(), - "ct_b64": base64.b64encode(ct).decode(), - "ct_hash_b64": base64.b64encode(ct_hash).decode(), - "pt_hash_b64": base64.b64encode(pt_hash).decode(), - "sig_b64": base64.b64encode(sig).decode(), - "pk_node_b64": pk_to_b64(sk_node.public_key()), - "file_hash_b64": base64.b64encode(file_hash).decode(), + "nonce": nonce, + "ct": ct, } @@ -312,6 +388,7 @@ class WebRTCTransport: "gek": gek, "shared_root": shared_root, "index": index, + "_peers": {}, } if groups: self._ctx["groups"] = groups -- cgit v1.2.3