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
|
"""
The group-key envelope: round trip, and every refusal it owes.
`groupbox.seal`/`unseal` is what puts `index_sync`, `index_delta` and the
`handshake_ack` configuration under a key the hub does not hold. The structural
tests here matter less than `test_index_no_cleartext.py`, which asserts the
property on a real frame; these pin the primitive.
"""
import msgpack
import pytest
from cryptography.exceptions import InvalidTag
from meshbay_common.crypto import generate_gek
from meshbay_common.groupbox import (
PURPOSE_ACK,
PURPOSE_INDEX,
group_key,
seal,
unseal,
)
PAYLOAD = {"version": 7, "entries": [{"name": "a.mkv", "size": 12}], "dirs": ["root"]}
@pytest.fixture
def gek():
return generate_gek()
def test_round_trip(gek):
for purpose in (PURPOSE_INDEX, PURPOSE_ACK):
sealed = seal(gek, purpose, "index_sync", "g1", PAYLOAD)
assert set(sealed) == {"nonce", "ct"}
assert unseal(gek, purpose, "index_sync", "g1", sealed) == PAYLOAD
def test_a_wrong_key_does_not_open(gek):
sealed = seal(gek, PURPOSE_INDEX, "index_sync", "g1", PAYLOAD)
with pytest.raises(InvalidTag):
unseal(generate_gek(), PURPOSE_INDEX, "index_sync", "g1", sealed)
def test_purposes_are_separate_key_spaces(gek):
"""
The reason there are two info strings rather than one key reused.
An ack sealed under the index subkey would otherwise be openable by anything
holding the index subkey — the confusion that treating "the index as chunk 0 of
a virtual index file" would create for chunk keys.
"""
assert group_key(gek, PURPOSE_INDEX) != group_key(gek, PURPOSE_ACK)
sealed = seal(gek, PURPOSE_INDEX, "index_sync", "g1", PAYLOAD)
with pytest.raises(InvalidTag):
unseal(gek, PURPOSE_ACK, "index_sync", "g1", sealed)
def test_a_body_cannot_be_replayed_as_another_message_type(gek):
"""The AAD's first half: an index_sync body is not an index_delta."""
sealed = seal(gek, PURPOSE_INDEX, "index_sync", "g1", PAYLOAD)
with pytest.raises(InvalidTag):
unseal(gek, PURPOSE_INDEX, "index_delta", "g1", sealed)
def test_a_body_cannot_be_moved_between_groups(gek):
"""
The AAD's second half. Two groups on one node share a GEK-holding process but
not a GEK; this closes the case where they do share one (a rotation in flight,
a test fixture, an operator reusing a key) as well.
"""
sealed = seal(gek, PURPOSE_INDEX, "index_sync", "group-a", PAYLOAD)
with pytest.raises(InvalidTag):
unseal(gek, PURPOSE_INDEX, "index_sync", "group-b", sealed)
def test_a_tampered_ciphertext_does_not_open(gek):
sealed = seal(gek, PURPOSE_INDEX, "index_sync", "g1", PAYLOAD)
sealed["ct"] = bytes([sealed["ct"][0] ^ 1]) + sealed["ct"][1:]
with pytest.raises(InvalidTag):
unseal(gek, PURPOSE_INDEX, "index_sync", "g1", sealed)
def test_a_message_that_is_not_sealed_is_refused_as_such(gek):
"""
Not an empty payload, and not a crash on a missing key — the two shapes a
caller might otherwise paper over.
"""
with pytest.raises(ValueError):
unseal(gek, PURPOSE_INDEX, "index_sync", "g1", {"entries": []})
def test_a_fresh_nonce_per_message(gek):
"""
Never derived from the payload: two identical payloads under one long-lived
subkey would then reuse a nonce, which for GCM is a total break.
"""
nonces = {seal(gek, PURPOSE_INDEX, "index_sync", "g1", PAYLOAD)["nonce"]
for _ in range(50)}
assert len(nonces) == 50
def test_no_key_is_an_error_not_a_plaintext_fallback(gek):
with pytest.raises(ValueError):
seal(None, PURPOSE_INDEX, "index_sync", "g1", PAYLOAD)
def test_an_unknown_purpose_is_refused(gek):
with pytest.raises(ValueError):
group_key(gek, "chat")
def test_the_envelope_carries_no_readable_payload(gek):
"""
The property, at the level of the primitive: what `seal` returns holds nothing
of what went in. `test_index_no_cleartext.py` asserts the same thing on the
real frames.
"""
payload = {"video_root": "holidays-2019-invoices", "entries": ["ledger.pdf"]}
frame = msgpack.packb(seal(gek, PURPOSE_ACK, "handshake_ack", "g1", payload))
for word in (b"holidays", b"invoices", b"ledger", b"video_root"):
assert word not in frame
|