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
|
"""
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
from spa_source import transport_argv
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,
req_id: int | None = None) -> str:
payload = {"version": version, "entries": [_entry(n) for n in names],
"dirs": ["library"], "roots": [{"name": "library"}]}
# `req_id` is what a current node stamps on a *reply*; the push it sends a
# newly connected peer answers no request and carries none. Both shapes
# arrive here, and only one of them may resolve a waiting fetchIndex.
return _frame({"type": "index_sync", "v": "1.0", "group_id": GROUP,
**({"req_id": req_id} if req_id is not None else {}),
**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), transport_argv()],
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]
def test_a_sealed_reply_is_opened_before_it_reaches_its_caller():
"""A stamped index_sync must not be short-circuited by its `req_id`.
Every other reply a node stamps is resolved straight out of the pending
map, which is the whole point of the id. An index message cannot be: it is
sealed, opening it is asynchronous, and `_dispatch` is not. Handing it over
on the strength of the id alone gives `fetchIndex` the envelope — nonce and
ciphertext, no entries — and never calls `onIndexSync` at all.
`req_id` is 0 here because it is the transport's first request, and a
falsy id is exactly the one a presence check gets wrong.
"""
out = _run([_sync_frame(GEK, "a-film.mkv", req_id=0)])
assert [e["event"] for e in out["events"]] == ["index_sync"], (
"the consumer was never told about an index that arrived as a reply")
assert out["events"][0]["entries"] == ["a-film.mkv"]
assert out["events"][0]["hasCiphertext"] is False
assert out["fetchIndex"]["state"] == "resolved"
assert out["fetchIndex"]["entries"] == ["a-film.mkv"], (
"the caller was handed the sealed envelope instead of the index")
|