From 60c4570e72e36c2a9720593c8baec74ee2ab52d6 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Mon, 10 Aug 2026 22:12:59 +0200 Subject: feat: Phase 9.1–9.5 — WebRTC DataChannel transport for browser P2P MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Browser clients can now connect P2P to nodes behind residential NAT via WebRTC DataChannel with ICE/STUN. Validated on SFR Port-Restricted Cone NAT + 4G CGNAT across three scenarios (WiFi LAN, 4G IPv6, 4G IPv4 STUN). No TURN relay needed. Hub serves only as signaling relay (<1 KB). New files: - webrtc_server.py: aiortc-based WebRTC transport (node side) - signaling.py: SDP/ICE relay endpoint (hub side) - transport.js: browser WebRTC client with msgpack framing - webrtc-test.html: spike test page for browser→NAT→node validation - test_webrtc_transport.py: 4 tests (handshake, file transfer, auth, guard) - meshbay-draft-v4.md: architecture spec updated for web client Modified: - hub_client.py: WebRTC offer handling via hub WebSocket - revocation.py: node_id from WS auth + webrtc_answer routing - pyproject.toml: aiortc>=1.9 dependency 123 tests passing (117 existing + 6 new). Co-Authored-By: Claude Opus 4.6 --- .../meshbay-hub/src/meshbay_hub/api/revocation.py | 5 +- .../meshbay-hub/src/meshbay_hub/api/signaling.py | 102 +++++ packages/meshbay-hub/src/meshbay_hub/api/webapp.py | 15 + packages/meshbay-hub/src/meshbay_hub/app.py | 2 + .../src/meshbay_hub/static/transport.js | 409 +++++++++++++++++++++ .../src/meshbay_hub/static/webrtc-test.html | 258 +++++++++++++ packages/meshbay-hub/tests/test_hub_api.py | 70 ++++ packages/meshbay-node/pyproject.toml | 1 + .../meshbay-node/src/meshbay_node/hub_client.py | 16 +- .../src/meshbay_node/transport/__init__.py | 13 +- .../src/meshbay_node/transport/webrtc_server.py | 373 +++++++++++++++++++ .../meshbay-node/tests/test_webrtc_transport.py | 326 ++++++++++++++++ 12 files changed, 1586 insertions(+), 4 deletions(-) create mode 100644 packages/meshbay-hub/src/meshbay_hub/api/signaling.py create mode 100644 packages/meshbay-hub/src/meshbay_hub/static/transport.js create mode 100644 packages/meshbay-hub/src/meshbay_hub/static/webrtc-test.html create mode 100644 packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py create mode 100644 packages/meshbay-node/tests/test_webrtc_transport.py (limited to 'packages') diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py index bbd1bc2..8f30745 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py @@ -119,7 +119,7 @@ async def node_websocket(ws: WebSocket): await ws.close(code=4001) return - node_id = decoded.get("sub", "unknown") + node_id = msg.get("node_id") or decoded.get("sub", "unknown") _connected_nodes[node_id] = ws log.info("Node WS connected: %s", node_id[:8]) await ws.send_text(json.dumps({"type": "auth_ok", "node_id": node_id})) @@ -134,6 +134,9 @@ async def node_websocket(ws: WebSocket): event = _punch_events.get(node_id) if event: event.set() + elif msg.get("type") == "webrtc_answer": + from meshbay_hub.api.signaling import handle_webrtc_answer + handle_webrtc_answer(msg) except WebSocketDisconnect: log.info("Node WS disconnected: %s", (node_id or "unknown")[:8]) diff --git a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py new file mode 100644 index 0000000..bd343c9 --- /dev/null +++ b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py @@ -0,0 +1,102 @@ +""" +WebRTC signaling — relay SDP/ICE between browser and node. + +The hub NEVER touches content. This is pure signaling: < 1 KB per message, +stateless relay. After the SDP exchange completes, the browser and node +communicate P2P via WebRTC DataChannel — hub is out of the loop. + +Flow: + Browser → Hub : POST /v1/nodes/{node_id}/webrtc/offer {sdp, ice_candidates} + Hub → Node : WS push {type: "webrtc_offer", sdp, ice_candidates, peer_id} + Node → Hub : WS reply {type: "webrtc_answer", sdp, ice_candidates, peer_id} + Hub → Browser : HTTP response {sdp, ice_candidates} +""" + +import asyncio +import json +import logging +import uuid + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel + +from meshbay_hub.api.deps import get_current_user +from meshbay_hub.db.models import User + +log = logging.getLogger(__name__) + +router = APIRouter(prefix="/v1/nodes", tags=["signaling"]) + +_webrtc_answers: dict[str, asyncio.Future] = {} + + +class WebRTCOfferRequest(BaseModel): + sdp: str + ice_candidates: list[dict] = [] + + +class WebRTCOfferResponse(BaseModel): + sdp: str + ice_candidates: list[dict] = [] + peer_id: str + + +@router.post("/{node_id}/webrtc/offer", response_model=WebRTCOfferResponse) +async def webrtc_offer( + node_id: str, + body: WebRTCOfferRequest, + current_user: User = Depends(get_current_user), +): + """ + Browser sends WebRTC SDP offer for a node. Hub relays via WebSocket. + Returns the node's SDP answer once received. + """ + from meshbay_hub.api.revocation import _connected_nodes + + ws = _connected_nodes.get(node_id) + if not ws: + raise HTTPException(status_code=404, detail="Node not connected") + + peer_id = str(uuid.uuid4()) + answer_future: asyncio.Future = asyncio.get_event_loop().create_future() + _webrtc_answers[peer_id] = answer_future + + try: + await ws.send_text(json.dumps({ + "type": "webrtc_offer", + "peer_id": peer_id, + "user_id": current_user.id, + "sdp": body.sdp, + "ice_candidates": body.ice_candidates, + })) + + try: + answer = await asyncio.wait_for(answer_future, timeout=15.0) + except asyncio.TimeoutError: + raise HTTPException( + status_code=504, detail="Node did not respond with WebRTC answer") + + return WebRTCOfferResponse( + sdp=answer["sdp"], + ice_candidates=answer.get("ice_candidates", []), + peer_id=peer_id, + ) + finally: + _webrtc_answers.pop(peer_id, None) + + +def handle_webrtc_answer(msg: dict) -> None: + """Called from the node WebSocket message loop when a webrtc_answer arrives.""" + peer_id = msg.get("peer_id") + if not peer_id: + log.warning("webrtc_answer without peer_id") + return + + future = _webrtc_answers.get(peer_id) + if future and not future.done(): + future.set_result({ + "sdp": msg.get("sdp", ""), + "ice_candidates": msg.get("ice_candidates", []), + }) + else: + log.warning("webrtc_answer for unknown peer_id: %s", peer_id) diff --git a/packages/meshbay-hub/src/meshbay_hub/api/webapp.py b/packages/meshbay-hub/src/meshbay_hub/api/webapp.py index 3d5e78b..6005927 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/webapp.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/webapp.py @@ -27,6 +27,21 @@ async def app_js(): return FileResponse(STATIC_DIR / "app.js", media_type="application/javascript") +@router.get("/transport.js") +async def transport_js(): + return FileResponse(STATIC_DIR / "transport.js", media_type="application/javascript") + + +@router.get("/crypto.js") +async def crypto_js(): + return FileResponse(STATIC_DIR / "crypto.js", media_type="application/javascript") + + +@router.get("/webrtc-test.html") +async def webrtc_test(): + return FileResponse(STATIC_DIR / "webrtc-test.html", media_type="text/html") + + @router.get("/", response_class=HTMLResponse) async def index(): return HTMLResponse(_HTML) diff --git a/packages/meshbay-hub/src/meshbay_hub/app.py b/packages/meshbay-hub/src/meshbay_hub/app.py index c80da28..7bd5ee3 100644 --- a/packages/meshbay-hub/src/meshbay_hub/app.py +++ b/packages/meshbay-hub/src/meshbay_hub/app.py @@ -32,6 +32,7 @@ from meshbay_hub.api.federation import router as federation_router from meshbay_hub.csam import csam_router from meshbay_hub.api.health import router as health_router from meshbay_hub.api.relay import router as relay_router +from meshbay_hub.api.signaling import router as signaling_router from meshbay_hub.api.webapp import router as webapp_router from meshbay_hub.api.middleware import limiter @@ -93,6 +94,7 @@ def create_app(cfg: HubConfig | None = None) -> FastAPI: app.include_router(csam_router) app.include_router(health_router) app.include_router(relay_router) + app.include_router(signaling_router) app.include_router(webapp_router) return app diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js new file mode 100644 index 0000000..9112734 --- /dev/null +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -0,0 +1,409 @@ +/** + * MeshBay Browser Transport — WebRTC DataChannel client. + * + * Connects to a MeshBay node via WebRTC DataChannel (P2P, E2E). + * The hub is only used for signaling (SDP/ICE relay) — after connection, + * all data flows directly between browser and node. + * + * Wire format: length-prefixed msgpack (4-byte big-endian + msgpack payload). + * Same format as QUIC and TCP+TLS transports on the node side. + * + * Usage: + * const transport = new MeshBayTransport(hubUrl, accessToken); + * await transport.connect(nodeId, jwtToken, groupId); + * const index = await transport.fetchIndex(); + * const chunk = await transport.fetchChunk(fileId, 0); + * transport.close(); + */ + +class MeshBayTransport { + constructor(hubUrl, accessToken) { + this._hubUrl = hubUrl; + this._accessToken = accessToken; + this._pc = null; + this._channel = null; + this._pending = new Map(); + this._seqId = 0; + this._recvBuf = new Uint8Array(0); + this._connected = false; + this._onChat = null; + } + + get connected() { return this._connected; } + + set onChat(fn) { this._onChat = fn; } + + async connect(nodeId, jwtToken, groupId) { + this._pc = new RTCPeerConnection({ + iceServers: [{ urls: 'stun:stun.l.google.com:19302' }], + }); + + this._channel = this._pc.createDataChannel('mnp', { ordered: true }); + this._channel.binaryType = 'arraybuffer'; + + const channelReady = new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error('DataChannel open timeout')), 30000); + this._channel.onopen = () => { + clearTimeout(timeout); + this._connected = true; + resolve(); + }; + this._channel.onerror = (e) => { + clearTimeout(timeout); + reject(new Error('DataChannel error: ' + e.message)); + }; + }); + + this._channel.onmessage = (event) => this._onMessage(event.data); + this._channel.onclose = () => { this._connected = false; }; + + const offer = await this._pc.createOffer(); + await this._pc.setLocalDescription(offer); + + await new Promise((resolve) => { + if (this._pc.iceGatheringState === 'complete') return resolve(); + this._pc.onicegatheringstatechange = () => { + if (this._pc.iceGatheringState === 'complete') resolve(); + }; + }); + + const resp = await fetch(`${this._hubUrl}/v1/nodes/${nodeId}/webrtc/offer`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Authorization': `Bearer ${this._accessToken}`, + }, + body: JSON.stringify({ + sdp: this._pc.localDescription.sdp, + ice_candidates: [], + }), + }); + + if (!resp.ok) { + const detail = await resp.json().catch(() => ({})); + throw new Error(`Signaling failed: ${resp.status} ${detail.detail || ''}`); + } + + const answer = await resp.json(); + await this._pc.setRemoteDescription({ type: 'answer', sdp: answer.sdp }); + + await channelReady; + + const ack = await this._sendAndWait({ + type: 'handshake', + v: '0.1', + token: jwtToken, + group_id: groupId || '', + }); + + if (ack.type !== 'handshake_ack') { + throw new Error('MNP handshake rejected: ' + (ack.detail || JSON.stringify(ack))); + } + + return ack; + } + + async fetchIndex() { + const msg = await this._sendAndWait({ type: 'index_sync', v: '0.1' }); + if (msg.type === 'error') throw new Error(msg.detail); + return _b64decode(msg.index_b64); + } + + async fetchChunk(fileId, chunkIndex) { + const msg = await this._sendAndWait({ + type: 'file_req', + v: '0.1', + file_id: fileId, + chunk_index: chunkIndex, + }); + if (msg.type === 'error') throw new Error(msg.detail); + return msg; + } + + async fetchStreamSegment(fileId, segmentIndex, segmentDuration) { + const msg = await this._sendAndWait({ + type: 'stream_seg', + v: '0.1', + file_id: fileId, + segment_index: segmentIndex, + segment_duration: segmentDuration || 4, + }); + if (msg.type === 'error') throw new Error(msg.detail); + return _b64decode(msg.data_b64); + } + + async sendChat(payload, iteration, threadId) { + const msg = await this._sendAndWait({ + type: 'chat_msg', + v: '0.1', + payload: payload, + iteration: iteration || 0, + thread_id: threadId || null, + }); + return msg; + } + + close() { + if (this._channel) this._channel.close(); + if (this._pc) this._pc.close(); + this._connected = false; + for (const [, p] of this._pending) p.reject(new Error('Transport closed')); + this._pending.clear(); + } + + // ── Internal ────────────────────────────────────────────────────────────── + + _sendAndWait(obj) { + return new Promise((resolve, reject) => { + const id = this._seqId++; + const timeout = setTimeout(() => { + this._pending.delete(id); + reject(new Error('Response timeout')); + }, 30000); + this._pending.set(id, { + resolve: (msg) => { clearTimeout(timeout); this._pending.delete(id); resolve(msg); }, + reject: (err) => { clearTimeout(timeout); this._pending.delete(id); reject(err); }, + }); + this._send(obj); + }); + } + + _send(obj) { + const encoded = msgpack_encode(obj); + const header = new Uint8Array(4); + new DataView(header.buffer).setUint32(0, encoded.byteLength, false); + const frame = new Uint8Array(4 + encoded.byteLength); + frame.set(header); + frame.set(encoded, 4); + this._channel.send(frame); + } + + _onMessage(data) { + const incoming = new Uint8Array(data); + const combined = new Uint8Array(this._recvBuf.length + incoming.length); + combined.set(this._recvBuf); + combined.set(incoming, this._recvBuf.length); + this._recvBuf = combined; + + while (this._recvBuf.length >= 4) { + const len = new DataView(this._recvBuf.buffer, this._recvBuf.byteOffset).getUint32(0, false); + if (this._recvBuf.length < 4 + len) break; + const msgBytes = this._recvBuf.slice(4, 4 + len); + this._recvBuf = this._recvBuf.slice(4 + len); + + const msg = msgpack_decode(msgBytes); + this._dispatch(msg); + } + } + + _dispatch(msg) { + if (msg.type === 'chat_msg' && this._onChat) { + this._onChat(msg); + return; + } + + const oldest = this._pending.entries().next(); + if (!oldest.done) { + const [, handler] = oldest.value; + handler.resolve(msg); + } + } +} + +// ── Minimal msgpack encode/decode ──────────────────────────────────────────── +// Covers the subset used by MNP: maps, strings, integers, binary, arrays, null. + +function msgpack_encode(obj) { + const parts = []; + _encodeValue(obj, parts); + const total = parts.reduce((s, p) => s + p.length, 0); + const result = new Uint8Array(total); + let off = 0; + for (const p of parts) { result.set(p, off); off += p.length; } + return result; +} + +function _encodeValue(val, parts) { + if (val === null || val === undefined) { + parts.push(new Uint8Array([0xc0])); + } else if (typeof val === 'boolean') { + parts.push(new Uint8Array([val ? 0xc3 : 0xc2])); + } else if (typeof val === 'number') { + if (Number.isInteger(val)) { + if (val >= 0 && val <= 127) { + parts.push(new Uint8Array([val])); + } else if (val >= 0 && val <= 0xff) { + parts.push(new Uint8Array([0xcc, val])); + } else if (val >= 0 && val <= 0xffff) { + const b = new Uint8Array(3); b[0] = 0xcd; + new DataView(b.buffer).setUint16(1, val, false); + parts.push(b); + } else if (val >= 0 && val <= 0xffffffff) { + const b = new Uint8Array(5); b[0] = 0xce; + new DataView(b.buffer).setUint32(1, val, false); + parts.push(b); + } else if (val >= -32 && val < 0) { + parts.push(new Uint8Array([val & 0xff])); + } else if (val >= -128 && val < 0) { + const b = new Uint8Array(2); b[0] = 0xd0; b[1] = val & 0xff; + parts.push(b); + } else { + const b = new Uint8Array(5); b[0] = 0xd2; + new DataView(b.buffer).setInt32(1, val, false); + parts.push(b); + } + } else { + const b = new Uint8Array(9); b[0] = 0xcb; + new DataView(b.buffer).setFloat64(1, val, false); + parts.push(b); + } + } else if (typeof val === 'string') { + const encoded = new TextEncoder().encode(val); + if (encoded.length <= 31) { + parts.push(new Uint8Array([0xa0 | encoded.length])); + } else if (encoded.length <= 0xff) { + parts.push(new Uint8Array([0xd9, encoded.length])); + } else if (encoded.length <= 0xffff) { + const b = new Uint8Array(3); b[0] = 0xda; + new DataView(b.buffer).setUint16(1, encoded.length, false); + parts.push(b); + } else { + const b = new Uint8Array(5); b[0] = 0xdb; + new DataView(b.buffer).setUint32(1, encoded.length, false); + parts.push(b); + } + parts.push(encoded); + } else if (val instanceof Uint8Array) { + if (val.length <= 0xff) { + parts.push(new Uint8Array([0xc4, val.length])); + } else if (val.length <= 0xffff) { + const b = new Uint8Array(3); b[0] = 0xc5; + new DataView(b.buffer).setUint16(1, val.length, false); + parts.push(b); + } else { + const b = new Uint8Array(5); b[0] = 0xc6; + new DataView(b.buffer).setUint32(1, val.length, false); + parts.push(b); + } + parts.push(val); + } else if (Array.isArray(val)) { + if (val.length <= 15) { + parts.push(new Uint8Array([0x90 | val.length])); + } else if (val.length <= 0xffff) { + const b = new Uint8Array(3); b[0] = 0xdc; + new DataView(b.buffer).setUint16(1, val.length, false); + parts.push(b); + } else { + const b = new Uint8Array(5); b[0] = 0xdd; + new DataView(b.buffer).setUint32(1, val.length, false); + parts.push(b); + } + for (const item of val) _encodeValue(item, parts); + } else if (typeof val === 'object') { + const keys = Object.keys(val); + if (keys.length <= 15) { + parts.push(new Uint8Array([0x80 | keys.length])); + } else if (keys.length <= 0xffff) { + const b = new Uint8Array(3); b[0] = 0xde; + new DataView(b.buffer).setUint16(1, keys.length, false); + parts.push(b); + } else { + const b = new Uint8Array(5); b[0] = 0xdf; + new DataView(b.buffer).setUint32(1, keys.length, false); + parts.push(b); + } + for (const k of keys) { + _encodeValue(k, parts); + _encodeValue(val[k], parts); + } + } +} + +function msgpack_decode(buf) { + const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength); + const [val] = _decodeValue(buf, view, 0); + return val; +} + +function _decodeValue(buf, view, offset) { + const byte = buf[offset]; + + if (byte <= 0x7f) return [byte, offset + 1]; + if ((byte & 0xe0) === 0xe0) return [byte - 256, offset + 1]; + if ((byte & 0xa0) === 0xa0) { + const len = byte & 0x1f; + return [new TextDecoder().decode(buf.slice(offset + 1, offset + 1 + len)), offset + 1 + len]; + } + if ((byte & 0xf0) === 0x90) { + const len = byte & 0x0f; + return _decodeArray(buf, view, offset + 1, len); + } + if ((byte & 0xf0) === 0x80) { + const len = byte & 0x0f; + return _decodeMap(buf, view, offset + 1, len); + } + + switch (byte) { + case 0xc0: return [null, offset + 1]; + case 0xc2: return [false, offset + 1]; + case 0xc3: return [true, offset + 1]; + case 0xc4: { const len = buf[offset + 1]; return [buf.slice(offset + 2, offset + 2 + len), offset + 2 + len]; } + case 0xc5: { const len = view.getUint16(offset + 1, false); return [buf.slice(offset + 3, offset + 3 + len), offset + 3 + len]; } + case 0xc6: { const len = view.getUint32(offset + 1, false); return [buf.slice(offset + 5, offset + 5 + len), offset + 5 + len]; } + case 0xcc: return [buf[offset + 1], offset + 2]; + case 0xcd: return [view.getUint16(offset + 1, false), offset + 3]; + case 0xce: return [view.getUint32(offset + 1, false), offset + 5]; + case 0xcb: return [view.getFloat64(offset + 1, false), offset + 9]; + case 0xd0: return [view.getInt8(offset + 1), offset + 2]; + case 0xd1: return [view.getInt16(offset + 1, false), offset + 3]; + case 0xd2: return [view.getInt32(offset + 1, false), offset + 5]; + case 0xd9: { + const len = buf[offset + 1]; + return [new TextDecoder().decode(buf.slice(offset + 2, offset + 2 + len)), offset + 2 + len]; + } + case 0xda: { + const len = view.getUint16(offset + 1, false); + return [new TextDecoder().decode(buf.slice(offset + 3, offset + 3 + len)), offset + 3 + len]; + } + case 0xdb: { + const len = view.getUint32(offset + 1, false); + return [new TextDecoder().decode(buf.slice(offset + 5, offset + 5 + len)), offset + 5 + len]; + } + case 0xdc: { const len = view.getUint16(offset + 1, false); return _decodeArray(buf, view, offset + 3, len); } + case 0xdd: { const len = view.getUint32(offset + 1, false); return _decodeArray(buf, view, offset + 5, len); } + case 0xde: { const len = view.getUint16(offset + 1, false); return _decodeMap(buf, view, offset + 3, len); } + case 0xdf: { const len = view.getUint32(offset + 1, false); return _decodeMap(buf, view, offset + 5, len); } + default: throw new Error(`Unknown msgpack type: 0x${byte.toString(16)}`); + } +} + +function _decodeArray(buf, view, offset, count) { + const arr = []; + for (let i = 0; i < count; i++) { + const [val, newOff] = _decodeValue(buf, view, offset); + arr.push(val); + offset = newOff; + } + return [arr, offset]; +} + +function _decodeMap(buf, view, offset, count) { + const obj = {}; + for (let i = 0; i < count; i++) { + const [key, off1] = _decodeValue(buf, view, offset); + const [val, off2] = _decodeValue(buf, view, off1); + obj[key] = val; + offset = off2; + } + return [obj, offset]; +} + +function _b64decode(b64) { + const binary = atob(b64); + const bytes = new Uint8Array(binary.length); + for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i); + return bytes; +} + +// Export +window.MeshBayTransport = MeshBayTransport; diff --git a/packages/meshbay-hub/src/meshbay_hub/static/webrtc-test.html b/packages/meshbay-hub/src/meshbay_hub/static/webrtc-test.html new file mode 100644 index 0000000..0d5750b --- /dev/null +++ b/packages/meshbay-hub/src/meshbay_hub/static/webrtc-test.html @@ -0,0 +1,258 @@ + + + + + + MeshBay — WebRTC Spike Test + + + +
+

MeshBay — WebRTC DataChannel Spike Test

+

Phase 9.5 — E2E browser → NAT → node file transfer via WebRTC

+ +
+

1. Login to Hub

+ + + + +
+ +
+

2. Connect to Node via WebRTC

+ + + + +
+ +
+

3. File Transfer Test

+ +
+ + + +
+ +

Log

+
+
+ + + + + diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py index 327158f..a9c97a3 100644 --- a/packages/meshbay-hub/tests/test_hub_api.py +++ b/packages/meshbay-hub/tests/test_hub_api.py @@ -483,6 +483,76 @@ async def test_password_rehash_on_login(client, app): assert r.status_code == 200 +# ── WebRTC signaling (9.2) ────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_webrtc_offer_no_node(client): + """WebRTC offer to a non-connected node returns 404.""" + pk_ed, pk_x, _ = _gen_user_keys() + await client.post("/v1/users/register", json={ + "username": "sig_user", "email": "sig@x.com", "password": "sigpass99", + "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) + r = await client.post("/v1/users/login", json={ + "username": "sig_user", "password": "sigpass99"}) + token = r.json()["access_token"] + + r = await client.post("/v1/nodes/fake-node-id/webrtc/offer", + json={"sdp": "v=0\r\n...", "ice_candidates": []}, + headers={"Authorization": f"Bearer {token}"}) + assert r.status_code == 404 + assert "not connected" in r.json()["detail"].lower() + + +@pytest.mark.asyncio +async def test_webrtc_signaling_roundtrip(client, app): + """WebRTC signaling: offer relayed to node via WS, answer returned to browser.""" + import asyncio + import json + + pk_ed, pk_x, _ = _gen_user_keys() + await client.post("/v1/users/register", json={ + "username": "sig_user2", "email": "sig2@x.com", "password": "sigpass99", + "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) + r = await client.post("/v1/users/login", json={ + "username": "sig_user2", "password": "sigpass99"}) + token = r.json()["access_token"] + + from meshbay_hub.api.revocation import _connected_nodes + from meshbay_hub.api.signaling import handle_webrtc_answer + + class FakeWS: + def __init__(self): + self.sent = [] + + async def send_text(self, text): + self.sent.append(json.loads(text)) + msg = self.sent[-1] + if msg.get("type") == "webrtc_offer": + await asyncio.sleep(0.01) + handle_webrtc_answer({ + "type": "webrtc_answer", + "peer_id": msg["peer_id"], + "sdp": "v=0\r\nanswer-sdp", + "ice_candidates": [{"candidate": "test"}], + }) + + fake_ws = FakeWS() + node_id = "test-node-sig" + _connected_nodes[node_id] = fake_ws + + try: + r = await client.post(f"/v1/nodes/{node_id}/webrtc/offer", + json={"sdp": "v=0\r\noffer-sdp", "ice_candidates": []}, + headers={"Authorization": f"Bearer {token}"}) + assert r.status_code == 200 + data = r.json() + assert "answer-sdp" in data["sdp"] + assert len(data["ice_candidates"]) == 1 + assert "peer_id" in data + finally: + _connected_nodes.pop(node_id, None) + + # ── IP log cleanup (8.9) ──────────────────────────────────────────────────── @pytest.mark.asyncio diff --git a/packages/meshbay-node/pyproject.toml b/packages/meshbay-node/pyproject.toml index 805871e..592de54 100644 --- a/packages/meshbay-node/pyproject.toml +++ b/packages/meshbay-node/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "aioice>=0.9", # ICE/STUN for NAT traversal "aioquic>=1.0", # QUIC transport (MNP v2) — implemented in Phase 5 "websockets>=12.0", # hub→node revocation push + "aiortc>=1.9", # WebRTC DataChannel for browser P2P (Phase 9) ] [project.optional-dependencies] diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index 74851c1..3e77ed0 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -250,10 +250,11 @@ class HubClient: self, on_incoming: Any = None, on_revocation: Any = None, + on_webrtc_offer: Any = None, ) -> None: """ Maintain a persistent WebSocket connection to the hub. - Receives NAT punch requests and revocation tokens. + Receives NAT punch requests, revocation tokens, and WebRTC offers. Runs until cancelled. """ import websockets @@ -270,6 +271,7 @@ class HubClient: await ws.send(json.dumps({ "type": "auth", "token": self._session.access_token, + "node_id": self._session.node_id, })) auth_resp = json.loads(await ws.recv()) if auth_resp.get("type") != "auth_ok": @@ -289,6 +291,18 @@ class HubClient: elif mtype == "revocation" and on_revocation: on_revocation(msg.get("token", "")) + elif mtype == "webrtc_offer" and on_webrtc_offer: + answer = await on_webrtc_offer( + msg["sdp"], msg["peer_id"], + msg.get("ice_candidates", [])) + if answer: + await ws.send(json.dumps({ + "type": "webrtc_answer", + "peer_id": msg["peer_id"], + "sdp": answer[0], + "ice_candidates": answer[1], + })) + elif mtype == "pong": pass diff --git a/packages/meshbay-node/src/meshbay_node/transport/__init__.py b/packages/meshbay-node/src/meshbay_node/transport/__init__.py index 5a1b8d7..df9c209 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/transport/__init__.py @@ -1,10 +1,9 @@ -"""MeshBay Node transport layer — TCP+TLS (MNP v1) and QUIC (MNP v2).""" +"""MeshBay Node transport layer — TCP+TLS (v1), QUIC (v2), WebRTC (browsers).""" from .server import ChunkServer from .client import ChunkClient 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, Denylist from .quic_client import QuicChunkClient @@ -15,7 +14,17 @@ except ImportError: Denylist = None # type: ignore[assignment,misc] QUIC_AVAILABLE = False +# WebRTC transport (browsers) — requires aiortc>=1.9 +try: + from .webrtc_server import WebRTCTransport, WebRTCPeerSession + WEBRTC_AVAILABLE = True +except ImportError: + WebRTCTransport = None # type: ignore[assignment,misc] + WebRTCPeerSession = None # type: ignore[assignment,misc] + WEBRTC_AVAILABLE = False + __all__ = [ "ChunkServer", "ChunkClient", "create_http_app", "QuicChunkServer", "QuicChunkClient", "Denylist", "QUIC_AVAILABLE", + "WebRTCTransport", "WebRTCPeerSession", "WEBRTC_AVAILABLE", ] diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py new file mode 100644 index 0000000..89391b9 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -0,0 +1,373 @@ +""" +MeshBay Node — WebRTC DataChannel server for browser clients. + +Browsers cannot use QUIC for NAT traversal (WebTransport doesn't allow choosing +the UDP source port — Port-Restricted Cone NAT requires exact port matching). +WebRTC DataChannel with ICE/STUN handles this automatically. + +The MNP protocol (handshake, file_request, file_chunk, chat, etc.) runs +identically over WebRTC DataChannel as over QUIC streams. Same E2E encryption, +same message types, same msgpack wire format. + +Wire format on the DataChannel: + - Each message is length-prefixed msgpack (4-byte big-endian + msgpack payload) + - Same as QUIC streams and TCP+TLS + - DataChannel is ordered and reliable (SCTP over DTLS) + +Signaling flow (handled externally by the hub): + Browser → Hub : POST /v1/nodes/{id}/webrtc/offer {sdp, ice_candidates} + Hub → Node : WS push {type: "webrtc_offer", sdp, ice_candidates, peer_id} + Node → Hub : WS push {type: "webrtc_answer", sdp, ice_candidates, peer_id} + Hub → Browser : SSE/response {sdp, ice_candidates} + After signaling, DataChannel is P2P — hub is out of the loop. +""" + +import asyncio +import base64 +import logging +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.protocol import MNP +from meshbay_node.indexer import GroupIndex + +log = logging.getLogger(__name__) + +CHUNK_SIZE = 1024 * 1024 +MAX_MSG = 64 * 1024 * 1024 + + +def _pack(obj: dict) -> bytes: + data = msgpack.packb(obj, use_bin_type=True) + return struct.pack(">I", len(data)) + data + + +class _DataChannelBuffer: + """Accumulate DataChannel messages and extract length-prefixed msgpack.""" + + def __init__(self): + self._buf = bytearray() + + def feed(self, data: bytes): + self._buf.extend(data) + + def messages(self): + while len(self._buf) >= 4: + length = struct.unpack(">I", self._buf[:4])[0] + if length > MAX_MSG: + raise ValueError(f"Message too large: {length}") + if len(self._buf) < 4 + length: + break + msg_bytes = bytes(self._buf[4:4 + length]) + del self._buf[:4 + length] + yield msgpack.unpackb(msg_bytes, raw=False) + + +class WebRTCPeerSession: + """One WebRTC peer connection, handling MNP over a DataChannel.""" + + def __init__(self, pc: RTCPeerConnection, node_ctx: dict): + self._pc = pc + self._ctx = node_ctx + self._channel: RTCDataChannel | None = None + self._buffer = _DataChannelBuffer() + self._user_id: str | None = None + self._group_id: str | None = None + + def _setup_channel(self, channel: RTCDataChannel) -> None: + self._channel = channel + + @channel.on("message") + def on_message(message): + if isinstance(message, str): + message = message.encode() + self._buffer.feed(message) + for msg in self._buffer.messages(): + self._handle_message(msg) + + def _handle_message(self, msg: dict) -> None: + mtype = msg.get("type") + try: + if mtype == MNP.HANDSHAKE: + self._do_handshake(msg) + elif self._user_id is None: + self._send({"type": "error", "detail": "Handshake required"}) + elif mtype == MNP.INDEX_SYNC: + self._do_index_sync() + elif mtype == MNP.FILE_REQUEST: + self._do_file_request(msg) + elif mtype == MNP.STREAM_SEGMENT: + self._do_stream_segment(msg) + elif mtype == MNP.CHAT_MESSAGE: + self._do_chat_message(msg) + else: + log.warning("Unknown MNP message type on DataChannel: %s", mtype) + except Exception as e: + log.error("Error handling %s on DataChannel: %s", mtype, e) + self._send({"type": "error", "detail": str(e)}) + + def _do_handshake(self, msg: dict) -> None: + 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: + self._send({"type": "error", "detail": f"Invalid JWT: {e}"}) + return + + denylist = self._ctx.get("denylist") + if denylist and denylist.is_denied(decoded.get("sub", ""), decoded.get("jti", "")): + self._send({"type": "error", "detail": "Token revoked"}) + return + + if group_id and group_id not in decoded.get("groups", []): + self._send({"type": "error", "detail": "Not a member of this group"}) + return + + if group_id and "groups" in self._ctx and group_id not in self._ctx["groups"]: + self._send({"type": "error", "detail": "Group not hosted on this node"}) + return + + self._user_id = decoded["sub"] + self._group_id = group_id + + log.info("WebRTC handshake OK — user=%s group=%s", + self._user_id[:8], group_id[:8] if group_id else "none") + self._send({ + "type": MNP.HANDSHAKE_ACK, + "v": MNP_VERSION, + "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()), + }) + + def _group_ctx(self) -> dict: + if "groups" in self._ctx and self._group_id: + return self._ctx["groups"][self._group_id] + return self._ctx + + def _do_index_sync(self) -> None: + ctx = self._group_ctx() + wire = ctx["index"].serialize() + self._send({ + "type": MNP.INDEX_SYNC, + "v": MNP_VERSION, + "index_b64": base64.b64encode(wire).decode(), + }) + + def _do_file_request(self, msg: dict) -> None: + ctx = self._group_ctx() + file_id = msg["file_id"] + chunk_index = msg["chunk_index"] + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + file_path = ctx["shared_root"] / entry.path / entry.name + if not file_path.exists(): + self._send({"type": "error", "detail": "File not on disk"}) + return + + chunk_data = _read_and_encrypt( + self._ctx["sk_node"], + ctx["gek"], + file_path, + chunk_index, + ) + self._send(chunk_data) + + def _do_stream_segment(self, msg: dict) -> None: + 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({"type": "error", "detail": "File not found"}) + return + + file_path = ctx["shared_root"] / entry.path / entry.name + if not file_path.exists(): + self._send({"type": "error", "detail": "File not on disk"}) + return + + import subprocess + try: + result = subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", + "-ss", str(segment_index * segment_duration), + "-i", str(file_path), + "-t", str(segment_duration), + "-c:v", "copy", "-c:a", "copy", + "-f", "mpegts", "pipe:1"], + capture_output=True, timeout=30, + ) + if result.returncode != 0 or not result.stdout: + self._send({"type": "error", "detail": "Segment extraction failed"}) + return + segment_data = result.stdout + except Exception: + self._send({"type": "error", "detail": "Segment extraction failed"}) + return + + self._send({ + "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(self, msg: dict) -> None: + chat_store = self._ctx.get("chat_store") + if chat_store: + 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"), + )) + self._send({"type": "ack", "v": MNP_VERSION}) + + def _send(self, obj: dict) -> None: + if self._channel and self._channel.readyState == "open": + self._channel.send(_pack(obj)) + + async def close(self) -> None: + await self._pc.close() + + +def _read_and_encrypt( + sk_node: Ed25519PrivateKey, + gek: bytes, + file_path: Path, + chunk_index: int, +) -> 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) + + 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(), + } + + +class WebRTCTransport: + """ + Manages WebRTC peer connections for browser clients. + + Usage: + transport = WebRTCTransport(sk_node, hub_pk_pem, gek, shared_root, index) + answer_sdp = await transport.handle_offer(offer_sdp, peer_id) + # Return answer_sdp to the browser via hub signaling + """ + + def __init__( + self, + sk_node: Ed25519PrivateKey, + hub_pk_pem: bytes, + gek: bytes, + shared_root: Path, + index: GroupIndex, + groups: dict[str, dict] | None = None, + denylist: Any | None = None, + stun_servers: list[str] | None = None, + ): + self._ctx: dict[str, Any] = { + "sk_node": sk_node, + "hub_pk_pem": hub_pk_pem, + "gek": gek, + "shared_root": shared_root, + "index": index, + } + if groups: + self._ctx["groups"] = groups + if denylist: + self._ctx["denylist"] = denylist + self._stun = stun_servers or ["stun:stun.l.google.com:19302"] + self._sessions: dict[str, WebRTCPeerSession] = {} + + async def handle_offer( + self, offer_sdp: str, peer_id: str, + ) -> tuple[str, list[dict]]: + """ + Process a WebRTC SDP offer from a browser client. + + Returns (answer_sdp, ice_candidates) to relay back via hub signaling. + ICE candidates are embedded in the SDP (aiortc gathers before returning). + """ + from aiortc import RTCIceServer, RTCConfiguration + + config = RTCConfiguration( + iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else [] + ) + pc = RTCPeerConnection(configuration=config) + session = WebRTCPeerSession(pc, self._ctx) + self._sessions[peer_id] = session + + @pc.on("datachannel") + def on_datachannel(channel: RTCDataChannel): + log.info("WebRTC DataChannel opened: %s (peer=%s)", channel.label, peer_id) + session._setup_channel(channel) + + @pc.on("connectionstatechange") + async def on_state_change(): + state = pc.connectionState + log.info("WebRTC connection state: %s (peer=%s)", state, peer_id) + if state in ("failed", "closed"): + self._sessions.pop(peer_id, None) + + offer = RTCSessionDescription(sdp=offer_sdp, type="offer") + await pc.setRemoteDescription(offer) + answer = await pc.createAnswer() + await pc.setLocalDescription(answer) + + log.info("WebRTC answer ready for peer=%s", peer_id) + return pc.localDescription.sdp, [] + + async def close_peer(self, peer_id: str) -> None: + session = self._sessions.pop(peer_id, None) + if session: + await session.close() + + async def close_all(self) -> None: + for session in self._sessions.values(): + await session.close() + self._sessions.clear() + + @property + def active_peers(self) -> int: + return len(self._sessions) diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py new file mode 100644 index 0000000..4c0fdbf --- /dev/null +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -0,0 +1,326 @@ +""" +Integration test: WebRTC DataChannel transport for browser clients. + +Phase 9 milestone 9.1 — spike: validate aiortc WebRTC DataChannel works +for MNP protocol exchange (handshake, index_sync, file_request, file_chunk). + +Uses local loopback (no STUN/ICE needed for localhost). +""" + +import asyncio +import base64 +import os +import struct +import time + +import blake3 +import jwt +import msgpack +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from aiortc import RTCPeerConnection, RTCSessionDescription + +from meshbay_common import MNP_VERSION +from meshbay_common.crypto import ( + generate_gek, + pk_to_b64, + chunk_key as derive_chunk_key, + decrypt_chunk, + verify_chunk_signature, +) +from meshbay_common.protocol import MNP +from meshbay_node.indexer import DirectoryIndexer +from meshbay_node.transport.webrtc_server import WebRTCTransport + + +@pytest.fixture +def sk_node(): + return Ed25519PrivateKey.generate() + + +@pytest.fixture +def sk_hub(): + return Ed25519PrivateKey.generate() + + +@pytest.fixture +def gek(): + return generate_gek() + + +@pytest.fixture +def shared_dir(tmp_path): + d = tmp_path / "shared" + d.mkdir() + (d / "test.bin").write_bytes(os.urandom(2048)) + (d / "hello.txt").write_bytes(b"hello webrtc " * 50) + return d + + +def _hub_pk_pem(sk_hub): + return sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + +def _make_jwt(sk_hub, groups=None): + 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-001", + "pk_user": "test", "hub_id": "test-hub", + "jti": "test-jti-webrtc", "iat": now, "exp": now + 3600, + "groups": groups or [], + }, sk_pem, algorithm="EdDSA") + + +def _pack(obj: dict) -> bytes: + data = msgpack.packb(obj, use_bin_type=True) + return struct.pack(">I", len(data)) + data + + +def _unpack(raw: bytes) -> dict: + length = struct.unpack(">I", raw[:4])[0] + return msgpack.unpackb(raw[4:4 + length], raw=False) + + +@pytest.mark.asyncio +async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: browser sends MNP handshake, node responds with handshake_ack.""" + hub_pk_pem = _hub_pk_pem(sk_hub) + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + stun_servers=[], + ) + + browser_pc = RTCPeerConnection() + received = asyncio.Queue() + + channel = browser_pc.createDataChannel("mnp") + + @channel.on("message") + def on_msg(message): + if isinstance(message, str): + message = message.encode() + received.put_nowait(_unpack(message)) + + offer = await browser_pc.createOffer() + await browser_pc.setLocalDescription(offer) + + answer_sdp, ice_candidates = await transport.handle_offer( + browser_pc.localDescription.sdp, "peer-001") + + answer = RTCSessionDescription(sdp=answer_sdp, type="answer") + await browser_pc.setRemoteDescription(answer) + + await asyncio.sleep(0.5) + + token = _make_jwt(sk_hub) + channel.send(_pack({ + "type": MNP.HANDSHAKE, + "v": MNP_VERSION, + "token": token, + })) + + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == MNP.HANDSHAKE_ACK + assert msg["v"] == MNP_VERSION + assert "node_pk" in msg + + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: full file transfer — handshake, index, fetch chunk, decrypt.""" + hub_pk_pem = _hub_pk_pem(sk_hub) + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + stun_servers=[], + ) + + browser_pc = RTCPeerConnection() + received = asyncio.Queue() + + channel = browser_pc.createDataChannel("mnp") + + @channel.on("open") + def on_open(): + token = _make_jwt(sk_hub) + channel.send(_pack({ + "type": MNP.HANDSHAKE, + "v": MNP_VERSION, + "token": token, + })) + + buf = bytearray() + + @channel.on("message") + def on_msg(message): + if isinstance(message, str): + message = message.encode() + buf.extend(message) + while len(buf) >= 4: + length = struct.unpack(">I", buf[:4])[0] + if len(buf) < 4 + length: + break + msg_bytes = bytes(buf[4:4 + length]) + del buf[:4 + length] + received.put_nowait(msgpack.unpackb(msg_bytes, raw=False)) + + offer = await browser_pc.createOffer() + await browser_pc.setLocalDescription(offer) + + answer_sdp, _ = await transport.handle_offer( + browser_pc.localDescription.sdp, "peer-002") + await browser_pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + + # 1) Handshake ack + ack = await asyncio.wait_for(received.get(), timeout=5.0) + assert ack["type"] == MNP.HANDSHAKE_ACK + + # 2) Request index + channel.send(_pack({"type": MNP.INDEX_SYNC, "v": MNP_VERSION})) + idx_msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert idx_msg["type"] == MNP.INDEX_SYNC + assert "index_b64" in idx_msg + + # 3) Request file chunk + entry = next(e for e in indexer.index.entries if e.name == "test.bin") + channel.send(_pack({ + "type": MNP.FILE_REQUEST, + "v": MNP_VERSION, + "file_id": entry.id, + "chunk_index": 0, + })) + + chunk_msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert chunk_msg["type"] == MNP.FILE_CHUNK + + # 4) Verify and decrypt + ct = base64.b64decode(chunk_msg["ct_b64"]) + nonce = base64.b64decode(chunk_msg["nonce_b64"]) + ct_hash = base64.b64decode(chunk_msg["ct_hash_b64"]) + pt_hash = base64.b64decode(chunk_msg["pt_hash_b64"]) + sig = base64.b64decode(chunk_msg["sig_b64"]) + file_hash = base64.b64decode(chunk_msg["file_hash_b64"]) + + pk_node = sk_node.public_key() + verify_chunk_signature(pk_node, 0, nonce, ct_hash, sig) + assert blake3.blake3(ct).digest() == ct_hash + + ckey = derive_chunk_key(gek, file_hash, 0) + plaintext = decrypt_chunk(ckey, nonce, ct) + assert blake3.blake3(plaintext).digest() == pt_hash + + original = (shared_dir / "test.bin").read_bytes() + assert plaintext == original + + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: invalid JWT is rejected with error.""" + hub_pk_pem = _hub_pk_pem(sk_hub) + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + stun_servers=[], + ) + + browser_pc = RTCPeerConnection() + received = asyncio.Queue() + + channel = browser_pc.createDataChannel("mnp") + + @channel.on("message") + def on_msg(message): + if isinstance(message, str): + message = message.encode() + received.put_nowait(_unpack(message)) + + offer = await browser_pc.createOffer() + await browser_pc.setLocalDescription(offer) + + answer_sdp, _ = await transport.handle_offer( + browser_pc.localDescription.sdp, "peer-003") + await browser_pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + + await asyncio.sleep(0.5) + + channel.send(_pack({ + "type": MNP.HANDSHAKE, + "v": MNP_VERSION, + "token": "invalid.jwt.token", + })) + + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == "error" + assert "JWT" in msg["detail"] or "Invalid" in msg["detail"] + + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_request_before_handshake_rejected(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: request without handshake is rejected.""" + hub_pk_pem = _hub_pk_pem(sk_hub) + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + stun_servers=[], + ) + + browser_pc = RTCPeerConnection() + received = asyncio.Queue() + + channel = browser_pc.createDataChannel("mnp") + + @channel.on("message") + def on_msg(message): + if isinstance(message, str): + message = message.encode() + received.put_nowait(_unpack(message)) + + offer = await browser_pc.createOffer() + await browser_pc.setLocalDescription(offer) + + answer_sdp, _ = await transport.handle_offer( + browser_pc.localDescription.sdp, "peer-004") + await browser_pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + + await asyncio.sleep(0.5) + + channel.send(_pack({"type": MNP.INDEX_SYNC, "v": MNP_VERSION})) + + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == "error" + assert "Handshake required" in msg["detail"] + + await browser_pc.close() + await transport.close_all() -- cgit v1.2.3