fix(users): roles/is_admin null -> 422, admin non si toglie da solo (#36 punto 4)

roles e is_admin sono NOT NULL nel DB ma Optional nello schema: {"roles":
null} o {"is_admin": null} passava la validazione Pydantic e falliva al
flush con un 500 non gestito. Ora 422 esplicito. Aggiunta anche la
protezione sull'ultimo admin: un admin non puo' togliersi is_admin da solo
(409) - unico caso raggiungibile via API dato che solo un admin puo'
chiamare l'endpoint, quindi rimuovere il flag a qualcun altro lascia sempre
il chiamante; il controllo sul conteggio resta comunque come difesa in
profondita'.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-21 22:45:43 +02:00
parent 8864b2021b
commit 3e99b64fc4
2 changed files with 92 additions and 1 deletions
+37 -1
View File
@@ -1,6 +1,6 @@
"""Users router - CRUD operations (admin only).""" """Users router - CRUD operations (admin only)."""
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from src.backend.database import get_db from src.backend.database import get_db
@@ -78,6 +78,21 @@ async def update_user(
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
update_data = data.model_dump(exclude_unset=True) update_data = data.model_dump(exclude_unset=True)
# roles and is_admin are NOT NULL columns: an explicit null passed the
# Optional[...] schema check but blew up at flush time as an unhandled
# 500 (#36 point 4).
if "roles" in update_data and update_data["roles"] is None:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="roles cannot be null",
)
if "is_admin" in update_data and update_data["is_admin"] is None:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="is_admin cannot be null",
)
# Validate roles if provided # Validate roles if provided
if "roles" in update_data: if "roles" in update_data:
for role in update_data["roles"]: for role in update_data["roles"]:
@@ -86,6 +101,27 @@ async def update_user(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Invalid role '{role}'. Valid roles: {VALID_ROLES}", detail=f"Invalid role '{role}'. Valid roles: {VALID_ROLES}",
) )
# Losing the last admin locks everyone out of user management; losing
# your own admin flag while acting as admin is the same mistake aimed
# at yourself. Neither is reversible without direct DB access.
if update_data.get("is_admin") is False and user.is_admin:
if user.id == admin.id:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Cannot remove your own admin privilege",
)
other_admins = await db.execute(
select(func.count()).select_from(User).where(
User.is_admin == True, User.id != user.id # noqa: E712
)
)
if other_admins.scalar_one() == 0:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Cannot remove admin privilege from the last admin",
)
for field, value in update_data.items(): for field, value in update_data.items():
setattr(user, field, value) setattr(user, field, value)
await db.flush() await db.flush()
+55
View File
@@ -107,6 +107,61 @@ class TestUpdateUser:
assert resp.json()["display_name"] == "Updated Maker" assert resp.json()["display_name"] == "Updated Maker"
class TestUpdateUserValidation:
"""gitea#36 point 4: roles/is_admin cannot be null, admin self/last-admin guard."""
async def test_roles_null_is_422(
self, client: AsyncClient, admin_user: User, maker_user: User,
):
"""roles is Optional in the schema but NOT NULL in the DB: an explicit
null used to reach setattr() and blow up as an unhandled 500."""
resp = await client.put(
f"/api/users/{maker_user.id}",
headers=auth_headers(admin_user),
json={"roles": None},
)
assert resp.status_code == 422
async def test_is_admin_null_is_422(
self, client: AsyncClient, admin_user: User, maker_user: User,
):
resp = await client.put(
f"/api/users/{maker_user.id}",
headers=auth_headers(admin_user),
json={"is_admin": None},
)
assert resp.status_code == 422
async def test_admin_cannot_remove_own_admin_flag(
self, client: AsyncClient, admin_user: User,
):
"""admin_user is the only admin: this is also the only reachable
"last admin" case, since only an admin can call this endpoint at
all — removing someone else's flag always leaves the caller."""
resp = await client.put(
f"/api/users/{admin_user.id}",
headers=auth_headers(admin_user),
json={"is_admin": False},
)
assert resp.status_code == 409
async def test_admin_can_demote_another_admin_when_not_last(
self, client: AsyncClient, admin_user: User, db_session: AsyncSession,
):
"""Demoting someone else is fine as long as an admin remains (the
caller): the guard must not be broader than the two stated cases."""
second_admin = await _create_user(
db_session, username="second_admin", roles=[], is_admin=True,
)
resp = await client.put(
f"/api/users/{second_admin.id}",
headers=auth_headers(admin_user),
json={"is_admin": False},
)
assert resp.status_code == 200
assert resp.json()["is_admin"] is False
class TestDeleteUser: class TestDeleteUser:
"""DELETE /api/users/{user_id} tests.""" """DELETE /api/users/{user_id} tests."""