aboutsummaryrefslogtreecommitdiffstats
path: root/packages
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-08-10 22:12:59 +0200
committerChristophe Besson <cbesson@gmail.com>2026-08-10 22:12:59 +0200
commit60c4570e72e36c2a9720593c8baec74ee2ab52d6 (patch)
treee833e222a6a95e64e5b24a0813882636044198c2 /packages
parent1a53eb4cc404ec94658fde0ae04cfe2ccf1810dc (diff)
downloadmeshbay-60c4570e72e36c2a9720593c8baec74ee2ab52d6.tar.gz
feat: Phase 9.1–9.5 — WebRTC DataChannel transport for browser P2P
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 <noreply@anthropic.com>
Diffstat (limited to 'packages')
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/api/revocation.py5
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/api/signaling.py102
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/api/webapp.py15
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/app.py2
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/static/transport.js409
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/static/webrtc-test.html258
-rw-r--r--packages/meshbay-hub/tests/test_hub_api.py70
-rw-r--r--packages/meshbay-node/pyproject.toml1
-rw-r--r--packages/meshbay-node/src/meshbay_node/hub_client.py16
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/__init__.py13
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py373
-rw-r--r--packages/meshbay-node/tests/test_webrtc_transport.py326
12 files changed, 1586 insertions, 4 deletions
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 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="utf-8">
+ <meta name="viewport" content="width=device-width, initial-scale=1">
+ <title>MeshBay — WebRTC Spike Test</title>
+ <style>
+ *, *::before, *::after { box-sizing: border-box; }
+ body { font-family: system-ui, sans-serif; margin: 0; background: #0f172a; color: #e2e8f0; }
+ .container { max-width: 800px; margin: 32px auto; padding: 0 16px; }
+ h1 { color: #38bdf8; font-size: 1.4em; }
+ h2 { color: #94a3b8; font-size: 1.1em; margin-top: 2em; }
+ .step { background: #1e293b; border: 1px solid #334155; border-radius: 8px;
+ padding: 16px; margin: 12px 0; }
+ .step.done { border-color: #22c55e; }
+ .step.fail { border-color: #ef4444; }
+ .step.active { border-color: #38bdf8; }
+ input { padding: 8px 12px; border: 1px solid #475569; border-radius: 6px;
+ background: #0f172a; color: #e2e8f0; font-size: 0.95em; margin: 4px; width: 240px; }
+ button { padding: 8px 20px; background: #0ea5e9; color: #fff; border: none;
+ border-radius: 6px; cursor: pointer; font-size: 0.95em; margin: 4px; }
+ button:hover { background: #0284c7; }
+ button:disabled { background: #475569; cursor: not-allowed; }
+ #log { background: #020617; border: 1px solid #1e293b; border-radius: 8px;
+ padding: 12px; font-family: monospace; font-size: 0.85em; line-height: 1.6;
+ max-height: 400px; overflow-y: auto; white-space: pre-wrap; }
+ .ok { color: #22c55e; }
+ .err { color: #ef4444; }
+ .info { color: #38bdf8; }
+ .warn { color: #f59e0b; }
+ .dim { color: #64748b; }
+ .badge { display: inline-block; background: #22c55e; color: #0f172a; padding: 2px 8px;
+ border-radius: 4px; font-size: 0.8em; font-weight: bold; margin-left: 8px; }
+ .badge.fail { background: #ef4444; color: #fff; }
+ </style>
+</head>
+<body>
+<div class="container">
+ <h1>MeshBay — WebRTC DataChannel Spike Test</h1>
+ <p class="dim">Phase 9.5 — E2E browser → NAT → node file transfer via WebRTC</p>
+
+ <div class="step" id="step-login">
+ <h2>1. Login to Hub</h2>
+ <input id="username" placeholder="Username" value="bob">
+ <input id="password" placeholder="Password" type="password" value="bob">
+ <button id="btn-login" onclick="doLogin()">Login</button>
+ <span id="login-status"></span>
+ </div>
+
+ <div class="step" id="step-connect">
+ <h2>2. Connect to Node via WebRTC</h2>
+ <input id="node-id" placeholder="Node ID">
+ <input id="group-id" placeholder="Group ID (optional)">
+ <button id="btn-connect" onclick="doConnect()" disabled>Connect</button>
+ <span id="connect-status"></span>
+ </div>
+
+ <div class="step" id="step-transfer">
+ <h2>3. File Transfer Test</h2>
+ <button id="btn-index" onclick="doFetchIndex()" disabled>Fetch Index</button>
+ <br>
+ <input id="file-id" placeholder="File ID (blake3 hex, from node log)">
+ <button id="btn-chunk" onclick="doFetchChunk()" disabled>Fetch Chunk</button>
+ <span id="transfer-status"></span>
+ </div>
+
+ <h2>Log</h2>
+ <div id="log"></div>
+</div>
+
+<script src="/transport.js"></script>
+<script>
+const HUB_URL = window.location.origin;
+const params = new URLSearchParams(window.location.search);
+let accessToken = null;
+let jwtToken = null;
+let transport = null;
+let fileIndex = null;
+
+// Pre-fill from URL params
+if (params.get('user')) document.getElementById('username').value = params.get('user');
+if (params.get('pass')) document.getElementById('password').value = params.get('pass');
+if (params.get('node')) document.getElementById('node-id').value = params.get('node');
+if (params.get('group')) document.getElementById('group-id').value = params.get('group');
+if (params.get('file')) document.getElementById('file-id').value = params.get('file');
+
+// Auto-run if all params provided
+if (params.get('auto')) {
+ setTimeout(async () => {
+ await doLogin();
+ if (accessToken) await doConnect();
+ if (transport && transport.connected) {
+ await doFetchIndex();
+ if (document.getElementById('file-id').value) await doFetchChunk();
+ }
+ }, 500);
+}
+
+function logMsg(cls, text) {
+ const el = document.getElementById('log');
+ const line = document.createElement('span');
+ line.className = cls;
+ line.textContent = text + '\n';
+ el.appendChild(line);
+ el.scrollTop = el.scrollHeight;
+}
+
+function setStep(id, state) {
+ const el = document.getElementById(id);
+ el.className = 'step ' + state;
+}
+
+async function doLogin() {
+ const user = document.getElementById('username').value;
+ const pass = document.getElementById('password').value;
+ logMsg('info', `Logging in as ${user}...`);
+ setStep('step-login', 'active');
+
+ try {
+ const resp = await fetch(`${HUB_URL}/v1/users/login`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ username: user, password: pass }),
+ });
+
+ if (!resp.ok) {
+ const err = await resp.json();
+ throw new Error(err.detail || resp.statusText);
+ }
+
+ const data = await resp.json();
+ accessToken = data.access_token;
+ jwtToken = data.access_token;
+ logMsg('ok', `Login OK — token: ${accessToken.substring(0, 20)}...`);
+ setStep('step-login', 'done');
+ document.getElementById('login-status').innerHTML = '<span class="badge">OK</span>';
+ document.getElementById('btn-connect').disabled = false;
+ } catch (e) {
+ logMsg('err', `Login FAILED: ${e.message}`);
+ setStep('step-login', 'fail');
+ document.getElementById('login-status').innerHTML = '<span class="badge fail">FAIL</span>';
+ }
+}
+
+async function doConnect() {
+ const nodeId = document.getElementById('node-id').value;
+ const groupId = document.getElementById('group-id').value;
+ if (!nodeId) { logMsg('warn', 'Enter a node ID'); return; }
+
+ logMsg('info', `Connecting to node ${nodeId.substring(0, 8)}... via WebRTC`);
+ setStep('step-connect', 'active');
+
+ try {
+ transport = new MeshBayTransport(HUB_URL, accessToken);
+
+ logMsg('dim', ' Creating RTCPeerConnection...');
+ logMsg('dim', ' Creating DataChannel "mnp"...');
+ logMsg('dim', ' Gathering ICE candidates...');
+ logMsg('dim', ' Sending SDP offer to hub...');
+
+ const t0 = performance.now();
+ const ack = await transport.connect(nodeId, jwtToken, groupId);
+ const elapsed = (performance.now() - t0).toFixed(0);
+
+ logMsg('ok', `WebRTC connected in ${elapsed}ms`);
+ logMsg('ok', ` MNP handshake_ack — node_pk: ${ack.node_pk?.substring(0, 16)}...`);
+ logMsg('ok', ` DataChannel: open, ordered, reliable`);
+ setStep('step-connect', 'done');
+ document.getElementById('connect-status').innerHTML = '<span class="badge">P2P OK</span>';
+ document.getElementById('btn-index').disabled = false;
+ document.getElementById('btn-chunk').disabled = false;
+ } catch (e) {
+ logMsg('err', `Connection FAILED: ${e.message}`);
+ setStep('step-connect', 'fail');
+ document.getElementById('connect-status').innerHTML = '<span class="badge fail">FAIL</span>';
+ }
+}
+
+async function doFetchIndex() {
+ logMsg('info', 'Fetching Mesh Group Index via DataChannel...');
+ try {
+ const t0 = performance.now();
+ const indexBytes = await transport.fetchIndex();
+ const elapsed = (performance.now() - t0).toFixed(0);
+
+ logMsg('ok', `Index received: ${indexBytes.byteLength} bytes in ${elapsed}ms`);
+
+ try {
+ const envelope = msgpack_decode(indexBytes);
+ logMsg('dim', ` type: ${envelope.type}, encrypted: ${envelope.encrypted}, version: ${envelope.version}`);
+ logMsg('dim', ` group_id: ${envelope.group_id}`);
+
+ if (envelope.encrypted) {
+ logMsg('warn', ` Index is GEK-encrypted — browser decryption not implemented in spike`);
+ logMsg('dim', ` ct_b64 length: ${envelope.ct_b64?.length || 0} chars`);
+ logMsg('info', ` Spike workaround: enter a file_id manually or use Fetch First Chunk`);
+ // Store envelope so chunk test can proceed with manual file_id
+ fileIndex = { entries: [], envelope };
+ } else {
+ // Public group: decompress and parse
+ logMsg('dim', ` Public index — data_b64 length: ${envelope.data_b64?.length || 0}`);
+ fileIndex = { entries: [], envelope };
+ }
+ } catch (pe) {
+ logMsg('warn', ` Could not parse index envelope: ${pe.message}`);
+ }
+ } catch (e) {
+ logMsg('err', `Index fetch FAILED: ${e.message}`);
+ }
+}
+
+async function doFetchChunk() {
+ let fileId = document.getElementById('file-id').value.trim();
+
+ if (!fileId) {
+ logMsg('warn', 'Enter a file_id (blake3 hex hash from node indexer log)');
+ logMsg('dim', ' Look for "Initial scan complete" in the node terminal');
+ logMsg('dim', ' Or run: python -c "import blake3; print(blake3.blake3(open(\'QE/demo-v3/shared_media/sample.txt\',\'rb\').read()).hexdigest())"');
+ return;
+ }
+
+ logMsg('info', `Fetching chunk 0 of file ${fileId.substring(0, 16)}... via DataChannel...`);
+
+ try {
+ const t0 = performance.now();
+ const chunkMsg = await transport.fetchChunk(fileId, 0);
+ const elapsed = (performance.now() - t0).toFixed(0);
+
+ if (chunkMsg.type === 'error') {
+ logMsg('err', `Chunk fetch error: ${chunkMsg.detail}`);
+ return;
+ }
+
+ logMsg('ok', `Chunk received in ${elapsed}ms:`);
+ logMsg('ok', ` type: ${chunkMsg.type}`);
+ logMsg('ok', ` chunk_index: ${chunkMsg.chunk_index}`);
+ logMsg('ok', ` plaintext_size: ${chunkMsg.plaintext_size} bytes`);
+ logMsg('ok', ` ct_b64 length: ${chunkMsg.ct_b64?.length || 0} chars`);
+ logMsg('ok', ` nonce_b64: ${chunkMsg.nonce_b64?.substring(0, 16)}...`);
+ logMsg('ok', ` sig_b64: ${chunkMsg.sig_b64?.substring(0, 16)}...`);
+
+ logMsg('', '');
+ logMsg('ok', '=== SPIKE TEST PASSED ===');
+ logMsg('ok', 'Browser connected to node via WebRTC DataChannel.');
+ logMsg('ok', 'MNP handshake, index sync, and file chunk transfer all work.');
+ logMsg('ok', 'Data flowed P2P — hub was only used for signaling.');
+
+ setStep('step-transfer', 'done');
+ document.getElementById('transfer-status').innerHTML = '<span class="badge">E2E OK</span>';
+ } catch (e) {
+ logMsg('err', `Chunk fetch FAILED: ${e.message}`);
+ setStep('step-transfer', 'fail');
+ document.getElementById('transfer-status').innerHTML = '<span class="badge fail">FAIL</span>';
+ }
+}
+</script>
+</body>
+</html>
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()