fix(hub): an administrator can erase an account that owns groups

An administrator's deletion answered 409 for any account owning a group,
so an erasure ordered by an authority had to wait on the person it was
about. It now deletes the account's groups with it, then pushes a signed
revocation for the account and for each group to every connected node:
an access token already issued stays valid on a node until it expires,
and the revocation is what makes the nodes refuse the account and close
the groups' sessions now. The action is written to the IP log, and the
confirmation dialog says the groups go too, in all ten catalogues.

The owner's own deletion is unchanged: refused while they own groups,
which they can hand over first (CGU 3.4, privacy statement).

Deleting a group had three partial cascades. The owner's route left
email_verifications behind, and the cleanup of unhosted groups left
notifications, invitations and reports - each an IntegrityError on
PostgreSQL, invisible on SQLite, which does not enforce foreign keys by
default. db/purge.py is now the one implementation: it finds every table
referencing groups.id from the schema, deletes the group's rows and
detaches content reports, which are evidence and outlive the group.

test_group_purge.py turns foreign-key enforcement on for its connection,
seeds every referencing table, and fails without the fix on all three
routes. MESHBAY_DESIGN.md 7.7 states the rule, and now lists the device
keys and swarm sources that e3c68b3 erases.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01D9MCBBWSm9GhBESmqzJxNy

1 files changed, 184 insertions, 0 deletions

diff --git a/packages/meshbay-hub/tests/test_group_purge.py b/packages/meshbay-hub/tests/test_group_purge.py
new file mode 100644
index 0000000..e05c374
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_group_purge.py
@@ -0,0 +1,184 @@
+"""
+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 sqlalchemy import delete, func, select, text
+from sqlalchemy.exc import IntegrityError
+
+from meshbay_hub.db.models import (
+    ContentReport, EmailVerification, Group, GroupMember, IPLog, Notification, User,
+)
+from meshbay_hub.db.purge import _referencing
+
+SEEDED = {"group_members", "notifications", "email_verifications", "content_reports"}
+
+
+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"))
+    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")
+    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")
+    await _account(client, "member")
+    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", "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")
+
+    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")
+    await _account(client, "guest")
+    gid = await _group_with_everything(client, db_session, owner_token, "guest", "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)