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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
|
"""
Unified handshake — properties every transport must inherit (11.5.4/5, C6, C3, L4).
These test the shared module rather than any one transport. The point of the module
is that WebRTC and QUIC cannot drift apart again: the handshake existed three times
over and only the newest copy enforced the GEK proof.
"""
import time
import jwt
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.handshake import (
HANDSHAKE_PREFIX,
NONCE_LEN,
ROLE_CLIENT,
ROLE_NODE,
AuthorizedPeer,
HandshakeError,
authorize_token,
handshake_transcript,
make_proof,
quic_binding,
verify_proof,
webrtc_binding,
)
GEK = b"\x11" * 32
GROUP = "g" * 32
NONCE_C = b"\x01" * NONCE_LEN
NONCE_S = b"\x02" * NONCE_LEN
BINDING = webrtc_binding(b"\xaa" * 32, b"\xbb" * 32)
# ── Token authorization ───────────────────────────────────────────────────────
@pytest.fixture
def hub_key():
sk = Ed25519PrivateKey.generate()
pem = sk.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
pub = sk.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
return pem, pub
def _token(sk_pem, **over):
now = int(time.time())
payload = {
"iss": "test-hub", "sub": "user-1", "jti": "jti-1",
"iat": now, "exp": now + 3600,
"groups": [GROUP], "scope": "user", "pk_user": "pk",
}
payload.update(over)
return jwt.encode(payload, sk_pem, algorithm="EdDSA")
def test_valid_token_authorizes(hub_key):
sk_pem, pk_pem = hub_key
peer = authorize_token(_token(sk_pem), pk_pem, group_id=GROUP)
assert isinstance(peer, AuthorizedPeer)
assert peer.user_id == "user-1"
def test_group_id_is_mandatory(hub_key):
"""
M1: group_id used to be optional, and omitting it skipped the membership check
entirely while falling back to the node's first group.
"""
sk_pem, pk_pem = hub_key
with pytest.raises(HandshakeError, match="group_id"):
authorize_token(_token(sk_pem), pk_pem, group_id="")
def test_non_member_refused(hub_key):
sk_pem, pk_pem = hub_key
token = _token(sk_pem, groups=["other-group"])
with pytest.raises(HandshakeError, match="Not a member"):
authorize_token(token, pk_pem, group_id=GROUP)
def test_node_scoped_token_refused_on_client_path(hub_key):
"""M9: a daemon's node-scoped token must not be usable as a client token."""
sk_pem, pk_pem = hub_key
token = _token(sk_pem, scope="node")
with pytest.raises(HandshakeError, match="scope"):
authorize_token(token, pk_pem, group_id=GROUP)
def test_unhosted_group_refused(hub_key):
sk_pem, pk_pem = hub_key
with pytest.raises(HandshakeError, match="not hosted"):
authorize_token(_token(sk_pem), pk_pem, group_id=GROUP,
hosted_groups={"some-other-group"})
def test_denylisted_token_refused(hub_key):
sk_pem, pk_pem = hub_key
class _Deny:
def is_denied(self, user_id, jti, group_id=""):
return group_id == GROUP
with pytest.raises(HandshakeError, match="revoked"):
authorize_token(_token(sk_pem), pk_pem, group_id=GROUP, denylist=_Deny())
def test_forged_token_refused(hub_key):
_, pk_pem = hub_key
other = Ed25519PrivateKey.generate().private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
with pytest.raises(HandshakeError, match="Invalid JWT"):
authorize_token(_token(other), pk_pem, group_id=GROUP)
# ── Proof transcript ──────────────────────────────────────────────────────────
def test_transcript_is_domain_separated():
assert handshake_transcript(
ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING
).startswith(HANDSHAKE_PREFIX)
def test_client_proof_is_not_a_node_proof():
"""
C3: the node proves itself with the same key over the same connection. Without
the role bound in, a client's proof would satisfy the node check and vice
versa, so an impersonating peer could simply echo it back.
"""
client = make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
assert not verify_proof(GEK, client, ROLE_NODE, GROUP, NONCE_C, NONCE_S, BINDING)
node = make_proof(GEK, ROLE_NODE, GROUP, NONCE_C, NONCE_S, BINDING)
assert not verify_proof(GEK, node, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
assert client != node
@pytest.mark.parametrize("field,value", [
("group_id", "other-group"),
("nonce_client", b"\x09" * NONCE_LEN),
("nonce_node", b"\x09" * NONCE_LEN),
("binding", webrtc_binding(b"\xcc" * 32, b"\xdd" * 32)),
])
def test_proof_binds_every_field(field, value):
base = dict(role=ROLE_CLIENT, group_id=GROUP, nonce_client=NONCE_C,
nonce_node=NONCE_S, binding=BINDING)
proof = make_proof(GEK, **base)
altered = dict(base, **{field: value})
assert not verify_proof(GEK, proof, **altered), (
f"proof ignores {field} — replayable across connections")
def test_proof_requires_channel_binding():
"""
L4/NS5: the old transcript was nonce ‖ offer_fp ‖ answer_fp, and a missing
fingerprint silently degraded it to nonce-only, dropping MitM detection.
"""
with pytest.raises(HandshakeError, match="binding"):
make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, b"")
assert not verify_proof(
GEK, b"\x00" * 32, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, b"")
def test_proof_requires_gek():
with pytest.raises(HandshakeError):
make_proof(b"", ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
def test_wrong_gek_fails():
proof = make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
assert not verify_proof(
b"\x22" * 32, proof, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
def test_transcript_is_unambiguous():
"""
L4: with bare concatenation, a crafted group id could impersonate the
following field and two different handshakes would produce identical bytes.
"""
a = handshake_transcript(ROLE_CLIENT, "gg", NONCE_C, NONCE_S, BINDING)
b = handshake_transcript(ROLE_CLIENT, "g", b"g" + NONCE_C, NONCE_S, BINDING)
assert a != b
def test_bindings_differ_by_transport():
"""A WebRTC proof must not be replayable on a QUIC connection."""
assert webrtc_binding(b"\xaa" * 32, b"\xbb" * 32) != quic_binding(b"cert-der")
def test_membership_refusal_carries_a_code_a_client_can_act_on():
"""
`groups` is baked into the token at login, so someone added to a group after
signing in is refused although they are a member. The client refreshes and
retries on this code — it must not have to match on the human wording, which
is exactly the kind of coupling that breaks when someone improves a message.
"""
import jwt as _jwt
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
sk = Ed25519PrivateKey.generate()
pem_priv = sk.private_bytes(
serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
serialization.NoEncryption())
pem_pub = sk.public_key().public_bytes(
serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
token = _jwt.encode({"sub": "u1", "jti": "j1", "scope": "user", "groups": []},
pem_priv, algorithm="EdDSA")
with pytest.raises(HandshakeError) as excinfo:
authorize_token(token, pem_pub, group_id="g" * 32)
assert excinfo.value.code == "not_a_member"
|