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
|
"""Tests for revocation — admin endpoint + token signing."""
import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from meshbay_common.crypto import pk_to_b64
from meshbay_hub.api.deps import set_admin_usernames
async def _register_and_login(client, username, pk_ed, pk_x):
await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@test.com",
"password": "testpass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x,
})
r = await client.post("/v1/users/login",
json={"username": username, "password": "testpass99"})
return r.json()["access_token"], r.json()["user_id"] if "user_id" in r.json() else None
@pytest.mark.asyncio
async def test_revoke_user_marks_db(client):
"""POST /v1/admin/revoke marks user as revoked in DB."""
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
sk_ed = Ed25519PrivateKey.generate()
sk_x = X25519PrivateKey.generate()
pk_ed = pk_to_b64(sk_ed.public_key())
pk_x = pk_to_b64(sk_x.public_key())
r = await client.post("/v1/users/register", json={
"username": "vic1_test", "email": "v@t.com", "password": "vicpass99",
"pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x,
})
victim_id = r.json()["user_id"]
# Admin (any registered user for now — production would add role check)
sk_admin_ed = Ed25519PrivateKey.generate()
sk_admin_x = X25519PrivateKey.generate()
admin_token, _ = await _register_and_login(
client, "admin1_test",
pk_to_b64(sk_admin_ed.public_key()),
pk_to_b64(sk_admin_x.public_key()),
)
set_admin_usernames(["admin1_test"])
r = await client.post("/v1/admin/revoke", json={
"target": "user", "target_id": victim_id, "reason": "spam",
}, headers={"Authorization": f"Bearer {admin_token}"})
assert r.status_code == 200
data = r.json()
assert data["status"] == "revoked"
assert "token" in data
# Victim can no longer login
r = await client.post("/v1/users/login", json={
"username": "vic1_test", "password": "vicpass99"})
assert r.status_code == 403
@pytest.mark.asyncio
async def test_revocation_token_verifiable_offline(client):
"""Revocation token is a valid JWT signed by hub Ed25519 key."""
sk_ed = Ed25519PrivateKey.generate()
sk_x = X25519PrivateKey.generate()
r = await client.post("/v1/users/register", json={
"username": "vic2_test", "email": "v2@t.com", "password": "vicpass99",
"pk_user_ed25519": pk_to_b64(sk_ed.public_key()),
"pk_user_x25519": pk_to_b64(sk_x.public_key()),
})
victim_id = r.json()["user_id"]
sk_admin_ed = Ed25519PrivateKey.generate()
sk_admin_x = X25519PrivateKey.generate()
admin_token, _ = await _register_and_login(
client, "admin2_test",
pk_to_b64(sk_admin_ed.public_key()),
pk_to_b64(sk_admin_x.public_key()),
)
set_admin_usernames(["admin2_test"])
r = await client.post("/v1/admin/revoke", json={
"target": "user", "target_id": victim_id, "reason": "test",
}, headers={"Authorization": f"Bearer {admin_token}"})
rev_token = r.json()["token"]
# Verify offline with hub's public key
r_pk = await client.get("/v1/hub/pubkey")
hub_pk_pem = r_pk.json()["pk_hub_pem"].encode()
decoded = jwt.decode(rev_token, hub_pk_pem, algorithms=["EdDSA"],
options={"verify_exp": False})
assert decoded["type"] == "revocation"
assert decoded["target"] == "user"
assert decoded["target_id"] == victim_id
assert "jti" in decoded
@pytest.mark.asyncio
async def test_revoke_group(client):
sk_ed = Ed25519PrivateKey.generate()
sk_x = X25519PrivateKey.generate()
admin_token, _ = await _register_and_login(
client, "admin3_test",
pk_to_b64(sk_ed.public_key()),
pk_to_b64(sk_x.public_key()),
)
set_admin_usernames(["admin3_test"])
hdrs = {"Authorization": f"Bearer {admin_token}"}
r = await client.post("/v1/groups", json={"name": "grp-to-revoke"}, headers=hdrs)
group_id = r.json()["group_id"]
r = await client.post("/v1/admin/revoke", json={
"target": "group", "target_id": group_id, "reason": "tos_violation",
}, headers=hdrs)
assert r.status_code == 200
assert r.json()["status"] == "revoked"
@pytest.mark.asyncio
async def test_group_status_message_names_the_real_state(client):
"""A member of a revoked group must not be told it was merely 'suspended'.
`GET /v1/groups/{id}/nodes` and `POST /join` used to answer "Group is
suspended" for any non-active status. Suspend is the reversible hub flag;
revoke is a signed instruction every node enforces. The client shows this
string verbatim, so it has to be the truth.
"""
sk_ed = Ed25519PrivateKey.generate()
sk_x = X25519PrivateKey.generate()
admin_token, _ = await _register_and_login(
client, "admin_msg",
pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key()))
set_admin_usernames(["admin_msg"])
hdrs = {"Authorization": f"Bearer {admin_token}"}
gid = (await client.post("/v1/groups", json={"name": "state-msg"},
headers=hdrs)).json()["group_id"]
# Suspend → the message says suspended.
await client.patch(f"/v1/admin/groups/{gid}", json={"status": "suspended"},
headers=hdrs)
r = await client.get(f"/v1/groups/{gid}/nodes", headers=hdrs)
assert r.status_code == 403 and r.json()["detail"] == "Group is suspended"
# Revoke → the message says revoked, not suspended.
await client.post("/v1/admin/revoke",
json={"target": "group", "target_id": gid, "reason": "x"},
headers=hdrs)
r = await client.get(f"/v1/groups/{gid}/nodes", headers=hdrs)
assert r.status_code == 403 and r.json()["detail"] == "Group is revoked"
|