1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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"])
|