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
|
"""
The browser half of MNP 2.0's sealed upload, measured rather than read.
`test_upload_sealed.py` (node side) proves the node opens what the shared
encoder produces and refuses everything else. This proves the *shipped browser
code* produces it — and, the part that matters more, that a caller of
`uploadFile` is still told the name the node stored the file under, which now
arrives sealed and would otherwise be `undefined` with nothing on screen to say
so: a chat attachment would point at a file that is not there.
Driven through `harness/upload_seal_probe.mjs`, which runs the shipped
`transport.js` over the shipped `crypto.js`. The node half in between is the
real `_do_file_upload`, writing to a real directory.
A source-reading test can see that `sealGroup` is called. Only this can see
whether what comes out of it opens.
"""
import json
import shutil
import subprocess
import tempfile
from pathlib import Path
import msgpack
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_common.protocol import MNP
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import RootSet
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
PROBE = Path(__file__).resolve().parent / "harness" / "upload_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-upload-probe"
CHUNK = 32
BODY = bytes(range(256)) * 3 # 768 bytes → 24 chunks of 32
def _run_probe(payload: dict) -> dict:
with tempfile.TemporaryDirectory() as d:
f = Path(d) / "input.json"
f.write_text(json.dumps(payload))
proc = subprocess.run(
["node", str(PROBE), str(STATIC), str(f)],
capture_output=True, text=True, timeout=60,
)
if proc.returncode != 0 or not proc.stdout:
pytest.fail(f"upload probe failed:\n{proc.stderr}")
return json.loads(proc.stdout)
def _node_session(tmp_path: Path, gek: bytes) -> WebRTCPeerSession:
root = tmp_path / "library"
root.mkdir()
index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = {
"roots": RootSet.build([{"path": str(root), "name": "library",
"writable": True}]),
"index": index, "sk_node": index.sk_node, "gek": gek,
}
session._group_id = GROUP
session._user_id = "prober"
session._pk_user = ""
session._uploads = {}
session.sent = []
session._send = session.sent.append
session._audit = lambda *a, **k: None
return session
def _probe_input(gek: bytes, mode: str, **extra) -> dict:
return {
"mode": mode, "gek": gek.hex(), "group_id": GROUP,
"node_version": "2.0", "chunk_size": CHUNK, "dir": "library",
"root": "library",
"file": {"name": "holiday.jpg", "data": BODY.hex()},
**extra,
}
@pytest.fixture(scope="module")
def _gek():
return generate_gek()
@pytest.fixture(scope="module")
def _sent(_gek):
"""Every frame the shipped `uploadFile` puts on the wire, unanswered."""
return _run_probe(_probe_input(_gek, "send"))
def test_the_browser_puts_no_filename_and_no_content_on_the_wire(_sent):
"""The whole point, measured on the bytes rather than read off the source."""
assert _sent["frames"], "the client sent nothing"
for hexframe in _sent["frames"]:
raw = bytes.fromhex(hexframe)
assert b"holiday.jpg" not in raw, "the filename is on the wire in clear"
msg = msgpack.unpackb(raw, raw=False)
assert set(msg) == {"type", "v", "upload_id", "chunk_index",
"total_chunks", "nonce", "ct"}
assert msg["type"] == MNP.FILE_UPLOAD
def test_a_real_node_opens_what_the_real_browser_sealed(tmp_path, _gek, _sent):
"""
End to end: the shipped browser encoder into the shipped node handler, with
the file that lands on disk as the assertion. A mismatch in the HKDF salt,
the AAD encoding or the payload shape shows up here as "did not open" — and
nowhere else until somebody tries to upload something.
"""
session = _node_session(tmp_path, _gek)
for hexframe in _sent["frames"]:
session._do_file_upload(msgpack.unpackb(bytes.fromhex(hexframe), raw=False))
errors = [m for m in session.sent if m.get("type") == "error"]
assert not errors, f"the node refused a frame the browser built: {errors[:1]}"
root = session._ctx["roots"].roots[0].path
assert (root / "holiday.jpg").read_bytes() == BODY
assert not list(root.glob("*.part")), "a temp file was left behind"
def test_the_caller_is_told_the_name_the_node_chose(tmp_path, _gek, _sent):
"""
`stored_as` is sealed now, so reading it takes a decrypt that can fail
silently. It must not: the node finds a free name rather than replacing
anything, and a chat attachment that never learns which name points at
nothing.
The acks below are the ones the node really produced — only their
`upload_id`, which is outside the seal, is retargeted to the second probe
run's own upload.
"""
session = _node_session(tmp_path, _gek)
# A file of that name is already there, so the node has to choose another.
(session._ctx["roots"].roots[0].path / "holiday.jpg").write_bytes(b"someone else's")
for hexframe in _sent["frames"]:
session._do_file_upload(msgpack.unpackb(bytes.fromhex(hexframe), raw=False))
acks = [msgpack.packb(m, use_bin_type=True).hex()
for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK]
assert len(acks) == len(_sent["frames"])
result = _run_probe(_probe_input(_gek, "receive", acks=acks))
assert result["state"] == "resolved", result.get("message")
assert result["stored"]["stored_as"] == "holiday (2).jpg"
assert result["stored"]["dir"] == "library"
def test_the_client_refuses_an_older_node_before_sending_a_chunk(_gek):
"""
A 1.x node would answer "Missing filename or data" — an error about the
wrong thing, naming no upload, which fails every upload in flight. Asked
first instead, and nothing goes on the wire.
"""
result = _run_probe(_probe_input(_gek, "receive", acks=[],
node_version="1.1"))
assert result["state"] == "rejected"
assert "older MeshBay" in result["message"]
assert result["frames"] == [], "a chunk was sent to a node that cannot open it"
|