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
|
"""
Leaving a group of your own accord, and the cap on public groups.
Sits beside `test_group_membership.py`, which covers the owner removing someone
else. The two are deliberately different endpoints rather than one with an
exception for the self case, so they are tested apart.
The refusals are the interesting half: the owner who cannot walk out and leave
the group unmanageable, and the eleventh public group. What leaving does *not*
do is asserted too — the hub can only drop a membership row. The node keeps the
identity it pinned, the keypair bundle and the uploaded files until its operator
removes them, and whoever left still holds the group key they were served.
"""
import base64
import hashlib
from datetime import datetime, timezone
import pytest
from sqlalchemy import select
from meshbay_hub.api.groups import MAX_PUBLIC_GROUPS
from meshbay_hub.db.models import Group, GroupMember, User
def _auth_key(password: str, username: str) -> str:
salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
return base64.b64encode(
hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()
async def _user(client, username, password="a-long-enough-passphrase"):
await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@example.com",
"auth_key": _auth_key(password, username)})
r = await client.post("/v1/users/login", json={
"username": username, "auth_key": _auth_key(password, username)})
return {"Authorization": f"Bearer {r.json()['access_token']}"}
async def _group(client, owner, name, visibility="private"):
# A public group must be open to join — invite-only is refused, because a
# group everyone can find and nobody can enter is a dead end.
body = {"name": name, "visibility": visibility}
if visibility == "public":
body["join_policy"] = "open"
r = await client.post("/v1/groups", json=body, headers=owner)
assert r.status_code == 201, r.text
return r.json()["group_id"]
# ── Leaving ───────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_a_member_can_leave(client, db_session):
owner = await _user(client, "owner1")
member = await _user(client, "member1")
gid = await _group(client, owner, "readers")
await client.post(f"/v1/groups/{gid}/members/member1", json={}, headers=owner)
# Marked hosted, or the member would not see the group in the first place
# and the assertion below would hold whether or not leaving worked.
g = await db_session.get(Group, gid)
g.hosted_at = datetime.now(timezone.utc)
await db_session.commit()
before = await client.get("/v1/groups/mine", headers=member)
assert [g["id"] for g in before.json()["groups"]] == [gid], "precondition"
r = await client.post(f"/v1/groups/{gid}/leave", headers=member)
assert r.status_code == 200, r.text
assert r.json()["status"] == "left"
mine = await client.get("/v1/groups/mine", headers=member)
assert [g["id"] for g in mine.json()["groups"]] == []
@pytest.mark.asyncio
async def test_leaving_removes_only_that_membership_row(client, db_session):
"""The group and everyone else in it are untouched — this is not a deletion."""
owner = await _user(client, "owner2")
member = await _user(client, "member2")
gid = await _group(client, owner, "still-here")
await client.post(f"/v1/groups/{gid}/members/member2", json={}, headers=owner)
await client.post(f"/v1/groups/{gid}/leave", headers=member)
assert await db_session.get(Group, gid) is not None
rows = (await db_session.execute(
select(GroupMember).where(GroupMember.group_id == gid))).scalars().all()
assert len(rows) == 1, "the owner should still be a member"
mine = await client.get("/v1/groups/mine", headers=owner)
assert [g["id"] for g in mine.json()["groups"]] == [gid]
@pytest.mark.asyncio
async def test_leaving_does_not_touch_the_account_or_its_other_groups(client, db_session):
owner = await _user(client, "owner3")
member = await _user(client, "member3")
elsewhere = await _user(client, "owner3b")
gid = await _group(client, owner, "leaving")
other = await _group(client, elsewhere, "staying")
await client.post(f"/v1/groups/{gid}/members/member3", json={}, headers=owner)
await client.post(f"/v1/groups/{other}/members/member3", json={}, headers=elsewhere)
await client.post(f"/v1/groups/{gid}/leave", headers=member)
user = (await db_session.execute(
select(User).where(User.username == "member3"))).scalar_one()
assert user.status == "active"
assert await db_session.get(GroupMember, (other, user.id)) is not None
@pytest.mark.asyncio
async def test_the_owner_cannot_leave_their_own_group(client):
"""It would leave the group with nobody able to admit, edit or delete it."""
owner = await _user(client, "owner4")
gid = await _group(client, owner, "orphan-risk")
r = await client.post(f"/v1/groups/{gid}/leave", headers=owner)
assert r.status_code == 409, r.text
assert "own this group" in r.json()["detail"]
mine = await client.get("/v1/groups/mine", headers=owner)
assert [g["id"] for g in mine.json()["groups"]] == [gid]
@pytest.mark.asyncio
async def test_leaving_twice_is_refused(client):
owner = await _user(client, "owner5")
member = await _user(client, "member5")
gid = await _group(client, owner, "once")
await client.post(f"/v1/groups/{gid}/members/member5", json={}, headers=owner)
assert (await client.post(f"/v1/groups/{gid}/leave",
headers=member)).status_code == 200
assert (await client.post(f"/v1/groups/{gid}/leave",
headers=member)).status_code == 404
@pytest.mark.asyncio
async def test_leaving_a_group_you_were_never_in_is_refused(client):
owner = await _user(client, "owner6")
stranger = await _user(client, "stranger6")
gid = await _group(client, owner, "not-yours")
r = await client.post(f"/v1/groups/{gid}/leave", headers=stranger)
assert r.status_code == 404
# ── Public group cap ──────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_public_groups_are_capped(client):
owner = await _user(client, "prolific")
for i in range(MAX_PUBLIC_GROUPS):
await _group(client, owner, f"public-{i}", visibility="public")
r = await client.post("/v1/groups",
json={"name": "one-too-many", "visibility": "public",
"join_policy": "open"},
headers=owner)
assert r.status_code == 409, r.text
detail = r.json()["detail"]
assert str(MAX_PUBLIC_GROUPS) in detail
assert "private" in detail, "the message should name the way out"
@pytest.mark.asyncio
async def test_private_groups_are_not_capped(client):
"""Private groups cost other people nothing — they are invisible to non-members."""
owner = await _user(client, "hoarder")
for i in range(MAX_PUBLIC_GROUPS + 5):
await _group(client, owner, f"private-{i}")
mine = await client.get("/v1/groups/mine", headers=owner)
assert len(mine.json()["groups"]) == MAX_PUBLIC_GROUPS + 5
@pytest.mark.asyncio
async def test_a_suspended_public_group_does_not_hold_a_slot(client, db_session):
"""Moderation already dealt with the owner; the slot should not punish twice."""
owner = await _user(client, "moderated")
ids = [await _group(client, owner, f"pub-{i}", visibility="public")
for i in range(MAX_PUBLIC_GROUPS)]
suspended = await db_session.get(Group, ids[0])
suspended.status = "suspended"
await db_session.commit()
r = await client.post("/v1/groups",
json={"name": "replacement", "visibility": "public",
"join_policy": "open"},
headers=owner)
assert r.status_code == 201, r.text
@pytest.mark.asyncio
async def test_the_cap_is_per_owner(client):
"""Being a member of someone else's public groups costs nothing."""
a = await _user(client, "ownera")
await _user(client, "ownerb")
b = await _user(client, "ownerb2")
for i in range(MAX_PUBLIC_GROUPS):
gid = await _group(client, a, f"a-pub-{i}", visibility="public")
await client.post(f"/v1/groups/{gid}/members/ownerb2", json={}, headers=a)
r = await client.post("/v1/groups",
json={"name": "b-first", "visibility": "public",
"join_policy": "open"},
headers=b)
assert r.status_code == 201, r.text
@pytest.mark.asyncio
async def test_hub_staff_are_exempt(client, db_session):
"""The cap is anti-spam, not a rule about running an instance."""
owner = await _user(client, "instanceadmin")
user = (await db_session.execute(
select(User).where(User.username == "instanceadmin"))).scalar_one()
user.role = "admin"
await db_session.commit()
for i in range(MAX_PUBLIC_GROUPS + 2):
r = await client.post("/v1/groups",
json={"name": f"admin-pub-{i}", "visibility": "public",
"join_policy": "open"},
headers=owner)
assert r.status_code == 201, r.text
# ── Presence in the group list ────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_the_group_list_reports_node_presence(client):
"""`node_online` rides on the request the sidebar already makes.
It reflects the hub's signaling registry, so it says a node is connected
*to the hub* — not that this browser can reach it, and not something a
dishonest hub could not fake. The client downgrades it on a connection it
tried and failed, which is the evidence that concerns the reader.
"""
owner = await _user(client, "watcher")
gid = await _group(client, owner, "quiet")
mine = await client.get("/v1/groups/mine", headers=owner)
entry = next(g for g in mine.json()["groups"] if g["id"] == gid)
assert entry["node_online"] is False
from meshbay_hub.api import revocation
revocation._node_groups["node-x"] = [gid]
try:
mine = await client.get("/v1/groups/mine", headers=owner)
entry = next(g for g in mine.json()["groups"] if g["id"] == gid)
assert entry["node_online"] is True
finally:
revocation._node_groups.pop("node-x", None)
|