From 8a4651e9d223de856ff085b329801998f95db138 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 30 Sep 2026 12:37:31 +0200 Subject: 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 --- packages/meshbay-hub/tests/test_admin_pins.py | 121 ++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 packages/meshbay-hub/tests/test_admin_pins.py (limited to 'packages/meshbay-hub/tests') 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) -- cgit v1.2.3