1 files changed, 131 insertions, 0 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..c3c8ff9
--- /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 = 65536, 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