""" Tier 2, the half that decides anything: the client walks the chain. The node relays evidence and asserts nothing (`test_group_roster.py`). What turns that into a property is here — `_verifyRoster` in the shipped `transport.js`, run under node against rosters built by the shipped Python, so neither side is a model of the other. The property, stated exactly: **once this client has seen an account, a node that later substitutes a key for it is detected.** Nothing is gained at first sight, where there is nothing to compare against. A device the node lists but cannot evidence never enters `verified`, which is what stops a fabricated key being laundered into the set merely by being mentioned. """ import base64 import json import shutil import subprocess import tempfile import time from pathlib import Path import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.device import device_add_transcript STATIC = (Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static") TRANSPORT = STATIC / "transport.js" CRYPTO = STATIC / "crypto.js" pytestmark = pytest.mark.skipif( shutil.which("node") is None or not TRANSPORT.exists(), reason="node or the SPA sources are not available") NODE_PK = "Tk9ERVBL" NONCE = b"\x11" * 32 _HARNESS = r""" const fs = require('fs'); // The same stub the other transport.js harnesses use (upload_seal_probe.mjs, // index_seal_probe.mjs): the module reads `location.hash` at load time for its // debug flag, and registers listeners. globalThis.window = globalThis; globalThis.addEventListener = () => {}; globalThis.removeEventListener = () => {}; globalThis.location = { hash: '' }; globalThis.document = { addEventListener() {}, removeEventListener() {}, visibilityState: 'visible', }; globalThis.localStorage = { _v: {}, getItem(k) { return this._v[k] ?? null; }, setItem(k, v) { this._v[k] = String(v); }, }; // crypto.js publishes onto window; transport.js reads it from there. new Function(fs.readFileSync(process.argv[2], 'utf8'))(); const T = new Function( fs.readFileSync(process.argv[3], 'utf8') + '\nreturn { _verifyRoster };')(); (async () => { const input = JSON.parse(fs.readFileSync(process.argv[4], 'utf8')); const out = await T._verifyRoster(input.payload, input.node_pk); const result = {}; for (const [user, e] of out.byAccount) { result[user] = { verified: e.verified, unevidenced: e.unevidenced, all: e.all }; } process.stdout.write(JSON.stringify(result)); })().catch(e => { console.error(e); process.exit(1); }); """ def _device(sk=None): sk = sk or Ed25519PrivateKey.generate() raw = sk.public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw) return sk, base64.b64encode(raw).decode() def _entry(user, pk_ed, pk_x="cGtY", *, added_by="", sk_signer=None, node_pk=NODE_PK): """One roster row, countersigned for real when a signer is given.""" ts = int(time.time()) row = {"user_id": user, "username": user, "pk_ed25519": pk_ed, "pk_x25519": pk_x, "added_by_pk": added_by, "add_sig": "", "add_nonce": "", "add_ts": 0, "pinned_at": ""} if sk_signer is not None: transcript = device_add_transcript( node_pk_b64=node_pk, user_id=user, pk_ed25519_b64=pk_ed, pk_x25519_b64=pk_x, nonce_node=NONCE, ts=ts) row["add_sig"] = base64.b64encode(sk_signer.sign(transcript)).decode() row["add_nonce"] = base64.b64encode(NONCE).decode() row["add_ts"] = ts return row def _verify(devices, node_pk=NODE_PK): with tempfile.TemporaryDirectory() as tmp: h = Path(tmp) / "h.js" h.write_text(_HARNESS) payload = Path(tmp) / "in.json" payload.write_text(json.dumps( {"payload": {"devices": devices, "node_pk": node_pk}, "node_pk": node_pk})) run = subprocess.run( ["node", str(h), str(CRYPTO), str(TRANSPORT), str(payload)], capture_output=True, timeout=60) assert run.returncode == 0, run.stderr.decode()[-2000:] return json.loads(run.stdout.decode()) def test_a_lone_first_device_is_the_trust_root(): """No countersignature and none possible — an operator code admitted it.""" _sk, pk = _device() out = _verify([_entry("alice", pk)]) assert out["alice"]["verified"] == [pk] assert out["alice"]["unevidenced"] == [] def test_a_countersigned_second_device_is_reached(): """The ordinary case: Alice adds a laptop, and nobody compares digits.""" sk_a, pk_a = _device() _sk_b, pk_b = _device() out = _verify([_entry("alice", pk_a), _entry("alice", pk_b, added_by=pk_a, sk_signer=sk_a)]) assert sorted(out["alice"]["verified"]) == sorted([pk_a, pk_b]) assert out["alice"]["unevidenced"] == [] def test_a_chain_of_three_is_walked_in_any_order(): """ A device may be countersigned by one that is itself countersigned, and the roster arrives in whatever order SQL returned. The walk repeats until it stops making progress rather than assuming an order. """ sk_a, pk_a = _device() sk_b, pk_b = _device() _sk_c, pk_c = _device() rows = [_entry("alice", pk_c, added_by=pk_b, sk_signer=sk_b), _entry("alice", pk_b, added_by=pk_a, sk_signer=sk_a), _entry("alice", pk_a)] out = _verify(rows) assert sorted(out["alice"]["verified"]) == sorted([pk_a, pk_b, pk_c]) def test_a_fabricated_device_is_not_verified(): """ The attack. A node writes a device of its own into Alice's row — it writes the roster, so it can. It cannot sign as a key it does not hold, so no chain reaches the key and it stays out of `verified`. """ _sk_a, pk_a = _device() _sk_evil, pk_evil = _device() out = _verify([_entry("alice", pk_a), _entry("alice", pk_evil, added_by=pk_a)]) # no signature assert out["alice"]["verified"] == [pk_a] assert out["alice"]["unevidenced"] == [pk_evil] def test_a_signature_by_the_wrong_key_is_not_verified(): """A real signature, from a key that is not the one it names.""" _sk_a, pk_a = _device() sk_other, _pk_other = _device() _sk_b, pk_b = _device() out = _verify([_entry("alice", pk_a), _entry("alice", pk_b, added_by=pk_a, sk_signer=sk_other)]) assert out["alice"]["verified"] == [pk_a] assert out["alice"]["unevidenced"] == [pk_b] def test_a_signature_for_another_node_does_not_transfer(): """ The transcript binds `node_pk`. A countersignature collected on one node must not admit the same key on another — which is what an operator running two nodes would otherwise be able to do to a member of both. """ sk_a, pk_a = _device() _sk_b, pk_b = _device() row = _entry("alice", pk_b, added_by=pk_a, sk_signer=sk_a, node_pk="QU5PVEhFUg==") out = _verify([_entry("alice", pk_a), row]) assert out["alice"]["unevidenced"] == [pk_b] def test_a_signature_for_another_account_does_not_transfer(): """The transcript binds the account too.""" sk_a, pk_a = _device() _sk_b, pk_b = _device() row = _entry("alice", pk_b, added_by=pk_a, sk_signer=sk_a) # Same signature, presented as admitting a device of Bob's. row["user_id"] = "bob" out = _verify([_entry("bob", pk_a), row]) assert out["bob"]["unevidenced"] == [pk_b] def test_an_orphan_chain_is_not_admitted_by_itself(): """ Two fabricated devices signing each other. Neither is reachable from a root, so a cycle admits nothing — the walk starts from what an operator code admitted, not from whatever claims to be signed. """ sk_x, pk_x = _device() sk_y, pk_y = _device() _sk_a, pk_a = _device() out = _verify([ _entry("alice", pk_a), _entry("alice", pk_x, added_by=pk_y, sk_signer=sk_y), _entry("alice", pk_y, added_by=pk_x, sk_signer=sk_x), ]) assert out["alice"]["verified"] == [pk_a] assert sorted(out["alice"]["unevidenced"]) == sorted([pk_x, pk_y]) def test_a_device_pinned_before_the_evidence_existed_reads_as_a_root(): """ Honest about what it is. Such a device has `added_by_pk` but no signature — it was countersigned, the proof was simply not kept. Treating it as verified would mean accepting an unsigned key; treating it as a root is trust-on-first-use, which is what it actually is. """ _sk_a, pk_a = _device() _sk_b, pk_b = _device() out = _verify([_entry("alice", pk_a), _entry("alice", pk_b, added_by=pk_a)]) assert pk_b in out["alice"]["unevidenced"]