""" 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}"