Let admins invalidate the password of selected users

app/core/users/cruds_users.py

@@ -283,6 +283,18 @@ async def update_user_password_by_id(
     await db.flush()
 
 
+async def update_should_user_change_password_by_id(
+    db: AsyncSession,
+    user_id: str,
+    should_change_password: bool = True,
+):
+    await db.execute(
+        update(models_users.CoreUser)
+        .where(models_users.CoreUser.id == user_id)
+        .values(should_change_password=should_change_password),
+    )
+
+
 async def remove_users_from_school(
     db: AsyncSession,
     school_id: UUID,

app/core/users/endpoints_users.py

@@ -406,6 +406,7 @@ async def activate_user(
         school_id=school_id,
         account_type=account_type,
         password_hash=password_hash,
+        should_change_password=False,
         name=user.name,
         firstname=user.firstname,
         nickname=user.nickname,
@@ -642,12 +643,31 @@ async def reset_password(
             ),
         )
 
+    user = await cruds_users.get_user_by_id(db=db, user_id=recover_request.user_id)
+    if user is None:
+        raise HTTPException(status_code=404, detail="Invalid user ID")
+    if user.should_change_password:
+        # we control whether we check if the new password is different
+        if security.verify_password(
+            reset_password_request.new_password,
+            user.password_hash,
+        ):
+            raise HTTPException(
+                status_code=403,
+                detail="The new password should not be identical to the current password",
+            )
+
     new_password_hash = security.get_password_hash(reset_password_request.new_password)
     await cruds_users.update_user_password_by_id(
         db=db,
         user_id=recover_request.user_id,
         new_password_hash=new_password_hash,
     )
+    await cruds_users.update_should_user_change_password_by_id(
+        db=db,
+        user_id=recover_request.user_id,
+        should_change_password=False,
+    )
 
     # As the user has reset its password, all other recovery requests can be deleted from the table
     await cruds_users.delete_recover_request_by_email(
@@ -666,6 +686,78 @@ async def reset_password(
     return standard_responses.Result()
 
 
+@router.post(
+    "/users/change-password",
+    response_model=standard_responses.Result,
+    status_code=201,
+)
+async def change_password(
+    change_password_request: schemas_users.ChangePasswordRequest,
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Change a user password.
+
+    This endpoint will check the **old_password**, see also the `/users/reset-password` endpoint if the user forgot their password.
+    """
+
+    user = await security.authenticate_user(
+        db=db,
+        email=change_password_request.email,
+        password=change_password_request.old_password,
+    )
+    if user is None:
+        raise HTTPException(status_code=403, detail="The old password is invalid")
+
+    if user.should_change_password:
+        # we control whether we check if the new password is different
+        if security.verify_password(
+            change_password_request.new_password,
+            user.password_hash,
+        ):
+            raise HTTPException(
+                status_code=403,
+                detail="The new password should not be identical to the current password",
+            )
+
+    new_password_hash = security.get_password_hash(change_password_request.new_password)
+    await cruds_users.update_user_password_by_id(
+        db=db,
+        user_id=user.id,
+        new_password_hash=new_password_hash,
+    )
+    await cruds_users.update_should_user_change_password_by_id(
+        db=db,
+        user_id=user.id,
+        should_change_password=False,
+    )
+
+    # Revoke existing auth refresh tokens
+    # to force the user to reauthenticate on all services and devices
+    # when their token expire
+    await cruds_auth.revoke_refresh_token_by_user_id(
+        db=db,
+        user_id=user.id,
+    )
+
+    return standard_responses.Result()
+
+
+@router.post("/users/invalidate-password/{user_id}", status_code=201)
+async def invalidate_password(
+    user_id: str,
+    user: models_users.CoreUser = Depends(is_user_in(GroupType.admin)),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Invalidate one user's password.
+    The concerned user should change their password with a different one to use our services again.
+
+    **This endpoint is only usable by administrators**
+    """
+    await cruds_users.update_should_user_change_password_by_id(db=db, user_id=user_id)
+
+
 @router.post(
     "/users/migrate-mail",
     status_code=204,
@@ -801,47 +893,6 @@ async def migrate_mail_confirm(
     return "The email address has been successfully updated"
 
 
-@router.post(
-    "/users/change-password",
-    response_model=standard_responses.Result,
-    status_code=201,
-)
-async def change_password(
-    change_password_request: schemas_users.ChangePasswordRequest,
-    db: AsyncSession = Depends(get_db),
-):
-    """
-    Change a user password.
-
-    This endpoint will check the **old_password**, see also the `/users/reset-password` endpoint if the user forgot their password.
-    """
-
-    user = await security.authenticate_user(
-        db=db,
-        email=change_password_request.email,
-        password=change_password_request.old_password,
-    )
-    if user is None:
-        raise HTTPException(status_code=403, detail="The old password is invalid")
-
-    new_password_hash = security.get_password_hash(change_password_request.new_password)
-    await cruds_users.update_user_password_by_id(
-        db=db,
-        user_id=user.id,
-        new_password_hash=new_password_hash,
-    )
-
-    # Revoke existing auth refresh tokens
-    # to force the user to reauthenticate on all services and devices
-    # when their token expire
-    await cruds_auth.revoke_refresh_token_by_user_id(
-        db=db,
-        user_id=user.id,
-    )
-
-    return standard_responses.Result()
-
-
 # We put the following endpoints at the end of the file to prevent them
 # from interacting with the previous endpoints
 # Ex: /users/activate is interpreted as a user whose id is "activate"

app/core/users/factory_users.py

@@ -93,6 +93,7 @@ async def create_core_users(cls, db: AsyncSession):
             user = CoreUser(
                 id=cls.other_users_id[i],
                 password_hash=password[i],
+                should_change_password=False,
                 firstname=firstname[i],
                 nickname=nickname[i],
                 name=name[i],
@@ -113,6 +114,7 @@ async def create_core_users(cls, db: AsyncSession):
                 password_hash=security.get_password_hash(
                     user_info.password or faker.password(16, True, True, True, True),
                 ),
+                should_change_password=False,
                 firstname=user_info.firstname,
                 nickname=user_info.nickname,
                 name=user_info.name,

app/core/users/models_users.py

@@ -24,6 +24,7 @@ class CoreUser(Base):
     email: Mapped[str] = mapped_column(unique=True, index=True)
     school_id: Mapped[UUID] = mapped_column(ForeignKey("core_school.id"))
     password_hash: Mapped[str]
+    should_change_password: Mapped[bool]
     # Depending on the account type, the user may have different rights and access to different features
     # External users may exist for:
     # - accounts meant to be used by external services based on Hyperion SSO or Hyperion backend

migrations/versions/43-password_invalidation.py

@@ -0,0 +1,53 @@
+"""password-invalidation
+
+Create Date: 2025-10-03 11:24:35.720759
+"""
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pytest_alembic import MigrationContext
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "cfba514689ed"
+down_revision: str | None = "c4812e1ab108"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "core_user",
+        sa.Column(
+            "should_change_password",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.sql.false(),
+        ),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("core_user", "should_change_password")
+    # ### end Alembic commands ###
+
+
+def pre_test_upgrade(
+    alembic_runner: "MigrationContext",
+    alembic_connection: sa.Connection,
+) -> None:
+    pass
+
+
+def test_upgrade(
+    alembic_runner: "MigrationContext",
+    alembic_connection: sa.Connection,
+) -> None:
+    pass

tests/commons.py

@@ -193,6 +193,7 @@ async def create_user_with_groups(
         email=email or (get_random_string() + "@etu.ec-lyon.fr"),
         school_id=school_id,
         password_hash=password_hash,
+        should_change_password=False,
         name=name or get_random_string(),
         firstname=firstname or get_random_string(),
         nickname=nickname,