diff --git a/src/app/api/api_v1/endpoints/users.py b/src/app/api/api_v1/endpoints/users.py index 5d2042d9..d424d546 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 last a year + (`JWT_UNLIMITED`, see `login_with_creds`). The new role only applies to tokens minted afterwards, so the + user has to log in again for the change to take effect. + """ + if user_id == token_payload.sub: + raise HTTPException(status.HTTP_403_FORBIDDEN, "Admins cannot change their own role: it would lock them out") + + 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") + + telemetry_client.capture( + token_payload.sub, event="user-role", properties={"user_id": user_id, "role": payload.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 099db419..71c11628 100644 --- a/src/app/crud/crud_user.py +++ b/src/app/crud/crud_user.py @@ -5,16 +5,16 @@ from typing import Union +from pydantic import BaseModel from sqlmodel.ext.asyncio.session import AsyncSession from app.crud.base import BaseCRUD from app.models import User -from app.schemas.users import CredHash __all__ = ["UserCRUD"] -class UserCRUD(BaseCRUD[User, User, CredHash]): +class UserCRUD(BaseCRUD[User, User, BaseModel]): 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 0d9db69f..8abc7f90 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 intentionally excluded: an admin demoting themselves could lock everyone out.""" + + 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 9a8b18eb..b257057a 100644 --- a/src/tests/endpoints/test_users.py +++ b/src/tests/endpoints/test_users.py @@ -4,6 +4,8 @@ from httpx import AsyncClient from sqlmodel.ext.asyncio.session import AsyncSession +from app.models import User, UserRole + @pytest.mark.parametrize( ("user_idx", "payload", "status_code", "status_detail"), @@ -174,6 +176,95 @@ 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 would lock them out"), + (0, 2, {"role": "camera"}, 422, None), + (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.asyncio +async def test_update_user_role_camera_token(async_client: AsyncClient, user_session: AsyncSession): + auth = pytest.get_token(1, ["camera"], 1) + response = await async_client.patch("/users/2/role", json={"role": "user"}, headers=auth) + assert response.status_code == 403, print(response.__dict__) + assert response.json()["detail"] == "Incompatible token scope." + + +@pytest.mark.asyncio +async def test_update_user_role_admin_target(async_client: AsyncClient, user_session: AsyncSession): + # It promote the user 2 to admin in the DB : then check that the endpoint refuses to touch it + db_user = await user_session.get(User, 2) + db_user.role = UserRole.ADMIN + user_session.add(db_user) + await user_session.commit() + + auth = pytest.get_token(1, ["admin"], 1) + response = await async_client.patch("/users/2/role", json={"role": "user"}, headers=auth) + assert response.status_code == 403, print(response.__dict__) + assert response.json()["detail"] == "Cannot change an admin's role" + + +@pytest.mark.asyncio +async def test_update_user_role_takes_effect_on_next_login(async_client: AsyncClient, user_session: AsyncSession): + auth = pytest.get_token(1, ["admin"], 1) + + async def login_scopes(login: str, password: str) -> list: + creds = await async_client.post("/login/creds", data={"username": login, "password": password}) + token = creds.json()["access_token"] + validate = await async_client.get("/login/validate", headers={"Authorization": f"Bearer {token}"}) + return validate.json()["scopes"] + + assert await login_scopes("third_login", "third_pwd") == ["user"] + + # promote and then check that the patch is persistent + patch = await async_client.patch("/users/3/role", json={"role": "agent"}, headers=auth) + assert patch.status_code == 200, print(patch.__dict__) + assert patch.json()["role"] == "agent" + + get = await async_client.get("/users/3", headers=auth) + assert get.json()["role"] == "agent" + + assert await login_scopes("third_login", "third_pwd") == ["agent"] + + @pytest.mark.parametrize( ("user_idx", "user_id", "payload", "status_code", "status_detail", "expected_idx"), [