diff options
Diffstat (limited to 'packages/meshbay-node/tests')
| -rw-r--r-- | packages/meshbay-node/tests/test_daemon.py | 44 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_hub_client.py | 76 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_webrtc_transport.py | 1066 |
3 files changed, 918 insertions, 268 deletions
diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py index faf12e3..1c5a07e 100644 --- a/packages/meshbay-node/tests/test_daemon.py +++ b/packages/meshbay-node/tests/test_daemon.py @@ -7,11 +7,13 @@ Hub interaction is mocked. """ import asyncio +import base64 import os import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from unittest.mock import AsyncMock, MagicMock, patch from meshbay_common.crypto import generate_gek @@ -20,6 +22,20 @@ from meshbay_node.daemon import NodeDaemon from meshbay_node.indexer import DirectoryIndexer +def _mock_keystore_keys(sk_ed): + """Create a mock keystore with real Ed25519 + X25519 key material.""" + sk_x = X25519PrivateKey.generate() + pk_x_raw = sk_x.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + + mock_keys = MagicMock() + mock_keys.sk_ed25519 = sk_ed + mock_keys.pk_ed25519_b64 = "test" + mock_keys.sk_x25519 = sk_x + mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode() + return mock_keys + + @pytest.fixture def sk_hub(): return Ed25519PrivateKey.generate() @@ -48,7 +64,7 @@ def shared_dir(tmp_path): @pytest.fixture def node_config(tmp_path, shared_dir): return Config( - hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"), + hub=HubConfig(url="http://localhost:9999", username="testuser"), node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000), groups=[GroupConfig( id="g" * 32, @@ -70,10 +86,7 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem) daemon = NodeDaemon(node_config) sk_node = Ed25519PrivateKey.generate() - mock_keys = MagicMock() - mock_keys.sk_ed25519 = sk_node - mock_keys.pk_ed25519_b64 = "test" - mock_keys.pk_x25519_b64 = "test" + mock_keys = _mock_keystore_keys(sk_node) mock_session = MagicMock() mock_session.node_id = "node123" @@ -85,7 +98,6 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem) hub_instance = AsyncMock() hub_instance.startup = AsyncMock(return_value=mock_session) - hub_instance.fetch_gek = AsyncMock(return_value=gek) hub_instance.maintain_ws = AsyncMock() hub_instance.send_ws = AsyncMock() hub_instance._ws = None @@ -139,7 +151,7 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem) async def test_daemon_no_groups_exits(tmp_path): """Daemon with no valid groups exits cleanly.""" config = Config( - hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"), + hub=HubConfig(url="http://localhost:9999", username="testuser"), node=NodeConfig(), groups=[GroupConfig(id="", name="empty", shared_dir="")], keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), @@ -148,17 +160,22 @@ async def test_daemon_no_groups_exits(tmp_path): daemon = NodeDaemon(config) sk_node = Ed25519PrivateKey.generate() - mock_keys = MagicMock() - mock_keys.sk_ed25519 = sk_node - mock_keys.pk_ed25519_b64 = "test" + mock_keys = _mock_keystore_keys(sk_node) mock_session = MagicMock() mock_session.node_id = "node123" mock_session.user_id = "user123" mock_session.hub_pk_pem = b"pem" + mock_server = AsyncMock() + mock_server.serve = AsyncMock() + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ - patch("meshbay_node.daemon.HubClient") as MockHub: + patch("meshbay_node.daemon.HubClient") as MockHub, \ + patch("meshbay_node.daemon.uvicorn") as mock_uvicorn: + + mock_uvicorn.Config = MagicMock() + mock_uvicorn.Server = MagicMock(return_value=mock_server) hub_instance = AsyncMock() hub_instance.startup = AsyncMock(return_value=mock_session) @@ -169,7 +186,6 @@ async def test_daemon_no_groups_exits(tmp_path): await daemon.run() - assert daemon._state["status"] == "starting" assert len(daemon._chat_stores) == 0 @@ -177,7 +193,7 @@ async def test_daemon_no_groups_exits(tmp_path): async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hub_pk_pem): """Index change callback pushes updated index to WebRTC peers.""" config = Config( - hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"), + hub=HubConfig(url="http://localhost:9999", username="testuser"), node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000), groups=[GroupConfig( id="a" * 32, @@ -228,7 +244,7 @@ async def test_daemon_index_change_skips_other_group_peers( ): """Index change only pushes to peers in the same group.""" config = Config( - hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"), + hub=HubConfig(url="http://localhost:9999", username="testuser"), node=NodeConfig(), groups=[], keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), diff --git a/packages/meshbay-node/tests/test_hub_client.py b/packages/meshbay-node/tests/test_hub_client.py index fe8a2af..2975ee8 100644 --- a/packages/meshbay-node/tests/test_hub_client.py +++ b/packages/meshbay-node/tests/test_hub_client.py @@ -2,7 +2,6 @@ Tests for meshbay_node.hub_client — uses httpx.MockTransport to avoid network. """ -import base64 import json import os import time @@ -16,7 +15,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from cryptography.hazmat.primitives import serialization -from meshbay_common.crypto import generate_gek, pk_to_b64, wrap_gek + from meshbay_node.hub_client import HubClient, HubConfig, HubSession from meshbay_node.keystore import NodeKeys @@ -51,16 +50,15 @@ def hub_config(tmp_path): return HubConfig( hub_url="http://fake-hub", username="testuser", - password="testpass99", cache_dir=tmp_path, ) -def make_token(sk_pem, user_id, pk_user_b64, hub_id="fake-hub", ttl=3600): +def make_node_token(sk_pem, user_id, pk_user_b64, hub_id="fake-hub", ttl=3600): now = int(time.time()) return jwt.encode({ "iss": hub_id, "sub": user_id, "pk_user": pk_user_b64, - "hub_id": hub_id, "jti": "test-jti", + "hub_id": hub_id, "jti": "test-jti", "scope": "node", "iat": now, "exp": now + ttl, }, sk_pem, algorithm="EdDSA") @@ -71,14 +69,14 @@ def make_token(sk_pem, user_id, pk_user_b64, hub_id="fake-hub", ttl=3600): async def test_login_verifies_jwt_offline(hub_keys, node_keys, hub_config): sk_hub, sk_hub_pem, pk_hub_pem = hub_keys user_id = "user-uuid-001" - token = make_token(sk_hub_pem, user_id, node_keys.pk_ed25519_b64) + token = make_node_token(sk_hub_pem, user_id, node_keys.pk_ed25519_b64) def handler(request): if request.url.path == "/v1/hub/pubkey": return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()}) - if request.url.path == "/v1/users/login": + if request.url.path == "/v1/nodes/auth": return httpx.Response(200, json={ - "access_token": token, "refresh_token": "rt-abc", "expires_in": 3600}) + "access_token": token, "token_type": "bearer", "expires_in": 3600}) return httpx.Response(404) transport = httpx.MockTransport(handler) @@ -96,18 +94,18 @@ async def test_login_verifies_jwt_offline(hub_keys, node_keys, hub_config): @pytest.mark.asyncio async def test_login_rejects_missing_jti(hub_keys, node_keys, hub_config): sk_hub, sk_hub_pem, pk_hub_pem = hub_keys - # Token without jti bad_token = jwt.encode({ "iss": "fake-hub", "sub": "uid", "pk_user": node_keys.pk_ed25519_b64, - "hub_id": "fake-hub", "iat": int(time.time()), "exp": int(time.time()) + 3600, + "hub_id": "fake-hub", "scope": "node", + "iat": int(time.time()), "exp": int(time.time()) + 3600, }, sk_hub_pem, algorithm="EdDSA") def handler(request): if request.url.path == "/v1/hub/pubkey": return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()}) - if request.url.path == "/v1/users/login": + if request.url.path == "/v1/nodes/auth": return httpx.Response(200, json={ - "access_token": bad_token, "refresh_token": "rt", "expires_in": 3600}) + "access_token": bad_token, "token_type": "bearer", "expires_in": 3600}) return httpx.Response(404) transport = httpx.MockTransport(handler) @@ -119,33 +117,16 @@ async def test_login_rejects_missing_jti(hub_keys, node_keys, hub_config): @pytest.mark.asyncio -async def test_register_idempotent(hub_keys, node_keys, hub_config): - def handler(request): - if request.url.path == "/v1/users/register": - return httpx.Response(409, json={"detail": "Username already taken"}) - return httpx.Response(404) - - transport = httpx.MockTransport(handler) - client = HubClient(hub_config, node_keys) - client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub") - - # Should not raise on 409 - result = await client.register() - assert result == "" - - -@pytest.mark.asyncio async def test_token_needs_refresh(hub_keys, node_keys, hub_config): sk_hub, sk_hub_pem, pk_hub_pem = hub_keys - # Token expiring in 60s (< TOKEN_REFRESH_MARGIN of 300s) - short_token = make_token(sk_hub_pem, "uid", node_keys.pk_ed25519_b64, ttl=60) + short_token = make_node_token(sk_hub_pem, "uid", node_keys.pk_ed25519_b64, ttl=60) def handler(request): if request.url.path == "/v1/hub/pubkey": return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()}) - if request.url.path == "/v1/users/login": + if request.url.path == "/v1/nodes/auth": return httpx.Response(200, json={ - "access_token": short_token, "refresh_token": "rt", "expires_in": 60}) + "access_token": short_token, "token_type": "bearer", "expires_in": 60}) return httpx.Response(404) transport = httpx.MockTransport(handler) @@ -157,37 +138,6 @@ async def test_token_needs_refresh(hub_keys, node_keys, hub_config): @pytest.mark.asyncio -async def test_fetch_gek(hub_keys, node_keys, hub_config): - """Admin wraps GEK for this node; client fetches and unwraps.""" - sk_hub, sk_hub_pem, pk_hub_pem = hub_keys - gek = generate_gek() - - # Simulate admin wrapping GEK for this node - pk_x_raw = base64.b64decode(node_keys.pk_x25519_b64) - bundle = wrap_gek(gek, pk_x_raw) - - token = make_token(sk_hub_pem, "uid", node_keys.pk_ed25519_b64) - - def handler(request): - if request.url.path == "/v1/hub/pubkey": - return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()}) - if request.url.path == "/v1/users/login": - return httpx.Response(200, json={ - "access_token": token, "refresh_token": "rt", "expires_in": 3600}) - if "/v1/groups/" in request.url.path and request.url.path.endswith("/gek"): - return httpx.Response(200, json=bundle) - return httpx.Response(404) - - transport = httpx.MockTransport(handler) - client = HubClient(hub_config, node_keys) - client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub") - - await client.login() - recovered = await client.fetch_gek("group-abc") - assert recovered == gek - - -@pytest.mark.asyncio async def test_hub_pk_cached(hub_keys, node_keys, hub_config, tmp_path): _, _, pk_hub_pem = hub_keys call_count = {"n": 0} diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index b6664e8..693a68b 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -9,6 +9,8 @@ Uses local loopback (no STUN/ICE needed for localhost). import asyncio import base64 +import hashlib +import hmac import os import struct import time @@ -24,9 +26,13 @@ from meshbay_common import MNP_VERSION from meshbay_common.crypto import ( generate_gek, pk_to_b64, + wrap_gek, + wrap_gek_aes, + unwrap_gek, ) from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes from meshbay_common.protocol import MNP +from meshbay_node.bundle_store import BundleStore from meshbay_node.indexer import DirectoryIndexer from meshbay_node.transport.webrtc_server import WebRTCTransport @@ -42,6 +48,11 @@ def sk_hub(): @pytest.fixture +def sk_user(): + return Ed25519PrivateKey.generate() + + +@pytest.fixture def gek(): return generate_gek() @@ -60,7 +71,7 @@ def _hub_pk_pem(sk_hub): serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) -def _make_jwt(sk_hub, groups=None): +def _make_jwt(sk_hub, groups=None, pk_user="test"): sk_pem = sk_hub.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, @@ -69,7 +80,7 @@ def _make_jwt(sk_hub, groups=None): now = int(time.time()) return jwt.encode({ "iss": "test-hub", "sub": "user-001", - "pk_user": "test", "hub_id": "test-hub", + "pk_user": pk_user, "hub_id": "test-hub", "jti": "test-jti-webrtc", "iat": now, "exp": now + 3600, "groups": groups or [], }, sk_pem, algorithm="EdDSA") @@ -85,6 +96,108 @@ def _unpack(raw: bytes) -> dict: return msgpack.unpackb(raw[4:4 + length], raw=False) +def _extract_dtls_fp(sdp: str) -> bytes: + for line in sdp.splitlines(): + if line.startswith("a=fingerprint:sha-256 "): + return bytes.fromhex(line.split(" ", 1)[1].replace(":", "")) + return b"" + + +async def _handshake_with_gek_proof(channel, received, sk_hub, gek, groups=None, + browser_pc=None): + """Send handshake, handle GEK challenge, return handshake_ack.""" + token = _make_jwt(sk_hub, groups=groups) + channel.send(_pack({ + "type": MNP.HANDSHAKE, + "v": MNP_VERSION, + "token": token, + })) + msg = await asyncio.wait_for(received.get(), timeout=5.0) + if msg["type"] == MNP.HANDSHAKE_CHALLENGE: + nonce = base64.b64decode(msg["nonce"]) + offer_fp = b"" + answer_fp = b"" + if browser_pc: + offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp) + answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp) + proof = hmac.new(gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest() + channel.send(_pack({ + "type": MNP.HANDSHAKE_RESPONSE, + "v": MNP_VERSION, + "proof": base64.b64encode(proof).decode(), + })) + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == MNP.HANDSHAKE_ACK + return msg + + +async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None): + """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue).""" + pc = RTCPeerConnection() + q = asyncio.Queue() + buf = bytearray() + ch = pc.createDataChannel("mnp") + ready = asyncio.Event() + + @ch.on("open") + def on_open(): + ready.set() + + @ch.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] + 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")) + await asyncio.wait_for(ready.wait(), timeout=5.0) + + pk_user = "test" + if sk_user: + pk_user = base64.b64encode( + sk_user.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + ).decode() + + 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": pk_user, "hub_id": "test-hub", + "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600, + "groups": [], + }, sk_h_pem, algorithm="EdDSA") + + ch.send(_pack({"type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token})) + msg = await asyncio.wait_for(q.get(), timeout=5.0) + if msg["type"] == MNP.HANDSHAKE_CHALLENGE: + nonce = base64.b64decode(msg["nonce"]) + offer_fp = _extract_dtls_fp(pc.localDescription.sdp) + answer_fp = _extract_dtls_fp(pc.remoteDescription.sdp) + proof = hmac.new(gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest() + ch.send(_pack({ + "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION, + "proof": base64.b64encode(proof).decode(), + })) + msg = await asyncio.wait_for(q.get(), timeout=5.0) + assert msg["type"] == MNP.HANDSHAKE_ACK + return pc, ch, q + + @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.""" @@ -120,15 +233,8 @@ async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir): 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 + msg = await _handshake_with_gek_proof(channel, received, sk_hub, gek, + browser_pc=browser_pc) assert msg["v"] == MNP_VERSION assert "node_pk" in msg @@ -153,15 +259,11 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir received = asyncio.Queue() channel = browser_pc.createDataChannel("mnp") + channel_ready = asyncio.Event() @channel.on("open") def on_open(): - token = _make_jwt(sk_hub) - channel.send(_pack({ - "type": MNP.HANDSHAKE, - "v": MNP_VERSION, - "token": token, - })) + channel_ready.set() buf = bytearray() @@ -186,8 +288,11 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) - # 1) Handshake ack - ack = await asyncio.wait_for(received.get(), timeout=5.0) + await asyncio.wait_for(channel_ready.wait(), timeout=5.0) + + # 1) Handshake with GEK proof + ack = await _handshake_with_gek_proof(channel, received, sk_hub, gek, + browser_pc=browser_pc) assert ack["type"] == MNP.HANDSHAKE_ACK # 2) Request index @@ -220,13 +325,6 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir 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() @@ -342,44 +440,8 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm ) 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 + browser_pc, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-chat") channel.send(_pack({ "type": MNP.CHAT_MESSAGE, @@ -421,41 +483,8 @@ async def test_webrtc_chat_history_no_store(sk_node, sk_hub, gek, shared_dir): 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 + browser_pc, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-no-store") channel.send(_pack({ "type": MNP.CHAT_HISTORY, "v": MNP_VERSION, "since": 0, "limit": 50, @@ -487,57 +516,8 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path) ) 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") + pc_a, ch_a, q_a = await _setup_peer(transport, sk_hub, gek, "peer-A", "user-A") + pc_b, ch_b, q_b = await _setup_peer(transport, sk_hub, gek, "peer-B", "user-B") ch_a.send(_pack({ "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "hi from A", @@ -617,18 +597,509 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir): stun_servers=[], ) + browser_pc, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-cleanup") + + 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, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-stream") + + 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() + + +@pytest.mark.asyncio +async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: wrong GEK proof is rejected — hub admin can't fake membership.""" + 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("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-fake") + await browser_pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + + await asyncio.sleep(0.5) + + token = _make_jwt(sk_hub) + channel.send(_pack({ + "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, + })) + + challenge = await asyncio.wait_for(received.get(), timeout=5.0) + assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE + + fake_gek = os.urandom(32) + nonce = base64.b64decode(challenge["nonce"]) + offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp) + answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp) + bad_proof = hmac.new(fake_gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest() + channel.send(_pack({ + "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION, + "proof": base64.b64encode(bad_proof).decode(), + })) + + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == "error" + assert "GEK proof failed" in msg["detail"] + + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_dtls_channel_binding_detects_mitm(sk_node, sk_hub, gek, shared_dir): + """WebRTC: DTLS channel binding detects fingerprint substitution (simulated MitM).""" + 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-mitm") + await browser_pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + + await asyncio.sleep(0.5) + + token = _make_jwt(sk_hub) + channel.send(_pack({ + "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, + })) + + challenge = await asyncio.wait_for(received.get(), timeout=5.0) + assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE + + nonce = base64.b64decode(challenge["nonce"]) + # Correct GEK but fake fingerprints — simulates MitM substituting DTLS certs + fake_fp = os.urandom(32) + proof = hmac.new(gek, nonce + fake_fp + fake_fp, hashlib.sha256).digest() + channel.send(_pack({ + "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION, + "proof": base64.b64encode(proof).decode(), + })) + + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == "error" + assert "GEK proof failed" in msg["detail"] + + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_admin_challenge_response(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: admin file delete requires Ed25519 challenge-response.""" + 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() + + sk_admin = Ed25519PrivateKey.generate() + + 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["admin_pk_ed25519"] = sk_admin.public_key() + transport._ctx["node_user_id"] = "user-001" + + browser_pc, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-admin") + + entry = indexer.index.entries[0] + channel.send(_pack({ + "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id, + })) + + challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE + assert challenge_msg["file_id"] == entry.id + + challenge = base64.b64decode(challenge_msg["challenge"]) + signature = sk_admin.sign(challenge) + channel.send(_pack({ + "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, + "file_id": entry.id, + "signature": base64.b64encode(signature).decode(), + })) + + ack = await asyncio.wait_for(received.get(), timeout=5.0) + assert ack["type"] == MNP.FILE_DELETE_ACK + assert ack["file_id"] == entry.id + + assert indexer.index.get_entry(entry.id) is None + + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: wrong Ed25519 signature is rejected — hub can't fake admin.""" + 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() + + sk_admin = Ed25519PrivateKey.generate() + sk_attacker = Ed25519PrivateKey.generate() + + 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["admin_pk_ed25519"] = sk_admin.public_key() + transport._ctx["node_user_id"] = "user-001" + + browser_pc, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-attacker") + + entry = indexer.index.entries[0] + channel.send(_pack({ + "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id, + })) + + challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE + + challenge = base64.b64decode(challenge_msg["challenge"]) + bad_sig = sk_attacker.sign(challenge) + channel.send(_pack({ + "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, + "file_id": entry.id, + "signature": base64.b64encode(bad_sig).decode(), + })) + + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == "error" + assert "signature" in msg["detail"].lower() or "verification" in msg["detail"].lower() + + assert indexer.index.get_entry(entry.id) is not None + + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: stream_request 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, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-mse") + + channel.send(_pack({ + "type": MNP.STREAM_REQUEST, "v": MNP_VERSION, + "file_id": "nonexistent-file-id", + })) + + 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() + + +@pytest.mark.asyncio +async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, shared_dir): + """Uploader must prove Ed25519 key ownership to delete — no uploader shortcut.""" + 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() + + sk_uploader = Ed25519PrivateKey.generate() + pk_uploader_b64 = base64.b64encode( + sk_uploader.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + ).decode() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + stun_servers=[], + ) + # No admin_pk configured — only uploader_pk should authorize deletion + + browser_pc, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-uploader-del", sk_user=sk_uploader) + + # Tag an existing entry with the uploader's public key + entry = indexer.index.entries[0] + entry.uploader_id = "user-001" + entry.uploader_pk = pk_uploader_b64 + + # Request deletion — should get a challenge (no shortcut) + channel.send(_pack({ + "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id, + })) + + challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE + assert challenge_msg["file_id"] == entry.id + + # Sign with uploader's Ed25519 key + challenge = base64.b64decode(challenge_msg["challenge"]) + signature = sk_uploader.sign(challenge) + channel.send(_pack({ + "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, + "file_id": entry.id, + "signature": base64.b64encode(signature).decode(), + })) + + ack = await asyncio.wait_for(received.get(), timeout=5.0) + assert ack["type"] == MNP.FILE_DELETE_ACK + assert ack["file_id"] == entry.id + + # Verify file was removed from index + assert indexer.index.get_entry(entry.id) is None + + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, shared_dir): + """Hub-forged JWT with same sub cannot delete — wrong Ed25519 key 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() + + # User A uploaded the file + sk_user_a = Ed25519PrivateKey.generate() + pk_a_b64 = base64.b64encode( + sk_user_a.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + ).decode() + + # User B is the attacker (different Ed25519 key, but hub forges JWT with same sub) + sk_user_b = Ed25519PrivateKey.generate() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + stun_servers=[], + ) + # No admin_pk — only uploader_pk matters + + # Tag entry with user A's public key + entry = indexer.index.entries[0] + entry.uploader_id = "user-001" + entry.uploader_pk = pk_a_b64 + + # Connect as user B (same jwt_sub "user-001" via hub forgery, but B's Ed25519 key) + browser_pc, channel, received = await _setup_peer( + transport, sk_hub, gek, "peer-impersonator", + jwt_sub="user-001", sk_user=sk_user_b) + + # Request deletion — should get a challenge + channel.send(_pack({ + "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id, + })) + + challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE + + # Sign with user B's key (wrong key) + challenge = base64.b64decode(challenge_msg["challenge"]) + bad_sig = sk_user_b.sign(challenge) + channel.send(_pack({ + "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, + "file_id": entry.id, + "signature": base64.b64encode(bad_sig).decode(), + })) + + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == "error" + assert "verification" in msg["detail"].lower() or "signature" in msg["detail"].lower() + + # File must still exist in the index + assert indexer.index.get_entry(entry.id) is not None + + await browser_pc.close() + await transport.close_all() + + +# ── GEK bundle P2P exchange tests ────────────────────────────────────────── + + +@pytest.fixture +def x25519_keypair(): + from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + sk = X25519PrivateKey.generate() + sk_raw = sk.private_bytes( + serialization.Encoding.Raw, serialization.PrivateFormat.Raw, + serialization.NoEncryption()) + pk_raw = sk.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + return sk_raw, pk_raw + + +@pytest.mark.asyncio +async def test_gek_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, + tmp_path, x25519_keypair): + """GEK bundle stored on node via DataChannel, then fetched during handshake.""" + 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() + + bundle_store = BundleStore(db_path=tmp_path / "bundles.db") + await bundle_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["bundle_store"] = bundle_store + + # Connect as admin and store a GEK bundle for user-002 + pc_admin, ch_admin, q_admin = await _setup_peer( + transport, sk_hub, gek, "peer-admin") + + sk_x_raw, pk_x_raw = x25519_keypair + bundle = wrap_gek(gek, pk_x_raw) + + ch_admin.send(_pack({ + "type": MNP.GEK_BUNDLE_STORE, + "v": MNP_VERSION, + "user_id": "user-002", + "group_id": "g", + "pk_eph_b64": bundle["pk_eph_b64"], + "nonce_b64": bundle["nonce_b64"], + "wrapped_b64": bundle["wrapped_b64"], + })) + ack = await asyncio.wait_for(q_admin.get(), timeout=5.0) + assert ack["type"] == "ack" + assert ack["detail"] == "gek_bundle_stored" + + # Verify bundle was persisted + stored = await bundle_store.fetch("g", "user-002") + assert stored is not None + assert stored["pk_eph_b64"] == bundle["pk_eph_b64"] + + # Unwrap to verify it's correct + recovered = unwrap_gek(stored, sk_x_raw, pk_x_raw) + assert recovered == gek + + await bundle_store.close() + await pc_admin.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_dir, + tmp_path, x25519_keypair): + """Browser fetches GEK bundle from node during the handshake challenge window.""" + 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() + + sk_x_raw, pk_x_raw = x25519_keypair + bundle_store = BundleStore(db_path=tmp_path / "bundles.db") + await bundle_store.open() + + # Pre-populate a bundle for user-001 in group "g" + bundle = wrap_gek(gek, pk_x_raw) + await bundle_store.store("g", "user-001", + bundle["pk_eph_b64"], bundle["nonce_b64"], + bundle["wrapped_b64"]) + + 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["bundle_store"] = bundle_store + + # Connect manually: handshake → challenge → gek_bundle_fetch → response + browser_pc = RTCPeerConnection() + received = asyncio.Queue() + buf = bytearray() channel = browser_pc.createDataChannel("mnp") + ready = asyncio.Event() @channel.on("open") def on_open(): - channel.send(_pack({ - "type": MNP.HANDSHAKE, "v": MNP_VERSION, - "token": _make_jwt(sk_hub), - })) + ready.set() @channel.on("message") def on_msg(message): @@ -646,49 +1117,162 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir): offer = await browser_pc.createOffer() await browser_pc.setLocalDescription(offer) answer_sdp, _ = await transport.handle_offer( - browser_pc.localDescription.sdp, "peer-cleanup") + browser_pc.localDescription.sdp, "peer-fetch") await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) + await asyncio.wait_for(ready.wait(), timeout=5.0) + + # Step 1: Send handshake with group_id so _pending_group is set + token = _make_jwt(sk_hub, groups=["g"]) + channel.send(_pack({ + "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g", + })) + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == MNP.HANDSHAKE_CHALLENGE + + # Step 2: Fetch GEK bundle from node (during challenge window) + channel.send(_pack({"type": MNP.GEK_BUNDLE_FETCH, "v": MNP_VERSION})) + bundle_resp = await asyncio.wait_for(received.get(), timeout=5.0) + assert bundle_resp["type"] == MNP.GEK_BUNDLE_RESP + assert bundle_resp["found"] is True + + # Step 3: Unwrap GEK and compute HMAC proof + recovered_gek = unwrap_gek(bundle_resp, sk_x_raw, pk_x_raw) + assert recovered_gek == gek + nonce = base64.b64decode(msg["nonce"]) + offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp) + answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp) + proof = hmac.new(recovered_gek, nonce + offer_fp + answer_fp, + hashlib.sha256).digest() + + # Step 4: Complete handshake + channel.send(_pack({ + "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION, + "proof": base64.b64encode(proof).decode(), + })) 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 bundle_store.close() + await browser_pc.close() + await transport.close_all() - await transport.close_peer("peer-cleanup") - assert "user-001" not in transport._ctx["_peers"] - assert transport.active_peers == 0 +# ── Keypair bundle P2P tests ───────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, tmp_path): + """Keypair bundle stored on node, then fetched during handshake window.""" + 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() + bundle_store = BundleStore(db_path=tmp_path / "bundles.db") + await bundle_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["bundle_store"] = bundle_store + + # Connect and store a keypair bundle + pc1, ch1, q1 = await _setup_peer(transport, sk_hub, gek, "peer-kp-store") + ch1.send(_pack({ + "type": MNP.KEYPAIR_BUNDLE_STORE, + "v": MNP_VERSION, + "bundle_enc": "encrypted-keypair-data-base64", + })) + ack = await asyncio.wait_for(q1.get(), timeout=5.0) + assert ack["type"] == "ack" + assert ack["detail"] == "keypair_bundle_stored" + + # Verify in DB + stored = await bundle_store.fetch_keypair("user-001") + assert stored == "encrypted-keypair-data-base64" + + await pc1.close() + + # New connection: fetch during handshake window + browser_pc = RTCPeerConnection() + received = asyncio.Queue() + buf = bytearray() + channel = browser_pc.createDataChannel("mnp") + ready = asyncio.Event() + + @channel.on("open") + def on_open(): + ready.set() + + @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-kp-fetch") + await browser_pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + await asyncio.wait_for(ready.wait(), timeout=5.0) + + token = _make_jwt(sk_hub, groups=["g"]) + channel.send(_pack({ + "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g", + })) + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == MNP.HANDSHAKE_CHALLENGE + + # Fetch keypair bundle during challenge window + channel.send(_pack({"type": MNP.KEYPAIR_BUNDLE_FETCH, "v": MNP_VERSION})) + kp_resp = await asyncio.wait_for(received.get(), timeout=5.0) + assert kp_resp["type"] == MNP.KEYPAIR_BUNDLE_RESP + assert kp_resp["found"] is True + assert kp_resp["bundle_enc"] == "encrypted-keypair-data-base64" + + await bundle_store.close() await browser_pc.close() + await transport.close_all() @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.""" +async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path): + """Keypair bundle fetch returns found=false when no bundle exists.""" 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() + bundle_store = BundleStore(db_path=tmp_path / "bundles.db") + await bundle_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["bundle_store"] = bundle_store browser_pc = RTCPeerConnection() received = asyncio.Queue() buf = bytearray() - channel = browser_pc.createDataChannel("mnp") + ready = asyncio.Event() @channel.on("open") def on_open(): - channel.send(_pack({ - "type": MNP.HANDSHAKE, "v": MNP_VERSION, - "token": _make_jwt(sk_hub), - })) + ready.set() @channel.on("message") def on_msg(message): @@ -706,52 +1290,150 @@ async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_d offer = await browser_pc.createOffer() await browser_pc.setLocalDescription(offer) answer_sdp, _ = await transport.handle_offer( - browser_pc.localDescription.sdp, "peer-stream") + browser_pc.localDescription.sdp, "peer-kp-none") await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) + await asyncio.wait_for(ready.wait(), timeout=5.0) - ack = await asyncio.wait_for(received.get(), timeout=5.0) - assert ack["type"] == MNP.HANDSHAKE_ACK + token = _make_jwt(sk_hub, groups=["g"]) + channel.send(_pack({ + "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g", + })) + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == MNP.HANDSHAKE_CHALLENGE + channel.send(_pack({"type": MNP.KEYPAIR_BUNDLE_FETCH, "v": MNP_VERSION})) + resp = await asyncio.wait_for(received.get(), timeout=5.0) + assert resp["type"] == MNP.KEYPAIR_BUNDLE_RESP + assert resp["found"] is False + + await bundle_store.close() + await browser_pc.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shared_dir, + tmp_path, x25519_keypair): + """Storing the node operator's GEK bundle auto-activates GEK (AES variant).""" + 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() + + sk_x_raw, pk_x_raw = x25519_keypair + bundle_store = BundleStore(db_path=tmp_path / "bundles.db") + await bundle_store.open() + + new_gek = generate_gek() + + 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["bundle_store"] = bundle_store + transport._ctx["node_user_id"] = "node-operator" + transport._ctx["sk_x25519_raw"] = sk_x_raw + transport._ctx["pk_x25519_raw"] = pk_x_raw + transport._ctx["pk_x25519_b64"] = base64.b64encode(pk_x_raw).decode() + + pc_admin, ch_admin, q_admin = await _setup_peer( + transport, sk_hub, gek, "peer-setup-admin") + + # Store GEK bundle wrapped with AES-GCM (browser-compatible) + node_bundle = wrap_gek_aes(new_gek, pk_x_raw) + ch_admin.send(_pack({ + "type": MNP.GEK_BUNDLE_STORE, + "v": MNP_VERSION, + "user_id": "node-operator", + "group_id": "g", + "pk_eph_b64": node_bundle["pk_eph_b64"], + "nonce_b64": node_bundle["nonce_b64"], + "wrapped_b64": node_bundle["wrapped_b64"], + })) + ack = await asyncio.wait_for(q_admin.get(), timeout=5.0) + assert ack["type"] == "ack" + + await asyncio.sleep(0.2) + + assert transport._ctx.get("gek") == new_gek + + await bundle_store.close() + await pc_admin.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir): + """WebRTC DataChannel: connection refused when GEK is not initialized.""" + hub_pk_pem = _hub_pk_pem(sk_hub) + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None) + await indexer.initial_scan() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=None, + 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-no-gek") + await browser_pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + + await asyncio.sleep(0.5) + + token = _make_jwt(sk_hub) channel.send(_pack({ - "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION, - "file_id": "nonexistent-file-id", - "segment_index": 0, "segment_duration": 4, + "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, })) msg = await asyncio.wait_for(received.get(), timeout=5.0) assert msg["type"] == "error" - assert "not found" in msg["detail"].lower() + assert "not initialized" in msg["detail"].lower() await browser_pc.close() await transport.close_all() @pytest.mark.asyncio -async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir): - """WebRTC DataChannel: stream_request for non-existent file returns error.""" +async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path): + """GEK bundle fetch returns found=false when no bundle exists.""" 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() + bundle_store = BundleStore(db_path=tmp_path / "bundles.db") + await bundle_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["bundle_store"] = bundle_store browser_pc = RTCPeerConnection() received = asyncio.Queue() buf = bytearray() - channel = browser_pc.createDataChannel("mnp") + ready = asyncio.Event() @channel.on("open") def on_open(): - channel.send(_pack({ - "type": MNP.HANDSHAKE, "v": MNP_VERSION, - "token": _make_jwt(sk_hub), - })) + ready.set() @channel.on("message") def on_msg(message): @@ -769,21 +1451,23 @@ async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_d offer = await browser_pc.createOffer() await browser_pc.setLocalDescription(offer) answer_sdp, _ = await transport.handle_offer( - browser_pc.localDescription.sdp, "peer-mse") + browser_pc.localDescription.sdp, "peer-nofound") await browser_pc.setRemoteDescription( RTCSessionDescription(sdp=answer_sdp, type="answer")) + await asyncio.wait_for(ready.wait(), timeout=5.0) - ack = await asyncio.wait_for(received.get(), timeout=5.0) - assert ack["type"] == MNP.HANDSHAKE_ACK - + token = _make_jwt(sk_hub, groups=["g"]) channel.send(_pack({ - "type": MNP.STREAM_REQUEST, "v": MNP_VERSION, - "file_id": "nonexistent-file-id", + "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g", })) - msg = await asyncio.wait_for(received.get(), timeout=5.0) - assert msg["type"] == "error" - assert "not found" in msg["detail"].lower() + assert msg["type"] == MNP.HANDSHAKE_CHALLENGE + + channel.send(_pack({"type": MNP.GEK_BUNDLE_FETCH, "v": MNP_VERSION})) + resp = await asyncio.wait_for(received.get(), timeout=5.0) + assert resp["type"] == MNP.GEK_BUNDLE_RESP + assert resp["found"] is False + await bundle_store.close() await browser_pc.close() await transport.close_all() |