1 files changed, 142 insertions, 0 deletions
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"])