""" Deleting a group, and deleting the account that owns it, with foreign keys on. PostgreSQL enforces foreign keys and SQLite does not unless asked, so a cascade that left one row pointing at a deleted group passed every test here and would fail in production with an IntegrityError. These tests turn enforcement on for their connection (the in-memory engine is a single shared one) and fill every table that references `groups.id` before deleting — the list of tables comes from the schema, so one added later is covered, and the seeding below has to be extended or `test_the_seeding_covers_every_reference` says so. An administrator can erase an account that owns groups; its groups go with it, and the connected nodes receive a signed revocation for the account and for each group. The owner's own deletion still asks them to hand the groups over first (test_account_deletion.py). """ import base64 import hashlib from datetime import UTC, datetime, timedelta import jwt import pytest from meshbay_hub.db.models import ( ContentReport, EmailVerification, Group, GroupInviteLink, GroupMember, IPLog, Notification, User, ) from meshbay_hub.db.purge import _referencing from sqlalchemy import delete, func, select, text from sqlalchemy.exc import IntegrityError SEEDED = {"group_members", "notifications", "email_verifications", "content_reports", "group_invite_links"} 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 _account(client, username: str) -> str: 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 r = await client.post("/v1/users/login", json={ "username": username, "auth_key": _auth_key(password, username)}) return r.json()["access_token"] async def _uid(db, username: str) -> str: return (await db.execute(select(User.id).where(User.username == username))).scalar_one() async def _enforce_foreign_keys(db): await db.execute(text("PRAGMA foreign_keys=ON")) await db.commit() assert (await db.execute(text("PRAGMA foreign_keys"))).scalar() == 1 async def _group_with_everything(client, db, owner_token: str, member: str, name: str) -> str: """A group, a member, and a row in every table that points at groups.""" r = await client.post("/v1/groups", json={"name": name}, headers={"Authorization": f"Bearer {owner_token}"}) assert r.status_code in (200, 201), r.text gid = r.json()["group_id"] r = await client.post(f"/v1/groups/{gid}/members/{member}", json={}, headers={"Authorization": f"Bearer {owner_token}"}) assert r.status_code in (200, 201), r.text member_id = await _uid(db, member) db.add(Notification(user_id=member_id, kind="chat", group_id=gid, title="a message")) db.add(EmailVerification(email_hash="0" * 64, code="123456", purpose="invitation", group_id=gid, expires_at=datetime.now(UTC) + timedelta(days=1))) db.add(ContentReport(reporter_id=member_id, content_hash="ab" * 32, group_id=gid, ip_address="192.0.2.1")) owner_id = (await db.execute(select(Group.admin_id).where(Group.id == gid))).scalar_one() db.add(GroupInviteLink(group_id=gid, created_by=owner_id, ticket_hash=gid[:8] * 8, email_masked="m***@e***.com", expires_at=datetime.now(UTC) + timedelta(days=1), redeemed_by=member_id, redeemed_at=datetime.now(UTC))) await db.commit() return gid async def _nothing_points_at(db, gid: str) -> None: db.expire_all() assert await db.get(Group, gid) is None, "the group survived" for table, column in _referencing(): n = (await db.execute( select(func.count()).select_from(table).where(column == gid))).scalar() assert n == 0, f"{table.name} still has {n} row(s) pointing at the deleted group" reports = (await db.execute(select(ContentReport))).scalars().all() assert reports and all(r.group_id is None for r in reports), \ "a report is evidence about content and must outlive the group, detached" def test_the_seeding_covers_every_reference(): """If a table gains a foreign key to groups, the tests below must seed it, or they stop proving the deletion survives enforcement.""" assert {t.name for t, _ in _referencing()} == SEEDED @pytest.mark.asyncio async def test_enforcement_is_really_on(client, db_session): """The control: without it, every test below would pass on SQLite's default and prove nothing about PostgreSQL.""" owner = await _account(client, "control_test") await _account(client, "control_member") gid = await _group_with_everything(client, db_session, owner, "control_member", "control") await _enforce_foreign_keys(db_session) with pytest.raises(IntegrityError): await db_session.execute(delete(Group).where(Group.id == gid)) await db_session.commit() await db_session.rollback() @pytest.mark.asyncio async def test_an_admin_erases_an_account_that_owns_groups(client, db_session, monkeypatch): sent = [] async def capture(token: str) -> int: sent.append(jwt.decode(token, options={"verify_signature": False})) return 0 monkeypatch.setattr("meshbay_hub.api.revocation.broadcast_revocation", capture) admin_token = await _account(client, "the_admin") owner_token = await _account(client, "owner_test") await _account(client, "member_test") admin = await db_session.get(User, await _uid(db_session, "the_admin")) admin.role = "admin" await db_session.commit() first = await _group_with_everything(client, db_session, owner_token, "member_test", "first") r = await client.post("/v1/groups", json={"name": "second"}, headers={"Authorization": f"Bearer {owner_token}"}) second = r.json()["group_id"] owner_id = await _uid(db_session, "owner_test") await _enforce_foreign_keys(db_session) r = await client.delete(f"/v1/admin/users/{owner_id}", headers={"Authorization": f"Bearer {admin_token}"}) assert r.status_code == 200, r.text assert {g["name"] for g in r.json()["groups_deleted"]} == {"first", "second"} await _nothing_points_at(db_session, first) assert await db_session.get(Group, second) is None owner = await db_session.get(User, owner_id) assert owner.status == "deleted" and owner.email == "" targets = {(p["target"], p["target_id"]) for p in sent} assert targets == {("user", owner_id), ("group", first), ("group", second)}, ( "the nodes must be told to refuse the account and close its groups") logged = (await db_session.execute( select(IPLog).where(IPLog.event == "admin_user_delete"))).scalars().all() assert len(logged) == 1 and "first" in logged[0].detail and "owner" in logged[0].detail @pytest.mark.asyncio async def test_the_owner_deletes_a_group_with_everything_pointing_at_it(client, db_session): """The owner's route left `email_verifications` behind — an IntegrityError on PostgreSQL the first time an invited group was deleted.""" owner_token = await _account(client, "keeper_test") await _account(client, "guest_test") gid = await _group_with_everything(client, db_session, owner_token, "guest_test", "doomed") await _enforce_foreign_keys(db_session) r = await client.delete(f"/v1/groups/{gid}", headers={"Authorization": f"Bearer {owner_token}"}) assert r.status_code == 200, r.text await _nothing_points_at(db_session, gid) assert (await db_session.execute( select(GroupMember).where(GroupMember.group_id == gid))).first() is None @pytest.mark.asyncio async def test_the_cleanup_of_unhosted_groups_removes_everything_too(client, db_session): """Its own cascade deleted memberships only.""" from meshbay_hub.tasks.cleanup import prune_unhosted_groups owner_token = await _account(client, "abandoner") await _account(client, "bystander") gid = await _group_with_everything(client, db_session, owner_token, "bystander", "never hosted") await _enforce_foreign_keys(db_session) pruned = await prune_unhosted_groups(db_session, grace_days=0) assert gid in {g for g, _ in pruned} await _nothing_points_at(db_session, gid)