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
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
|
"""Tests for node Ed25519 authentication and JWT scope enforcement.
Node auth flow: register user (browser keys) → login → link node key → node auth.
The node's Ed25519 key is separate from the user's browser key.
"""
import base64
import time
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from cryptography.hazmat.primitives import serialization
def _gen_ed25519():
sk = Ed25519PrivateKey.generate()
pk_raw = sk.public_key().public_bytes(
serialization.Encoding.Raw, serialization.PublicFormat.Raw)
return sk, base64.b64encode(pk_raw).decode()
def _gen_x25519():
sk = X25519PrivateKey.generate()
pk_raw = sk.public_key().public_bytes(
serialization.Encoding.Raw, serialization.PublicFormat.Raw)
return sk, base64.b64encode(pk_raw).decode()
async def _register(client, username, pk_ed_b64, pk_x_b64):
r = await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@test.local",
"password": "testpass99",
"pk_user_ed25519": pk_ed_b64, "pk_user_x25519": pk_x_b64,
})
assert r.status_code == 201
return r.json()["user_id"]
async def _user_login(client, username, password="testpass99"):
r = await client.post("/v1/users/login", json={
"username": username, "password": password,
})
assert r.status_code == 200
return r.json()["access_token"]
async def _link_node_key(client, user_token, pk_node_ed_b64):
r = await client.put("/v1/users/me/node_key", json={
"pk_node_ed25519": pk_node_ed_b64,
}, headers={"Authorization": f"Bearer {user_token}"})
assert r.status_code == 200
return r
async def _node_auth(client, username, sk_node_ed):
timestamp = int(time.time())
message = f"meshbay:node_auth:{username}:{timestamp}".encode()
signature = sk_node_ed.sign(message)
r = await client.post("/v1/nodes/auth", json={
"username": username,
"timestamp": timestamp,
"signature": base64.b64encode(signature).decode(),
})
return r
async def _setup_node_user(client, username):
"""Register user with browser keys, login, link node key. Returns (node_sk, node_token)."""
_, pk_ed = _gen_ed25519()
_, pk_x = _gen_x25519()
await _register(client, username, pk_ed, pk_x)
user_token = await _user_login(client, username)
sk_node, pk_node = _gen_ed25519()
await _link_node_key(client, user_token, pk_node)
return sk_node, user_token
# ── Core auth tests ──────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_node_auth_success(client):
sk_node, _ = await _setup_node_user(client, "nodeuser")
r = await _node_auth(client, "nodeuser", sk_node)
assert r.status_code == 200
data = r.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
@pytest.mark.asyncio
async def test_node_auth_wrong_key_rejected(client):
await _setup_node_user(client, "nodeuser2")
wrong_sk, _ = _gen_ed25519()
r = await _node_auth(client, "nodeuser2", wrong_sk)
assert r.status_code == 401
@pytest.mark.asyncio
async def test_node_auth_stale_timestamp(client):
sk_node, _ = await _setup_node_user(client, "nodeuser3")
timestamp = int(time.time()) - 120
message = f"meshbay:node_auth:nodeuser3:{timestamp}".encode()
signature = sk_node.sign(message)
r = await client.post("/v1/nodes/auth", json={
"username": "nodeuser3",
"timestamp": timestamp,
"signature": base64.b64encode(signature).decode(),
})
assert r.status_code == 401
@pytest.mark.asyncio
async def test_node_auth_fails_without_node_key(client):
"""Node auth must fail if no node key has been registered."""
_, pk_ed = _gen_ed25519()
_, pk_x = _gen_x25519()
await _register(client, "nokey_user", pk_ed, pk_x)
sk_any, _ = _gen_ed25519()
r = await _node_auth(client, "nokey_user", sk_any)
assert r.status_code == 401
assert "No node key registered" in r.json()["detail"]
# ── Scope enforcement tests ──────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_node_scope_blocks_group_create(client):
sk_node, _ = await _setup_node_user(client, "operator")
r = await _node_auth(client, "operator", sk_node)
node_token = r.json()["access_token"]
r = await client.post("/v1/groups", json={
"name": "forbidden-group", "visibility": "private",
}, headers={"Authorization": f"Bearer {node_token}"})
assert r.status_code == 403
assert "Node-scoped" in r.json()["detail"]
@pytest.mark.asyncio
async def test_node_scope_blocks_add_member(client):
sk_node, user_token = await _setup_node_user(client, "op1")
r = await client.post("/v1/groups", json={
"name": "mygroup", "visibility": "private", "join_policy": "invite",
}, headers={"Authorization": f"Bearer {user_token}"})
assert r.status_code == 201
gid = r.json()["group_id"]
_, pk2 = _gen_ed25519()
_, px2 = _gen_x25519()
await _register(client, "member1", pk2, px2)
r = await _node_auth(client, "op1", sk_node)
node_token = r.json()["access_token"]
r = await client.post(f"/v1/groups/{gid}/members/member1",
headers={"Authorization": f"Bearer {node_token}"})
assert r.status_code == 403
@pytest.mark.asyncio
async def test_node_scope_blocks_delete_group(client):
sk_node, user_token = await _setup_node_user(client, "op2")
r = await client.post("/v1/groups", json={
"name": "deleteme", "visibility": "public", "join_policy": "open",
}, headers={"Authorization": f"Bearer {user_token}"})
gid = r.json()["group_id"]
r = await _node_auth(client, "op2", sk_node)
node_token = r.json()["access_token"]
r = await client.delete(f"/v1/groups/{gid}",
headers={"Authorization": f"Bearer {node_token}"})
assert r.status_code == 403
@pytest.mark.asyncio
async def test_node_scope_allows_read_members(client):
sk_node, user_token = await _setup_node_user(client, "op3")
r = await client.post("/v1/groups", json={
"name": "readgroup", "visibility": "private", "join_policy": "invite",
}, headers={"Authorization": f"Bearer {user_token}"})
gid = r.json()["group_id"]
r = await _node_auth(client, "op3", sk_node)
node_token = r.json()["access_token"]
r = await client.get(f"/v1/groups/{gid}/members",
headers={"Authorization": f"Bearer {node_token}"})
assert r.status_code == 200
assert len(r.json()["members"]) == 1
@pytest.mark.asyncio
async def test_node_scope_allows_pubkey_lookup(client):
sk_node, _ = await _setup_node_user(client, "op4")
r = await _node_auth(client, "op4", sk_node)
node_token = r.json()["access_token"]
r = await client.get("/v1/users/op4/pubkeys",
headers={"Authorization": f"Bearer {node_token}"})
assert r.status_code == 200
# An account id and the node's linking key — no user identity keys (H3).
assert "pk_ed25519" not in r.json()
assert r.json()["pk_node_ed25519"] is not None
@pytest.mark.asyncio
async def test_user_scope_still_works(client):
"""Verify that user-scoped tokens (browser login) still have full access."""
_, pk_ed = _gen_ed25519()
_, pk_x = _gen_x25519()
await _register(client, "webuser", pk_ed, pk_x)
token = await _user_login(client, "webuser")
r = await client.post("/v1/groups", json={
"name": "browser-group", "visibility": "public", "join_policy": "open",
}, headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 201
# ── Node key registration tests ──────────────────────────────────────────────
@pytest.mark.asyncio
async def test_link_node_key(client):
_, pk_ed = _gen_ed25519()
_, pk_x = _gen_x25519()
await _register(client, "linkuser", pk_ed, pk_x)
token = await _user_login(client, "linkuser")
_, pk_node = _gen_ed25519()
r = await client.put("/v1/users/me/node_key", json={
"pk_node_ed25519": pk_node,
}, headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 200
assert r.json()["pk_node_ed25519"] == pk_node
@pytest.mark.asyncio
async def test_link_node_key_invalid_format(client):
_, pk_ed = _gen_ed25519()
_, pk_x = _gen_x25519()
await _register(client, "badkey", pk_ed, pk_x)
token = await _user_login(client, "badkey")
r = await client.put("/v1/users/me/node_key", json={
"pk_node_ed25519": "not-valid-base64!!!",
}, headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 400
@pytest.mark.asyncio
async def test_link_node_key_blocked_for_node_scope(client):
"""Node-scoped tokens must not be able to change the node key."""
sk_node, _ = await _setup_node_user(client, "sneaky")
r = await _node_auth(client, "sneaky", sk_node)
node_token = r.json()["access_token"]
_, pk_evil = _gen_ed25519()
r = await client.put("/v1/users/me/node_key", json={
"pk_node_ed25519": pk_evil,
}, headers={"Authorization": f"Bearer {node_token}"})
assert r.status_code == 403
|