fix(hub): the admin allow-list grants an account, not a username

Each name in admin_usernames is pinned to the first active account seen
holding it (admin_pins), so a name freed by a deletion grants nothing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 121 insertions, 0 deletions

diff --git a/packages/meshbay-hub/tests/test_admin_pins.py b/packages/meshbay-hub/tests/test_admin_pins.py
new file mode 100644
index 0000000..7e7affc
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_admin_pins.py
@@ -0,0 +1,121 @@
+"""
+The `hub.toml` admin allow-list grants an account, not a username.
+
+Deleting an account releases its name, so a list of names granted the admin role
+to whoever registered a freed one next. Each listed name is now pinned to the
+first active account seen holding it, and only that account is the admin.
+"""
+
+import hashlib
+
+import pytest
+from meshbay_hub.api.deps import set_admin_usernames
+from meshbay_hub.app import _pin_config_admins
+from meshbay_hub.db.models import AdminPin, User
+from sqlalchemy import select
+
+
+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()
+
+
+PASSWORD = "a-long-enough-passphrase"
+
+
+async def _register(client, username, email=None):
+    r = await client.post("/v1/users/register", json={
+        "username": username, "email": email or 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)})
+    assert login.status_code == 200, login.text
+    return {"Authorization": f"Bearer {login.json()['access_token']}"}
+
+
+async def _delete_own(client, headers, username):
+    r = await client.request("DELETE", "/v1/users/me", headers=headers,
+                             json={"auth_key": _auth_key(PASSWORD, username)})
+    assert r.status_code == 200, r.text
+
+
+async def _is_admin(client, headers) -> bool:
+    r = await client.get("/v1/admin/stats", headers=headers)
+    assert r.status_code in (200, 403), r.text
+    return r.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_a_released_name_registered_again_is_not_an_admin(client):
+    set_admin_usernames(["the_operator"])
+    first = await _register(client, "the_operator")
+    assert await _is_admin(client, first)
+
+    await _delete_own(client, first, "the_operator")
+    second = await _register(client, "the_operator", email="someone@example.com")
+    assert not await _is_admin(client, second)
+
+
+@pytest.mark.asyncio
+async def test_nor_after_a_restart(client, db_session):
+    """Startup must not pin the name again to whoever holds it now."""
+    set_admin_usernames(["the_operator"])
+    first = await _register(client, "the_operator")
+    assert await _is_admin(client, first)
+    await _delete_own(client, first, "the_operator")
+    second = await _register(client, "the_operator", email="someone@example.com")
+
+    await _pin_config_admins(["the_operator"])
+
+    assert not await _is_admin(client, second)
+    impostor = (await db_session.execute(
+        select(User).where(User.username == "the_operator"))).scalar_one()
+    assert impostor.role == "user"
+
+
+@pytest.mark.asyncio
+async def test_startup_pins_an_existing_account_before_its_name_is_freed(client, db_session):
+    """The upgrade path: the listed account exists before any pin does."""
+    first = await _register(client, "the_operator")
+    set_admin_usernames(["the_operator"])
+    await _pin_config_admins(["the_operator"])
+
+    pin = await db_session.get(AdminPin, "the_operator")
+    owner = (await db_session.execute(
+        select(User).where(User.username == "the_operator"))).scalar_one()
+    assert pin is not None and pin.user_id == owner.id
+    assert owner.role == "admin"
+
+    await _delete_own(client, first, "the_operator")
+    second = await _register(client, "the_operator", email="someone@example.com")
+    assert not await _is_admin(client, second)
+
+
+@pytest.mark.asyncio
+async def test_a_name_registered_while_the_hub_runs_is_admin_at_once(client):
+    """No restart between listing a name and its first sign-in, as before pins."""
+    set_admin_usernames(["late_operator"])
+    await _pin_config_admins(["late_operator"])
+    headers = await _register(client, "late_operator")
+    assert await _is_admin(client, headers)
+
+
+@pytest.mark.asyncio
+async def test_unlisting_then_relisting_hands_the_name_to_its_new_holder(client, db_session):
+    set_admin_usernames(["the_operator"])
+    first = await _register(client, "the_operator")
+    assert await _is_admin(client, first)
+    await _delete_own(client, first, "the_operator")
+    second = await _register(client, "the_operator", email="successor@example.com")
+
+    set_admin_usernames([])
+    await _pin_config_admins([])
+    assert await db_session.get(AdminPin, "the_operator") is None
+
+    set_admin_usernames(["the_operator"])
+    await _pin_config_admins(["the_operator"])
+    assert await _is_admin(client, second)