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
|
"""
Account deletion, by the owner and by an administrator.
Deletion is the one action here that cannot be undone from the UI, so the tests
state what survives it as carefully as what does not. Two things survive on
purpose: the IP log, which exists for a year to answer legal requests and would
be useless if it could no longer say whose connection it recorded, and everything
on a node — files and pinned identities live on machines the hub does not
command.
"""
import hashlib
import pytest
from sqlalchemy import select
from meshbay_hub.db.models import GroupMember, Notification, RefreshToken, User
def _auth_key(password: str, username: str) -> str:
import base64
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 _register(client, username, password="a-long-enough-passphrase"):
r = await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@example.com",
"auth_key": _auth_key(password, username),
})
assert r.status_code in (200, 201), r.text
login = await client.post("/v1/users/login", json={
"username": username, "auth_key": _auth_key(password, username)})
return login.json()["access_token"], password
@pytest.mark.asyncio
async def test_owner_can_delete_their_account(client, db_session):
token, password = await _register(client, "leaver")
headers = {"Authorization": f"Bearer {token}"}
r = await client.request("DELETE", "/v1/users/me", headers=headers,
json={"auth_key": _auth_key(password, "leaver")})
assert r.status_code == 200, r.text
user = (await db_session.execute(
select(User).where(User.status == "deleted"))).scalar_one()
assert user.username.startswith("deleted-")
assert user.email == ""
assert user.pw_hash == b""
assert user.pk_node_ed25519 is None
@pytest.mark.asyncio
async def test_deleting_needs_the_passphrase_not_just_a_session(client):
"""
A live token may be a borrowed laptop or a tab left open. Something
irreversible asks again.
"""
token, _ = await _register(client, "careful")
r = await client.request("DELETE", "/v1/users/me",
headers={"Authorization": f"Bearer {token}"},
json={"auth_key": _auth_key("wrong one", "careful")})
assert r.status_code == 403
me = await client.get("/v1/users/me",
headers={"Authorization": f"Bearer {token}"})
assert me.status_code == 200, "the account must survive a failed attempt"
@pytest.mark.asyncio
async def test_the_username_is_released(client):
token, password = await _register(client, "recycled")
await client.request("DELETE", "/v1/users/me",
headers={"Authorization": f"Bearer {token}"},
json={"auth_key": _auth_key(password, "recycled")})
again = await client.post("/v1/users/register", json={
"username": "recycled", "email": "new@example.com",
"auth_key": _auth_key("another passphrase entirely", "recycled"),
})
assert again.status_code in (200, 201), "the name should be free again"
@pytest.mark.asyncio
async def test_owning_a_group_blocks_deletion(client):
"""
Deleting an account that owns groups would strand their members, so it is
refused with the list rather than cascading into other people's data.
"""
token, password = await _register(client, "owner")
headers = {"Authorization": f"Bearer {token}"}
r = await client.post("/v1/groups", json={"name": "orphans"}, headers=headers)
assert r.status_code in (200, 201), r.text
r = await client.request("DELETE", "/v1/users/me", headers=headers,
json={"auth_key": _auth_key(password, "owner")})
assert r.status_code == 409
assert "orphans" in r.json()["detail"]
@pytest.mark.asyncio
async def test_deletion_clears_memberships_notifications_and_tokens(
client, db_session):
token, password = await _register(client, "member1")
owner_token, _ = await _register(client, "grouper")
g = await client.post("/v1/groups", json={"name": "shared"},
headers={"Authorization": f"Bearer {owner_token}"})
gid = g.json()["group_id"]
await client.post(f"/v1/groups/{gid}/members/member1", json={},
headers={"Authorization": f"Bearer {owner_token}"})
uid = (await db_session.execute(
select(User.id).where(User.username == "member1"))).scalar_one()
await client.request("DELETE", "/v1/users/me",
headers={"Authorization": f"Bearer {token}"},
json={"auth_key": _auth_key(password, "member1")})
for model in (GroupMember, Notification, RefreshToken):
rows = (await db_session.execute(
select(model).where(model.user_id == uid))).scalars().all()
assert rows == [], f"{model.__name__} survived the deletion"
@pytest.mark.asyncio
async def test_the_ip_log_survives_and_stays_attributable(client, db_session):
"""
Kept on purpose. These rows exist for a year to answer legal requests, and
detaching them would keep the data while losing the only thing it is for.
"""
from meshbay_hub.db.models import IPLog
token, password = await _register(client, "logged")
uid = (await db_session.execute(
select(User.id).where(User.username == "logged"))).scalar_one()
before = (await db_session.execute(
select(IPLog).where(IPLog.user_id == uid))).scalars().all()
assert before, "registration should have been logged"
await client.request("DELETE", "/v1/users/me",
headers={"Authorization": f"Bearer {token}"},
json={"auth_key": _auth_key(password, "logged")})
after = (await db_session.execute(
select(IPLog).where(IPLog.user_id == uid))).scalars().all()
assert len(after) >= len(before), "the compliance log must survive deletion"
@pytest.mark.asyncio
async def test_a_deleted_account_cannot_keep_using_its_token(client):
"""
Refresh tokens are removed, but an access token lives up to an hour. The
status check refuses it straight away — a deleted account must not keep
reading groups until its token happens to expire.
"""
token, password = await _register(client, "gone")
headers = {"Authorization": f"Bearer {token}"}
r = await client.request("DELETE", "/v1/users/me", headers=headers,
json={"auth_key": _auth_key(password, "gone")})
assert r.status_code == 200
after = await client.get("/v1/groups/mine", headers=headers)
assert after.status_code in (401, 403), "the session outlived the account"
@pytest.mark.asyncio
async def test_only_an_admin_may_delete_someone_else(client, db_session):
token, _ = await _register(client, "ordinary")
victim_token, _ = await _register(client, "victim")
victim_id = (await db_session.execute(
select(User.id).where(User.username == "victim"))).scalar_one()
r = await client.delete(f"/v1/admin/users/{victim_id}",
headers={"Authorization": f"Bearer {token}"})
assert r.status_code in (401, 403), "a plain user must not delete accounts"
me = await client.get("/v1/users/me",
headers={"Authorization": f"Bearer {victim_token}"})
assert me.status_code == 200
@pytest.mark.asyncio
async def test_the_log_still_says_who_it_was(client, db_session):
"""
The point of keeping the log is being able to answer who did what. Taking
the name from a join meant the answer became "deleted-3f9a1c" the moment
anyone deleted their account — for exactly the records that get asked about.
"""
from meshbay_hub.db.models import IPLog
token, password = await _register(client, "traceable")
uid = (await db_session.execute(
select(User.id).where(User.username == "traceable"))).scalar_one()
await client.request("DELETE", "/v1/users/me",
headers={"Authorization": f"Bearer {token}"},
json={"auth_key": _auth_key(password, "traceable")})
rows = (await db_session.execute(
select(IPLog).where(IPLog.user_id == uid))).scalars().all()
assert rows, "registration should have been logged"
assert all(r.username == "traceable" for r in rows), \
"the log lost the name it exists to record"
admin_token, _ = await _register(client, "logreader")
from meshbay_hub.db.models import User as U
admin = (await db_session.execute(
select(U).where(U.username == "logreader"))).scalar_one()
admin.role = "admin"
await db_session.commit()
r = await client.get(f"/v1/admin/logs?user_id={uid}",
headers={"Authorization": f"Bearer {admin_token}"})
assert r.status_code == 200, r.text
names = {e["username"] for e in r.json()["logs"]}
assert names == {"traceable"}, f"admin view shows {names}"
|