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
|
"""
The test that asserts the property, rather than the mechanism.
Worth more than checking that a `ct` field is present: this fails for any future
change that puts a name back in the clear, including one nobody thought of as an
index message. Findings C1 (the node HTTP API served the index and plaintext files
on 0.0.0.0 with no authentication) and C6 (the TCP transport accepted a bare JWT
with no GEK proof) were both "a peer that had not completed the handshake was served
data"; sealed, that bug leaks ciphertext instead of a library's filenames.
The distinctive strings below are invented and could not occur by chance in msgpack
framing or in a field name.
"""
import msgpack
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_ACK, PURPOSE_INDEX, seal, unseal
from meshbay_common.protocol import MNP
from meshbay_node.indexer import DirectoryIndexer, GroupIndex
from meshbay_node.transport.wire import index_delta_message, index_sync_message
# A filename and a folder name that appear nowhere else in the tree.
SECRET_FILE = "quixotry-ledger-2019.pdf"
SECRET_DIR = "zarfwidget-archive"
SECRET_ROOT = "/srv/vasculum-private/library"
@pytest.fixture
def gek():
return generate_gek()
@pytest.fixture
async def indexer(tmp_path, gek):
shared = tmp_path / "shared"
(shared / SECRET_DIR).mkdir(parents=True)
(shared / SECRET_DIR / SECRET_FILE).write_bytes(b"x" * 64)
roots = one_root(shared, name="library")
idx = DirectoryIndexer(
roots=roots, group_id="g-1", sk_node=Ed25519PrivateKey.generate(), gek=gek)
await idx.initial_scan()
return idx
def _assert_absent(frame: bytes, *words: str) -> None:
for word in words:
assert word.encode() not in frame, f"{word!r} travels in the clear"
@pytest.mark.asyncio
async def test_index_sync_frame_carries_no_filename(indexer):
frame = msgpack.packb(index_sync_message(indexer.index, indexer.roots),
use_bin_type=True)
# Not only the whole name — a fragment of it would be just as much of a leak.
_assert_absent(frame, SECRET_FILE, "quixotry", SECRET_DIR, "zarfwidget",
"entries", "dirs")
# And the routing fields are still readable, or nothing could be dispatched.
msg = msgpack.unpackb(frame, raw=False)
assert msg["type"] == MNP.INDEX_SYNC
assert msg["group_id"] == "g-1"
assert set(msg) == {"type", "v", "group_id", "nonce", "ct"}
@pytest.mark.asyncio
async def test_index_sync_payload_still_says_everything(indexer, gek):
"""Sealed, not lost: every field a client reads is inside."""
msg = index_sync_message(indexer.index, indexer.roots)
payload = unseal(gek, PURPOSE_INDEX, MNP.INDEX_SYNC, "g-1", msg)
assert [e["name"] for e in payload["entries"]] == [SECRET_FILE]
assert any(d.endswith(SECRET_DIR) for d in payload["dirs"])
assert payload["roots"]
# Moved inside deliberately (D4): there is no reason to act on a version
# carried by a message we have not authenticated.
assert payload["version"] == indexer.index.version
assert "version" not in msg
@pytest.mark.asyncio
async def test_index_delta_frame_carries_no_filename(indexer, gek):
"""
The delta is where a cleartext path would most easily survive: it used to be
hand-built in the daemon, a third construction site for an index message.
"""
before = GroupIndex._snapshot(
"g-1", indexer.index.sk_node, gek, indexer.index.version - 1, {})
delta = indexer.index.diff(before)
assert delta.additions
frame = msgpack.packb(index_delta_message(indexer.index, delta),
use_bin_type=True)
_assert_absent(frame, SECRET_FILE, "quixotry", "additions", "deletions")
payload = unseal(gek, PURPOSE_INDEX, MNP.INDEX_DELTA, "g-1",
msgpack.unpackb(frame, raw=False))
assert [e["name"] for e in payload["additions"]] == [SECRET_FILE]
assert payload["base_version"] == delta.base_version
def test_handshake_ack_frame_carries_no_configuration(gek):
"""
The ack is the line that matters most, and it is an integrity gap as much as a
confidentiality one: the node signs `handshake_transcript(...)`, which names no
ack field, so every value below was authenticated by the DTLS channel alone.
"""
config = {
"is_node_admin": True,
"video_root": SECRET_ROOT,
"enabled_apps": ["files", "videos"],
}
ack = {
"type": MNP.HANDSHAKE_ACK,
"v": "1.0",
"node_pk": "Tk9ERVBL",
"proof": "cHJvb2Y=",
"sig": "c2ln",
**seal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, "g-1", config),
}
frame = msgpack.packb(ack, use_bin_type=True)
_assert_absent(frame, SECRET_ROOT, "vasculum", "video_root",
"enabled_apps", "is_node_admin")
# What a client needs in order to authenticate the node is still in clear —
# it verifies those *before* it would trust a decryption.
msg = msgpack.unpackb(frame, raw=False)
assert msg["node_pk"] and msg["proof"] and msg["sig"]
assert unseal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, "g-1", msg) == config
def test_index_progress_stays_clear_and_stays_counters():
"""
Decision D3: `index_progress` is deliberately *not* sealed — counters only,
pushed every couple of seconds for the whole length of a scan, so sealing it
would buy a rough library size and cost a decrypt per push.
The field list is re-derived from the daemon's own source rather than restated
here, so this fails the day the message grows something that names anything —
`IndexProgress` already carries a `current_dir` the push deliberately omits,
and adding it would be one line. That is the moment the trade above is void.
"""
import ast
import inspect
import textwrap
from meshbay_node.daemon import NodeDaemon
source = inspect.getsource(NodeDaemon._push_index_progress)
tree = ast.parse(textwrap.dedent(source))
dicts = [n for n in ast.walk(tree) if isinstance(n, ast.Dict)]
assert len(dicts) == 1, "more than one message built here — re-read this test"
keys = {k.value for k in dicts[0].keys}
assert keys == {"type", "v", "group_id",
"scanning", "scanned_bytes", "total_bytes"}, (
f"index_progress now carries {keys} — re-read decision D3 before shipping it")
assert "seal(" not in source
assert "D3" in source, "the reason it is not sealed must stay next to the code"
|