fix(auth): revoke API tokens when an admin is demoted (#6831) (#6832)

core/auth.py

@@ -463,6 +463,17 @@ class AuthManager:
                 admin_count = sum(1 for d in self.users.values() if d.get("is_admin"))
                 if admin_count <= 1:
                     return SetAdminResult.LAST_ADMIN
+                # Token creation is admin-only and a demoted user can neither
+                # list nor revoke their tokens, so purge them with the role.
+                # Done before the flag flips: if the token store fails the
+                # exception propagates and the user is still an admin.
+                from core.database import get_db_session, ApiToken
+                with get_db_session() as db:
+                    removed_tokens = db.query(ApiToken).filter(ApiToken.owner == username).delete()
+                if removed_tokens:
+                    logger.info(
+                        f"Revoked {removed_tokens} API token(s) owned by demoted admin '{username}'"
+                    )
             # Write order matters for lock-free readers: get_privileges()
             # reads without _config_lock and trusts is_admin, so the admin
             # flag must be flipped while the stored map is safe to expose —

routes/auth_routes.py

@@ -668,7 +668,18 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter:
         user = _get_current_user(request)
         if not user or not auth_manager.is_admin(user):
             raise HTTPException(403, "Admin only")
+        try:
             result = auth_manager.set_admin(username, body.is_admin, user)
+        finally:
+            # A demotion purges the user's ApiToken rows; dirty the bearer
+            # cache so already-cached tokens stop authenticating.
+            if not body.is_admin:
+                invalidator = getattr(
+                    getattr(getattr(request, "app", None), "state", None),
+                    "invalidate_token_cache", None,
+                )
+                if callable(invalidator):
+                    invalidator()
         if result is SetAdminResult.USER_NOT_FOUND:
             raise HTTPException(404, "User not found")
         if result is SetAdminResult.NOT_AUTHORIZED:

tests/test_demote_admin_revokes_api_tokens.py

@@ -0,0 +1,82 @@
+"""Demoting an admin purges their API tokens (issue #6831).
+
+Token management is admin-only and scoped to the token's owner, so once the
+owner is demoted nobody can list or revoke their tokens any more.
+"""
+
+import contextlib
+import sys
+import types
+
+import pytest
+
+from tests.test_delete_user_revokes_api_tokens import (
+    _auth_module,
+    _FakeApiToken,
+    _FakeSession,
+)
+
+
+@pytest.fixture
+def manager(tmp_path, monkeypatch):
+    auth_mod = _auth_module()
+    monkeypatch.setattr(auth_mod, "_hash_password", lambda password: f"hash:{password}")
+    monkeypatch.setattr(
+        auth_mod, "_verify_password", lambda password, hashed: hashed == f"hash:{password}"
+    )
+    mgr = auth_mod.AuthManager(str(tmp_path / "auth.json"))
+    assert mgr.create_user("admin", "secret-admin-pw", is_admin=True)
+    assert mgr.create_user("adm2", "secret-adm2-pw", is_admin=True)
+    assert mgr.create_user("bob", "secret-bob-pw", is_admin=False)
+    return mgr
+
+
+@pytest.fixture
+def db_calls(monkeypatch):
+    calls = []
+
+    @contextlib.contextmanager
+    def _fake_db_session():
+        yield _FakeSession(calls)
+
+    db_stub = types.ModuleType("core.database")
+    db_stub.get_db_session = _fake_db_session
+    db_stub.ApiToken = _FakeApiToken
+    monkeypatch.setitem(sys.modules, "core.database", db_stub)
+    return calls
+
+
+def test_demote_revokes_api_tokens(manager, db_calls):
+    result = manager.set_admin("adm2", False, "admin")
+
+    assert result.value == "ok"
+    assert manager.is_admin("adm2") is False
+    assert [("owner ==", "adm2")] in db_calls
+
+
+def test_promote_leaves_api_tokens_alone(manager, db_calls):
+    manager.set_admin("bob", True, "admin")
+
+    assert db_calls == []
+
+
+def test_refused_demotion_leaves_api_tokens_alone(manager, db_calls):
+    assert manager.set_admin("adm2", False, "bob").value == "not_authorized"
+    assert db_calls == []
+
+
+def test_demote_fails_closed_when_token_purge_fails(manager, monkeypatch):
+    @contextlib.contextmanager
+    def _failing_db_session():
+        raise RuntimeError("database unavailable")
+        yield
+
+    db_stub = types.ModuleType("core.database")
+    db_stub.get_db_session = _failing_db_session
+    db_stub.ApiToken = _FakeApiToken
+    monkeypatch.setitem(sys.modules, "core.database", db_stub)
+
+    with pytest.raises(RuntimeError):
+        manager.set_admin("adm2", False, "admin")
+
+    assert manager.is_admin("adm2") is True