""" 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 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, ) from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes 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 "entries" in idx_msg assert len(idx_msg["entries"]) > 0 # 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 (binary fields — no base64, minimal envelope) ct = chunk_msg["ct"] nonce = chunk_msg["nonce"] file_hash = bytes.fromhex(entry.id) ckey = chunk_key_aes(gek, file_hash, 0) plaintext = decrypt_chunk_aes(ckey, nonce, ct) original = (shared_dir / "test.bin").read_bytes() assert plaintext == original # 5) Request GEK over DataChannel channel.send(_pack({"type": MNP.GEK_REQUEST, "v": MNP_VERSION})) gek_msg = await asyncio.wait_for(received.get(), timeout=5.0) assert gek_msg["type"] == MNP.GEK_RESPONSE received_gek = base64.b64decode(gek_msg["gek_b64"]) assert received_gek == gek 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() @pytest.mark.asyncio async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tmp_path): """WebRTC DataChannel: send chat message, then retrieve history.""" from meshbay_node.chat.store import ChatStore 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() chat_store = ChatStore(db_path=tmp_path / "chat_test.db") await chat_store.open() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, shared_root=shared_dir, index=indexer.index, stun_servers=[], ) transport._ctx["chat_store"] = chat_store browser_pc = RTCPeerConnection() received = asyncio.Queue() buf = bytearray() 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, })) @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-chat") await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) ack = await asyncio.wait_for(received.get(), timeout=5.0) assert ack["type"] == MNP.HANDSHAKE_ACK channel.send(_pack({ "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "hello from browser", })) chat_ack = await asyncio.wait_for(received.get(), timeout=5.0) assert chat_ack["type"] == "ack" await asyncio.sleep(0.2) channel.send(_pack({ "type": MNP.CHAT_HISTORY, "v": MNP_VERSION, "since": 0, "limit": 50, })) hist = await asyncio.wait_for(received.get(), timeout=5.0) assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE assert len(hist["messages"]) == 1 assert hist["messages"][0]["payload"] == "hello from browser" assert hist["messages"][0]["sender_id"] == "user-001" await chat_store.close() await browser_pc.close() await transport.close_all() @pytest.mark.asyncio async def test_webrtc_chat_history_no_store(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: chat history without chat_store returns empty list.""" 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() buf = bytearray() channel = browser_pc.createDataChannel("mnp") @channel.on("open") def on_open(): channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": _make_jwt(sk_hub), })) @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-no-store") await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) ack = await asyncio.wait_for(received.get(), timeout=5.0) assert ack["type"] == MNP.HANDSHAKE_ACK channel.send(_pack({ "type": MNP.CHAT_HISTORY, "v": MNP_VERSION, "since": 0, "limit": 50, })) hist = await asyncio.wait_for(received.get(), timeout=5.0) assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE assert hist["messages"] == [] await browser_pc.close() await transport.close_all() @pytest.mark.asyncio async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path): """WebRTC DataChannel: chat message from peer A is broadcast to peer B.""" from meshbay_node.chat.store import ChatStore 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() chat_store = ChatStore(db_path=tmp_path / "chat_bc.db") await chat_store.open() transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, shared_root=shared_dir, index=indexer.index, stun_servers=[], ) transport._ctx["chat_store"] = chat_store async def _connect_peer(peer_id, jwt_sub, groups=None): pc = RTCPeerConnection() q = asyncio.Queue() b = bytearray() ch = pc.createDataChannel("mnp") sk_h_pem = sk_hub.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ) now = int(time.time()) token = jwt.encode({ "iss": "test-hub", "sub": jwt_sub, "pk_user": "test", "hub_id": "test-hub", "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600, "groups": groups or [], }, sk_h_pem, algorithm="EdDSA") @ch.on("open") def on_open(): ch.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, })) @ch.on("message") def on_msg(message): if isinstance(message, str): message = message.encode() b.extend(message) while len(b) >= 4: length = struct.unpack(">I", b[:4])[0] if len(b) < 4 + length: break msg_bytes = bytes(b[4:4 + length]) del b[:4 + length] q.put_nowait(msgpack.unpackb(msg_bytes, raw=False)) offer = await pc.createOffer() await pc.setLocalDescription(offer) answer_sdp, _ = await transport.handle_offer( pc.localDescription.sdp, peer_id) await pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) ack = await asyncio.wait_for(q.get(), timeout=5.0) assert ack["type"] == MNP.HANDSHAKE_ACK return pc, ch, q pc_a, ch_a, q_a = await _connect_peer("peer-A", "user-A") pc_b, ch_b, q_b = await _connect_peer("peer-B", "user-B") ch_a.send(_pack({ "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "hi from A", })) ack_a = await asyncio.wait_for(q_a.get(), timeout=5.0) assert ack_a["type"] == "ack" broadcast = await asyncio.wait_for(q_b.get(), timeout=5.0) assert broadcast["type"] == MNP.CHAT_MESSAGE assert broadcast["sender_id"] == "user-A" assert broadcast["payload"] == "hi from A" await chat_store.close() await pc_a.close() await pc_b.close() await transport.close_all() @pytest.mark.asyncio async def test_webrtc_group_membership_enforced(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: JWT without matching group claim 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-group-test") await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) await asyncio.sleep(0.5) token = _make_jwt(sk_hub, groups=["other-group"]) channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "my-group", })) msg = await asyncio.wait_for(received.get(), timeout=5.0) assert msg["type"] == "error" assert "Not a member" in msg["detail"] await browser_pc.close() await transport.close_all() @pytest.mark.asyncio async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: peer removed from _peers dict on session close.""" 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() buf = bytearray() channel = browser_pc.createDataChannel("mnp") @channel.on("open") def on_open(): channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": _make_jwt(sk_hub), })) @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-cleanup") await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) ack = await asyncio.wait_for(received.get(), timeout=5.0) assert ack["type"] == MNP.HANDSHAKE_ACK assert "user-001" in transport._ctx["_peers"] assert transport.active_peers == 1 await transport.close_peer("peer-cleanup") assert "user-001" not in transport._ctx["_peers"] assert transport.active_peers == 0 await browser_pc.close() @pytest.mark.asyncio async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: stream_segment for non-existent file returns 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() buf = bytearray() channel = browser_pc.createDataChannel("mnp") @channel.on("open") def on_open(): channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": _make_jwt(sk_hub), })) @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-stream") await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) ack = await asyncio.wait_for(received.get(), timeout=5.0) assert ack["type"] == MNP.HANDSHAKE_ACK channel.send(_pack({ "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION, "file_id": "nonexistent-file-id", "segment_index": 0, "segment_duration": 4, })) msg = await asyncio.wait_for(received.get(), timeout=5.0) assert msg["type"] == "error" assert "not found" in msg["detail"].lower() await browser_pc.close() await transport.close_all()