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
|
"""
Integration tests for the Hub API.
Uses SQLite in-memory + httpx.AsyncClient — no PostgreSQL, no network.
"""
import base64
import pytest
import pytest_asyncio
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from cryptography.hazmat.primitives import serialization
from meshbay_common.crypto import generate_gek, pk_to_b64, wrap_gek
def _gen_user_keys():
sk_ed = Ed25519PrivateKey.generate()
sk_x = X25519PrivateKey.generate()
return (
pk_to_b64(sk_ed.public_key()),
pk_to_b64(sk_x.public_key()),
sk_x,
)
# ── Hub info ──────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_hub_info(client):
r = await client.get("/v1/hub/info")
assert r.status_code == 200
data = r.json()
assert "mnp_version" in data
assert "mhp_version" in data
@pytest.mark.asyncio
async def test_hub_pubkey(client):
r = await client.get("/v1/hub/pubkey")
assert r.status_code == 200
pem = r.json()["pk_hub_pem"]
assert pem.startswith("-----BEGIN PUBLIC KEY-----")
# ── Users ─────────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_register_and_login(client):
pk_ed, pk_x, _ = _gen_user_keys()
r = await client.post("/v1/users/register", json={
"username": "alice", "email": "alice@example.com",
"password": "alicepass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x,
})
assert r.status_code == 201
assert "user_id" in r.json()
r = await client.post("/v1/users/login", json={
"username": "alice", "password": "alicepass99"})
assert r.status_code == 200
data = r.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["token_type"] == "bearer"
@pytest.mark.asyncio
async def test_register_duplicate_rejected(client):
pk_ed, pk_x, _ = _gen_user_keys()
body = {"username": "bob", "email": "bob@example.com",
"password": "bobpass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}
await client.post("/v1/users/register", json=body)
r = await client.post("/v1/users/register", json=body)
assert r.status_code == 409
@pytest.mark.asyncio
async def test_wrong_password_rejected(client):
pk_ed, pk_x, _ = _gen_user_keys()
await client.post("/v1/users/register", json={
"username": "carol", "email": "carol@example.com",
"password": "carolpass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
r = await client.post("/v1/users/login", json={
"username": "carol", "password": "wrongpass"})
assert r.status_code == 401
@pytest.mark.asyncio
async def test_jwt_offline_verify(client, hub_key_path):
"""JWT returned by login must be verifiable offline with hub's public key."""
import jwt as pyjwt
pk_ed, pk_x, _ = _gen_user_keys()
await client.post("/v1/users/register", json={
"username": "dave", "email": "dave@example.com",
"password": "davepass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
r = await client.post("/v1/users/login", json={
"username": "dave", "password": "davepass99"})
token = r.json()["access_token"]
r_pk = await client.get("/v1/hub/pubkey")
hub_pk_pem = r_pk.json()["pk_hub_pem"].encode()
decoded = pyjwt.decode(token, hub_pk_pem, algorithms=["EdDSA"])
assert decoded["pk_user"] == pk_ed
assert "jti" in decoded # mandatory
@pytest.mark.asyncio
async def test_token_refresh(client):
pk_ed, pk_x, _ = _gen_user_keys()
await client.post("/v1/users/register", json={
"username": "eve", "email": "eve@example.com",
"password": "evepass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
r = await client.post("/v1/users/login", json={
"username": "eve", "password": "evepass99"})
rt = r.json()["refresh_token"]
at = r.json()["access_token"]
r2 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt})
assert r2.status_code == 200
assert r2.json()["access_token"] != at # new token (different jti)
@pytest.mark.asyncio
async def test_get_user_pubkeys(client):
pk_ed, pk_x, _ = _gen_user_keys()
await client.post("/v1/users/register", json={
"username": "frank", "email": "frank@example.com",
"password": "frankpass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
login = await client.post("/v1/users/login", json={
"username": "frank", "password": "frankpass99"})
token = login.json()["access_token"]
r = await client.get("/v1/users/frank/pubkeys",
headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 200
assert r.json()["pk_ed25519"] == pk_ed
assert r.json()["pk_x25519"] == pk_x
# ── Nodes ─────────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_announce_and_get_node(client):
pk_ed, pk_x, _ = _gen_user_keys()
await client.post("/v1/users/register", json={
"username": "node1", "email": "n@example.com",
"password": "nodepass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
login = await client.post("/v1/users/login", json={
"username": "node1", "password": "nodepass99"})
token = login.json()["access_token"]
hdrs = {"Authorization": f"Bearer {token}"}
r = await client.post("/v1/nodes/announce",
json={"pk_node": pk_ed, "endpoint_hint": "1.2.3.4:19000"},
headers=hdrs)
assert r.status_code == 201
node_id = r.json()["node_id"]
r2 = await client.get(f"/v1/nodes/{node_id}", headers=hdrs)
assert r2.status_code == 200
assert r2.json()["pk_node"] == pk_ed
assert r2.json()["endpoint_hint"] == "1.2.3.4:19000"
# ── Groups + GEK bundles ──────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_group_gek_roundtrip(client):
"""Admin creates group, wraps GEK for member, member retrieves and can unwrap."""
import jwt as pyjwt
from meshbay_common.crypto import unwrap_gek
# Register admin (alice2) and member (bob2)
pk_ed_a, pk_x_a, _ = _gen_user_keys()
pk_ed_b, pk_x_b, sk_x_b = _gen_user_keys()
for uname, email, pwd, pk_ed, pk_x in [
("alice2", "a2@x.com", "alicepass99", pk_ed_a, pk_x_a),
("bob2", "b2@x.com", "bobpass99", pk_ed_b, pk_x_b),
]:
await client.post("/v1/users/register", json={
"username": uname, "email": email, "password": pwd,
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
def _token(uname, pwd):
async def _inner():
r = await client.post("/v1/users/login",
json={"username": uname, "password": pwd})
return r.json()["access_token"]
return _inner
alice_token = (await client.post("/v1/users/login",
json={"username": "alice2", "password": "alicepass99"})).json()["access_token"]
bob_token = (await client.post("/v1/users/login",
json={"username": "bob2", "password": "bobpass99"})).json()["access_token"]
a_hdrs = {"Authorization": f"Bearer {alice_token}"}
b_hdrs = {"Authorization": f"Bearer {bob_token}"}
# Alice creates group
r = await client.post("/v1/groups", json={"name": "mygroup"}, headers=a_hdrs)
assert r.status_code == 201
group_id = r.json()["group_id"]
# Alice generates GEK and wraps it for bob
gek = generate_gek()
pk_bob_raw = base64.b64decode(pk_x_b)
bundle = wrap_gek(gek, pk_bob_raw)
r = await client.post(f"/v1/groups/{group_id}/members/bob2/gek",
json=bundle, headers=a_hdrs)
assert r.status_code == 201
# Bob retrieves his bundle
r = await client.get(f"/v1/groups/{group_id}/gek", headers=b_hdrs)
assert r.status_code == 200
retrieved = r.json()
# Bob unwraps — must recover original GEK
sk_b_raw = sk_x_b.private_bytes(
serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
serialization.NoEncryption())
recovered = unwrap_gek(retrieved, sk_b_raw, pk_bob_raw)
assert recovered == gek
@pytest.mark.asyncio
async def test_non_admin_cannot_add_member(client):
pk_ed_a, pk_x_a, _ = _gen_user_keys()
pk_ed_b, pk_x_b, _ = _gen_user_keys()
for uname, email, pwd, pk_ed, pk_x in [
("charlie", "c@x.com", "charliepass", pk_ed_a, pk_x_a),
("dan", "d@x.com", "danpass1234", pk_ed_b, pk_x_b),
]:
await client.post("/v1/users/register", json={
"username": uname, "email": email, "password": pwd,
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
charlie_token = (await client.post("/v1/users/login",
json={"username": "charlie", "password": "charliepass"})).json()["access_token"]
dan_token = (await client.post("/v1/users/login",
json={"username": "dan", "password": "danpass1234"})).json()["access_token"]
r = await client.post("/v1/groups", json={"name": "charlies-group"},
headers={"Authorization": f"Bearer {charlie_token}"})
group_id = r.json()["group_id"]
gek = generate_gek()
bundle = wrap_gek(gek, base64.b64decode(pk_x_b))
# Dan (non-admin) tries to add a member → 403
r = await client.post(f"/v1/groups/{group_id}/members/charlie/gek",
json=bundle,
headers={"Authorization": f"Bearer {dan_token}"})
assert r.status_code == 403
|