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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
"""
Cross-language parity: the browser's transcripts must be byte-identical to Python's.
The handshake proof and the admin signature are computed independently on both sides
and compared by producing the same bytes. Nothing on the wire carries the transcript,
which is the point — but it also means a one-byte disagreement between `crypto.js` and
`meshbay_common` is invisible to every other test and produces a total outage: no
browser can complete a handshake with any node.
The rest of the suite runs in Python only, so nothing else crosses this boundary. These
tests drive the real `crypto.js` under node and compare against the real Python.
Skipped when node is unavailable; that is a coverage gap, not a pass.
"""
import json
import shutil
import subprocess
from pathlib import Path
import pytest
from meshbay_common.adminop import admin_transcript
from meshbay_common.handshake import handshake_transcript, webrtc_binding
CRYPTO_JS = (Path(__file__).resolve().parents[2]
/ "meshbay-hub" / "src" / "meshbay_hub" / "static" / "crypto.js")
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not CRYPTO_JS.exists(),
reason="node or crypto.js unavailable — parity cannot be checked",
)
# (role, group_id, nonce_c hex, nonce_s hex, offer_fp hex, answer_fp hex)
HANDSHAKE_VECTORS = [
("client", "g" * 32, "01" * 32, "02" * 32, "aa" * 32, "bb" * 32),
("node", "g" * 32, "01" * 32, "02" * 32, "aa" * 32, "bb" * 32),
# Short and empty group ids — length prefixing must keep these distinct.
("client", "g", "03" * 32, "04" * 32, "cc" * 32, "dd" * 32),
("client", "", "03" * 32, "04" * 32, "cc" * 32, "dd" * 32),
# Non-ASCII: JS TextEncoder and Python .encode() must agree on UTF-8.
("client", "groupe-café-日本", "05" * 32, "06" * 32, "ee" * 32, "ff" * 32),
# Fields that could run together under naive concatenation.
("client", "gg", "07" * 32, "08" * 32, "11" * 32, "22" * 32),
]
# (op, node_pk_b64, group_id, subject, nonce hex, ts)
ADMIN_VECTORS = [
("file_delete", "Tk9ERVBL", "g" * 32, "file-1", "01" * 32, 1_700_000_000),
("gek_bundle_store", "Tk9ERVBL", "g" * 32, "user-2", "02" * 32, 1_700_000_001),
("file_delete", "Tk9ERVBL", "", "", "03" * 32, 0),
("file_delete", "Tk9ERVBL", "café", "fichier é.mp4", "04" * 32, 1_700_000_002),
]
_HARNESS = r"""
const fs = require('fs');
// crypto.js ends with `window.MeshBayCrypto = {...}` and references SubtleCrypto in
// functions we do not call. A stub is enough to evaluate the module body.
globalThis.window = {};
globalThis.crypto = globalThis.crypto || {};
const src = fs.readFileSync(process.argv[2], 'utf8');
const load = new Function(
src + '\nreturn { handshakeTranscript, adminTranscript, webrtcBinding, b64encode };');
const M = load();
const hex = (s) => {
const out = new Uint8Array(s.length / 2);
for (let i = 0; i < s.length; i += 2) out[i / 2] = parseInt(s.substr(i, 2), 16);
return out;
};
const toHex = (u8) =>
Array.from(u8).map(b => b.toString(16).padStart(2, '0')).join('');
const input = JSON.parse(fs.readFileSync(process.argv[3], 'utf8'));
const out = { handshake: [], admin: [] };
for (const v of input.handshake) {
const binding = M.webrtcBinding(hex(v.offer_fp), hex(v.answer_fp));
out.handshake.push(toHex(M.handshakeTranscript(
v.role, v.group_id, hex(v.nonce_c), hex(v.nonce_s), binding)));
}
for (const v of input.admin) {
out.admin.push(toHex(M.adminTranscript(
v.op, v.node_pk, v.group_id, v.subject, M.b64encode(hex(v.nonce)), v.ts)));
}
process.stdout.write(JSON.stringify(out));
"""
@pytest.fixture(scope="module")
def js_output(tmp_path_factory):
"""Run the real crypto.js under node and return its transcripts as hex."""
d = tmp_path_factory.mktemp("parity")
harness = d / "harness.js"
harness.write_text(_HARNESS)
payload = d / "vectors.json"
payload.write_text(json.dumps({
"handshake": [
{"role": r, "group_id": g, "nonce_c": nc,
"nonce_s": ns, "offer_fp": ofp, "answer_fp": afp}
for r, g, nc, ns, ofp, afp in HANDSHAKE_VECTORS
],
"admin": [
{"op": op, "node_pk": pk, "group_id": g,
"subject": s, "nonce": n, "ts": ts}
for op, pk, g, s, n, ts in ADMIN_VECTORS
],
}))
proc = subprocess.run(
["node", str(harness), str(CRYPTO_JS), str(payload)],
capture_output=True, text=True, timeout=60,
)
if proc.returncode != 0:
pytest.fail(f"node harness failed:\n{proc.stderr}")
return json.loads(proc.stdout)
@pytest.mark.parametrize("idx,vector", list(enumerate(HANDSHAKE_VECTORS)))
def test_handshake_transcript_parity(idx, vector, js_output):
"""
A mismatch here means no browser can complete a handshake with any node —
the GEK proof would never verify, and no other test would notice.
"""
role, group_id, nonce_c, nonce_s, offer_fp, answer_fp = vector
expected = handshake_transcript(
role=role,
group_id=group_id,
nonce_client=bytes.fromhex(nonce_c),
nonce_node=bytes.fromhex(nonce_s),
binding=webrtc_binding(bytes.fromhex(offer_fp), bytes.fromhex(answer_fp)),
)
assert js_output["handshake"][idx] == expected.hex(), (
f"crypto.js and meshbay_common.handshake disagree for role={role!r} "
f"group={group_id!r}"
)
@pytest.mark.parametrize("idx,vector", list(enumerate(ADMIN_VECTORS)))
def test_admin_transcript_parity(idx, vector, js_output):
"""
A mismatch here means the browser signs bytes the node did not ask for, so every
file deletion and GEK bundle store is rejected.
"""
op, node_pk, group_id, subject, nonce, ts = vector
expected = admin_transcript(
op=op,
node_pk_b64=node_pk,
group_id=group_id,
subject=subject,
nonce=bytes.fromhex(nonce),
ts=ts,
)
assert js_output["admin"][idx] == expected.hex(), (
f"crypto.js and meshbay_common.adminop disagree for op={op!r} "
f"subject={subject!r}"
)
def test_length_prefixing_actually_disambiguates(js_output):
"""
The reason both sides length-prefix: two different field splits must not collide.
Verified across the language boundary, since a JS implementation that concatenated
naively would still agree with itself.
"""
a = js_output["handshake"][2] # group_id "g"
b = js_output["handshake"][3] # group_id ""
assert a != b, "JS transcripts collide across different group ids"
|