From 4e33fca55e48bbf6233e950355d31c603d88a6e6 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Fri, 2 Oct 2026 11:56:03 +0200 Subject: test(hub): shared keyring and transcript vectors One file every bundle/transcript implementation must reproduce, generated from the desktop keyring; checked against it and against the specification. Co-Authored-By: Claude Opus 5.5 --- packages/meshbay-hub/tests/test_keyring_vectors.py | 142 +++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 packages/meshbay-hub/tests/test_keyring_vectors.py (limited to 'packages/meshbay-hub/tests/test_keyring_vectors.py') diff --git a/packages/meshbay-hub/tests/test_keyring_vectors.py b/packages/meshbay-hub/tests/test_keyring_vectors.py new file mode 100644 index 0000000..3928965 --- /dev/null +++ b/packages/meshbay-hub/tests/test_keyring_vectors.py @@ -0,0 +1,142 @@ +""" +The keypair-bundle and transcript vectors (`tests/vectors/keyring.json`). + +One file that every implementation of the bundle format and of the signed +transcripts has to reproduce: the page (`keyderive.js`), the desktop keyring +(`keyring.js`, `transcripts.js`), the Python reference below, and the Android +keyring, whose unit tests read the same file. Pairwise parity tests grow with +the square of the implementations; a shared file grows by one consumer. + +Two things are checked here. The file is what the shipped desktop code writes, +so it cannot drift from the code it describes. And the specification +(`docs/MESHBAY_DESIGN.md` §3.7, written out with argon2-cffi and +`cryptography`, sharing nothing with the generator) arrives at the same bytes. +""" + +import base64 +import hashlib +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + +VECTORS = Path(__file__).resolve().parent / "vectors" +FILE = VECTORS / "keyring.json" +GENERATOR = VECTORS / "gen_keyring_vectors.js" +KEYRING = Path(__file__).resolve().parents[2] / "meshbay-client" / "src" / "keyring.js" + +try: + from argon2.low_level import Type, hash_secret_raw + HAVE_ARGON2 = True +except ImportError: + HAVE_ARGON2 = False + + +def _vectors() -> dict: + return json.loads(FILE.read_text(encoding="utf-8")) + + +@pytest.mark.skipif(shutil.which("node") is None or not KEYRING.exists(), + reason="node or the desktop client sources are unavailable") +def test_the_file_is_what_the_shipped_keyring_writes(): + """Regenerated from keyring.js and transcripts.js, byte for byte. A change + to either that alters a bundle or a transcript fails here first — which is + the moment to decide whether it is a format change every client must make.""" + out = subprocess.run(["node", str(GENERATOR)], capture_output=True, text=True, + timeout=120, check=True).stdout + assert json.loads(out) == _vectors(), ( + "keyring.js or transcripts.js no longer produce tests/vectors/keyring.json; " + "if the format change is deliberate, regenerate it and update every client") + + +def _hkdf(ikm: bytes, info: str) -> bytes: + from cryptography.hazmat.primitives import hashes + from cryptography.hazmat.primitives.kdf.hkdf import HKDF + return HKDF(hashes.SHA256(), 32, None, info.encode()).derive(ikm) + + +@pytest.fixture(scope="module") +def spec(): + """The chain from the specification: passphrase → Argon2id → M → keys.""" + if not HAVE_ARGON2: + pytest.skip("argon2-cffi is unavailable") + v = _vectors() + i, p = v["input"], v["kdf"]["argon2_params"] + salt = hashlib.sha256(f"meshbay:bundle:v2:{i['username']}".encode()).digest()[:16] + a = hash_secret_raw(i["password"].encode(), salt, time_cost=p["passes"], + memory_cost=p["memory"], parallelism=p["parallelism"], + hash_len=p["tagLength"], type=Type.ID) + m = _hkdf(a + base64.b64decode(i["pepperB64"]), f"meshbay:bundle-master:v3|{i['userId']}") + return {"v": v, "salt": salt, "a": a, "m": m, + "node_key": _hkdf(m, f"meshbay:bundle:v3|node|{i['nodePk']}"), + "recovery_key": _hkdf(bytes.fromhex(v["kdf"]["mnemonic_bytes_hex"]), + f"meshbay:recovery:v1:{i['username']}")} + + +def test_the_specification_derives_the_same_keys(spec): + k = spec["v"]["kdf"] + assert spec["salt"].hex() == k["salt_hex"] + assert spec["a"].hex() == k["argon2_hex"] + assert spec["m"].hex() == k["master_hex"] + assert hashlib.sha256(spec["m"]).hexdigest()[:16] == k["master_fingerprint"] + assert spec["node_key"].hex() == k["node_key_hex"] + playlist = _hkdf(spec["m"], "meshbay:playlists:v2") + assert base64.b64encode(playlist).decode() == k["playlist_key_b64"] + assert spec["recovery_key"].hex() == k["recovery_key_hex"] + + +def test_the_specification_seals_the_same_bundles(spec): + """MBK3 = magic ‖ pepper version ‖ nonce ‖ AES-GCM(JSON, aad). With the nonce + pinned, sealing is deterministic, so every client must write these bytes.""" + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + v = spec["v"] + i, b = v["input"], v["bundles"] + nonce = bytes.fromhex(i["fixedNonceHex"]) + plain, aad = b["sealed_json_plaintext"].encode(), b["aad"].encode() + + def seal(key: bytes, version: int) -> str: + return base64.b64encode(b"MBK3" + bytes([version]) + nonce + + AESGCM(key).encrypt(nonce, plain, aad)).decode() + + assert seal(spec["node_key"], i["pepperVersion"]) == b["fixed_nonce_bundle_b64"] + assert seal(spec["recovery_key"], 0) == b["recovery_fixed_nonce_b64"] + raw = base64.b64decode(b["legacy_mbk2_b64"]) # TRANSITIONAL: MBK2, no AAD + assert AESGCM(spec["a"]).decrypt(raw[4:16], raw[16:], None) == plain + + +def test_the_identity_signs_and_agrees_as_recorded(): + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives.asymmetric.x25519 import ( + X25519PrivateKey, + X25519PublicKey, + ) + v = _vectors() + i = v["input"] + ed = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(i["edSeedHex"])) + x = X25519PrivateKey.from_private_bytes(bytes.fromhex(i["xSeedHex"])) + + def raw(k) -> str: + return base64.b64encode(k.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw)).decode() + + assert raw(ed) == v["identity"]["public"]["pkEdB64"] + assert raw(x) == v["identity"]["public"]["pkXB64"] + peer = X25519PublicKey.from_public_bytes(bytes.fromhex(i["peerXPubHex"])) + assert base64.b64encode(x.exchange(peer)).decode() == v["agreement"]["shared_b64"] + for kind, t in v["transcripts"].items(): + sig = base64.b64encode(ed.sign(bytes.fromhex(t["transcript_hex"]))).decode() + assert sig == t["signature_b64"], kind + + +def test_every_signed_kind_and_a_refusal_for_each_check_is_recorded(): + """A consumer that passes an empty list proves nothing; pin what is there.""" + v = _vectors() + assert set(v["transcripts"]) == {"join", "device_hello", "device_request", + "device_add", "device_revoke", "chat", "admin"} + labels = {r["label"] for r in v["refusals"]} + assert {"another node", "another account", "stale timestamp", "unknown kind", + "an op that widens sharing (gone from MNP 6.0)"} <= labels + assert all(r["error"].startswith("Refused: ") for r in v["refusals"]) -- cgit v1.2.3