From 32040377975b2e285a1172db206f2fc86868d653 Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Sun, 4 Oct 2026 10:18:15 -0400 Subject: [PATCH] fix(auth): validate reset passwords and make the last-admin guard atomic, with tests (#5106) --- backend/endpoints/auth.py | 11 + backend/endpoints/user.py | 44 ++-- backend/exceptions/database_exceptions.py | 9 + backend/handler/auth/base_handler.py | 21 +- backend/handler/database/users_handler.py | 29 +++ .../tests/endpoints/sockets/test_activity.py | 221 ++++++++++++++++ backend/tests/endpoints/test_auth.py | 212 ++++++++++++++++ backend/tests/endpoints/test_user.py | 240 ++++++++++++++++++ .../handler/database/test_users_handler.py | 102 ++++++++ 9 files changed, 860 insertions(+), 29 deletions(-) create mode 100644 backend/tests/endpoints/sockets/test_activity.py create mode 100644 backend/tests/endpoints/test_auth.py create mode 100644 backend/tests/endpoints/test_user.py create mode 100644 backend/tests/handler/database/test_users_handler.py diff --git a/backend/endpoints/auth.py b/backend/endpoints/auth.py index 227790efa..c7b0b75a6 100644 --- a/backend/endpoints/auth.py +++ b/backend/endpoints/auth.py @@ -43,6 +43,7 @@ from models.audit_event import AuditAction from models.user import User from utils.auth import create_or_find_web_device from utils.router import APIRouter +from utils.validation import ValidationError, validate_password router = APIRouter( tags=["auth"], @@ -462,6 +463,16 @@ async def reset_password( Returns: None: Returns 200 OK status """ + # Ahead of the token check, which spends the link: a rejected password + # leaves it usable for another try. + try: + validate_password(new_password) + except ValidationError as exc: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=exc.message, + ) from exc + user = auth_handler.verify_password_reset_token(token) await auth_handler.set_user_new_password(user, new_password) diff --git a/backend/endpoints/user.py b/backend/endpoints/user.py index 3b6880f4b..ff5e26324 100644 --- a/backend/endpoints/user.py +++ b/backend/endpoints/user.py @@ -10,6 +10,7 @@ from decorators.auth import protected_route from endpoints.forms.identity import UserForm from endpoints.permissions import emit_permissions_changed from endpoints.responses.identity import InviteLinkSchema, UserSchema +from exceptions.database_exceptions import LastAdminError from handler.audit_handler import ( AuditActor, AuditTarget, @@ -482,18 +483,7 @@ async def update_user( # You can't change your own role if form_data.role and request.user.id != id: - new_role = Role.coerce(form_data.role) - # You can't demote the last admin (mirrors the delete guard). - if ( - db_user.role == Role.ADMIN - and new_role != Role.ADMIN - and len(db_user_handler.get_admin_users()) == 1 - ): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="You cannot demote the last admin user", - ) - cleaned_data["role"] = new_role + cleaned_data["role"] = Role.coerce(form_data.role) # You can't disable yourself if form_data.enabled is not None and request.user.id != id: @@ -541,11 +531,18 @@ async def update_user( "hashed_password" ) - await auth_handler.apply_user_update( - id, - cleaned_data, - revoke_sessions_for=previous_username if creds_updated else None, - ) + try: + await auth_handler.apply_user_update( + id, + cleaned_data, + revoke_sessions_for=previous_username if creds_updated else None, + keep_an_admin=True, + ) + except LastAdminError as exc: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="You cannot demote the last admin user", + ) from exc if creds_updated and request.user.id == id: request.session.clear() @@ -607,15 +604,14 @@ async def delete_user( if request.user.id == id: raise HTTPException(status_code=400, detail="You cannot delete yourself") - # You can't delete the last admin user - if user.role == Role.ADMIN and len(db_user_handler.get_admin_users()) == 1: - raise HTTPException( - status_code=400, detail="You cannot delete the last admin user" - ) - token_ids = db_client_token_handler.get_token_ids_by_users([id]) device_ids = [device.id for device in db_device_handler.get_devices(user_id=id)] - db_user_handler.delete_user(id) + try: + db_user_handler.delete_user(id, keep_an_admin=True) + except LastAdminError as exc: + raise HTTPException( + status_code=400, detail="You cannot delete the last admin user" + ) from exc await close_client_token_sockets(token_ids) for device_id in device_ids: await device_install_handler.discard_for_device(device_id) diff --git a/backend/exceptions/database_exceptions.py b/backend/exceptions/database_exceptions.py index d68106d5e..01800892e 100644 --- a/backend/exceptions/database_exceptions.py +++ b/backend/exceptions/database_exceptions.py @@ -7,3 +7,12 @@ class RomFileOwnerChangedError(Exception): def __repr__(self) -> str: return self.message + + +class LastAdminError(Exception): + def __init__(self, user_id: int) -> None: + self.message = f"User {user_id} is the last admin" + super().__init__(self.message) + + def __repr__(self) -> str: + return self.message diff --git a/backend/handler/auth/base_handler.py b/backend/handler/auth/base_handler.py index fdf84b2e3..3d83a0654 100644 --- a/backend/handler/auth/base_handler.py +++ b/backend/handler/auth/base_handler.py @@ -303,6 +303,7 @@ class AuthHandler: user_id: int, data: dict[str, Any], revoke_sessions_for: str | None = None, + keep_an_admin: bool = False, ) -> None: """ Write an update to a user, revoking that account's sessions around it. @@ -311,15 +312,25 @@ class AuthHandler: data (dict[str, Any]): The fields to write. revoke_sessions_for (str | None): Username the sessions are keyed by, or None to write without revoking. + keep_an_admin (bool): Refuse, with LastAdminError, a role change + that would leave no admin. """ from handler.database import db_user_handler + from handler.database.base_handler import sync_session - if revoke_sessions_for: - # Ahead of the write: an unreachable Redis then aborts the change - # rather than committing it with the account's sessions left live. - await RedisSessionMiddleware.clear_user_sessions(revoke_sessions_for) + with sync_session.begin() as session: + # First, so a refused update hasn't already logged the account out. + if keep_an_admin: + db_user_handler.refuse_removing_the_last_admin( + user_id, data, session=session + ) - db_user_handler.update_user(user_id, data) + if revoke_sessions_for: + # Ahead of the write: an unreachable Redis then aborts the change + # rather than committing it with the account's sessions left live. + await RedisSessionMiddleware.clear_user_sessions(revoke_sessions_for) + + db_user_handler.update_user(user_id, data, session=session) if revoke_sessions_for: # After it, for a login the old password was still good for. The diff --git a/backend/handler/database/users_handler.py b/backend/handler/database/users_handler.py index 7cebbaacb..9f29348d6 100644 --- a/backend/handler/database/users_handler.py +++ b/backend/handler/database/users_handler.py @@ -6,6 +6,7 @@ from sqlalchemy.orm import QueryableAttribute, Session, load_only from sqlalchemy.sql import Delete, Select, Update from decorators.database import INJECTED_SESSION, begin_session +from exceptions.database_exceptions import LastAdminError from models.user import Role, User from .base_handler import DBBaseHandler @@ -83,13 +84,37 @@ class DBUsersHandler(DBBaseHandler): ) -> User | None: return session.get(User, id) + @begin_session + def refuse_removing_the_last_admin( + self, + id: int, + data: dict[str, Any] | None = None, + *, + session: Session = INJECTED_SESSION, + ) -> None: + """Raise LastAdminError if writing `data` to `id`, or deleting it when None, would leave no admin, locking the admin rows until the transaction ends.""" + if data is not None and data.get("role", Role.ADMIN) == Role.ADMIN: + return + admin_ids = session.scalars( + select(User.id) + .where(User.role == Role.ADMIN) + .order_by(User.id) + .with_for_update() + ).all() + if list(admin_ids) == [id]: + raise LastAdminError(id) + @begin_session def update_user( self, id: int, data: dict[str, Any], + *, + keep_an_admin: bool = False, session: Session = INJECTED_SESSION, ) -> User: + if keep_an_admin: + self.refuse_removing_the_last_admin(id, data, session=session) session.execute( update(User) .where(User.id == id) @@ -126,8 +151,12 @@ class DBUsersHandler(DBBaseHandler): def delete_user( self, id: int, + *, + keep_an_admin: bool = False, session: Session = INJECTED_SESSION, ) -> Result[*tuple[Any, ...]]: + if keep_an_admin: + self.refuse_removing_the_last_admin(id, session=session) return session.execute( delete(User) .where(User.id == id) diff --git a/backend/tests/endpoints/sockets/test_activity.py b/backend/tests/endpoints/sockets/test_activity.py new file mode 100644 index 000000000..d7dcd114e --- /dev/null +++ b/backend/tests/endpoints/sockets/test_activity.py @@ -0,0 +1,221 @@ +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from endpoints.sockets import activity +from handler.activity_handler import activity_handler +from handler.database import db_user_handler +from handler.socket_handler import socket_handler + +SID = "sid-1" + + +@pytest.fixture +def sessions(mocker) -> dict[str, dict[str, Any]]: + stored: dict[str, dict[str, Any]] = {} + + async def get_session(sid: str) -> dict[str, Any]: + if sid not in stored: + raise KeyError(sid) + return dict(stored[sid]) + + async def save_session(sid: str, session: dict[str, Any]) -> None: + stored[sid] = dict(session) + + mocker.patch.object(socket_handler.socket_server, "get_session", get_session) + mocker.patch.object(socket_handler.socket_server, "save_session", save_session) + return stored + + +@pytest.fixture +def entry() -> MagicMock: + return MagicMock(name="entry") + + +@pytest.fixture +def build_entry(mocker, entry: MagicMock): + return mocker.patch.object( + activity_handler, "build_entry", AsyncMock(return_value=entry) + ) + + +@pytest.fixture +def publish_active(mocker): + return mocker.patch.object(activity_handler, "publish_active", AsyncMock()) + + +@pytest.fixture +def publish_clear(mocker): + return mocker.patch.object(activity_handler, "publish_clear", AsyncMock()) + + +@pytest.fixture +async def signed_in(sessions) -> None: + await activity.store_authenticated_user(SID, 7) + + +class TestStart: + async def test_publishes_for_the_socket_user_not_the_payload( + self, signed_in, build_entry, publish_active, entry, sessions + ): + await activity.activity_start( + SID, {"rom_id": 42, "device_id": "deck", "user_id": 99} # type: ignore[typeddict-unknown-key] + ) + + build_entry.assert_awaited_once_with( + user_id=7, device_id="deck", rom_id=42, preserve_started_at=False + ) + publish_active.assert_awaited_once_with(entry) + assert sessions[SID]["activity_user_id"] == 7 + assert sessions[SID]["activity_device_id"] == "deck" + + async def test_an_unauthenticated_socket_is_ignored( + self, sessions, build_entry, publish_active + ): + await activity.activity_start(SID, {"rom_id": 42, "device_id": "deck"}) + + build_entry.assert_not_awaited() + publish_active.assert_not_awaited() + + @pytest.mark.parametrize( + "payload", + [ + None, + "rom 42", + {"device_id": "deck"}, + {"rom_id": 42}, + {"rom_id": 42, "device_id": ""}, + {"rom_id": 42, "device_id": 5}, + {"rom_id": "forty-two", "device_id": "deck"}, + {"rom_id": [42], "device_id": "deck"}, + ], + ids=[ + "none", + "not_a_dict", + "no_rom", + "no_device", + "empty_device", + "non_string_device", + "non_numeric_rom", + "list_rom", + ], + ) + async def test_an_invalid_payload_is_ignored( + self, signed_in, build_entry, publish_active, payload: Any + ): + await activity.activity_start(SID, payload) + + build_entry.assert_not_awaited() + publish_active.assert_not_awaited() + + async def test_a_numeric_string_rom_id_is_accepted( + self, signed_in, build_entry, publish_active + ): + await activity.activity_start(SID, {"rom_id": "42", "device_id": "deck"}) # type: ignore[typeddict-item] + + call = build_entry.await_args + assert call is not None + assert call.kwargs["rom_id"] == 42 + + async def test_a_missing_rom_or_user_publishes_nothing( + self, signed_in, build_entry, publish_active, sessions + ): + build_entry.return_value = None + + await activity.activity_start(SID, {"rom_id": 42, "device_id": "deck"}) + + publish_active.assert_not_awaited() + assert "activity_device_id" not in sessions[SID] + + +class TestHeartbeat: + async def test_keeps_the_original_start_time( + self, signed_in, build_entry, publish_active, entry + ): + await activity.activity_heartbeat(SID, {"rom_id": 42, "device_id": "deck"}) + + build_entry.assert_awaited_once_with( + user_id=7, device_id="deck", rom_id=42, preserve_started_at=True + ) + publish_active.assert_awaited_once_with(entry) + + async def test_an_unauthenticated_socket_is_ignored( + self, sessions, build_entry, publish_active + ): + await activity.activity_heartbeat(SID, {"rom_id": 42, "device_id": "deck"}) + + build_entry.assert_not_awaited() + + async def test_a_missing_rom_or_user_publishes_nothing( + self, signed_in, build_entry, publish_active + ): + build_entry.return_value = None + + await activity.activity_heartbeat(SID, {"rom_id": 42, "device_id": "deck"}) + + publish_active.assert_not_awaited() + + +class TestStop: + async def test_clears_the_named_device(self, signed_in, publish_clear): + await activity.activity_stop(SID, {"device_id": "deck"}) + + publish_clear.assert_awaited_once_with(7, "deck") + + async def test_falls_back_to_the_device_it_started_on( + self, signed_in, build_entry, publish_active, publish_clear + ): + await activity.activity_start(SID, {"rom_id": 42, "device_id": "deck"}) + + await activity.activity_stop(SID) + + publish_clear.assert_awaited_once_with(7, "deck") + + async def test_without_any_device_does_nothing(self, signed_in, publish_clear): + await activity.activity_stop(SID, {}) + + publish_clear.assert_not_awaited() + + async def test_an_unauthenticated_socket_is_ignored(self, sessions, publish_clear): + await activity.activity_stop(SID, {"device_id": "deck"}) + + publish_clear.assert_not_awaited() + + +class TestDisconnect: + async def test_clears_the_activity_the_socket_started( + self, signed_in, build_entry, publish_active, publish_clear + ): + await activity.activity_start(SID, {"rom_id": 42, "device_id": "deck"}) + + await activity.activity_on_disconnect(SID) + + publish_clear.assert_awaited_once_with(7, "deck") + + async def test_a_socket_that_never_played_clears_nothing( + self, signed_in, publish_clear + ): + await activity.activity_on_disconnect(SID) + + publish_clear.assert_not_awaited() + + async def test_an_unknown_socket_clears_nothing(self, sessions, publish_clear): + await activity.activity_on_disconnect("gone") + + publish_clear.assert_not_awaited() + + +async def test_the_authenticated_user_is_looked_up_from_the_session(signed_in, mocker): + user = MagicMock(id=7) + get_user = mocker.patch.object(db_user_handler, "get_user", return_value=user) + + assert await activity.get_authenticated_user(SID) is user + get_user.assert_called_once_with(7) + + +async def test_an_unauthenticated_socket_has_no_user(sessions, mocker): + get_user = mocker.patch.object(db_user_handler, "get_user") + + assert await activity.get_authenticated_user(SID) is None + get_user.assert_not_called() diff --git a/backend/tests/endpoints/test_auth.py b/backend/tests/endpoints/test_auth.py new file mode 100644 index 000000000..2bdf8a49f --- /dev/null +++ b/backend/tests/endpoints/test_auth.py @@ -0,0 +1,212 @@ +import base64 +import time +from http import HTTPStatus +from unittest import mock + +import httpx2 +import pytest +from fastapi import status +from fastapi.testclient import TestClient +from joserfc import jwt +from joserfc.jwk import OctKey +from tests.audit_events import recorded_events + +from handler.auth import base_handler as auth_handler_module +from handler.auth.base_handler import auth_handler, oidc_handler +from handler.auth.constants import ALGORITHM, SESSION_COOKIE_NAME +from handler.database import db_user_handler +from handler.redis_handler import redis_client +from models.user import User + + +def _basic(username: str, password: str) -> dict[str, str]: + encoded = base64.b64encode(f"{username}:{password}".encode()).decode() + return {"Authorization": f"Basic {encoded}"} + + +def _reset_password(client, token: str, new_password: str = "a_new_password"): + return client.post( + "/api/reset-password", json={"token": token, "new_password": new_password} + ) + + +def test_reset_password_with_a_valid_token(client, admin_user: User): + token = auth_handler.generate_password_reset_token(admin_user) + + response = _reset_password(client, token) + + assert response.status_code == status.HTTP_200_OK + old = client.post("/api/login", headers=_basic("test_admin", "test_admin_password")) + new = client.post("/api/login", headers=_basic("test_admin", "a_new_password")) + assert old.status_code == status.HTTP_401_UNAUTHORIZED + assert new.status_code == status.HTTP_200_OK + assert "auth.password_reset" in [event.action for event in recorded_events()] + + +def test_reset_password_token_works_once(client, admin_user: User): + token = auth_handler.generate_password_reset_token(admin_user) + assert _reset_password(client, token).status_code == status.HTTP_200_OK + + response = _reset_password(client, token, "another_password") + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert client.post( + "/api/login", headers=_basic("test_admin", "another_password") + ).status_code == (status.HTTP_401_UNAUTHORIZED) + + +@pytest.mark.parametrize("new_password", ["", "short", "pässword"]) +def test_reset_password_rejects_an_invalid_password_and_keeps_the_link( + client, admin_user: User, new_password: str +): + token = auth_handler.generate_password_reset_token(admin_user) + + response = _reset_password(client, token, new_password) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert client.post( + "/api/login", headers=_basic("test_admin", "test_admin_password") + ).status_code == (status.HTTP_200_OK) + assert _reset_password(client, token).status_code == status.HTTP_200_OK + + +def test_reset_password_rejects_a_malformed_token(client): + assert _reset_password(client, "not-a-jwt").status_code == ( + status.HTTP_400_BAD_REQUEST + ) + + +def test_reset_password_rejects_a_token_signed_with_another_key( + client, admin_user: User +): + forged = jwt.encode( + {"alg": ALGORITHM}, + {"sub": admin_user.username, "type": "reset", "jti": "forged"}, + OctKey.import_key("not-the-server-secret-key-not-the-server"), + ) + redis_client.set("reset-jti:forged", "valid", ex=60) + + assert _reset_password(client, forged).status_code == (status.HTTP_400_BAD_REQUEST) + + +def test_reset_password_rejects_a_token_for_another_purpose(client, admin_user: User): + invite = jwt.encode( + {"alg": ALGORITHM}, + { + "sub": admin_user.username, + "type": "invite", + "jti": "invite", + "exp": int(time.time()) + 60, + }, + auth_handler_module.oct_key, + ) + redis_client.set("reset-jti:invite", "valid", ex=60) + + response = _reset_password(client, invite) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"] == "Invalid token purpose" + + +def test_reset_password_rejects_an_expired_token(client, admin_user: User): + expired = jwt.encode( + {"alg": ALGORITHM}, + { + "sub": admin_user.username, + "type": "reset", + "jti": "expired", + "exp": int(time.time()) - 1, + }, + auth_handler_module.oct_key, + ) + redis_client.set("reset-jti:expired", "valid", ex=60) + + response = _reset_password(client, expired) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"] == "Token has expired" + + +def test_reset_password_for_a_deleted_user_is_not_found(client, viewer_user: User): + token = auth_handler.generate_password_reset_token(viewer_user) + db_user_handler.delete_user(viewer_user.id) + + assert _reset_password(client, token).status_code == status.HTTP_404_NOT_FOUND + + +def _oidc_callback(client: TestClient, user: User | None) -> httpx2.Response: + fake_oauth = mock.MagicMock() + fake_oauth.openid.authorize_access_token = mock.AsyncMock( + return_value={"id_token": "provider-id-token"} + ) + with ( + mock.patch("endpoints.auth.OIDC_ENABLED", True), + mock.patch("endpoints.auth.OIDC_RP_INITIATED_LOGOUT", True), + mock.patch("endpoints.auth.oauth", fake_oauth), + mock.patch.object( + oidc_handler, + "get_current_active_user_from_openid_token", + mock.AsyncMock(return_value=(user, {})), + ), + ): + return client.get("/api/oauth/openid?code=c&state=s", follow_redirects=False) + + +def test_oidc_callback_logs_the_user_in(client: TestClient, admin_user: User): + response = _oidc_callback(client, admin_user) + + assert response.status_code == HTTPStatus.TEMPORARY_REDIRECT + assert response.headers["location"] == "/" + session_cookie = response.cookies.get(SESSION_COOKIE_NAME) + assert session_cookie is not None + cookie_header = {"Cookie": f"{SESSION_COOKIE_NAME}={session_cookie}"} + me = client.get("/api/users/me", headers=cookie_header) + assert me.status_code == status.HTTP_200_OK + assert me.json()["username"] == "test_admin" + + updated = db_user_handler.get_user(admin_user.id) + assert updated and updated.last_login is not None + [event] = recorded_events() + assert (event.action, event.data["method"]) == ("auth.login", "oidc") + assert event.device_id is not None + + with ( + mock.patch("endpoints.auth.OIDC_RP_INITIATED_LOGOUT", True), + mock.patch( + "endpoints.auth.OIDC_END_SESSION_ENDPOINT", "https://idp.example/logout" + ), + ): + logout = client.post("/api/logout", headers=cookie_header) + assert "id_token_hint=provider-id-token" in logout.json()["oidc_logout_url"] + + +def test_oidc_callback_for_an_unknown_user_is_unauthorized(client: TestClient): + response = _oidc_callback(client, None) + + assert response.status_code == status.HTTP_401_UNAUTHORIZED + assert SESSION_COOKIE_NAME not in response.cookies + [event] = recorded_events() + assert (event.action, event.data["reason"]) == ("auth.login_failed", "credentials") + + +def test_oidc_callback_for_a_disabled_user_is_unauthorized( + client: TestClient, viewer_user: User +): + db_user_handler.update_user(viewer_user.id, {"enabled": False}) + disabled = db_user_handler.get_user(viewer_user.id) + + response = _oidc_callback(client, disabled) + + assert response.status_code == status.HTTP_401_UNAUTHORIZED + assert SESSION_COOKIE_NAME not in response.cookies + [event] = recorded_events() + assert (event.action, event.data["reason"]) == ("auth.login_failed", "disabled") + + +@pytest.mark.parametrize("path", ["/api/login/openid", "/api/oauth/openid"]) +def test_oidc_endpoints_refuse_when_oidc_is_disabled(client: TestClient, path: str): + with mock.patch("endpoints.auth.OIDC_ENABLED", False): + response = client.get(path, follow_redirects=False) + + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + assert response.json()["detail"] == "OAuth disabled" diff --git a/backend/tests/endpoints/test_user.py b/backend/tests/endpoints/test_user.py new file mode 100644 index 000000000..9a0598ac5 --- /dev/null +++ b/backend/tests/endpoints/test_user.py @@ -0,0 +1,240 @@ +from datetime import timedelta + +from fastapi import status + +from exceptions.database_exceptions import LastAdminError +from handler.auth.base_handler import oauth_handler +from handler.auth.middleware.redis_session_middleware import RedisSessionMiddleware +from handler.database import db_permission_handler, db_user_handler +from models.permission import PermAction, PermEntity +from models.user import Role, User + + +def _bearer(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +def test_an_unknown_user_is_not_found(client, access_token: str): + response = client.put( + "/api/users/999999", data={"ra_username": "x"}, headers=_bearer(access_token) + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + +def test_non_admin_cannot_edit_another_user( + client, viewer_access_token: str, admin_user: User +): + response = client.put( + f"/api/users/{admin_user.id}", + data={"password": "hijacked_password"}, + headers=_bearer(viewer_access_token), + ) + + assert response.status_code == status.HTTP_403_FORBIDDEN + unchanged = db_user_handler.get_user(admin_user.id) + assert unchanged and unchanged.hashed_password == admin_user.hashed_password + + +def test_non_admin_can_edit_themselves( + client, viewer_access_token: str, viewer_user: User +): + response = client.put( + f"/api/users/{viewer_user.id}", + data={"email": "Viewer@Example.com"}, + headers=_bearer(viewer_access_token), + ) + + assert response.status_code == status.HTTP_200_OK + assert response.json()["email"] == "viewer@example.com" + + +def test_non_admin_cannot_promote_or_disable_themselves( + client, viewer_access_token: str, viewer_user: User +): + client.put( + f"/api/users/{viewer_user.id}", + data={"role": "admin", "enabled": "false"}, + headers=_bearer(viewer_access_token), + ) + + updated = db_user_handler.get_user(viewer_user.id) + assert updated and (updated.role, updated.enabled) == (Role.USER, True) + + +def test_admin_cannot_demote_or_disable_themselves( + client, access_token: str, admin_user: User +): + client.put( + f"/api/users/{admin_user.id}", + data={"role": "user", "enabled": "false"}, + headers=_bearer(access_token), + ) + + updated = db_user_handler.get_user(admin_user.id) + assert updated and (updated.role, updated.enabled) == (Role.ADMIN, True) + + +def test_admin_can_demote_another_admin(client, access_token: str, admin_user: User): + other = db_user_handler.add_user( + User(username="other_admin", hashed_password="x", role=Role.ADMIN) + ) + + response = client.put( + f"/api/users/{other.id}", data={"role": "user"}, headers=_bearer(access_token) + ) + + assert response.status_code == status.HTTP_200_OK + assert response.json()["role"] == Role.USER + + +def test_rename_is_lowercased(client, access_token: str, viewer_user: User): + response = client.put( + f"/api/users/{viewer_user.id}", + data={"username": "Renamed_Viewer"}, + headers=_bearer(access_token), + ) + + assert response.status_code == status.HTTP_200_OK + assert response.json()["username"] == "renamed_viewer" + + +def test_rename_to_a_taken_username_is_rejected( + client, access_token: str, viewer_user: User, admin_user: User +): + response = client.put( + f"/api/users/{viewer_user.id}", + data={"username": admin_user.username}, + headers=_bearer(access_token), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + unchanged = db_user_handler.get_user(viewer_user.id) + assert unchanged and unchanged.username == "test_viewer" + + +def test_rename_to_an_invalid_username_is_rejected( + client, access_token: str, viewer_user: User +): + response = client.put( + f"/api/users/{viewer_user.id}", + data={"username": "ab"}, + headers=_bearer(access_token), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + + +def test_update_to_a_taken_email_is_rejected( + client, access_token: str, viewer_user: User, editor_user: User +): + db_user_handler.update_user(editor_user.id, {"email": "taken@example.com"}) + + response = client.put( + f"/api/users/{viewer_user.id}", + data={"email": "taken@example.com"}, + headers=_bearer(access_token), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + + +def test_update_with_an_empty_email_clears_it( + client, access_token: str, viewer_user: User +): + db_user_handler.update_user(viewer_user.id, {"email": "viewer@example.com"}) + + response = client.put( + f"/api/users/{viewer_user.id}", + data={"email": ""}, + headers=_bearer(access_token), + ) + + assert response.status_code == status.HTTP_200_OK + assert response.json()["email"] is None + + +def test_update_to_a_short_password_is_rejected( + client, access_token: str, viewer_user: User +): + response = client.put( + f"/api/users/{viewer_user.id}", + data={"password": "short"}, + headers=_bearer(access_token), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + + +def test_a_delegated_user_manager_cannot_delete_the_last_admin( + client, admin_user: User, viewer_user: User +): + db_permission_handler.replace_user_overrides( + viewer_user.id, [(PermEntity.USERS, PermAction.WRITE, True, False)] + ) + manager = db_user_handler.get_user(viewer_user.id) + assert manager + token = oauth_handler.create_access_token( + data={ + "sub": manager.username, + "iss": "romm:oauth", + "scopes": " ".join(manager.oauth_scopes), + }, + expires_delta=timedelta(minutes=5), + ) + + response = client.delete(f"/api/users/{admin_user.id}", headers=_bearer(token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"] == "You cannot delete the last admin user" + assert db_user_handler.get_user(admin_user.id) is not None + + +def test_a_demotion_that_would_leave_no_admin_is_refused( + client, access_token: str, admin_user: User, mocker +): + other = db_user_handler.add_user( + User(username="other_admin", hashed_password="x", role=Role.ADMIN) + ) + # Only reachable when another request demotes this admin at the same time. + mocker.patch.object( + db_user_handler, + "refuse_removing_the_last_admin", + side_effect=LastAdminError(other.id), + ) + + response = client.put( + f"/api/users/{other.id}", data={"role": "user"}, headers=_bearer(access_token) + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"] == "You cannot demote the last admin user" + unchanged = db_user_handler.get_user(other.id) + assert unchanged and unchanged.role == Role.ADMIN + + +def test_a_refused_demotion_does_not_log_the_admin_out( + client, access_token: str, admin_user: User, mocker +): + other = db_user_handler.add_user( + User(username="other_admin", hashed_password="x", role=Role.ADMIN) + ) + mocker.patch.object( + db_user_handler, + "refuse_removing_the_last_admin", + side_effect=LastAdminError(other.id), + ) + clear_sessions = mocker.patch.object( + RedisSessionMiddleware, "clear_user_sessions", mocker.AsyncMock() + ) + + response = client.put( + f"/api/users/{other.id}", + data={"role": "user", "password": "a_new_password"}, + headers=_bearer(access_token), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + clear_sessions.assert_not_awaited() + unchanged = db_user_handler.get_user(other.id) + assert unchanged and unchanged.hashed_password == "x" diff --git a/backend/tests/handler/database/test_users_handler.py b/backend/tests/handler/database/test_users_handler.py new file mode 100644 index 000000000..9c8cad870 --- /dev/null +++ b/backend/tests/handler/database/test_users_handler.py @@ -0,0 +1,102 @@ +import threading + +import pytest + +from exceptions.database_exceptions import LastAdminError +from handler.database import db_user_handler +from handler.database.base_handler import sync_session +from models.user import Role, User + + +@pytest.fixture +def other_admin() -> User: + return db_user_handler.add_user( + User(username="other_admin", hashed_password="x", role=Role.ADMIN) + ) + + +def _role(user: User) -> Role: + current = db_user_handler.get_user(user.id) + assert current + return current.role + + +class TestKeepAnAdmin: + def test_the_last_admin_cannot_be_demoted(self, admin_user: User): + with pytest.raises(LastAdminError): + db_user_handler.update_user( + admin_user.id, {"role": Role.USER}, keep_an_admin=True + ) + + assert _role(admin_user) == Role.ADMIN + + def test_the_last_admin_cannot_be_deleted(self, admin_user: User): + with pytest.raises(LastAdminError): + db_user_handler.delete_user(admin_user.id, keep_an_admin=True) + + assert db_user_handler.get_user(admin_user.id) is not None + + def test_one_of_two_admins_can_be_demoted( + self, admin_user: User, other_admin: User + ): + db_user_handler.update_user( + other_admin.id, {"role": Role.USER}, keep_an_admin=True + ) + + assert _role(other_admin) == Role.USER + + def test_one_of_two_admins_can_be_deleted( + self, admin_user: User, other_admin: User + ): + db_user_handler.delete_user(other_admin.id, keep_an_admin=True) + + assert db_user_handler.get_user(other_admin.id) is None + + def test_other_edits_of_the_last_admin_go_through(self, admin_user: User): + db_user_handler.update_user( + admin_user.id, {"ra_username": "player"}, keep_an_admin=True + ) + + def test_a_user_can_be_deleted_with_one_admin_left( + self, admin_user: User, viewer_user: User + ): + db_user_handler.delete_user(viewer_user.id, keep_an_admin=True) + + assert db_user_handler.get_user(viewer_user.id) is None + + def test_without_the_flag_the_last_admin_can_be_demoted(self, admin_user: User): + # The OIDC role sync relies on this: the provider's claims win. + db_user_handler.update_user(admin_user.id, {"role": Role.USER}) + + assert _role(admin_user) == Role.USER + + def test_two_admins_demoting_each_other_leave_one( + self, admin_user: User, other_admin: User + ): + errors: list[Exception] = [] + + def demote_admin_user() -> None: + try: + db_user_handler.update_user( + admin_user.id, {"role": Role.USER}, keep_an_admin=True + ) + except Exception as exc: + errors.append(exc) + + # The first demotion holds its locks while the second one starts. + with sync_session.begin() as session: + db_user_handler.update_user( + other_admin.id, + {"role": Role.USER}, + keep_an_admin=True, + session=session, + ) + second = threading.Thread(target=demote_admin_user) + second.start() + second.join(timeout=1) + assert second.is_alive(), "the second demotion did not wait for the lock" + second.join(timeout=30) + + assert not second.is_alive() + assert [type(error) for error in errors] == [LastAdminError] + assert (_role(admin_user), _role(other_admin)) == (Role.ADMIN, Role.USER)