diff options
Diffstat (limited to 'packages/meshbay-node/tests')
| -rw-r--r-- | packages/meshbay-node/tests/test_daemon.py | 77 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_http_server.py | 227 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_quic_transport.py | 13 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_roster_pairing.py | 744 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_security_regressions.py | 604 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_transport.py | 222 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_webrtc_transport.py | 368 |
7 files changed, 1684 insertions, 571 deletions
diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py index 1c5a07e..60ef11f 100644 --- a/packages/meshbay-node/tests/test_daemon.py +++ b/packages/meshbay-node/tests/test_daemon.py @@ -21,7 +21,6 @@ from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, Keys 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() @@ -35,23 +34,33 @@ def _mock_keystore_keys(sk_ed): mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode() return mock_keys +def _free_port() -> int: + """ + A port nobody else in the session is on. + + These tests start the real admin UI server. Hardcoding 28000 made them fail + with EADDRINUSE whenever another test file had a node running — which is why + the full suite failed while each file passed on its own. + """ + import socket + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + @pytest.fixture def sk_hub(): return Ed25519PrivateKey.generate() - @pytest.fixture def hub_pk_pem(sk_hub): return sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - @pytest.fixture def gek(): return generate_gek() - @pytest.fixture def shared_dir(tmp_path): d = tmp_path / "shared" @@ -60,26 +69,22 @@ def shared_dir(tmp_path): (d / "hello.txt").write_bytes(b"hello daemon test " * 50) return d - @pytest.fixture def node_config(tmp_path, shared_dir): return Config( hub=HubConfig(url="http://localhost:9999", username="testuser"), - node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), groups=[GroupConfig( id="g" * 32, name="test-group", shared_dir=str(shared_dir), visibility="private", - port=29000, quic_port=29010, - http_port=29001, )], keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), data_dir=tmp_path / "data", ) - @pytest.mark.asyncio async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem): """Daemon creates ChatStore for each group and shuts down cleanly.""" @@ -130,7 +135,14 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem) assert daemon._chat_stores[group_id]._db is not None if daemon._webrtc: - assert "chat_store" in daemon._webrtc._ctx + # Finding H1: chat_store must live in the per-group context, never on + # the shared transport context. Hoisting the first group's store + # transport-wide sent every group's chat to one database and served it + # back to members of every other group. + assert "chat_store" not in daemon._webrtc._ctx + groups_ctx = daemon._webrtc._ctx["groups"] + assert groups_ctx[group_id]["chat_store"] is daemon._chat_stores[group_id] + assert "hub_ws" in daemon._webrtc._ctx assert "node_user_id" in daemon._webrtc._ctx assert daemon._webrtc._ctx["node_user_id"] == "user123" @@ -146,7 +158,6 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem) for store in daemon._chat_stores.values(): assert store._db is None - @pytest.mark.asyncio async def test_daemon_no_groups_exits(tmp_path): """Daemon with no valid groups exits cleanly.""" @@ -188,19 +199,18 @@ async def test_daemon_no_groups_exits(tmp_path): assert len(daemon._chat_stores) == 0 - @pytest.mark.asyncio 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"), - node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), groups=[GroupConfig( id="a" * 32, name="test-group", shared_dir=str(shared_dir), visibility="private", - port=29000, quic_port=29010, http_port=29001, + quic_port=29010, )], keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), data_dir=tmp_path / "data", @@ -232,10 +242,45 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu assert msg["group_id"] == "a" * 32 assert len(msg["entries"]) == indexer.index.count + # Finding H7: this group is private, so its content hashes must NOT be + # registered with the hub. The test previously asserted the opposite — + # publishing a fingerprint of every private file was treated as expected + # behaviour. Index push to members is unaffected (asserted above). + await asyncio.sleep(0.1) + daemon._hub.register_swarm.assert_not_called() + +@pytest.mark.asyncio +async def test_daemon_index_change_registers_swarm_for_public_group( + tmp_path, shared_dir, gek, hub_pk_pem): + """Public groups still register content hashes with the hub swarm (H7).""" + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), + groups=[GroupConfig( + id="a" * 32, + name="public-group", + shared_dir=str(shared_dir), + visibility="public", + quic_port=29010, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._hub = AsyncMock() + daemon._hub.register_swarm = AsyncMock(return_value=2) + daemon._state["endpoint_hint"] = "node123" + + indexer = DirectoryIndexer( + root=shared_dir, group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.1) daemon._hub.register_swarm.assert_called_once() - call_args = daemon._hub.register_swarm.call_args - assert len(call_args[0][0]) == indexer.index.count + assert len(daemon._hub.register_swarm.call_args[0][0]) == indexer.index.count @pytest.mark.asyncio diff --git a/packages/meshbay-node/tests/test_http_server.py b/packages/meshbay-node/tests/test_http_server.py deleted file mode 100644 index d4ccc32..0000000 --- a/packages/meshbay-node/tests/test_http_server.py +++ /dev/null @@ -1,227 +0,0 @@ -"""Tests for the node HTTP file API.""" - -import asyncio -import base64 -import json -import os -import time -import pytest -import jwt -import httpx -from pathlib import Path -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey -from cryptography.hazmat.primitives import serialization - -from meshbay_common.crypto import generate_gek, pk_to_b64 -from meshbay_node.indexer import DirectoryIndexer -from meshbay_node.transport.http_server import create_http_app - - -@pytest.fixture -def sk_node(): - return Ed25519PrivateKey.generate() - -@pytest.fixture -def sk_hub(): - return Ed25519PrivateKey.generate() - -@pytest.fixture -def hub_pk_pem(sk_hub): - return sk_hub.public_key().public_bytes( - serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - -@pytest.fixture -def gek(): - return generate_gek() - -@pytest.fixture -def shared_dir(tmp_path): - d = tmp_path / "shared" - d.mkdir() - (d / "video.mp4").write_bytes(os.urandom(3 * 1024 * 1024)) # 3MB - (d / "doc.pdf").write_bytes(os.urandom(512 * 1024)) - (d / "song.mp3").write_bytes(os.urandom(256 * 1024)) - return d - -def make_token(sk_hub, pk_node_b64, ttl=3600): - 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": pk_node_b64, "hub_id": "test-hub", - "jti": "test-jti", "iat": now, "exp": now + ttl, - }, sk_pem, algorithm="EdDSA") - - -@pytest.mark.asyncio -async def test_node_info(sk_node, sk_hub, hub_pk_pem, gek, shared_dir): - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) - await indexer.initial_scan() - - app = create_http_app( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, - shared_root=shared_dir, index=indexer.index, - group_id="test-group", group_name="Test Group", - ) - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" - ) as c: - r = await c.get("/") - assert r.status_code == 200 - data = r.json() - assert data["group_id"] == "test-group" - assert data["file_count"] == 3 - assert "pk_node" in data - - -@pytest.mark.asyncio -async def test_public_index(sk_node, sk_hub, hub_pk_pem, shared_dir): - """Public group: index accessible without auth.""" - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None) - await indexer.initial_scan() - - app = create_http_app( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, - shared_root=shared_dir, index=indexer.index, - group_id="pub-group", group_name="Public Group", - gek=None, - ) - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" - ) as c: - r = await c.get("/index") - assert r.status_code == 200 - data = r.json() - assert len(data["entries"]) == 3 - names = {e["name"] for e in data["entries"]} - assert "video.mp4" in names - assert "doc.pdf" in names - - -@pytest.mark.asyncio -async def test_file_download(sk_node, sk_hub, hub_pk_pem, shared_dir): - """Full file download via HTTP.""" - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None) - await indexer.initial_scan() - - app = create_http_app( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, - shared_root=shared_dir, index=indexer.index, - group_id="g", group_name="G", - ) - entry = next(e for e in indexer.index.entries if e.name == "doc.pdf") - original = (shared_dir / "doc.pdf").read_bytes() - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" - ) as c: - r = await c.get(f"/file/{entry.id}") - assert r.status_code == 200 - assert r.content == original - - -@pytest.mark.asyncio -async def test_chunk_public_group(sk_node, sk_hub, hub_pk_pem, shared_dir): - """Public group chunk: plaintext, signed, auth required.""" - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None) - await indexer.initial_scan() - - app = create_http_app( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, - shared_root=shared_dir, index=indexer.index, - group_id="g", group_name="G", gek=None, - ) - entry = next(e for e in indexer.index.entries if e.name == "video.mp4") - token = make_token(sk_hub, pk_to_b64(sk_node.public_key())) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" - ) as c: - r = await c.get(f"/file/{entry.id}/0", - headers={"Authorization": f"Bearer {token}"}) - assert r.status_code == 200 - chunk = r.json() - assert chunk["encrypted"] is False - assert chunk["chunk_index"] == 0 - assert "data_b64" in chunk - - # Verify the chunk data matches original - original = (shared_dir / "video.mp4").read_bytes() - data = base64.b64decode(chunk["data_b64"]) - assert data == original[:len(data)] - - -@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.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) - await indexer.initial_scan() - - app = create_http_app( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, - shared_root=shared_dir, index=indexer.index, - group_id="g", group_name="G", gek=gek, - ) - entry = next(e for e in indexer.index.entries if e.name == "doc.pdf") - token = make_token(sk_hub, pk_to_b64(sk_node.public_key())) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" - ) as c: - r = await c.get(f"/file/{entry.id}/0", - headers={"Authorization": f"Bearer {token}"}) - assert r.status_code == 200 - chunk = r.json() - assert chunk["encrypted"] is True - - # Decrypt and verify - file_hash = base64.b64decode(chunk["file_hash_b64"]) - nonce = base64.b64decode(chunk["nonce_b64"]) - ct = base64.b64decode(chunk["ct_b64"]) - ckey = derive_chunk_key(gek, file_hash, 0) - plaintext = decrypt_chunk(ckey, nonce, ct) - original = (shared_dir / "doc.pdf").read_bytes() - assert plaintext == original[:len(plaintext)] - - -@pytest.mark.asyncio -async def test_chunk_requires_auth(sk_node, hub_pk_pem, shared_dir): - """Chunk endpoint rejects unauthenticated requests.""" - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None) - await indexer.initial_scan() - app = create_http_app( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, - shared_root=shared_dir, index=indexer.index, - group_id="g", group_name="G", - ) - entry = indexer.index.entries[0] - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" - ) as c: - r = await c.get(f"/file/{entry.id}/0") # no token - assert r.status_code == 401 - - -@pytest.mark.asyncio -async def test_unknown_file_404(sk_node, hub_pk_pem, sk_hub, shared_dir): - indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None) - await indexer.initial_scan() - app = create_http_app( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, - shared_root=shared_dir, index=indexer.index, - group_id="g", group_name="G", - ) - token = make_token(sk_hub, pk_to_b64(sk_node.public_key())) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" - ) as c: - r = await c.get("/file/nonexistent-hash/0", - headers={"Authorization": f"Bearer {token}"}) - assert r.status_code == 404 diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py index 0c1a1cd..93ab1b0 100644 --- a/packages/meshbay-node/tests/test_quic_transport.py +++ b/packages/meshbay-node/tests/test_quic_transport.py @@ -49,7 +49,8 @@ def make_jwt(sk_hub, pk_node_b64, ttl=3600, groups=None): "iss": "test-hub", "sub": "user-001", "pk_user": pk_node_b64, "hub_id": "test-hub", "jti": "test-jti", "iat": now, "exp": now + ttl, - "groups": groups or [], + # group_id is mandatory (M1), so default tokens are members of "g". + "groups": groups if groups is not None else ["g"], }, sk_pem, algorithm="EdDSA") @@ -80,6 +81,7 @@ async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path): host="127.0.0.1", port=19100, jwt_token=token, gek=gek, pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="g", ) as client: chunk0 = await client.fetch_chunk(entry.id, chunk_index=0) chunk1 = await client.fetch_chunk(entry.id, chunk_index=1) @@ -116,6 +118,7 @@ async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path): host="127.0.0.1", port=19101, jwt_token=token, gek=gek, pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="g", ) as client: wire = await client.fetch_index() recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) @@ -220,10 +223,12 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat async with QuicChunkClient( host="127.0.0.1", port=19104, jwt_token=token, gek=gek, pk_node_b64=pk_b64, + group_id="g", ) as client: wire = await client.fetch_index() assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 saved_ticket = client.session_ticket + saved_cert = client.peer_cert_der # Allow server to process the close await asyncio.sleep(0.1) @@ -233,6 +238,10 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat host="127.0.0.1", port=19104, jwt_token=token, gek=gek, pk_node_b64=pk_b64, session_ticket=saved_ticket, + # A resumed session carries no certificate, so the binding anchor from the + # original handshake travels with the ticket (11.5.6). + peer_cert_der=saved_cert, + group_id="g", ) as client: wire = await client.fetch_index() assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 @@ -269,6 +278,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p async with QuicChunkClient( host="127.0.0.1", port=19105, jwt_token=token, gek=gek, pk_node_b64=pk_b64, + group_id="g", ) as client: wire = await client.fetch_index() assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 @@ -281,6 +291,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p async with QuicChunkClient( host="127.0.0.1", port=19105, jwt_token=token, gek=gek, pk_node_b64=pk_b64, + group_id="g", ) as client: await client.fetch_index() diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py new file mode 100644 index 0000000..665c060 --- /dev/null +++ b/packages/meshbay-node/tests/test_roster_pairing.py @@ -0,0 +1,744 @@ +""" +Roster and operator pairing (M3, and the mechanism that will close H3). + +Negative assertions, per the posture set in Phase 11.5: each test states an attack +or a mistake that must not work. The one to keep an eye on is +`test_daemon_does_not_auto_pin_keystore_key` — the auto-pin is what made node +sovereignty inert as shipped, and it fails closed, so nothing else in the suite +notices if it comes back. + +See `docs/invite-pairing-v1.md`. +""" + +import base64 +import time +from pathlib import Path + +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 meshbay_common.crypto import generate_gek, pk_to_b64, unwrap_gek_aes +from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import Roster, hash_code, normalize_code +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + +@pytest.fixture +async def roster(tmp_path): + r = Roster(db_path=tmp_path / "roster.db") + await r.open() + yield r + await r.close() + + +def _keypair_full(): + """(sk_ed, pk_ed_b64, pk_x_b64, sk_x) — the X25519 secret is needed to unwrap.""" + sk_ed = Ed25519PrivateKey.generate() + sk_x = X25519PrivateKey.generate() + pk_ed_b64 = pk_to_b64(sk_ed.public_key()) + pk_x_b64 = base64.b64encode( + sk_x.public_key().public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + ).decode() + return sk_ed, pk_ed_b64, pk_x_b64, sk_x + + +def _keypair(): + sk_ed, pk_ed_b64, pk_x_b64, _ = _keypair_full() + return sk_ed, pk_ed_b64, pk_x_b64 + + +def _session(tmp_path: Path, roster, user_id: str = "grenet", + group_id: str | None = None, gek: bytes | None = None, + join_policy: str = "invite") -> WebRTCPeerSession: + """A peer session with the join path wired and sending stubbed out.""" + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = { + "shared_root": shared_root, + "index": index, + "sk_node": index.sk_node, + "roster": roster, + } + if group_id: + session._ctx["groups"] = { + group_id: { + "gek": gek, + "shared_root": shared_root, + "index": index, + "join_policy": join_policy, + }, + } + session._group_id = group_id + session._user_id = user_id + session._username = user_id + session._pk_user = "" + session._uploads = {} + session._join_attempts = 0 + session._nonce_node = b"\x11" * 32 + session._remote_ip = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="", user_id="grenet", + group_id="", nonce=None, ts=None): + ts = int(time.time()) if ts is None else ts + transcript = join_transcript( + node_pk_b64=session._node_pk_b64(), + group_id=group_id, + user_id=user_id, + pk_ed25519_b64=pk_ed_b64, + pk_x25519_b64=pk_x_b64, + nonce_node=nonce if nonce is not None else session._nonce_node, + ts=ts, + ) + return { + "type": "join_request", + "group_id": group_id, + "pk_ed25519": pk_ed_b64, + "pk_x25519": pk_x_b64, + "code": code, + "ts": ts, + "sig": base64.b64encode(sk_ed.sign(transcript)).decode(), + } + + +def _last(session): + return session.sent[-1] if session.sent else {} + + +# ── Roster ──────────────────────────────────────────────────────────────────── + +async def test_invite_is_single_use(roster): + code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + assert await roster.consume_invite(code, "grenet") is not None + assert await roster.consume_invite(code, "grenet") is None, ( + "a pairing code must not be redeemable twice") + + +async def test_invite_is_bound_to_one_account(roster): + """A leaked code must be useless to whoever finds it.""" + code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + assert await roster.consume_invite(code, "eve") is None + assert await roster.consume_invite(code, "grenet") is not None + + +async def test_expired_invite_is_refused(roster): + code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli", ttl=-1) + assert await roster.consume_invite(code, "grenet") is None + + +async def test_reinvite_supersedes_the_previous_code(roster): + first = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + second = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + assert await roster.consume_invite(first, "grenet") is None + assert await roster.consume_invite(second, "grenet") is not None + + +async def test_codes_are_not_stored_in_the_clear(roster, tmp_path): + code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + rows = await roster.list_invites() + assert rows and rows[0]["code_hash"] != normalize_code(code) + assert rows[0]["code_hash"] == hash_code(code) + + +def test_code_normalization_absorbs_human_error(): + """Someone reading a code aloud must not be able to get it wrong.""" + assert normalize_code("k7m2-qx4p") == normalize_code("K7M2QX4P") + assert normalize_code("O1IL") == "0111" + assert normalize_code(" k7m2 qx4p ") == "K7M2QX4P" + + +async def test_operator_pks_reflect_unpinning(roster): + _, pk_ed_b64, pk_x_b64 = _keypair() + await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code") + await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") + assert await roster.operator_pks() == [pk_ed_b64] + + await roster.unpin("grenet") + assert await roster.operator_pks() == [], ( + "authority must disappear with the pin, without a daemon restart") + + +# ── Join / pairing over MNP ─────────────────────────────────────────────────── + +async def test_pairing_with_a_valid_code_pins_the_identity(tmp_path, roster): + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code)) + + assert _last(session).get("ok") is True + pinned = await roster.get_identity("grenet") + assert pinned["pk_ed25519"] == pk_ed_b64 + assert await roster.operator_pks() == [pk_ed_b64] + + +async def test_pairing_without_a_code_is_refused(tmp_path, roster): + """Fails closed: an unknown identity gets nothing until someone authorizes it.""" + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + + await session._do_join_request(_join_msg(session, sk_ed, pk_ed_b64, pk_x_b64)) + + assert _last(session).get("ok") is False + assert _last(session).get("reason") == "code_required" + assert await roster.get_identity("grenet") is None + + +async def test_wrong_code_pins_nothing(tmp_path, roster): + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="ZZZZ-ZZZZ")) + + assert _last(session).get("reason") == "code_invalid" + assert await roster.get_identity("grenet") is None + + +async def test_signature_must_cover_the_presented_keys(tmp_path, roster): + """ + The heart of it: the X25519 key is only trustworthy because the Ed25519 + identity signed it. Swapping in another encryption key after signing must fail. + """ + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + + msg = _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code) + _, _, attacker_pk_x = _keypair() + msg["pk_x25519"] = attacker_pk_x + + await session._do_join_request(msg) + + assert _last(session).get("reason") == "signature_invalid" + assert await roster.get_identity("grenet") is None + + +async def test_join_cannot_be_replayed_onto_another_connection(tmp_path, roster): + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + + # Signed against a nonce this connection never issued. + msg = _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code, + nonce=b"\x99" * 32) + await session._do_join_request(msg) + + assert _last(session).get("reason") == "signature_invalid" + assert await roster.get_identity("grenet") is None + + +async def test_pinned_identity_presenting_a_new_key_is_refused(tmp_path, roster): + """ + 11.5.8's rule, applied to people: a changed key is refused outright rather + than warned about, and clearing it is a deliberate operator action. + """ + session = _session(tmp_path, roster) + _, old_pk_ed, old_pk_x = _keypair() + await roster.pin_identity("grenet", "grenet", old_pk_ed, old_pk_x, "code") + + sk_ed2, new_pk_ed, new_pk_x = _keypair() + await session._do_join_request( + _join_msg(session, sk_ed2, new_pk_ed, new_pk_x, code="ANY-CODE")) + + assert _last(session).get("reason") == "key_changed" + assert (await roster.get_identity("grenet"))["pk_ed25519"] == old_pk_ed + + +async def test_attempts_are_bounded(tmp_path, roster): + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") + + for _ in range(6): + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA")) + + assert any(m.get("detail") == "Too many attempts" for m in session.sent), ( + "a connection must not be able to sit there guessing codes") + + +async def test_failures_are_counted_across_connections(tmp_path, roster): + """ + The adversary who can mint a token for any account is the hub, and it can + reconnect at will — so a per-connection budget alone would bound nothing. + """ + shared_ctx = None + for _ in range(6): + session = _session(tmp_path, roster) + if shared_ctx is None: + shared_ctx = session._ctx + else: + session._ctx = shared_ctx # same node, new connection + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + for _ in range(4): + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA")) + + assert any(m.get("detail") == "Pairing temporarily locked" + for m in session.sent), ( + "reconnecting must not reset the pairing budget") + + +async def test_group_id_cannot_name_another_group(tmp_path, roster): + session = _session(tmp_path, roster) + session._group_id = "a" * 32 + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, group_id="b" * 32)) + + assert _last(session).get("reason") == "group_mismatch" + + +# ── H3: the node wraps the group key, and only for people it admitted ───────── + +GROUP = "g" * 32 + + +async def test_node_wraps_the_gek_for_the_key_the_member_proved(tmp_path, roster): + """ + The H3 fix. Nobody fetches a public key from the hub: the node encrypts the + group key for the X25519 key the joiner signed with their pinned identity, so + a hub substituting a key of its own has nothing to substitute into. + """ + gek = generate_gek() + session = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek) + sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full() + + code = await roster.create_invite(GROUP, "bob", ROLE_MEMBER, "grenet") + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code, + user_id="bob", group_id=GROUP)) + + reply = _last(session) + assert reply["ok"] is True and reply["gek"] is True + + pk_x_raw = base64.b64decode(pk_x_b64) + sk_x_raw = sk_x.private_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PrivateFormat.Raw, + encryption_algorithm=serialization.NoEncryption(), + ) + assert unwrap_gek_aes(reply, sk_x_raw, pk_x_raw) == gek + + +async def test_hub_membership_alone_yields_no_key(tmp_path, roster): + """ + A hub can invent an account, add it to a group and mint it a token. What it + cannot do is put it on the node's roster — so the key never leaves. + """ + gek = generate_gek() + session = _session(tmp_path, roster, user_id="eve", group_id=GROUP, gek=gek) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + + # Pinned on this node (say, for another group) but never admitted to this one. + await roster.pin_identity("eve", "eve", pk_ed_b64, pk_x_b64, "code") + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + user_id="eve", group_id=GROUP)) + + reply = _last(session) + assert reply.get("gek") is False + assert reply.get("reason") == "not_authorized_for_group" + assert "wrapped_b64" not in reply + + +async def test_open_join_group_admits_without_a_code(tmp_path, roster): + """§3.4: where anyone may join, a code protects nothing and is not required.""" + gek = generate_gek() + session = _session(tmp_path, roster, user_id="newcomer", group_id=GROUP, + gek=gek, join_policy="open") + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + user_id="newcomer", group_id=GROUP)) + + reply = _last(session) + assert reply["ok"] is True and reply["gek"] is True + pinned = await roster.get_identity("newcomer") + assert pinned["pinned_via"] == "tofu" + + +async def test_invite_only_group_still_demands_a_code(tmp_path, roster): + """Being public (discoverable) is not being open (admitting anyone).""" + gek = generate_gek() + session = _session(tmp_path, roster, user_id="newcomer", group_id=GROUP, + gek=gek, join_policy="invite") + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + user_id="newcomer", group_id=GROUP)) + + assert _last(session).get("reason") == "code_required" + assert await roster.get_identity("newcomer") is None + + +async def test_unknown_group_is_invite_only(tmp_path, roster): + """ + Fail closed: a group whose policy the node cannot read is treated as + invite-only, never as open. + """ + session = _session(tmp_path, roster, user_id="newcomer") + session._group_id = "unconfigured-group" + assert session._group_join_policy("unconfigured-group") == "invite" + assert session._group_join_policy("") == "invite" + + +def test_join_policy_is_carried_from_node_config(): + """ + The policy reaches the transport from node.toml. If it ever came from the hub + instead, a hub could declare any group open and be handed its key. + """ + daemon_src = (Path(__file__).parent.parent + / "src" / "meshbay_node" / "daemon.py").read_text() + assert '"join_policy": group_cfg.join_policy' in daemon_src + + config_src = (Path(__file__).parent.parent + / "src" / "meshbay_node" / "config.py").read_text() + assert "join_policy" in config_src, "GroupConfig must carry the admission policy" + + +async def test_revoked_member_stops_receiving_the_key(tmp_path, roster): + """ + Wrapping on demand is what makes revocation work. A stored bundle survived + revocation; this does not. (Rotating the GEK is still required — the + ex-member has the old one.) + """ + gek = generate_gek() + session = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code") + await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "grenet") + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + user_id="bob", group_id=GROUP)) + assert _last(session)["gek"] is True + + await roster.set_status(GROUP, "bob", "revoked") + session.sent.clear() + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + user_id="bob", group_id=GROUP)) + assert _last(session).get("gek") is False + + +# ── What a first-time joiner can know ───────────────────────────────────────── + +def test_challenge_carries_node_pk_in_source(): + """ + Belt and braces for the above: the field must be in the message the node + builds, whatever the surrounding handshake does. + """ + source = (Path(__file__).parent.parent + / "src" / "meshbay_node" / "transport" / "webrtc_server.py").read_text() + challenge = source[source.find("MNP.HANDSHAKE_CHALLENGE,"):] + challenge = challenge[:challenge.find("})")] + assert "node_pk" in challenge, ( + "the challenge must announce the node key — a first-time joiner cannot " + "learn it any other way, and join_request signs it") + + +async def test_a_key_pinned_by_one_node_is_worthless_at_another(tmp_path, roster): + """ + The whole point of per-node identity: node A's operator who cracks the bundle + on their own disk holds a key node B has never seen. Presenting it there is a + first contact like any other — it needs a code from B's operator. + """ + gek = generate_gek() + node_b = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek) + + # The key bob uses at node A. Node B's roster knows nothing about it. + sk_ed_a, pk_ed_a, pk_x_a = _keypair() + + await node_b._do_join_request( + _join_msg(node_b, sk_ed_a, pk_ed_a, pk_x_a, + user_id="bob", group_id=GROUP)) + + assert _last(node_b).get("reason") == "code_required" + assert await roster.get_identity("bob") is None + + +async def test_the_stolen_key_cannot_be_forced_in_with_someone_elses_code( + tmp_path, roster): + """And a code issued for another account does not help either.""" + gek = generate_gek() + session = _session(tmp_path, roster, user_id="eve", group_id=GROUP, gek=gek) + sk_ed, pk_ed, pk_x = _keypair() + code = await roster.create_invite(GROUP, "bob", ROLE_MEMBER, "grenet") + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed, pk_x, code=code, + user_id="eve", group_id=GROUP)) + + assert _last(session).get("reason") == "code_invalid" + assert await roster.get_identity("eve") is None + + +# ── Code lifetimes ──────────────────────────────────────────────────────────── + +async def test_invitations_outlive_pairing_codes(roster): + """ + An invitation crosses a human conversation; a pairing code crosses an SSH + session. A day was long enough for the second and not for the first — a code + that dies over a weekend means someone has to be at a browser to reissue it. + """ + from meshbay_node.roster import DEFAULT_INVITE_TTL, DEFAULT_PAIR_TTL + + assert DEFAULT_INVITE_TTL == 7 * 24 * 3600 + assert DEFAULT_PAIR_TTL == 24 * 3600 + assert DEFAULT_INVITE_TTL > DEFAULT_PAIR_TTL + + +def test_code_lifetimes_are_configurable(tmp_path): + """The operator decides, not the default.""" + from meshbay_node.config import load_config + + path = tmp_path / "node.toml" + path.write_text( + '[hub]\nurl = "https://example.org"\nusername = "grenet"\n' + "[node]\ninvite_ttl_hours = 72\npair_ttl_hours = 2\n" + ) + cfg = load_config(path) + assert cfg.node.invite_ttl_hours == 72 + assert cfg.node.pair_ttl_hours == 2 + + default = load_config(tmp_path / "missing.toml") + assert default.node.invite_ttl_hours == 168 + assert default.node.pair_ttl_hours == 24 + + +async def test_expiry_is_enforced_at_redemption(tmp_path, roster): + """Purging is housekeeping; the check that matters happens on use.""" + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli", ttl=-1) + + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code)) + + assert _last(session).get("reason") == "code_invalid" + assert await roster.get_identity("grenet") is None + + +# ── M3: where node authority comes from ─────────────────────────────────────── + +async def test_admin_signature_verified_against_the_paired_key(tmp_path, roster): + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code") + await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") + + transcript = b"meshbay:admin:v1 whatever" + assert await session._verify_admin_sig(transcript, sk_ed.sign(transcript)) + + stranger = Ed25519PrivateKey.generate() + assert not await session._verify_admin_sig( + transcript, stranger.sign(transcript)) + + +async def test_unpinned_operator_loses_authority_immediately(tmp_path, roster): + """No caching: revoking a paired browser must not need a daemon restart.""" + session = _session(tmp_path, roster) + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code") + await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") + + transcript = b"meshbay:admin:v1 whatever" + assert await session._verify_admin_sig(transcript, sk_ed.sign(transcript)) + + await roster.unpin("grenet") + assert not await session._verify_admin_sig(transcript, sk_ed.sign(transcript)) + + +# ── Operator surface (slice 3) ──────────────────────────────────────────────── + +def _ui_client(tmp_path, roster, **extra): + from fastapi.testclient import TestClient + + from meshbay_node.config import Config + from meshbay_node.ui.app import create_ui_app + + state = { + "status": "running", "groups_ctx": {GROUP: {"gek": b"k" * 32}}, + "indexes": {}, "ui_token": "tok", "roster": roster, + "node_user_id": "grenet", "config": Config(), + } + state.update(extra) + return TestClient(create_ui_app(state)), state + + +async def test_revoke_endpoint_stops_authorization(tmp_path, roster): + client, _ = _ui_client(tmp_path, roster) + _, pk_ed_b64, pk_x_b64 = _keypair() + await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code") + await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "grenet") + assert await roster.is_authorized(GROUP, "bob") + + resp = client.post(f"/api/members/bob/revoke?group_id={GROUP}&t=tok") + assert resp.status_code == 200 + assert "gek-init" in resp.json()["reminder"], ( + "revocation must remind the operator to rotate the key they still hold") + assert not await roster.is_authorized(GROUP, "bob") + + +async def test_unpin_endpoint_allows_repairing(tmp_path, roster): + client, _ = _ui_client(tmp_path, roster) + _, pk_ed_b64, pk_x_b64 = _keypair() + await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code") + + assert client.post("/api/members/bob/unpin?t=tok").status_code == 200 + assert await roster.get_identity("bob") is None + assert client.post("/api/members/bob/unpin?t=tok").status_code == 404 + + +async def test_operator_surface_needs_the_session_token(tmp_path, roster): + """11.5.3 applies to every one of these: they change who may hold the key.""" + client, _ = _ui_client(tmp_path, roster) + for path in ("/api/roster", + "/api/operator/pair", + f"/api/members/bob/revoke?group_id={GROUP}", + "/api/members/bob/unpin", + f"/api/groups/{GROUP}/invites?username=bob"): + method = client.get if path == "/api/roster" else client.post + assert method(path).status_code == 403, f"{path} reachable without a token" + + +async def test_cli_invite_asks_the_hub_for_an_account_never_a_key(tmp_path, roster): + """ + The CLI resolves a username to an account id through the hub, and stops there. + A key fetched from the hub is what H3 was; an account id is not a secret and + a wrong one produces an invite whose code the hub never learns. + """ + class _Hub: + _session = object() + + async def get_user_pubkeys(self, username): + return {"user_id": f"id-of-{username}", + "pk_x25519": "SHOULD-NOT-BE-USED", + "pk_ed25519": "SHOULD-NOT-BE-USED"} + + client, _ = _ui_client(tmp_path, roster, hub=_Hub()) + resp = client.post(f"/api/groups/{GROUP}/invites?username=bob&t=tok") + assert resp.status_code == 200 + body = resp.json() + assert body["user_id"] == "id-of-bob" + + invites = await roster.list_invites() + assert [i["user_id"] for i in invites] == ["id-of-bob"] + # Whatever the hub said about keys was never stored anywhere. + assert "SHOULD-NOT-BE-USED" not in str(invites) + assert await roster.get_identity("id-of-bob") is None + + +def _run_cli(monkeypatch, tmp_path, argv, responses): + """Drive the real CLI with the daemon API stubbed, capturing the calls.""" + import sys as _sys + + from meshbay_node import daemon as _daemon + + calls = [] + + def fake_api(cfg, path, method="GET", timeout=30): + calls.append((method, path)) + for key, value in responses.items(): + if key in path: + return value + return {} + + monkeypatch.setattr(_daemon, "_daemon_api", fake_api) + + conf = tmp_path / "node.toml" + conf.write_text( + f'data_dir = "{tmp_path}"\n' + '[hub]\nurl = "https://example.org"\nusername = "grenet"\n' + f'[[groups]]\nid = "{GROUP}"\nname = "demo"\n' + f'shared_dir = "{tmp_path}"\n' + ) + monkeypatch.setattr(_sys, "argv", + ["meshbay-node", *argv, "--config", str(conf)]) + try: + _daemon.main() + except SystemExit as e: + calls.append(("exit", e.code)) + return calls + + +def test_cli_member_commands_reach_the_right_endpoints(monkeypatch, tmp_path, capsys): + resolved = {"user_id": "u-bob", "source": "roster"} + + calls = _run_cli(monkeypatch, tmp_path, ["member", "revoke", "bob"], + {"/api/resolve": resolved, + "revoke": {"status": "revoked", "reminder": "gek-init"}}) + assert ("POST", f"/api/members/u-bob/revoke?group_id={GROUP}") in calls + # The operator is told the revocation does not take back the key they hold. + assert "rotate" in capsys.readouterr().out.lower() + + calls = _run_cli(monkeypatch, tmp_path, ["member", "unpin", "bob"], + {"/api/resolve": resolved, "unpin": {"status": "unpinned"}}) + assert ("POST", "/api/members/u-bob/unpin") in calls + + +def test_cli_resolves_a_name_before_acting(monkeypatch, tmp_path): + """ + The name has to be turned into an account first, and the node's own roster is + asked before the hub. A JWT carries no username, so an identity pinned without + an invitation has none — the hub fallback is what keeps it manageable. + """ + calls = _run_cli(monkeypatch, tmp_path, ["member", "revoke", "bob"], + {"/api/resolve": {"user_id": "u-bob", "source": "hub"}, + "revoke": {"status": "revoked", "reminder": "gek-init"}}) + + assert ("GET", "/api/resolve?username=bob") == calls[0], ( + "the CLI must resolve the name before acting on anyone") + assert ("POST", f"/api/members/u-bob/revoke?group_id={GROUP}") in calls + +def test_daemon_does_not_auto_pin_keystore_key(): + """ + M3: the daemon used to auto-pin its own keystore key as the admin key, while + the browser signs with the user's identity key. Different keys, so every + privileged operation failed closed with a signature error that looked like a + bug elsewhere — and the demo only worked because a deploy script overwrote it. + + Authority now comes from the roster, or from an explicit node.toml value. + """ + source = (Path(__file__).parent.parent + / "src" / "meshbay_node" / "daemon.py").read_text() + assert "Auto-pinning admin key" not in source + assert "_resolve_admin_pk" not in source, ( + "the auto-pin resolver is back — node authority must be established " + "locally by pairing, never inferred from the node's own keystore (M3)") + + +def test_admin_authority_is_never_fetched_from_the_hub(): + """ + The fix M3 invites: ask the hub which key belongs to the operator. That would + hand a malicious hub the node — the same substitution as H3, one level deeper. + """ + source = (Path(__file__).parent.parent + / "src" / "meshbay_node" / "daemon.py").read_text() + admin_region = source[source.find("_legacy_admin_pk"):] + assert "pubkeys" not in admin_region.split("def ")[1], ( + "node authority must never be resolved through a hub lookup") diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py new file mode 100644 index 0000000..dcd9cf6 --- /dev/null +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -0,0 +1,604 @@ +""" +Phase 11.5 security regression tests. + +Each test here encodes a finding from `second-review.md`. They are negative tests: +they assert that an attack does NOT work. The pre-11.5 code passed 209 feature +tests while every one of these attacks succeeded — the suite only ever exercised +happy paths, never an authorization boundary. + +If one of these starts failing, a fix has been reverted. Do not "fix" the test. +""" + +import base64 +import struct +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.protocol import IndexEntry +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + + +def _safe_name_re(): + """ + Imported lazily so that a missing allowlist fails the two tests that need it, + rather than aborting collection of the whole module and hiding every other + finding's result. + """ + from meshbay_node.transport.webrtc_server import SAFE_UPLOAD_NAME + return SAFE_UPLOAD_NAME + + +# ── C1: the unauthenticated HTTP file API must stay deleted ─────────────────── + +def test_http_file_api_is_gone(): + """ + C1: transport/http_server.py served GET /index and GET /file/{id} on 0.0.0.0 + with no authentication, for private groups too. It was deleted rather than + patched. Re-adding any module that serves file bytes outside the MNP handshake + reintroduces a full confidentiality bypass. + """ + with pytest.raises(ImportError): + import meshbay_node.transport.http_server # noqa: F401 + + import meshbay_node.transport as transport + assert not hasattr(transport, "create_http_app") + + +def test_tcp_transport_is_gone(): + """C6: the TCP+TLS server accepted a bare JWT with no GEK proof.""" + with pytest.raises(ImportError): + import meshbay_node.transport.server # noqa: F401 + + import meshbay_node.transport as transport + assert not hasattr(transport, "ChunkServer") + + +def test_daemon_exposes_no_plaintext_listener(): + """ + C1: the daemon must not bind anything that serves content without a handshake. + NodeConfig no longer carries an HTTP port at all. + """ + from meshbay_node.config import NodeConfig, GroupConfig + + assert "http_port" not in NodeConfig.__dataclass_fields__ + assert "http_port" not in GroupConfig.__dataclass_fields__ + assert "port" not in NodeConfig.__dataclass_fields__ + + +# ── C5a: upload filename allowlist ─────────────────────────────────────────── + +@pytest.mark.parametrize("name", [ + "../../etc/passwd", + "..\\windows\\system32", + "/absolute/path", + "<img src=x onerror=alert(1)>", # the H2 stored-XSS vector + 'name";DROP TABLE x;--', + ".hidden", + "", + "a" * 200, + "file\x00.mp4", + "sub/dir/file.mp4", +]) +def test_upload_rejects_unsafe_filenames(name): + """C5a/H2: only a conservative allowlist may reach the filesystem.""" + assert not _safe_name_re().match(name), f"should be rejected: {name!r}" + + +@pytest.mark.parametrize("name", [ + "movie.mp4", + "My Holiday Video.mkv", + "report-2026.pdf", + "track_01.flac", +]) +def test_upload_accepts_ordinary_filenames(name): + """The allowlist must not break normal use.""" + assert _safe_name_re().match(name), f"should be accepted: {name!r}" + + +def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: + """A peer session wired to a real shared root, with sending stubbed out.""" + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + ctx = {"shared_root": shared_root, "index": index, "sk_node": index.sk_node} + + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = None + session._user_id = user_id + session._pk_user = "" + session._uploads = {} + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def test_upload_cannot_overwrite_another_members_file(tmp_path): + """ + C5a: uploads used to land in the shared root under a client-chosen name and + overwrite whatever was there. That let any member destroy the operator's files, + and — by becoming the recorded uploader of the replaced file — delete them + through the uploader path, bypassing the Ed25519 admin challenge entirely. + """ + victim = _session(tmp_path, "victim-user") + shared_root = victim._ctx["shared_root"] + + original = shared_root / "important.mp4" + original.write_bytes(b"operator's original content") + + attacker = _session(tmp_path, "attacker-user") + attacker._do_file_upload({ + "filename": "important.mp4", + "chunk_index": 0, + "total_chunks": 1, + "data": base64.b64encode(b"attacker content").decode(), + }) + + assert original.read_bytes() == b"operator's original content" + uploaded = shared_root / ".uploads" / "attacker-user" / "important.mp4" + assert uploaded.exists(), "upload should be quarantined, not dropped" + assert uploaded.read_bytes() == b"attacker content" + + +def test_upload_rejects_out_of_order_chunks(tmp_path): + """C5a: chunk_index > 0 used to append blindly to any .part file on disk.""" + session = _session(tmp_path, "user-1") + session._do_file_upload({ + "filename": "movie.mp4", "chunk_index": 3, "total_chunks": 5, + "data": base64.b64encode(b"spliced").decode(), + }) + assert any(m.get("type") == "error" for m in session.sent) + + +def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): + """C5a: even the original uploader goes through a fresh name, not an overwrite.""" + session = _session(tmp_path, "user-1") + payload = {"filename": "movie.mp4", "chunk_index": 0, "total_chunks": 1, + "data": base64.b64encode(b"first").decode()} + session._do_file_upload(dict(payload)) + session.sent.clear() + + session._do_file_upload(dict(payload)) + assert any(m.get("type") == "error" for m in session.sent) + stored = session._ctx["shared_root"] / ".uploads" / "user-1" / "movie.mp4" + assert stored.read_bytes() == b"first" + + +# ── H1: group isolation ────────────────────────────────────────────────────── + +def test_chat_store_and_peers_are_per_group(tmp_path): + """ + H1: chat_store and the peer registry were read from the shared transport + context, so on a multi-group node every group's messages went to the first + group's database and were served back to members of every other group. + """ + index_a = GroupIndex(group_id="a" * 32, sk_node=Ed25519PrivateKey.generate()) + index_b = GroupIndex(group_id="b" * 32, sk_node=Ed25519PrivateKey.generate()) + groups = { + "a" * 32: {"chat_store": "STORE_A", "index": index_a, "shared_root": tmp_path}, + "b" * 32: {"chat_store": "STORE_B", "index": index_b, "shared_root": tmp_path}, + } + ctx = {"groups": groups} + + sess_a = WebRTCPeerSession.__new__(WebRTCPeerSession) + sess_a._ctx, sess_a._group_id, sess_a._user_id = ctx, "a" * 32, "alice" + + sess_b = WebRTCPeerSession.__new__(WebRTCPeerSession) + sess_b._ctx, sess_b._group_id, sess_b._user_id = ctx, "b" * 32, "bob" + + assert sess_a._group_ctx()["chat_store"] == "STORE_A" + assert sess_b._group_ctx()["chat_store"] == "STORE_B" + + sess_a._peer_registry()["alice"] = sess_a + sess_b._peer_registry()["bob"] = sess_b + + # Alice's broadcast target set must not contain Bob, who is in another group. + assert "bob" not in sess_a._peer_registry() + assert "alice" not in sess_b._peer_registry() + + sess_a._user_names()["alice"] = "Alice" + assert "alice" not in sess_b._user_names() + + +def test_daemon_sets_no_global_chat_store(tmp_path): + """H1: the daemon must not hoist one group's chat store onto the transport.""" + source = (Path(__file__).parent.parent + / "src" / "meshbay_node" / "daemon.py").read_text() + assert '_ctx["chat_store"]' not in source, ( + "daemon must not assign a transport-wide chat_store — it leaks chat " + "across groups (H1)" + ) + + +# ── H2: node admin UI escaping ─────────────────────────────────────────────── + +def test_no_member_can_hand_the_node_key_material(tmp_path): + """ + C5b, strengthened by the invite redesign (docs/invite-pairing-v1.md). + + This test used to assert that `gek_bundle_store` answered with an admin + challenge and stored nothing without an operator signature. The message is now + gone entirely: the node holds the GEK and wraps it itself, so no member ever + submits key material, authorized or not. Deleting the path is a stronger + guarantee than gating it, which is why the assertion changed rather than the + behaviour regressing. + """ + from meshbay_common.protocol import MNP as _MNP + + assert not hasattr(_MNP, "GEK_BUNDLE_STORE"), ( + "the member-supplied bundle message is back — the node must never accept " + "key material over MNP (C5b)" + ) + + source = (Path(__file__).parent.parent + / "src" / "meshbay_node" / "transport" / "webrtc_server.py").read_text() + assert "_do_gek_bundle_store" not in source + assert "_admin_exec_bundle_store" not in source + + +def test_unknown_message_stores_nothing(tmp_path): + """A peer sending the retired message must not reach any storage path.""" + session = _session(tmp_path, "ordinary-member") + session._group_id = None + session._admin_ops = {} + + stored = [] + + class _Store: + async def store(self, *args): + stored.append(args) + + session._ctx["bundle_store"] = _Store() + session._handle_message({ + "type": "gek_bundle_store", + "user_id": "victim", "group_id": "g" * 32, + "pk_eph_b64": "AA==", "nonce_b64": "AA==", "wrapped_b64": "AA==", + }) + + assert stored == [], "a retired message type still reached the bundle store" + + +def test_gek_auto_activation_is_gone(): + """ + C5b: the node used to unwrap and adopt any bundle addressed to the operator. + Since the operator's X25519 public key is public, any member could hand the + node a GEK of their choosing. Nothing arriving over MNP may set a live GEK. + """ + source = (Path(__file__).parent.parent / "src" / "meshbay_node" + / "transport" / "webrtc_server.py").read_text() + assert "_try_activate_gek" not in source + assert 'unwrap_gek_aes' not in source, ( + "the MNP path must not unwrap a GEK — activation is local-admin only" + ) + + +# ── H5: admin challenge is bound, not a blind signing oracle ───────────────── + +def _transcript(**kw): + from meshbay_common.adminop import admin_transcript + base = dict(op="file_delete", node_pk_b64="NODEPK", group_id="g" * 32, + subject="file-1", nonce=b"\x01" * 32, ts=1_700_000_000) + base.update(kw) + return admin_transcript(**base) + + +def test_admin_transcript_is_domain_separated(): + """H5: signatures here can never be valid in another MeshBay protocol.""" + assert _transcript().startswith(b"meshbay:admin:v1") + + +@pytest.mark.parametrize("field,value", [ + ("op", "invite_create"), + ("subject", "file-2"), + ("node_pk_b64", "OTHERNODE"), + ("group_id", "h" * 32), + ("nonce", b"\x02" * 32), + ("ts", 1_700_000_001), +]) +def test_admin_transcript_binds_every_field(field, value): + """ + H5: a signature must not carry over to another operation, subject, node, + group, challenge or moment in time. + """ + assert _transcript() != _transcript(**{field: value}), ( + f"transcript ignores {field} — signature would be reusable" + ) + + +def test_admin_transcript_is_unambiguous(): + """ + H5/L4: fields are length-prefixed. With plain concatenation a crafted subject + could impersonate the following field and two different operations would + produce identical signed bytes. + """ + a = _transcript(subject="file-1", group_id="g") + b = _transcript(subject="1", group_id="gfile-") + assert a != b, "concatenation is ambiguous — length prefixes missing" + + +def test_admin_signature_does_not_transfer_between_operations(tmp_path): + """ + H5: the concrete attack. A signature collected to delete a file must not + authorize storing a GEK bundle. + """ + from meshbay_common.adminop import OP_FILE_DELETE, OP_INVITE_CREATE + + sk_admin = Ed25519PrivateKey.generate() + delete_transcript = _transcript(op=OP_FILE_DELETE) + signature = sk_admin.sign(delete_transcript) + + invite_transcript = _transcript(op=OP_INVITE_CREATE) + with pytest.raises(Exception): + sk_admin.public_key().verify(signature, invite_transcript) + + +def test_admin_challenge_expires(tmp_path): + """H5: a stale challenge must not be usable.""" + import time as _time + from meshbay_common.adminop import ADMIN_CHALLENGE_TTL, OP_FILE_DELETE + + session = _session(tmp_path, "operator") + session._group_id = None + session._admin_ops = { + "op-1": { + "op": OP_FILE_DELETE, "subject": "file-1", "nonce": b"\x00" * 32, + "ts": int(_time.time()) - ADMIN_CHALLENGE_TTL - 5, "payload": {}, + } + } + session._do_admin_response({"op_id": "op-1", "signature": ""}) + assert any(m.get("type") == "error" and "expired" in m.get("detail", "").lower() + for m in session.sent) + + +def test_denylist_persists_and_honours_groups(tmp_path): + """ + H4: revocations lived only in memory, so a node restart silently un-revoked + everyone, and 'group' targets were dropped entirely — the hub signed and + broadcast them, the node's handler understood only 'user' and 'jti'. + """ + from meshbay_node.transport import Denylist + + path = tmp_path / "denylist.json" + first = Denylist(path=path) + first.deny_group("g-revoked") + first.deny_user("u-revoked") + first.deny_jti("j-revoked") + + # A fresh instance stands in for a daemon restart. + reloaded = Denylist(path=path) + assert reloaded.is_denied("", "", "g-revoked"), "group revocation not honoured" + assert reloaded.is_denied("u-revoked", "") + assert reloaded.is_denied("", "j-revoked") + assert not reloaded.is_denied("someone", "other", "g-allowed") + + +def test_swarm_registration_skips_private_groups(): + """ + H7: the daemon registered content hashes for every group, private included, + handing the hub a fingerprint of every private file. The bug was masked by a + mis-mounted route, so fixing the route without this filter would have turned a + dormant leak into a live one. + """ + source = (Path(__file__).parent.parent / "src" / "meshbay_node" + / "daemon.py").read_text() + assert 'visibility' in source and '_register_swarm' in source + # Both registration sites must gate on public visibility. + for marker in ['gctx.get("visibility") != "public"', + 'group_cfg.visibility == "public"']: + assert marker in source, f"swarm registration not gated: {marker}" + + +def test_keystore_argon2_is_production_strength(): + """M2: the keystore protects the node's private keys and sat at 64 MB.""" + from meshbay_common.crypto import ARGON2_MEMORY_COST + assert ARGON2_MEMORY_COST >= 262144 + + +def test_keystore_records_argon2_params_for_migration(tmp_path): + """ + M2: raising the parameters must not orphan existing keystores, so each + envelope records the parameters it was written with. + """ + import json + from meshbay_node.keystore import create_keystore, load_keystore + + path = tmp_path / "keystore.enc" + created = create_keystore(path=path, password="correct horse battery") + envelope = json.loads(path.read_text()) + assert envelope["argon2"]["memory_cost"] >= 262144 + + reopened = load_keystore(path=path, password="correct horse battery") + assert reopened.pk_ed25519_b64 == created.pk_ed25519_b64 + + +def test_legacy_keystore_still_opens(tmp_path): + """M2: a keystore written under the 64 MB profile must still unlock.""" + import base64 as _b64 + import json + import msgpack + from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + from meshbay_common.crypto import ( + LEGACY_ARGON2_ITERATIONS, LEGACY_ARGON2_LANES, LEGACY_ARGON2_MEMORY_COST, + derive_keystore_key, encrypt_keystore, pk_to_b64, sk_to_b64, + ) + from meshbay_node.keystore import load_keystore + + sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate() + payload = msgpack.packb({ + "sk_ed25519_b64": sk_to_b64(sk_ed), + "sk_x25519_b64": sk_to_b64(sk_x), + }, use_bin_type=True) + + salt = b"\x01" * 16 + key = derive_keystore_key( + "legacy-pass", salt, + iterations=LEGACY_ARGON2_ITERATIONS, + memory_cost=LEGACY_ARGON2_MEMORY_COST, + lanes=LEGACY_ARGON2_LANES, + ) + iv, ct, tag = encrypt_keystore(payload, key) + + path = tmp_path / "legacy.enc" + # No "argon2" key — exactly how pre-M2 envelopes look. + path.write_text(json.dumps({ + "version": 1, + "argon2_salt_b64": _b64.b64encode(salt).decode(), + "iv_b64": _b64.b64encode(iv).decode(), + "tag_b64": _b64.b64encode(tag).decode(), + "ciphertext_b64": _b64.b64encode(ct).decode(), + })) + + keys = load_keystore(path=path, password="legacy-pass") + assert keys.pk_ed25519_b64 == pk_to_b64(sk_ed.public_key()) + assert keys.pk_x25519_b64 == pk_to_b64(sk_x.public_key()) + + +def test_dead_gek_protocol_constants_removed(): + """L1: the node never serves a GEK; the message types should not suggest it.""" + from meshbay_common.protocol import MNP + assert not hasattr(MNP, "GEK_REQUEST") + assert not hasattr(MNP, "GEK_RESPONSE") + + +def test_peer_errors_do_not_leak_internals(): + """ + L3: arbitrary exception text carries filesystem paths and internal state, so + the catch-all handler must not relay it. + + Deliberately narrow: HandshakeError messages ARE sent to the peer, because a + client needs to know why it was refused, and those strings are authored for + that purpose. The check targets the generic `except Exception as e` path. + """ + source = (Path(__file__).parent.parent / "src" / "meshbay_node" + / "transport" / "webrtc_server.py").read_text() + assert '"detail": str(e)' not in source, ( + "generic exception text relayed to peer — use a fixed message" + ) + # And the catch-all must still exist, sending something opaque. + assert '"detail": "Request failed"' in source + + +def test_pre_handshake_message_budget_is_small(): + """ + H6: the frame limit was a flat 64 MB applied before authentication, so an + unauthenticated peer could announce a huge frame and dribble bytes into it. + """ + from meshbay_node.transport.webrtc_server import ( + MAX_MSG, PRE_HANDSHAKE_MAX_MSG, _DataChannelBuffer, + ) + assert PRE_HANDSHAKE_MAX_MSG <= 1024 * 1024 + assert PRE_HANDSHAKE_MAX_MSG < MAX_MSG + + buf = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG) + buf.feed(struct.pack(">I", PRE_HANDSHAKE_MAX_MSG + 1) + b"x") + with pytest.raises(ValueError): + list(buf.messages()) + + +def test_stream_segment_is_not_synchronous(): + """ + H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop, + stalling every peer on the node for up to thirty seconds per request. + + Asserts the property (the worker is a coroutine, ffmpeg is spawned through + asyncio) rather than grepping for "subprocess.run" — which also matches the + comment that documents the old behaviour. + """ + import ast + import inspect + from meshbay_node.transport.webrtc_server import WebRTCPeerSession + + assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async) + + source = (Path(__file__).parent.parent / "src" / "meshbay_node" + / "transport" / "webrtc_server.py").read_text() + tree = ast.parse(source) + blocking = [ + node for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "run" + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "subprocess" + ] + assert not blocking, "blocking subprocess.run() in the event loop" + assert "_transcode_sem" in source, "ffmpeg spawns must be capped" + + +def test_pre_proof_fetches_are_bounded(): + """C4: the pre-proof bundle window is a disclosure surface; bound it.""" + from meshbay_node.transport.webrtc_server import MAX_PRE_PROOF_FETCHES + assert 0 < MAX_PRE_PROOF_FETCHES <= 10 + + +def test_node_admin_ui_requires_token(): + """ + 11.5.3: "localhost only" is not authentication. Any local process — or a + rebound browser page — could re-initialise a group's GEK and read the audit log. + """ + from fastapi.testclient import TestClient + from meshbay_node.ui.app import create_ui_app + + app = create_ui_app({"status": "running", "groups_ctx": {}, + "indexes": {}, "ui_token": "secret-token"}) + client = TestClient(app) + + assert client.get("/api/status").status_code == 403 + assert client.get("/api/status?t=wrong").status_code == 403 + assert client.get("/api/config?t=wrong").status_code == 403 + assert client.get("/api/status?t=secret-token").status_code == 200 + assert client.get( + "/api/status", headers={"X-MeshBay-Token": "secret-token"} + ).status_code == 200 + + +def test_admin_ui_escapes_filenames(tmp_path): + """ + H2: filenames are chosen by any group member and were rendered into the + localhost admin UI unescaped, giving script execution against an + unauthenticated admin API. + """ + from meshbay_node.ui.app import _render_page + + payload = '<img src=x onerror="fetch(1)">' + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + index.add_entry(IndexEntry( + id="0" * 64, name=payload, path="", size=1, type="video", added_at=0, + )) + + html = _render_page({ + "status": "running", + "groups_ctx": {"g" * 32: {"index": index, "shared_root": tmp_path}}, + "indexes": {"g" * 32: index}, + }) + + assert payload not in html, "filename rendered unescaped — stored XSS (H2)" + assert "<img" in html, "filename should appear escaped" + + +def test_admin_ui_escapes_roster_usernames(tmp_path): + """ + H2 again, for the roster: usernames originate at the hub and land on the + operator's own admin page, which can re-key groups and read the audit log. + """ + from meshbay_node.ui.app import _render_page + + payload = '<img src=x onerror="fetch(1)">' + html = _render_page( + {"status": "running", "groups_ctx": {}, "indexes": {}}, + { + "identities": {"u1": {"user_id": "u1", "username": payload, + "pk_ed25519": "AAA", "pinned_at": "now", + "pinned_via": "code"}}, + "members": [{"group_id": "", "user_id": "u1", "role": "operator", + "status": "active"}], + "invites": [], + }, + ) + + assert payload not in html, "username rendered unescaped — stored XSS (H2)" + assert "<img" in html diff --git a/packages/meshbay-node/tests/test_transport.py b/packages/meshbay-node/tests/test_transport.py deleted file mode 100644 index 0e70d72..0000000 --- a/packages/meshbay-node/tests/test_transport.py +++ /dev/null @@ -1,222 +0,0 @@ -""" -Integration test: ChunkServer ↔ ChunkClient over TLS. - -Starts a real TLS server on localhost, connects a client, -fetches index and a chunk, verifies signature+hash+decryption. -""" - -import asyncio -import base64 -import os -import time -import jwt -import pytest -from pathlib import Path -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey -from cryptography.hazmat.primitives import serialization - -from meshbay_common.crypto import generate_gek, pk_to_b64 -from meshbay_node.indexer import DirectoryIndexer, GroupIndex -from meshbay_node.transport.server import ChunkServer -from meshbay_node.transport.client import ChunkClient - - -@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.mp4").write_bytes(os.urandom(2 * 1024 * 1024)) # 2 MB - (d / "small.txt").write_bytes(b"hello meshbay " * 100) - return d - -def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600, 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_id, - "pk_user": pk_node_b64, "hub_id": "test-hub", - "jti": "test-jti", - "iat": now, "exp": now + ttl, - "groups": groups or [], - }, sk_pem, algorithm="EdDSA") - - -@pytest.mark.asyncio -async def test_chunk_server_client_roundtrip( - sk_node, sk_hub, gek, shared_dir, tmp_path): - """Full integration: server serves a chunk, client verifies and decrypts.""" - - # Build index - indexer = DirectoryIndexer( - root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) - await indexer.initial_scan() - assert indexer.index.count == 2 - - # Hub PK for JWT verification - hub_pk_pem = sk_hub.public_key().public_bytes( - serialization.Encoding.PEM, - serialization.PublicFormat.SubjectPublicKeyInfo) - - # TLS cert in tmp dir - cert_path = tmp_path / "node.crt" - key_path = tmp_path / "node.key" - - server = ChunkServer( - sk_node=sk_node, - hub_pk_pem=hub_pk_pem, - gek=gek, - shared_root=shared_dir, - index=indexer.index, - host="127.0.0.1", - port=0, # OS picks a free port - cert_path=cert_path, - key_path=key_path, - ) - await server.start() - port = server._server.sockets[0].getsockname()[1] - - token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key())) - - # Find the large test file in the index - entry = next(e for e in indexer.index.entries if e.name == "test.mp4") - - async with ChunkClient( - host="127.0.0.1", - port=port, - jwt_token=token, - gek=gek, - pk_node_b64=pk_to_b64(sk_node.public_key()), - ) as client: - # Fetch first chunk - chunk0 = await client.fetch_chunk(entry.id, chunk_index=0) - assert len(chunk0) == 1024 * 1024 # first 1MB of 2MB file - - # Fetch second chunk - chunk1 = await client.fetch_chunk(entry.id, chunk_index=1) - assert len(chunk1) == 1024 * 1024 # second 1MB - - # Reassembled file matches original - original = (shared_dir / "test.mp4").read_bytes() - assert chunk0 + chunk1 == original - - await server.stop() - - -@pytest.mark.asyncio -async def test_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path): - hub_pk_pem = sk_hub.public_key().public_bytes( - serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - - indexer = DirectoryIndexer(root=shared_dir, group_id="g", - sk_node=sk_node, gek=gek) - await indexer.initial_scan() - - cert_path = tmp_path / "node.crt" - key_path = tmp_path / "node.key" - - server = ChunkServer( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, - host="127.0.0.1", port=0, - cert_path=cert_path, key_path=key_path, - ) - await server.start() - port = server._server.sockets[0].getsockname()[1] - - # Use a different hub key to sign the token - sk_other_hub = Ed25519PrivateKey.generate() - bad_token = make_jwt(sk_other_hub, pk_to_b64(sk_node.public_key())) - - with pytest.raises(Exception): - async with ChunkClient( - host="127.0.0.1", port=port, - jwt_token=bad_token, gek=gek, - pk_node_b64=pk_to_b64(sk_node.public_key()), - ) as client: - pass - - await server.stop() - - -@pytest.mark.asyncio -async def test_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path): - """TCP+TLS server rejects a client whose JWT groups don't include the requested group_id.""" - hub_pk_pem = sk_hub.public_key().public_bytes( - serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - - indexer = DirectoryIndexer(root=shared_dir, group_id="g", - sk_node=sk_node, gek=gek) - await indexer.initial_scan() - - cert_path = tmp_path / "node.crt" - key_path = tmp_path / "node.key" - - server = ChunkServer( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, - host="127.0.0.1", port=0, - cert_path=cert_path, key_path=key_path, - ) - await server.start() - port = server._server.sockets[0].getsockname()[1] - - token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) - - with pytest.raises(ConnectionError, match="rejected"): - async with ChunkClient( - host="127.0.0.1", port=port, - jwt_token=token, gek=gek, - pk_node_b64=pk_to_b64(sk_node.public_key()), - group_id="group-b", - ) as client: - pass - - await server.stop() - - -@pytest.mark.asyncio -async def test_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path): - hub_pk_pem = sk_hub.public_key().public_bytes( - serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) - indexer = DirectoryIndexer(root=shared_dir, group_id="g", - sk_node=sk_node, gek=gek) - await indexer.initial_scan() - - cert_path = tmp_path / "node.crt" - key_path = tmp_path / "node.key" - - server = ChunkServer( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - shared_root=shared_dir, index=indexer.index, - host="127.0.0.1", port=0, - cert_path=cert_path, key_path=key_path, - ) - await server.start() - port = server._server.sockets[0].getsockname()[1] - token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key())) - - async with ChunkClient( - host="127.0.0.1", port=port, jwt_token=token, - gek=gek, pk_node_b64=pk_to_b64(sk_node.public_key()), - ) as client: - wire = await client.fetch_index() - recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) - assert recovered.count == 2 - - await server.stop() diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index 693a68b..93cd3fd 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -19,7 +19,9 @@ import jwt import msgpack import pytest from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, Ed25519PublicKey, +) from aiortc import RTCPeerConnection, RTCSessionDescription from meshbay_common import MNP_VERSION @@ -29,10 +31,24 @@ from meshbay_common.crypto import ( wrap_gek, wrap_gek_aes, unwrap_gek, + unwrap_gek_aes, ) from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes from meshbay_common.protocol import MNP +TEST_GROUP = "g" + +from meshbay_common.handshake import ( + NONCE_LEN, ROLE_CLIENT, ROLE_NODE, handshake_transcript, + make_proof, verify_proof, webrtc_binding, +) +from meshbay_common.adminop import ( + OP_FILE_DELETE, + OP_INVITE_CREATE, + admin_transcript, +) +from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript from meshbay_node.bundle_store import BundleStore +from meshbay_node.roster import Roster from meshbay_node.indexer import DirectoryIndexer from meshbay_node.transport.webrtc_server import WebRTCTransport @@ -72,6 +88,9 @@ def _hub_pk_pem(sk_hub): def _make_jwt(sk_hub, groups=None, pk_user="test"): + # group_id is mandatory now (M1), so the default token must be a member + # of the group the tests connect to. Tests that exercise refusal pass + # groups=[...] explicitly. sk_pem = sk_hub.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, @@ -82,10 +101,25 @@ def _make_jwt(sk_hub, groups=None, pk_user="test"): "iss": "test-hub", "sub": "user-001", "pk_user": pk_user, "hub_id": "test-hub", "jti": "test-jti-webrtc", "iat": now, "exp": now + 3600, - "groups": groups or [], + "groups": groups if groups is not None else [TEST_GROUP], }, sk_pem, algorithm="EdDSA") +def _transcript_from(challenge_msg: dict) -> bytes: + """ + Rebuild the signed transcript from an admin_challenge, the way a real client + does — from the announced fields, never from opaque bytes on the wire (H5). + """ + return admin_transcript( + op=challenge_msg["op"], + node_pk_b64=challenge_msg["node_pk"], + group_id=challenge_msg["group_id"], + subject=challenge_msg["subject"], + nonce=base64.b64decode(challenge_msg["nonce"]), + ts=challenge_msg["ts"], + ) + + def _pack(obj: dict) -> bytes: data = msgpack.packb(obj, use_bin_type=True) return struct.pack(">I", len(data)) + data @@ -103,36 +137,80 @@ def _extract_dtls_fp(sdp: str) -> bytes: 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) +async def _do_mnp_handshake(channel, received, token, gek, pc, group_id): + """ + Client half of the unified handshake (11.5.4): client nonce, length-prefixed + role-bound transcript, and verification of the node's own proof + signature. + """ + nonce_c = os.urandom(NONCE_LEN) channel.send(_pack({ - "type": MNP.HANDSHAKE, - "v": MNP_VERSION, - "token": token, + "type": MNP.HANDSHAKE, "v": MNP_VERSION, + "token": token, "group_id": group_id, + "nonce": base64.b64encode(nonce_c).decode(), })) 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) + if msg["type"] != MNP.HANDSHAKE_CHALLENGE: + return msg + + nonce_s = base64.b64decode(msg["nonce"]) + binding = webrtc_binding( + _extract_dtls_fp(pc.localDescription.sdp), + _extract_dtls_fp(pc.remoteDescription.sdp), + ) + proof = make_proof(gek, ROLE_CLIENT, group_id, nonce_c, nonce_s, binding) + 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) + + if ack.get("type") == MNP.HANDSHAKE_ACK: + # The client must authenticate the node too (C3). + assert verify_proof( + gek, base64.b64decode(ack["proof"]), ROLE_NODE, + group_id, nonce_c, nonce_s, binding), "node proof invalid" + transcript = handshake_transcript( + ROLE_NODE, group_id, nonce_c, nonce_s, binding) + Ed25519PublicKey.from_public_bytes( + base64.b64decode(ack["node_pk"]) + ).verify(base64.b64decode(ack["sig"]), transcript) + return ack + + +async def _handshake_with_gek_proof(channel, received, sk_hub, gek, groups=None, + browser_pc=None, group_id=TEST_GROUP): + """Send handshake, handle GEK challenge, return handshake_ack.""" + token = _make_jwt(sk_hub, groups=groups or [group_id]) + msg = await _do_mnp_handshake( + channel, received, token, gek, browser_pc, group_id) 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).""" +def _token(sk_hub, jwt_sub, peer_id, group_id, pk_user="test"): + """A hub-issued user token, as the browser would present it.""" + sk_h_pem = sk_hub.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + now = int(time.time()) + return 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": [group_id], "scope": "user", + }, sk_h_pem, algorithm="EdDSA") + + +async def _open_channel(transport, peer_id): + """ + Signaling only: a live DataChannel with no MNP handshake performed. + + Separate from `_setup_peer` because someone joining a group for the first time + cannot complete the handshake — they have no GEK to prove — and the join has to + happen in that window. + """ pc = RTCPeerConnection() q = asyncio.Queue() buf = bytearray() @@ -161,6 +239,13 @@ async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_us 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) + return pc, ch, q + + +async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None, + group_id=TEST_GROUP): + """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue).""" + pc, ch, q = await _open_channel(transport, peer_id) pk_user = "test" if sk_user: @@ -169,31 +254,9 @@ async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_us 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") + token = _token(sk_hub, jwt_sub, peer_id, group_id, pk_user) - 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) + msg = await _do_mnp_handshake(ch, q, token, gek, pc, group_id) assert msg["type"] == MNP.HANDSHAKE_ACK return pc, ch, q @@ -676,6 +739,8 @@ async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir) token = _make_jwt(sk_hub) channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, + "group_id": TEST_GROUP, + "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(), })) challenge = await asyncio.wait_for(received.get(), timeout=5.0) @@ -734,6 +799,8 @@ async def test_webrtc_dtls_channel_binding_detects_mitm(sk_node, sk_hub, gek, sh token = _make_jwt(sk_hub) channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, + "group_id": TEST_GROUP, + "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(), })) challenge = await asyncio.wait_for(received.get(), timeout=5.0) @@ -783,13 +850,13 @@ async def test_webrtc_admin_challenge_response(sk_node, sk_hub, gek, shared_dir) 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 + assert challenge_msg["op"] == OP_FILE_DELETE + assert challenge_msg["subject"] == entry.id - challenge = base64.b64decode(challenge_msg["challenge"]) - signature = sk_admin.sign(challenge) + signature = sk_admin.sign(_transcript_from(challenge_msg)) channel.send(_pack({ "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, - "file_id": entry.id, + "op_id": challenge_msg["op_id"], "signature": base64.b64encode(signature).decode(), })) @@ -832,11 +899,10 @@ async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_ 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) + bad_sig = sk_attacker.sign(_transcript_from(challenge_msg)) channel.send(_pack({ "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, - "file_id": entry.id, + "op_id": challenge_msg["op_id"], "signature": base64.b64encode(bad_sig).decode(), })) @@ -914,14 +980,14 @@ async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, s 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 + assert challenge_msg["op"] == OP_FILE_DELETE + assert challenge_msg["subject"] == entry.id # Sign with uploader's Ed25519 key - challenge = base64.b64decode(challenge_msg["challenge"]) - signature = sk_uploader.sign(challenge) + signature = sk_uploader.sign(_transcript_from(challenge_msg)) channel.send(_pack({ "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, - "file_id": entry.id, + "op_id": challenge_msg["op_id"], "signature": base64.b64encode(signature).decode(), })) @@ -979,11 +1045,10 @@ async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, share 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) + bad_sig = sk_user_b.sign(_transcript_from(challenge_msg)) channel.send(_pack({ "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, - "file_id": entry.id, + "op_id": challenge_msg["op_id"], "signature": base64.b64encode(bad_sig).decode(), })) @@ -1014,54 +1079,122 @@ def x25519_keypair(): @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.""" +async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_dir, + tmp_path, x25519_keypair): + """ + The whole invite flow over a real DataChannel, end to end. + + The operator asks for a code; the invitee — who has never held the group key + and therefore cannot complete the GEK proof — redeems it in the pre-proof + window and the node wraps the key for the X25519 key they just proved they + hold. At no point is a public key fetched from the hub, which is the point: + that lookup was H3. + """ 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() + roster = Roster(db_path=tmp_path / "roster.db") + await roster.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 + transport._ctx["roster"] = roster + transport._ctx["has_admin_authority"] = True + transport._ctx["groups"] = { + TEST_GROUP: {"gek": gek, "shared_root": shared_dir, "index": indexer.index}, + } + + # A paired operator, as `meshbay-node operator pair` would have left it. + sk_admin = Ed25519PrivateKey.generate() + admin_pk_b64 = pk_to_b64(sk_admin.public_key()) + await roster.pin_identity("user-001", "grenet", admin_pk_b64, "AA==", "code") + await roster.set_member("", "user-001", ROLE_OPERATOR, "active", "local-cli") - # 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) + # 1. The operator asks the node for an invitation code. + ch_admin.send(_pack({ + "type": MNP.INVITE_CREATE, "v": MNP_VERSION, + "user_id": "user-002", "group_id": TEST_GROUP, "username": "bob", + })) + challenge_msg = await asyncio.wait_for(q_admin.get(), timeout=5.0) + assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE + assert challenge_msg["op"] == OP_INVITE_CREATE + assert challenge_msg["subject"] == "user-002" 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"], + "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, + "op_id": challenge_msg["op_id"], + "signature": base64.b64encode( + sk_admin.sign(_transcript_from(challenge_msg))).decode(), })) - ack = await asyncio.wait_for(q_admin.get(), timeout=5.0) - assert ack["type"] == "ack" - assert ack["detail"] == "gek_bundle_stored" + invite = await asyncio.wait_for(q_admin.get(), timeout=5.0) + assert invite["type"] == MNP.INVITE_RESULT + code = invite["code"] + assert code and len(code) == 9 # XXXX-XXXX - # 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"] + # 2. Bob connects. He cannot prove GEK possession — he has never had it — so + # he redeems the code in the pre-proof window instead. + sk_x_raw, pk_x_raw = x25519_keypair + sk_bob_ed = Ed25519PrivateKey.generate() + pc_bob, ch_bob, q_bob = await _open_channel(transport, "peer-bob") - # Unwrap to verify it's correct - recovered = unwrap_gek(stored, sk_x_raw, pk_x_raw) - assert recovered == gek + nonce_c = os.urandom(NONCE_LEN) + ch_bob.send(_pack({ + "type": MNP.HANDSHAKE, "v": MNP_VERSION, + "token": _token(sk_hub, "user-002", "peer-bob", TEST_GROUP), + "group_id": TEST_GROUP, + "nonce": base64.b64encode(nonce_c).decode(), + })) + challenge = await asyncio.wait_for(q_bob.get(), timeout=5.0) + assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE + nonce_s = base64.b64decode(challenge["nonce"]) - await bundle_store.close() + # Bob signs a transcript naming the node, and he cannot complete the handshake + # that would prove its key — he has no GEK yet. So he has to be able to learn + # it from the challenge; taking it from the test's own knowledge of sk_node + # would hide the fact that a real client cannot. + assert challenge["node_pk"] == pk_to_b64(sk_node.public_key()), ( + "the challenge must announce the node key to a first-time joiner") + node_pk_b64 = challenge["node_pk"] + + pk_ed_b64 = pk_to_b64(sk_bob_ed.public_key()) + pk_x_b64 = base64.b64encode(pk_x_raw).decode() + ts = int(time.time()) + transcript = join_transcript( + node_pk_b64=node_pk_b64, + group_id=TEST_GROUP, user_id="user-002", + pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, + nonce_node=nonce_s, ts=ts, + ) + ch_bob.send(_pack({ + "type": MNP.JOIN_REQUEST, "v": MNP_VERSION, + "group_id": TEST_GROUP, + "pk_ed25519": pk_ed_b64, "pk_x25519": pk_x_b64, + "code": code, "ts": ts, + "sig": base64.b64encode(sk_bob_ed.sign(transcript)).decode(), + })) + + result = await asyncio.wait_for(q_bob.get(), timeout=5.0) + assert result["type"] == MNP.JOIN_RESULT + assert result["ok"] is True + assert result["gek"] is True + assert result["role"] == ROLE_MEMBER + + # 3. The key really is the group key, and only Bob's secret opens it. + assert unwrap_gek_aes(result, sk_x_raw, pk_x_raw) == gek + + # 4. The code is spent. + assert await roster.consume_invite(code, "user-002") is None + + await roster.close() await pc_admin.close() + await pc_bob.close() await transport.close_all() @@ -1124,8 +1257,10 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di # Step 1: Send handshake with group_id so _pending_group is set token = _make_jwt(sk_hub, groups=["g"]) + nonce_c = os.urandom(NONCE_LEN) channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g", + "nonce": base64.b64encode(nonce_c).decode(), })) msg = await asyncio.wait_for(received.get(), timeout=5.0) assert msg["type"] == MNP.HANDSHAKE_CHALLENGE @@ -1140,11 +1275,12 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di 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() + nonce_s = base64.b64decode(msg["nonce"]) + binding = webrtc_binding( + _extract_dtls_fp(browser_pc.localDescription.sdp), + _extract_dtls_fp(browser_pc.remoteDescription.sdp), + ) + proof = make_proof(recovered_gek, ROLE_CLIENT, "g", nonce_c, nonce_s, binding) # Step 4: Complete handshake channel.send(_pack({ @@ -1229,8 +1365,10 @@ async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, await asyncio.wait_for(ready.wait(), timeout=5.0) token = _make_jwt(sk_hub, groups=["g"]) + nonce_c = os.urandom(NONCE_LEN) channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g", + "nonce": base64.b64encode(nonce_c).decode(), })) msg = await asyncio.wait_for(received.get(), timeout=5.0) assert msg["type"] == MNP.HANDSHAKE_CHALLENGE @@ -1296,8 +1434,10 @@ async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, await asyncio.wait_for(ready.wait(), timeout=5.0) token = _make_jwt(sk_hub, groups=["g"]) + nonce_c = os.urandom(NONCE_LEN) channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g", + "nonce": base64.b64encode(nonce_c).decode(), })) msg = await asyncio.wait_for(received.get(), timeout=5.0) assert msg["type"] == MNP.HANDSHAKE_CHALLENGE @@ -1313,9 +1453,18 @@ async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, @pytest.mark.asyncio -async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shared_dir, +async def test_gek_not_auto_activated_on_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).""" + """ + A GEK bundle arriving over MNP must NOT become the node's live key (C5b). + + This test previously asserted the opposite: storing a bundle addressed to the + node operator auto-activated it, with no signature required. Because the + operator's X25519 public key is public — the node publishes it in handshake_ack + — any group member could wrap a key of their own choosing for it and take over + the group, locking every legitimate member out. GEK activation now happens only + through the node's local admin UI or CLI. + """ 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() @@ -1324,7 +1473,8 @@ async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shar bundle_store = BundleStore(db_path=tmp_path / "bundles.db") await bundle_store.open() - new_gek = generate_gek() + attacker_gek = generate_gek() + assert attacker_gek != gek transport = WebRTCTransport( sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, @@ -1336,14 +1486,18 @@ async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shar 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() + transport._ctx["admin_pk_ed25519"] = Ed25519PrivateKey.generate().public_key() 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) + # An ordinary member wraps a key of their choosing for the operator's public + # key and offers it to the node. The message that used to carry this no longer + # exists (the node wraps the GEK itself now), so it reaches no handler at all — + # a stronger outcome than the admin challenge this test used to assert. + node_bundle = wrap_gek_aes(attacker_gek, pk_x_raw) ch_admin.send(_pack({ - "type": MNP.GEK_BUNDLE_STORE, + "type": "gek_bundle_store", "v": MNP_VERSION, "user_id": "node-operator", "group_id": "g", @@ -1351,12 +1505,12 @@ async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shar "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) + await asyncio.sleep(0.5) + assert q_admin.empty(), "the retired bundle message still gets a response" - assert transport._ctx.get("gek") == new_gek + assert transport._ctx.get("gek") == gek, "group key was seized over MNP (C5b)" + assert await bundle_store.fetch("g", "node-operator") is None await bundle_store.close() await pc_admin.close() @@ -1398,6 +1552,8 @@ async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir): token = _make_jwt(sk_hub) channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, + "group_id": TEST_GROUP, + "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(), })) msg = await asyncio.wait_for(received.get(), timeout=5.0) @@ -1457,8 +1613,10 @@ async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_ await asyncio.wait_for(ready.wait(), timeout=5.0) token = _make_jwt(sk_hub, groups=["g"]) + nonce_c = os.urandom(NONCE_LEN) channel.send(_pack({ "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g", + "nonce": base64.b64encode(nonce_c).decode(), })) msg = await asyncio.wait_for(received.get(), timeout=5.0) assert msg["type"] == MNP.HANDSHAKE_CHALLENGE |