From 3e0993fcebe6fcd399afbbbb9feee748e713d981 Mon Sep 17 00:00:00 2001 From: Cr0c0 Date: Sat, 25 Jul 2026 11:51:16 +0200 Subject: [PATCH] feat(api): Add the admin only endpoint to update the user role --- src/app/api/api_v1/endpoints/users.py | 29 +++++++++++++++++- src/app/crud/crud_user.py | 4 +-- src/app/schemas/users.py | 10 ++++++- src/tests/endpoints/test_users.py | 43 +++++++++++++++++++++++++++ 4 files changed, 82 insertions(+), 4 deletions(-) diff --git a/src/app/api/api_v1/endpoints/users.py b/src/app/api/api_v1/endpoints/users.py index 5d2042d93..e61ac7ade 100644 --- a/src/app/api/api_v1/endpoints/users.py +++ b/src/app/api/api_v1/endpoints/users.py @@ -12,7 +12,7 @@ from app.crud import UserCRUD from app.models import User, UserRole from app.schemas.login import TokenPayload -from app.schemas.users import Cred, CredHash, UserCreate +from app.schemas.users import Cred, CredHash, RoleUpdate, UserCreate from app.services.telemetry import telemetry_client router = APIRouter() @@ -85,6 +85,33 @@ async def update_user_password( return await users.update(user_id, CredHash(hashed_password=pwd)) +@router.patch("/{user_id}/role", status_code=status.HTTP_200_OK, summary="Updates a user's role") +async def update_user_role( + payload: RoleUpdate, + user_id: int = Path(..., gt=0), + users: UserCRUD = Depends(get_user_crud), + token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN]), +) -> User: + """Promote or demote a user between the `agent` role and the `user` role. + Admins are out of scope: neither the requester nor the target user can have their admin role changed here. + + Beware that the role is baked into the access tokens that were already issued, and those are long-lived + (see `JWT_EXPIRE_MINUTES`). The new role only applies to tokens minted afterwards, so the user has to log in + again for the change to take effect. + """ + telemetry_client.capture( + token_payload.sub, event="user-role", properties={"user_id": user_id, "role": payload.role} + ) + if user_id == token_payload.sub: + raise HTTPException(status.HTTP_403_FORBIDDEN, "Admins cannot change their own role : it can lead to deadlock") + + user = cast(User, await users.get(user_id, strict=True)) + if user.role == UserRole.ADMIN: + raise HTTPException(status.HTTP_403_FORBIDDEN, "Cannot change an admin's role") + + return await users.update(user_id, payload) + + @router.delete("/{user_id}", status_code=status.HTTP_200_OK, summary="Delete a user") async def delete_user( user_id: int = Path(..., gt=0), diff --git a/src/app/crud/crud_user.py b/src/app/crud/crud_user.py index 099db4192..19d0bc61b 100644 --- a/src/app/crud/crud_user.py +++ b/src/app/crud/crud_user.py @@ -9,12 +9,12 @@ from app.crud.base import BaseCRUD from app.models import User -from app.schemas.users import CredHash +from app.schemas.users import CredHash, RoleUpdate __all__ = ["UserCRUD"] -class UserCRUD(BaseCRUD[User, User, CredHash]): +class UserCRUD(BaseCRUD[User, User, Union[CredHash, RoleUpdate]]): def __init__(self, session: AsyncSession) -> None: super().__init__(session, User) diff --git a/src/app/schemas/users.py b/src/app/schemas/users.py index 0d9db69f9..1e7987a86 100644 --- a/src/app/schemas/users.py +++ b/src/app/schemas/users.py @@ -3,11 +3,13 @@ # This program is licensed under the Apache License 2.0. # See LICENSE or go to for full license details. +from typing import Literal + from pydantic import BaseModel, Field from app.models import UserRole -__all__ = ["Cred", "CredHash", "UserCreate", "UserCreation"] +__all__ = ["Cred", "CredHash", "RoleUpdate", "UserCreate", "UserCreation"] # Accesses @@ -27,6 +29,12 @@ class Role(BaseModel): role: UserRole = Field(UserRole.USER) +class RoleUpdate(BaseModel): + """Admin is excluded : if an admin retrograde himself that can lead to deadlock""" + + role: Literal[UserRole.AGENT, UserRole.USER] = Field(..., examples=["agent"]) + + class UserCreate(Role): login: str = Field(..., min_length=3, max_length=50, examples=["JohnDoe"]) password: str = Field(..., min_length=3, examples=["PickARobustOne"]) diff --git a/src/tests/endpoints/test_users.py b/src/tests/endpoints/test_users.py index 9a8b18eb3..04ee1b9a1 100644 --- a/src/tests/endpoints/test_users.py +++ b/src/tests/endpoints/test_users.py @@ -174,6 +174,49 @@ async def test_delete_user( assert response.json() is None +@pytest.mark.parametrize( + ("user_idx", "user_id", "payload", "status_code", "status_detail"), + [ + (None, 2, {"role": "user"}, 401, "Not authenticated"), + (1, 3, {"role": "agent"}, 403, "Incompatible token scope."), + (2, 2, {"role": "user"}, 403, "Incompatible token scope."), + (0, 0, {"role": "user"}, 422, None), + (0, 2, {"role": "admin"}, 422, None), + (0, 2, {}, 422, None), + (0, 400, {"role": "user"}, 404, "Table User has no corresponding entry."), + (0, 1, {"role": "user"}, 403, "Admins cannot change their own role : it can lead to deadlock"), + (0, 2, {"role": "user"}, 200, None), + (0, 3, {"role": "agent"}, 200, None), + (0, 3, {"role": "user"}, 200, None), # setting role to the actual role + ], +) +@pytest.mark.asyncio +async def test_update_user_role( + async_client: AsyncClient, + user_session: AsyncSession, + user_idx: Union[int, None], + user_id: int, + payload: Dict[str, Any], + status_code: int, + status_detail: Union[str, None], +): + auth = None + if isinstance(user_idx, int): + auth = pytest.get_token( + pytest.user_table[user_idx]["id"], + pytest.user_table[user_idx]["role"].split(), + pytest.user_table[user_idx]["organization_id"], + ) + + response = await async_client.patch(f"/users/{user_id}/role", json=payload, headers=auth) + assert response.status_code == status_code, print(response.__dict__) + if isinstance(status_detail, str): + assert response.json()["detail"] == status_detail + if response.status_code // 100 == 2: + expected = next(entry for entry in pytest.user_table if entry["id"] == user_id) + assert response.json() == {**expected, "role": payload["role"]} + + @pytest.mark.parametrize( ("user_idx", "user_id", "payload", "status_code", "status_detail", "expected_idx"), [