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
|
"""
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.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_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)
assert msg["type"] == MNP.INDEX_SYNC
assert msg["group_id"] == "g"
assert [e["name"] for e in msg["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 msg["dirs"])
assert msg["roots"]
|