fix(auth): only block bootstrap login when admin users exist

The previous check counted ALL users (func.count(User.id)), which
incorrectly blocked bootstrap login if any non-admin user (e.g.,
Google OAuth sign-in) already existed. The fix queries specifically
for admin users using the legacy is_admin flag OR the admin role
assignment in user_roles.

Also add tests for the non-admin-user-only scenario and fix the
stale test_non_admin_user_cannot_login_as_admin assertion (401
not 403 — bootstrap path is still active when no admin exists).
This commit is contained in:
Hermes Agent
2026-07-30 09:44:42 +00:00
parent f29640eb47
commit ffbc19f16f
2 changed files with 83 additions and 8 deletions
+20 -6
View File
@@ -2,7 +2,7 @@
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel from pydantic import BaseModel
from sqlalchemy import select, func from sqlalchemy import select
from sqlalchemy.orm import joinedload from sqlalchemy.orm import joinedload
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
@@ -62,14 +62,28 @@ async def login(payload: LoginRequest):
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Bootstrap admin not configured") raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Bootstrap admin not configured")
from app.database import async_session from app.database import async_session
from app.user_models import Role from app.user_models import Role, user_roles
# Check if any admins already exist in the DB # Check if any admin users already exist in the DB
async with async_session() as session: async with async_session() as session:
admin_count_result = await session.execute(select(func.count(User.id))) # Check for admin users: legacy is_admin flag OR admin role assignment
admin_count = admin_count_result.scalar() admin_role_subq = (
select(1)
.select_from(user_roles)
.join(Role, user_roles.c.role_id == Role.id)
.where(user_roles.c.user_id == User.id)
.where(Role.name == "admin")
.exists()
)
if admin_count > 0: result = await session.execute(
select(1).where(
(User.is_admin == True) | admin_role_subq
).limit(1)
)
has_admins = result.scalar() is not None
if has_admins:
# Admins exist — bootstrap creds are locked. Try normal password login. # Admins exist — bootstrap creds are locked. Try normal password login.
if (payload.username == settings.ADMIN_USERNAME and if (payload.username == settings.ADMIN_USERNAME and
payload.password == settings.ADMIN_PASSWORD): payload.password == settings.ADMIN_PASSWORD):
+63 -2
View File
@@ -115,6 +115,21 @@ async def db_with_non_admin_user():
yield yield
@pytest.fixture
async def db_with_google_user_only():
"""Seed the DB with a non-admin Google OAuth user (no admin users)."""
async with async_session() as session:
user = User(
email="john@gmail.com",
display_name="John",
auth_provider="google",
is_admin=False,
)
session.add(user)
await session.commit()
yield
class TestBootstrapLoginNoAdmins: class TestBootstrapLoginNoAdmins:
"""Scenario 1: Login with hardcoded creds when NO admins exist → PASS.""" """Scenario 1: Login with hardcoded creds when NO admins exist → PASS."""
@@ -293,8 +308,8 @@ class TestEdgeCases:
"/api/auth/login", "/api/auth/login",
json={"username": "regular", "password": "regularpass"}, json={"username": "regular", "password": "regularpass"},
) )
# Non-admin users should be rejected # Non-admin users should be rejected — bootstrap path is active since no admin exists
assert resp.status_code == 403 assert resp.status_code == 401
async def test_user_lookup_by_display_name(self, db_with_admin_different_password): async def test_user_lookup_by_display_name(self, db_with_admin_different_password):
"""Login should work when username matches display_name.""" """Login should work when username matches display_name."""
@@ -353,6 +368,52 @@ class TestEdgeCases:
assert verify_password("testpass123", user.password_hash) assert verify_password("testpass123", user.password_hash)
class TestBootstrapWithNonAdminUsers:
"""Bootstrap login should still work when only non-admin users exist."""
async def test_bootstrap_login_works_when_google_users_exist(self, db_with_google_user_only):
"""Bootstrap credentials should work even if non-admin Google users exist."""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
resp = await client.post(
"/api/auth/login",
json={"username": "testadmin", "password": "testpass123"},
)
assert resp.status_code == 200
data = resp.json()
assert "access_token" in data
async def test_bootstrap_creates_admin_when_google_users_exist(self, db_with_google_user_only):
"""After bootstrap login with existing non-admin users, the admin user should exist."""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await client.post(
"/api/auth/login",
json={"username": "testadmin", "password": "testpass123"},
)
from sqlalchemy import select as sa_select
async with async_session() as session:
# Admin user should have been created
result = await session.execute(
sa_select(User).where(
User.email == "testadmin@local",
User.auth_provider == "local",
)
)
user = result.scalar_one_or_none()
assert user is not None
assert user.is_admin is True
# The original Google user should still exist and not be admin
result = await session.execute(
sa_select(User).where(User.email == "john@gmail.com")
)
google_user = result.scalar_one_or_none()
assert google_user is not None
assert google_user.is_admin is False
class TestPasswordHashing: class TestPasswordHashing:
"""Unit tests for password hashing utilities.""" """Unit tests for password hashing utilities."""