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
|
"""
The account recovery key (docs/MESHBAY_DESIGN.md §3.6).
`generateRecoveryKey` / `deriveRecoveryKey` in keyderive.js are run here under
node against the real WebCrypto, rather than reimplemented: the mnemonic has to
round-trip its bytes exactly, and the derived key has to be deterministic per
account and domain-separated between accounts, or a recovery would hand back a
key that opens nothing.
"""
import json
import shutil
import subprocess
from pathlib import Path
import pytest
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
KEYDERIVE = STATIC / "keyderive.js"
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not KEYDERIVE.exists(),
reason="node or keyderive.js is unavailable",
)
_HARNESS = r"""
const fs = require('fs');
const webcrypto = require('crypto').webcrypto;
global.self = global;
global.window = global;
global.crypto = webcrypto;
// keyderive.js is a classic script ending in `window.MeshBayKeys = {...}`.
eval(fs.readFileSync(process.argv[2], 'utf8'));
const K = window.MeshBayKeys;
const hex = (buf) => Buffer.from(buf).toString('hex');
// deriveRecoveryKey yields a non-extractable AES-GCM key, so two keys are
// compared by encrypting a fixed block with a fixed IV: same key => same bytes.
const fp = async (key) => hex(await webcrypto.subtle.encrypt(
{ name: 'AES-GCM', iv: new Uint8Array(12) }, key, new Uint8Array(16)));
(async () => {
const out = {};
// 1. mnemonic round-trips the exact 32 bytes, 200 random draws: the key
// derived from the mnemonic string must match the key from the raw bytes.
let roundTripOk = true;
for (let i = 0; i < 200; i++) {
const rk = K.generateRecoveryKey(); // { rawB64, mnemonic }
const raw = Uint8Array.from(atob(rk.rawB64), c => c.charCodeAt(0));
const a = await fp(await K.deriveRecoveryKey(rk.mnemonic, 'u'));
const b = await fp(await K.deriveRecoveryKey(raw, 'u'));
if (a !== b) { roundTripOk = false; break; }
}
out.round_trip_ok = roundTripOk;
// 2. deterministic per account, different per account.
const rk = K.generateRecoveryKey();
const raw = Uint8Array.from(atob(rk.rawB64), c => c.charCodeAt(0));
const k1 = await fp(await K.deriveRecoveryKey(raw, 'alice'));
const k1again = await fp(await K.deriveRecoveryKey(raw, 'alice'));
const k2 = await fp(await K.deriveRecoveryKey(raw, 'bob'));
out.deterministic = (k1 === k1again);
out.domain_separated = (k1 !== k2);
// 3. mnemonic is grouped Base32, 52 significant chars for 32 bytes.
out.mnemonic_shape_ok =
/^[A-Z2-7]{4}( [A-Z2-7]{1,4})+$/.test(rk.mnemonic) &&
rk.mnemonic.replace(/ /g, '').length === 52;
// 4. a garbled key is rejected, not silently truncated.
let rejected = false;
try { await K.deriveRecoveryKey('too short', 'u'); }
catch { rejected = true; }
out.rejects_short = rejected;
// 5. the building block connect()'s Flow B fallback relies on: a bundle
// wrapped under one recovery key does not open under a wrong one, and does
// open under the right one.
{
const kA = await K.deriveRecoveryKey(raw, 'acc-A');
const kB = await K.deriveRecoveryKey(raw, 'acc-B');
const skEd = new Uint8Array([1, 2, 3]);
const skX = new Uint8Array([4, 5, 6]);
const blob = await K.encryptBundleWithKey(skEd, skX, kA);
let wrongRejected = false;
try { await K.decryptBundleWithKey(blob, kB); } catch { wrongRejected = true; }
const opened = await K.decryptBundleWithKey(blob, kA);
out.recovery_wrap_isolates = wrongRejected
&& opened.skEd === btoa(String.fromCharCode(1, 2, 3))
&& opened.skX === btoa(String.fromCharCode(4, 5, 6));
}
process.stdout.write(JSON.stringify(out));
})().catch(e => { console.error(e); process.exit(1); });
"""
@pytest.fixture(scope="module")
def result(tmp_path_factory):
d = tmp_path_factory.mktemp("recovery")
harness = d / "harness.cjs"
harness.write_text(_HARNESS)
proc = subprocess.run(
["node", str(harness), str(KEYDERIVE)],
capture_output=True, text=True, timeout=120,
)
if proc.returncode != 0:
pytest.fail(f"node harness failed:\n{proc.stderr[-2000:]}")
return json.loads(proc.stdout)
def test_mnemonic_round_trips_the_exact_bytes(result):
assert result["round_trip_ok"]
def test_derived_key_is_deterministic_per_account(result):
assert result["deterministic"]
def test_derived_key_is_domain_separated_between_accounts(result):
assert result["domain_separated"]
def test_mnemonic_is_grouped_base32(result):
assert result["mnemonic_shape_ok"]
def test_a_malformed_recovery_key_is_rejected(result):
assert result["rejects_short"]
def test_a_recovery_wrapped_bundle_only_opens_under_the_matching_key(result):
assert result["recovery_wrap_isolates"]
|