diff options
Diffstat (limited to 'packages/meshbay-hub/tests')
| -rw-r--r-- | packages/meshbay-hub/tests/test_bundle_kdf_parity.py | 131 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_hub_api.py | 66 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_node_auth.py | 3 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_node_ws_auth.py | 346 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_spa_ordering.py | 103 |
5 files changed, 632 insertions, 17 deletions
diff --git a/packages/meshbay-hub/tests/test_bundle_kdf_parity.py b/packages/meshbay-hub/tests/test_bundle_kdf_parity.py new file mode 100644 index 0000000..27e10d4 --- /dev/null +++ b/packages/meshbay-hub/tests/test_bundle_kdf_parity.py @@ -0,0 +1,131 @@ +""" +Cross-language parity for the keypair bundle KDF. + +The bundle is the one thing a user carries between browsers, and the passphrase +is all that stands between it and whoever holds the disk of a node they joined +(finding C4). It moved from PBKDF2-SHA512 to Argon2id for that reason — PBKDF2 is +compute-only, which is what makes it cheap on a GPU. + +Two implementations now have to agree byte for byte: the vendored WebAssembly the +browser runs, and `argon2-cffi` used by the QE harness. A disagreement would not +show up as an error — it would show up as a bundle nobody can open, which is +somebody's account gone. + +Skipped when node or argon2-cffi is missing; that is a coverage gap, not a pass. +""" + +import hashlib +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + +STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" +VENDOR = STATIC / "vendor" + +try: + from argon2.low_level import Type, hash_secret_raw + HAVE_ARGON2 = True +except ImportError: + HAVE_ARGON2 = False + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None + or not (VENDOR / "argon2.min.js").exists() + or not HAVE_ARGON2, + reason="node, the vendored argon2, or argon2-cffi is unavailable", +) + +# Parameters must match keyderive.js. If someone tunes them there and not here, +# this test fails — which is the point: changing them silently orphans every +# bundle already written. +MEM_KIB, TIME_COST, LANES = 131072, 3, 1 + +CASES = ["alice", "grenet", "utilisateur-é", ""] +PASSWORDS = ["correct horse battery staple", "p", "üñïçø∂é ✓ 🔐"] + +_HARNESS = r""" +const fs = require('fs'), webcrypto = require('crypto').webcrypto; +global.self = global; global.crypto = webcrypto; +// The browser uses the copy inlined in the bundle; under node the emscripten +// loader looks for a file, so hand it the same bytes explicitly. +global.Module = { wasmBinary: fs.readFileSync(process.argv[2]) }; +const argon2 = require(process.argv[3]); + +(async () => { + const input = JSON.parse(fs.readFileSync(process.argv[4], 'utf8')); + const out = []; + for (const v of input) { + const salt = new Uint8Array(await webcrypto.subtle.digest( + 'SHA-256', new TextEncoder().encode(`meshbay:bundle:v2:${v.username}`) + )).slice(0, 16); + const r = await argon2.hash({ + pass: v.password, salt, + time: v.time, mem: v.mem, parallelism: v.lanes, + hashLen: 32, type: argon2.ArgonType.Argon2id, + }); + out.push(Buffer.from(r.hash).toString('hex')); + } + process.stdout.write(JSON.stringify(out)); +})(); +""" + + +@pytest.fixture(scope="module") +def js_hashes(tmp_path_factory): + d = tmp_path_factory.mktemp("kdf") + harness = d / "harness.cjs" + harness.write_text(_HARNESS) + vectors = [ + {"username": u, "password": p, + "mem": MEM_KIB, "time": TIME_COST, "lanes": LANES} + for u in CASES for p in PASSWORDS + ] + payload = d / "vectors.json" + payload.write_text(json.dumps(vectors)) + + proc = subprocess.run( + ["node", str(harness), str(VENDOR / "argon2.wasm"), + str(VENDOR / "argon2.min.js"), str(payload)], + capture_output=True, text=True, timeout=300, + ) + if proc.returncode != 0: + pytest.fail(f"node harness failed:\n{proc.stderr[-2000:]}") + return vectors, json.loads(proc.stdout) + + +def _python_hash(username: str, password: str) -> str: + salt = hashlib.sha256(f"meshbay:bundle:v2:{username}".encode()).digest()[:16] + return hash_secret_raw( + password.encode(), salt, time_cost=TIME_COST, memory_cost=MEM_KIB, + parallelism=LANES, hash_len=32, type=Type.ID, + ).hex() + + +def test_bundle_key_matches_across_languages(js_hashes): + vectors, js = js_hashes + for i, v in enumerate(vectors): + assert js[i] == _python_hash(v["username"], v["password"]), ( + f"argon2id disagrees for username={v['username']!r} — a bundle " + f"written by one implementation would be unreadable by the other" + ) + + +def test_the_salt_separates_users(js_hashes): + """Two accounts with the same passphrase must not share a bundle key.""" + assert _python_hash("alice", "same passphrase") != \ + _python_hash("bob", "same passphrase") + + +def test_parameters_still_match_the_client(): + """ + The numbers live in keyderive.js; this test is the second copy. Tuning one + without the other orphans every bundle already written, so make it fail. + """ + source = (STATIC / "keyderive.js").read_text() + assert f"ARGON2_MEM_KIB = {MEM_KIB}" in source + assert f"ARGON2_TIME = {TIME_COST}" in source + assert f"ARGON2_LANES = {LANES}" in source + assert "meshbay:bundle:v2:" in source diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py index a8232c1..5a2cf86 100644 --- a/packages/meshbay-hub/tests/test_hub_api.py +++ b/packages/meshbay-hub/tests/test_hub_api.py @@ -24,6 +24,35 @@ def _gen_user_keys(): ) + +async def _announce_signed(client, token: str) -> tuple[str, str]: + """ + Announce a node with proof of possession (M8). + + The node key is independent of the user's identity key, so this mints a fresh + one and signs the domain-separated announce message with it. + """ + import base64 as _b64, time as _t + + me = await client.get("/v1/users/me", + headers={"Authorization": f"Bearer {token}"}) + user_id = me.json()["user_id"] + + sk_node = Ed25519PrivateKey.generate() + pk_node = pk_to_b64(sk_node.public_key()) + ts = _t.time().__trunc__() + msg = f"meshbay:node_announce:{user_id}:{pk_node}:{ts}".encode() + + r = await client.post("/v1/nodes/announce", json={ + "pk_node": pk_node, + "endpoint_hint": "1.2.3.4:19000", + "timestamp": ts, + "signature": _b64.b64encode(sk_node.sign(msg)).decode(), + }, headers={"Authorization": f"Bearer {token}"}) + assert r.status_code == 201, r.text + return r.json()["node_id"], pk_node + + # ── Hub info ────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @@ -115,8 +144,11 @@ async def test_jwt_offline_verify(client, hub_key_path): hub_pk_pem = r_pk.json()["pk_hub_pem"].encode() decoded = pyjwt.decode(token, hub_pk_pem, algorithms=["EdDSA"]) - assert decoded["pk_user"] == pk_ed assert "jti" in decoded # mandatory + # The token carries no user key. It used to, and the node recorded it as the + # uploader's identity — so whoever issued tokens decided who could delete a + # file. The hub certifies accounts; nodes pin keys. + assert "pk_user" not in decoded @pytest.mark.asyncio @@ -165,11 +197,17 @@ async def test_refresh_token_rotation_old_rejected(client): @pytest.mark.asyncio async def test_get_user_pubkeys(client): + """ + The endpoint resolves an account; it is not a key directory any more. + + Publishing user identity keys is what finding H3 exploited — the invite flow + wrapped the group key for whatever came back. Keys are now generated per node + and pinned there, so there is nothing here to substitute. + """ pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "frank", "email": "frank@example.com", - "password": "frankpass99", - "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) + "password": "frankpass99"}) login = await client.post("/v1/users/login", json={ "username": "frank", "password": "frankpass99"}) token = login.json()["access_token"] @@ -177,8 +215,10 @@ async def test_get_user_pubkeys(client): r = await client.get("/v1/users/frank/pubkeys", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 - assert r.json()["pk_ed25519"] == pk_ed - assert r.json()["pk_x25519"] == pk_x + body = r.json() + assert body["user_id"] and body["username"] == "frank" + assert "pk_ed25519" not in body, "user identity keys must not be published (H3)" + assert "pk_x25519" not in body, "user identity keys must not be published (H3)" # ── Nodes ───────────────────────────────────────────────────────────────────── @@ -195,15 +235,12 @@ async def test_announce_and_get_node(client): token = login.json()["access_token"] hdrs = {"Authorization": f"Bearer {token}"} - r = await client.post("/v1/nodes/announce", - json={"pk_node": pk_ed, "endpoint_hint": "1.2.3.4:19000"}, - headers=hdrs) - assert r.status_code == 201 - node_id = r.json()["node_id"] + node_id, pk_node = await _announce_signed(client, token) r2 = await client.get(f"/v1/nodes/{node_id}", headers=hdrs) assert r2.status_code == 200 - assert r2.json()["pk_node"] == pk_ed + # The node key is independent of the user identity key (M8). + assert r2.json()["pk_node"] == pk_node assert r2.json()["endpoint_hint"] == "1.2.3.4:19000" @@ -409,10 +446,7 @@ async def test_group_online_nodes(client): json={"username": "gn_user", "password": "gnpass999"})).json()["access_token"] # Announce a node - r = await client.post("/v1/nodes/announce", json={ - "pk_node": pk_ed, "endpoint_hint": "1.2.3.4:19000"}, - headers={"Authorization": f"Bearer {token}"}) - node_id = r.json()["node_id"] + node_id, pk_node = await _announce_signed(client, token) # No nodes online yet r = await client.get(f"/v1/groups/{group_id}/nodes", @@ -433,7 +467,7 @@ async def test_group_online_nodes(client): nodes = r.json()["nodes"] assert len(nodes) == 1 assert nodes[0]["node_id"] == node_id - assert nodes[0]["pk_node"] == pk_ed + assert nodes[0]["pk_node"] == pk_node finally: _connected_nodes.pop(node_id, None) _node_groups.pop(node_id, None) diff --git a/packages/meshbay-hub/tests/test_node_auth.py b/packages/meshbay-hub/tests/test_node_auth.py index e629d20..72ce412 100644 --- a/packages/meshbay-hub/tests/test_node_auth.py +++ b/packages/meshbay-hub/tests/test_node_auth.py @@ -214,7 +214,8 @@ async def test_node_scope_allows_pubkey_lookup(client): r = await client.get("/v1/users/op4/pubkeys", headers={"Authorization": f"Bearer {node_token}"}) assert r.status_code == 200 - assert "pk_ed25519" in r.json() + # An account id and the node's linking key — no user identity keys (H3). + assert "pk_ed25519" not in r.json() assert r.json()["pk_node_ed25519"] is not None diff --git a/packages/meshbay-hub/tests/test_node_ws_auth.py b/packages/meshbay-hub/tests/test_node_ws_auth.py new file mode 100644 index 0000000..1391722 --- /dev/null +++ b/packages/meshbay-hub/tests/test_node_ws_auth.py @@ -0,0 +1,346 @@ +""" +Phase 11.5 security regression tests — node WebSocket registration (finding C2). + +The hub relays every WebRTC offer for a node to whoever holds that node's entry in +`_connected_nodes`. That registration used to be established from a client-supplied +`node_id` with no ownership check, so any registered user could take over a victim +node's signaling and become the endpoint browsers connect to. + +These exercise `_authorize_node_ws` directly rather than through a socket: it is the +function that makes the authorization decision, and the hub test harness uses +ASGITransport, which has no WebSocket support. +""" + +import base64 + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + +from meshbay_common.crypto import pk_to_b64 + + +async def _make_user(client, username: str) -> dict: + """Register + log in a user, returning ids, token and keys.""" + sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate() + pk_ed, pk_x = pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key()) + + r = await client.post("/v1/users/register", json={ + "username": username, + "email": f"{username}@example.test", + "auth_key": base64.b64encode(b"k" * 32).decode(), + "pk_user_ed25519": pk_ed, + "pk_user_x25519": pk_x, + }) + assert r.status_code == 201, r.text + user_id = r.json()["user_id"] + + r = await client.post("/v1/users/login", json={ + "username": username, + "auth_key": base64.b64encode(b"k" * 32).decode(), + }) + assert r.status_code == 200, r.text + return {"user_id": user_id, "token": r.json()["access_token"], + "pk_ed": pk_ed, "sk_ed": sk_ed} + + +async def _announce_node(client, user: dict) -> str: + # Announce now requires proof of possession of the node key (M8). + import time as _t + ts = int(_t.time()) + msg = f"meshbay:node_announce:{user['user_id']}:{user['pk_ed']}:{ts}".encode() + r = await client.post( + "/v1/nodes/announce", + json={ + "pk_node": user["pk_ed"], "endpoint_hint": "test", + "timestamp": ts, + "signature": base64.b64encode(user["sk_ed"].sign(msg)).decode(), + }, + headers={"Authorization": f"Bearer {user['token']}"}, + ) + assert r.status_code == 201, r.text + return r.json()["node_id"] + + +def _node_token(user: dict) -> str: + from meshbay_hub.auth import issue_access_token + return issue_access_token(user["user_id"], scope="node") + + +@pytest.mark.asyncio +async def test_ws_rejects_user_scoped_token(client): + """C2: a browser token must never be able to register as a node.""" + from meshbay_hub.api.revocation import _authorize_node_ws + + victim = await _make_user(client, "victim1") + node_id = await _announce_node(client, victim) + + resolved, detail = await _authorize_node_ws(victim["token"], node_id, None) + assert resolved is None + assert "node-scoped" in detail.lower() + + +@pytest.mark.asyncio +async def test_ws_rejects_foreign_node_id(client): + """ + C2: the impersonation itself. An attacker with a perfectly valid node-scoped + token of their own must not be able to claim someone else's node_id. + """ + from meshbay_hub.api.revocation import _authorize_node_ws + + victim = await _make_user(client, "victim2") + attacker = await _make_user(client, "attacker2") + victim_node = await _announce_node(client, victim) + await _announce_node(client, attacker) + + resolved, detail = await _authorize_node_ws( + _node_token(attacker), victim_node, None) + assert resolved is None, "attacker hijacked the victim's node registration (C2)" + assert "does not belong" in detail.lower() + + +@pytest.mark.asyncio +async def test_ws_rejects_unknown_node_id(client): + """C2: an invented node_id must not register either.""" + from meshbay_hub.api.revocation import _authorize_node_ws + + user = await _make_user(client, "user3") + resolved, _ = await _authorize_node_ws(_node_token(user), "no-such-node", None) + assert resolved is None + + +@pytest.mark.asyncio +async def test_ws_rejects_missing_node_id(client): + """C2: identity may not fall back to the token subject.""" + from meshbay_hub.api.revocation import _authorize_node_ws + + user = await _make_user(client, "user4") + resolved, _ = await _authorize_node_ws(_node_token(user), "", None) + assert resolved is None + + +@pytest.mark.asyncio +async def test_ws_accepts_own_node(client): + """The legitimate path still works.""" + from meshbay_hub.api.revocation import _authorize_node_ws + + user = await _make_user(client, "owner5") + node_id = await _announce_node(client, user) + + resolved, groups = await _authorize_node_ws(_node_token(user), node_id, None) + assert resolved == node_id + assert groups == [] + + +@pytest.mark.asyncio +async def test_ws_group_claims_cannot_widen_beyond_membership(client): + """ + C2: `group_ids` used to be taken verbatim, letting a node advertise itself as + an online source for any group on the hub and attract clients to it. + """ + from meshbay_hub.api.revocation import _authorize_node_ws + + user = await _make_user(client, "owner6") + node_id = await _announce_node(client, user) + + r = await client.post( + "/v1/groups", + json={"name": "mine", "visibility": "private", "join_policy": "invite"}, + headers={"Authorization": f"Bearer {user['token']}"}, + ) + assert r.status_code == 201, r.text + own_group = r.json()["group_id"] + + resolved, groups = await _authorize_node_ws( + _node_token(user), node_id, [own_group, "someone-elses-group"]) + + assert resolved == node_id + assert groups == [own_group], "node advertised a group it is not a member of" + + +@pytest.mark.asyncio +async def test_signaling_rejects_non_member(client): + """ + H6/H4: POST /v1/nodes/{id}/webrtc/offer was reachable by any authenticated + user for any node, with no membership check and no rate limit. Each call makes + the target node allocate an aiortc PeerConnection and gather ICE, so it was a + remote resource-exhaustion primitive against a third party's machine. + """ + from meshbay_hub.api import revocation as rev + + owner = await _make_user(client, "owner8") + outsider = await _make_user(client, "outsider8") + node_id = await _announce_node(client, owner) + + r = await client.post( + "/v1/groups", + json={"name": "private-g", "visibility": "private", "join_policy": "invite"}, + headers={"Authorization": f"Bearer {owner['token']}"}, + ) + group_id = r.json()["group_id"] + + # Pretend the node is connected and hosting that group. + class _FakeWS: + async def send_text(self, _): + raise AssertionError("offer relayed to node despite non-membership") + + rev._connected_nodes[node_id] = _FakeWS() + rev._node_groups[node_id] = [group_id] + try: + resp = await client.post( + f"/v1/nodes/{node_id}/webrtc/offer", + json={"sdp": "v=0", "ice_candidates": []}, + headers={"Authorization": f"Bearer {outsider['token']}"}, + ) + assert resp.status_code == 403, resp.text + finally: + rev._connected_nodes.pop(node_id, None) + rev._node_groups.pop(node_id, None) + + +@pytest.mark.asyncio +async def test_signaling_rejects_oversized_sdp(client): + """H6: an SDP offer is ~2 KB; unbounded input is a memory amplifier.""" + user = await _make_user(client, "user9") + resp = await client.post( + "/v1/nodes/whatever/webrtc/offer", + json={"sdp": "v=0" + ("x" * 200_000), "ice_candidates": []}, + headers={"Authorization": f"Bearer {user['token']}"}, + ) + assert resp.status_code == 413 + + +@pytest.mark.asyncio +async def test_incoming_rejects_foreign_peer_ip(client): + """ + H6: peer_ip was taken verbatim, letting any user make an arbitrary node emit + UDP packets to an address of their choosing — reflection via someone else's + machine. The probe target must be the caller's own address. + """ + from meshbay_hub.api import revocation as rev + + owner = await _make_user(client, "owner10") + node_id = await _announce_node(client, owner) + + class _FakeWS: + async def send_text(self, _): + raise AssertionError("punch relayed with attacker-chosen peer_ip") + + rev._connected_nodes[node_id] = _FakeWS() + try: + resp = await client.post( + f"/v1/nodes/{node_id}/incoming", + json={"peer_ip": "198.51.100.7", "peer_port": 9999}, + headers={"Authorization": f"Bearer {owner['token']}"}, + ) + assert resp.status_code == 403, resp.text + finally: + rev._connected_nodes.pop(node_id, None) + + +@pytest.mark.asyncio +async def test_ws_node_may_narrow_its_group_set(client): + """A node hosting a subset of the operator's groups may say so.""" + from meshbay_hub.api.revocation import _authorize_node_ws + + user = await _make_user(client, "owner7") + node_id = await _announce_node(client, user) + + created = [] + for name in ("g-one", "g-two"): + r = await client.post( + "/v1/groups", + json={"name": name, "visibility": "private", "join_policy": "invite"}, + headers={"Authorization": f"Bearer {user['token']}"}, + ) + created.append(r.json()["group_id"]) + + resolved, groups = await _authorize_node_ws( + _node_token(user), node_id, [created[0]]) + assert resolved == node_id + assert groups == [created[0]] + + +# ── M8: announce proof of possession ───────────────────────────────────────── + +def _announce_payload(user_id: str, sk, pk_b64: str, ts: int | None = None): + import time as _t + ts = ts if ts is not None else int(_t.time()) + msg = f"meshbay:node_announce:{user_id}:{pk_b64}:{ts}".encode() + return { + "pk_node": pk_b64, + "endpoint_hint": "test", + "timestamp": ts, + "signature": base64.b64encode(sk.sign(msg)).decode(), + } + + +@pytest.mark.asyncio +async def test_announce_requires_proof_of_possession(client): + """ + M8: /v1/nodes/announce accepted any pk_node with no proof the announcer held + the private key, so a user could announce a record carrying someone else's + node key. + """ + user = await _make_user(client, "ann1") + r = await client.post( + "/v1/nodes/announce", + json={"pk_node": user["pk_ed"], "endpoint_hint": "test"}, + headers={"Authorization": f"Bearer {user['token']}"}, + ) + assert r.status_code == 400, r.text + + +@pytest.mark.asyncio +async def test_announce_rejects_foreign_key(client): + """M8: announcing someone else's public key must fail — no matching private key.""" + user = await _make_user(client, "ann2") + victim_sk = Ed25519PrivateKey.generate() + victim_pk = pk_to_b64(victim_sk.public_key()) + + attacker_sk = Ed25519PrivateKey.generate() + payload = _announce_payload(user["user_id"], attacker_sk, victim_pk) + + r = await client.post( + "/v1/nodes/announce", json=payload, + headers={"Authorization": f"Bearer {user['token']}"}, + ) + assert r.status_code == 401, r.text + + +@pytest.mark.asyncio +async def test_announce_rejects_stale_timestamp(client): + """M8: a captured announce must not be replayable later.""" + import time as _t + user = await _make_user(client, "ann3") + sk = Ed25519PrivateKey.generate() + payload = _announce_payload( + user["user_id"], sk, pk_to_b64(sk.public_key()), ts=int(_t.time()) - 3600) + + r = await client.post( + "/v1/nodes/announce", json=payload, + headers={"Authorization": f"Bearer {user['token']}"}, + ) + assert r.status_code == 401, r.text + + +@pytest.mark.asyncio +async def test_announce_with_valid_proof_succeeds_and_is_idempotent(client): + """The legitimate path works, and re-announcing updates rather than piling up rows.""" + user = await _make_user(client, "ann4") + sk = Ed25519PrivateKey.generate() + pk_b64 = pk_to_b64(sk.public_key()) + + first = await client.post( + "/v1/nodes/announce", json=_announce_payload(user["user_id"], sk, pk_b64), + headers={"Authorization": f"Bearer {user['token']}"}, + ) + assert first.status_code == 201, first.text + + second = await client.post( + "/v1/nodes/announce", json=_announce_payload(user["user_id"], sk, pk_b64), + headers={"Authorization": f"Bearer {user['token']}"}, + ) + assert second.status_code == 201, second.text + assert second.json()["node_id"] == first.json()["node_id"], ( + "re-announcing the same key must not create a second node record (M8)") diff --git a/packages/meshbay-hub/tests/test_spa_ordering.py b/packages/meshbay-hub/tests/test_spa_ordering.py new file mode 100644 index 0000000..0ef34fc --- /dev/null +++ b/packages/meshbay-hub/tests/test_spa_ordering.py @@ -0,0 +1,103 @@ +""" +Ordering guards for the SPA's connect() flow. + +These are source-level checks, which is not how one would normally test +behaviour. They exist because a specific class of bug shipped to a live browser +twice and no other test could see it: `connect()` is a long sequence in which +later steps read values earlier steps set, and the Python end-to-end client in +QE/deploy/ cannot catch a mistake there — it is a different implementation, +written in the right order by construction, so it passes while the browser fails. + +Concretely: join_request signs a transcript over the node key and the node nonce, +and runs *before* the GEK proof, because a first-time member has no GEK to prove. +Both values were being read further down, next to the proof that also uses them, +so every invited member hit "Handshake incomplete — reconnect and retry". + +If you restructure connect(), these will fail. Check the invariant still holds — +that nothing reads a value assigned later — and then move the markers. +""" + +from pathlib import Path + +import pytest + +STATIC = (Path(__file__).resolve().parents[1] + / "src" / "meshbay_hub" / "static") +TRANSPORT = STATIC / "transport.js" + +pytestmark = pytest.mark.skipif( + not TRANSPORT.exists(), reason="SPA sources not present") + + +def _positions(*needles: str) -> list[int]: + source = TRANSPORT.read_text() + out = [] + for needle in needles: + idx = source.find(needle) + assert idx != -1, f"{needle!r} is gone from transport.js — update this test" + out.append(idx) + return out + + +def test_challenge_values_are_captured_before_joining(): + """ + joinGroup() signs over node_pk and nonce_node, so both must be recorded when + the challenge arrives — not later, beside the proof. + """ + # Deliberately loose markers: what matters is where the assignment happens, + # not how it is spelled, so a reordering fails on the ordering assertion + # below rather than on a missing string. + node_pk, nonce_node, join_call = _positions( + "this.nodePk = reply.node_pk", + "this._nonceNode = ", + "await this.joinGroup(", + ) + assert node_pk < join_call, ( + "node_pk is read from the challenge after joinGroup() runs — the join " + "would sign a transcript naming nothing") + assert nonce_node < join_call, ( + "nonce_node is captured after joinGroup() runs — the join would not be " + "bound to this connection") + + +def test_join_happens_before_the_gek_proof(): + """ + The whole point of joining in the pre-proof window: someone who has never + held the group key cannot produce a proof, so the key has to arrive first. + """ + join_call, proof = _positions( + "await this.joinGroup(", + "await C.handshakeProof(", + ) + assert join_call < proof, ( + "the join must happen before the GEK proof — a first-time member has no " + "key to prove with") + + +def test_keys_are_recovered_before_the_join_is_attempted(): + """ + A second browser holds nothing but a password. It recovers its identity keys + from the node's encrypted keypair bundle, and only then can it sign a join — + so the recovery has to come first. Getting this order wrong is invisible on + the browser that registered, and breaks every other one. + """ + recover, join_call = _positions( + "type: 'keypair_bundle_fetch'", + "await this.joinGroup(", + ) + assert recover < join_call, ( + "the keypair bundle must be fetched before joinGroup() — otherwise a " + "browser that did not register has no key to sign the join with") + + +def test_the_ack_still_verifies_the_announced_node_key(): + """ + Taking node_pk from the challenge is only safe because the ack proves it and + the client compares the two. Losing that check would leave the announcement + trusted on its own. + """ + source = TRANSPORT.read_text() + assert "Node identity changed during the handshake" in source, ( + "the challenge's node_pk must be checked against the ack's") + assert "verifyNodeSignature" in source, ( + "the ack's signature over the handshake transcript must still be verified") |