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
|
"""Tests for revocation — admin endpoint + token signing."""
import time
import pytest
import jwt
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", "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",
pk_to_b64(sk_admin_ed.public_key()),
pk_to_b64(sk_admin_x.public_key()),
)
set_admin_usernames(["admin1"])
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", "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", "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",
pk_to_b64(sk_admin_ed.public_key()),
pk_to_b64(sk_admin_x.public_key()),
)
set_admin_usernames(["admin2"])
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",
pk_to_b64(sk_ed.public_key()),
pk_to_b64(sk_x.public_key()),
)
set_admin_usernames(["admin3"])
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"
|