""" 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"])