diff options
Diffstat (limited to 'packages/meshbay-node/tests')
| -rw-r--r-- | packages/meshbay-node/tests/test_http_server.py | 2 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_indexer.py | 2 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_webrtc_transport.py | 439 |
3 files changed, 422 insertions, 21 deletions
diff --git a/packages/meshbay-node/tests/test_http_server.py b/packages/meshbay-node/tests/test_http_server.py index 1483a1d..d4ccc32 100644 --- a/packages/meshbay-node/tests/test_http_server.py +++ b/packages/meshbay-node/tests/test_http_server.py @@ -156,7 +156,7 @@ async def test_chunk_public_group(sk_node, sk_hub, hub_pk_pem, shared_dir): @pytest.mark.asyncio async def test_chunk_private_group(sk_node, sk_hub, hub_pk_pem, gek, shared_dir): """Private group chunk: encrypted with GEK.""" - from meshbay_common.crypto import chunk_key as derive_chunk_key, decrypt_chunk + from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk import blake3 indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index 60f9b01..380d752 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -175,7 +175,7 @@ async def test_on_change_callback(shared_dir, sk_node, gek): await asyncio.sleep(0.1) (shared_dir / "newfile.mp4").write_bytes(os.urandom(256)) - await asyncio.sleep(0.5) # let watchdog detect the change + await asyncio.sleep(3.0) # watchdog detect + 2s debounce await indexer.stop() assert len(changes) >= 1, "on_change should have been called" diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index 4c0fdbf..d3847ff 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -13,7 +13,6 @@ import os import struct import time -import blake3 import jwt import msgpack import pytest @@ -25,10 +24,8 @@ 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.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 @@ -197,7 +194,8 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir 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 + 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") @@ -211,25 +209,24 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir 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"]) + # 4) Verify and decrypt (binary fields — no base64, minimal envelope) + ct = chunk_msg["ct"] + nonce = chunk_msg["nonce"] + file_hash = bytes.fromhex(entry.id) - 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 + 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() @@ -324,3 +321,407 @@ async def test_webrtc_request_before_handshake_rejected(sk_node, sk_hub, gek, sh 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() |