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
|
"""Revoke is one door, it is admin-only, and it broadcasts.
Two coupled gaps used to sit in the moderation surface (docs/MESHBAY_DESIGN.md
§7.5):
* `admin_patch_group` let a *moderator* set a group to `revoked`, while the
user handler makes revoke admin-only;
* a `revoked` set through either PATCH was **never broadcast** to nodes —
unlike `POST /v1/admin/revoke` and account deletion — so it behaved like
`suspended` on nodes while claiming to be the signed, node-enforced state.
Revoke now has one path, `POST /v1/admin/revoke` (admin-only, signs and
broadcasts). PATCH refuses `revoked` and refuses to move an entity *out* of
`revoked` unless the caller is an admin.
"""
import pytest
from meshbay_hub.api.deps import set_admin_usernames
async def _register(client, username):
r = await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@test.local", "auth_key": "k" * 44})
assert r.status_code == 201
return r.json()["user_id"]
async def _login(client, username):
r = await client.post("/v1/users/login", json={
"username": username, "auth_key": "k" * 44})
assert r.status_code == 200
return r.json()["access_token"]
def _h(token):
return {"Authorization": f"Bearer {token}"}
async def _admin(client, name="admin_rev_test"):
await _register(client, name)
set_admin_usernames([name])
return await _login(client, name)
async def _moderator(client, admin_token, name="mod_rev_test"):
uid = await _register(client, name)
r = await client.patch(f"/v1/admin/users/{uid}", json={"role": "moderator"},
headers=_h(admin_token))
assert r.status_code == 200
return uid, await _login(client, name)
async def _a_group(client, owner="owner_rev_test"):
await _register(client, owner)
tok = await _login(client, owner)
r = await client.post("/v1/groups", headers=_h(tok),
json={"name": "g", "visibility": "private", "join_policy": "invite"})
assert r.status_code == 201
return r.json()["group_id"]
async def _group_status(client, admin_token, group_id):
data = (await client.get("/v1/admin/groups?limit=200", headers=_h(admin_token))).json()
return next(g["status"] for g in data["groups"] if g["id"] == group_id)
# ── PATCH cannot revoke ──────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_moderator_cannot_revoke_a_group_via_patch(client):
admin_token = await _admin(client)
_, mod_token = await _moderator(client, admin_token)
gid = await _a_group(client)
r = await client.patch(f"/v1/admin/groups/{gid}", json={"status": "revoked"},
headers=_h(mod_token))
assert r.status_code == 403
assert await _group_status(client, admin_token, gid) == "active"
@pytest.mark.asyncio
async def test_admin_patch_revoked_group_is_redirected_not_silently_applied(client):
admin_token = await _admin(client)
gid = await _a_group(client)
r = await client.patch(f"/v1/admin/groups/{gid}", json={"status": "revoked"},
headers=_h(admin_token))
assert r.status_code == 400
assert "revoke" in r.json()["detail"].lower()
# And it was not quietly applied.
assert await _group_status(client, admin_token, gid) == "active"
@pytest.mark.asyncio
async def test_admin_patch_revoked_user_is_redirected(client):
admin_token = await _admin(client)
victim = await _register(client, "vic_rev_test")
r = await client.patch(f"/v1/admin/users/{victim}", json={"status": "revoked"},
headers=_h(admin_token))
assert r.status_code == 400
# ── The one door: /v1/admin/revoke, admin-only, and it broadcasts ────────────
@pytest.mark.asyncio
async def test_admin_revoke_group_broadcasts_and_sets_status(client):
admin_token = await _admin(client)
gid = await _a_group(client)
r = await client.post("/v1/admin/revoke", headers=_h(admin_token),
json={"target": "group", "target_id": gid})
assert r.status_code == 200
body = r.json()
assert body["status"] == "revoked"
assert "nodes_notified" in body # it went through the broadcast path
assert await _group_status(client, admin_token, gid) == "revoked"
@pytest.mark.asyncio
async def test_moderator_cannot_reach_the_revoke_endpoint(client):
admin_token = await _admin(client)
_, mod_token = await _moderator(client, admin_token)
gid = await _a_group(client)
r = await client.post("/v1/admin/revoke", headers=_h(mod_token),
json={"target": "group", "target_id": gid})
assert r.status_code == 403
# ── Leaving `revoked` is an admin's call ─────────────────────────────────────
@pytest.mark.asyncio
async def test_moderator_cannot_unrevoke_a_group(client):
admin_token = await _admin(client)
_, mod_token = await _moderator(client, admin_token)
gid = await _a_group(client)
await client.post("/v1/admin/revoke", headers=_h(admin_token),
json={"target": "group", "target_id": gid})
r = await client.patch(f"/v1/admin/groups/{gid}", json={"status": "active"},
headers=_h(mod_token))
assert r.status_code == 403
assert await _group_status(client, admin_token, gid) == "revoked"
@pytest.mark.asyncio
async def test_moderator_cannot_unrevoke_a_user(client):
admin_token = await _admin(client)
_, mod_token = await _moderator(client, admin_token)
victim = await _register(client, "vic2_rev_test")
await client.post("/v1/admin/revoke", headers=_h(admin_token),
json={"target": "user", "target_id": victim})
r = await client.patch(f"/v1/admin/users/{victim}", json={"status": "active"},
headers=_h(mod_token))
assert r.status_code == 403
# ── Regression: suspend/unsuspend by a moderator still works ─────────────────
@pytest.mark.asyncio
async def test_moderator_can_still_suspend_and_restore(client):
admin_token = await _admin(client)
_, mod_token = await _moderator(client, admin_token)
gid = await _a_group(client)
assert (await client.patch(f"/v1/admin/groups/{gid}", json={"status": "suspended"},
headers=_h(mod_token))).status_code == 200
assert (await client.patch(f"/v1/admin/groups/{gid}", json={"status": "active"},
headers=_h(mod_token))).status_code == 200
|