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
|
"""
The browser half of MNP 1.0's sealed index, measured rather than read.
`test_index_no_cleartext.py` proves the node sends no filename in the clear. This
proves the client can still read one — and, the part that matters more, that it
*stops* when it cannot instead of reporting an empty group.
Driven through `harness/index_seal_probe.mjs`, which runs the shipped
`transport.js` over the shipped `crypto.js` and is fed real frames built here.
A source-reading test could show that `openGroup` is called; only this can show
what a waiting `fetchIndex()` is told when it throws.
"""
import json
import shutil
import struct
import subprocess
import tempfile
from pathlib import Path
import msgpack
import pytest
from meshbay_common.crypto import generate_gek
from meshbay_common.groupbox import PURPOSE_INDEX, seal
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
PROBE = Path(__file__).resolve().parent / "harness" / "index_seal_probe.mjs"
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not PROBE.exists(),
reason="node unavailable — the client half cannot be measured",
)
GROUP = "g-probe"
GEK = generate_gek()
def _entry(name: str) -> dict:
return {"id": "ab" * 32, "name": name, "path": "library", "size": 10,
"type": "file", "added_at": 0, "uploader_id": ""}
def _frame(msg: dict) -> str:
body = msgpack.packb(msg, use_bin_type=True)
return (struct.pack(">I", len(body)) + body).hex()
def _sync_frame(gek: bytes, *names: str, version: int = 3) -> str:
payload = {"version": version, "entries": [_entry(n) for n in names],
"dirs": ["library"], "roots": [{"name": "library"}]}
return _frame({"type": "index_sync", "v": "1.0", "group_id": GROUP,
**seal(gek, PURPOSE_INDEX, "index_sync", GROUP, payload)})
def _delta_frame(gek: bytes, name: str, base: int, version: int) -> str:
payload = {"base_version": base, "version": version,
"additions": [_entry(name)], "deletions": [], "updates": []}
return _frame({"type": "index_delta", "v": "1.0", "group_id": GROUP,
**seal(gek, PURPOSE_INDEX, "index_delta", GROUP, payload)})
def _run(frames: list[str], gek: bytes = GEK) -> dict:
with tempfile.TemporaryDirectory() as tmp:
vectors = Path(tmp) / "vectors.json"
vectors.write_text(json.dumps(
{"gek": gek.hex(), "group_id": GROUP, "frames": frames}))
proc = subprocess.run(
["node", str(PROBE), str(STATIC), str(vectors)],
capture_output=True, text=True, timeout=120)
if proc.returncode != 0:
pytest.fail(f"probe failed:\n{proc.stderr}")
return json.loads(proc.stdout)
def test_a_sealed_index_reaches_the_consumer_intact():
out = _run([_sync_frame(GEK, "a-film.mkv", "another.mkv")])
assert [e["event"] for e in out["events"]] == ["index_sync"]
sync = out["events"][0]
assert sync["entries"] == ["a-film.mkv", "another.mkv"]
assert sync["dirs"] == ["library"]
# Moved inside the payload (D4) and still delivered flat, so no consumer had
# to change: it reads the same message it always read.
assert sync["version"] == 3
assert not sync["hasCiphertext"], "the envelope's own fields leaked to consumers"
# And the waiting caller gets the opened form, not the envelope.
assert out["fetchIndex"]["state"] == "resolved"
assert out["fetchIndex"]["entries"] == ["a-film.mkv", "another.mkv"]
def test_an_index_that_does_not_open_ends_the_session():
"""
§3.4, and the reason it is a rule rather than a preference. An empty
`entries` is a legitimate state — a group whose operator has shared nothing
yet — so a client that fell back to one would show the same screen for
"nothing here" and for "we could not decrypt anything this node sent".
"""
out = _run([_sync_frame(generate_gek(), "a-film.mkv")])
kinds = [e["event"] for e in out["events"]]
assert "index_sync" not in kinds, "a failed decrypt was reported as an index"
assert kinds == ["session_failed"]
assert "index_sync" in out["events"][0]["message"], (
"the failure must name the message type that could not be opened")
# The caller is told, rather than left to time out 30 s later.
assert out["fetchIndex"]["state"] == "rejected"
assert "index_sync" in out["fetchIndex"]["message"]
assert out["closed"] == 1, "the session carried on after an unopenable message"
def test_a_delta_that_does_not_open_ends_the_session_too():
"""
The delta has no caller waiting on it — it is pushed — so a silent failure
here would leave a browser showing a stale index with nothing wrong on
screen, which is the worst of the three shapes.
"""
out = _run([_sync_frame(GEK, "a-film.mkv"),
_delta_frame(generate_gek(), "new.mkv", 3, 4)])
assert [e["event"] for e in out["events"]] == ["index_sync", "session_failed"]
assert "index_delta" in out["events"][1]["message"]
assert out["closed"] == 1
def test_deltas_are_applied_in_arrival_order():
"""
Opening is asynchronous and `_dispatch` is not. Two messages opened
independently settle in whichever order WebCrypto finishes them, and a delta
applied before the one it follows is a wrong view of the group that nothing
reports. Three deltas in one burst is the cheapest way to force the race.
"""
frames = [_sync_frame(GEK, "a-film.mkv")]
frames += [_delta_frame(GEK, f"added-{i}.mkv", 3 + i, 4 + i) for i in range(3)]
out = _run(frames)
assert [e["event"] for e in out["events"]] == [
"index_sync", "index_delta", "index_delta", "index_delta"]
assert [e["additions"][0] for e in out["events"][1:]] == [
"added-0.mkv", "added-1.mkv", "added-2.mkv"]
assert [e["base_version"] for e in out["events"][1:]] == [3, 4, 5]
|