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
176
177
178
179
180
181
182
|
"""
The transports must produce the same wire, message for message.
`file_chunk` and `index_sync` were each built twice — once in `webrtc_server.py`, once
in `quic_server.py` — and the two copies disagreed. WebRTC sent binary, unsigned chunks
carrying a `file_id`; QUIC sent base64 fields, two hashes and a per-chunk Ed25519
signature, and no `file_id` at all. `index_sync` was plain entries on one transport and
a `GroupIndex.serialize()` envelope on the other. One type, two shapes, and nothing
that failed when they drifted.
This is the same guard the unified handshake has (`meshbay_common/handshake.py`): the
encoders now live in one place, and these tests fail if a transport grows its own copy
again.
"""
import inspect
import pytest
from conftest import one_root
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_common.groupbox import PURPOSE_INDEX, unseal
from meshbay_common.protocol import MNP, file_chunk_plaintext, file_chunk_wire
from meshbay_node.indexer import DirectoryIndexer
from meshbay_node.transport import quic_server, webrtc_server
from meshbay_node.transport.wire import index_sync_message
@pytest.fixture
def gek():
return generate_gek()
@pytest.fixture
def shared_dir(tmp_path):
d = tmp_path / "shared"
d.mkdir()
(d / "film.mkv").write_bytes(b"payload " * 500)
(d / "sub").mkdir()
return d
def test_both_transports_use_the_one_chunk_encoder():
"""Neither server may encrypt a chunk itself."""
for module in (webrtc_server, quic_server):
source = inspect.getsource(module)
assert "file_chunk_wire" in source, f"{module.__name__} bypasses the shared encoder"
assert "encrypt_chunk_aes(" not in source, (
f"{module.__name__} encrypts a chunk on its own — that is how the two "
f"copies diverged the first time")
assert "chunk_key_aes(" not in source, (
f"{module.__name__} derives a chunk key on its own")
def test_both_transports_use_the_one_index_builder():
for module in (webrtc_server, quic_server):
source = inspect.getsource(module)
assert "index_sync_message" in source, (
f"{module.__name__} builds index_sync itself")
assert "index_b64" not in inspect.getsource(quic_server), (
"QUIC is serializing the index again — that was the fork")
def test_neither_transport_seals_by_hand():
"""
`groupbox` is the only sealer, the same rule `file_chunk_wire` already has.
A server reaching for AESGCM or HKDF directly is a second envelope waiting to
disagree with the first about a nonce length, an info string or an AAD.
"""
for module in (webrtc_server, quic_server):
source = inspect.getsource(module)
assert "seal(" in source, f"{module.__name__} sends an unsealed ack"
assert "AESGCM(" not in source, (
f"{module.__name__} builds its own AEAD instead of using groupbox")
assert "HKDF(" not in source, (
f"{module.__name__} derives its own subkey instead of using groupbox")
def test_the_daemon_does_not_build_an_index_message_itself():
"""
The delta was hand-built in `_broadcast_index_change` — the third construction
site for an index message, and the one that would have kept sending cleartext
while the other two were sealed.
"""
from meshbay_node import daemon
source = inspect.getsource(daemon)
assert "index_delta_message" in source and "index_sync_message" in source
assert '"type": MNP.INDEX_DELTA' not in source, (
"the daemon builds index_delta by hand again")
assert '"type": MNP.INDEX_SYNC' not in source, (
"the daemon builds index_sync by hand again")
def test_chunk_wire_shape_is_identical_across_transports(gek, shared_dir):
"""The two servers' read-and-encrypt helpers agree on every field but the nonce."""
path = shared_dir / "film.mkv"
file_hash = bytes.fromhex("ab" * 32)
from_webrtc = webrtc_server._read_and_encrypt(gek, path, 0, file_hash, "ab" * 32)
from_quic = quic_server._read_and_encrypt(gek, path, 0, file_hash, "ab" * 32)
assert from_webrtc.keys() == from_quic.keys()
assert set(from_webrtc) == {
"type", "v", "file_id", "chunk_index", "plaintext_size", "nonce", "ct"}
assert from_webrtc["type"] == from_quic["type"] == MNP.FILE_CHUNK
for field in ("v", "file_id", "chunk_index", "plaintext_size"):
assert from_webrtc[field] == from_quic[field]
# Binary, not base64 — the 33% Phase 9.15 removed, and the QUIC copy kept.
assert isinstance(from_webrtc["nonce"], bytes)
assert isinstance(from_webrtc["ct"], bytes)
# A fresh nonce per encryption, so the ciphertexts differ while the plaintext
# both sides recover does not.
assert from_webrtc["ct"] != from_quic["ct"]
plaintext = path.read_bytes()
for msg in (from_webrtc, from_quic):
assert file_chunk_plaintext(gek, msg) == plaintext
def test_chunk_round_trip_rejects_a_tampered_ciphertext(gek):
msg = file_chunk_wire(gek, b"the payload", 3, bytes.fromhex("cd" * 32), "cd" * 32)
msg["ct"] = bytes([msg["ct"][0] ^ 1]) + msg["ct"][1:]
with pytest.raises(Exception):
file_chunk_plaintext(gek, msg)
def test_chunk_key_is_bound_to_file_and_index(gek):
"""A chunk cannot be replayed as another chunk, or as one of another file."""
file_hash = bytes.fromhex("ef" * 32)
msg = file_chunk_wire(gek, b"the payload", 7, file_hash, "ef" * 32)
moved = dict(msg, chunk_index=8)
with pytest.raises(Exception):
file_chunk_plaintext(gek, moved)
with pytest.raises(Exception):
file_chunk_plaintext(gek, msg, file_hash=bytes.fromhex("11" * 32))
@pytest.mark.asyncio
async def test_index_sync_shape(gek, shared_dir):
sk_node = Ed25519PrivateKey.generate()
roots = one_root(shared_dir)
indexer = DirectoryIndexer(roots=roots, group_id="g", sk_node=sk_node, gek=gek)
await indexer.initial_scan()
msg = index_sync_message(indexer.index, roots)
# In clear: what a receiver needs to route and version-check before it can
# decrypt, and nothing else.
assert msg["type"] == MNP.INDEX_SYNC
assert msg["group_id"] == "g"
assert set(msg) == {"type", "v", "group_id", "nonce", "ct"}
payload = unseal(gek, PURPOSE_INDEX, MNP.INDEX_SYNC, "g", msg)
assert [e["name"] for e in payload["entries"]] == ["film.mkv"]
# Directories are not index entries, so they travel separately — including the
# empty one, which no entry's path would have revealed.
assert any(d.endswith("sub") for d in payload["dirs"])
assert payload["roots"]
@pytest.mark.asyncio
async def test_a_wrong_key_raises_rather_than_reporting_an_empty_group(gek, shared_dir):
"""
§3.4, at the level a client would hit it. An index that fails to open must not
become an empty index: "the group has no files" is a legitimate state, so a
fallback there is indistinguishable from the truth — which is exactly what
makes it worse than a stop.
"""
sk_node = Ed25519PrivateKey.generate()
roots = one_root(shared_dir)
indexer = DirectoryIndexer(roots=roots, group_id="g", sk_node=sk_node, gek=gek)
await indexer.initial_scan()
msg = index_sync_message(indexer.index, roots)
with pytest.raises(Exception) as caught:
unseal(generate_gek(), PURPOSE_INDEX, MNP.INDEX_SYNC, "g", msg)
# Assert on the refusal, not on a degraded result.
assert not isinstance(caught.value, dict)
|