Add: clear user password history

app/alembic/versions/a99f866a7e3a_add_user_pwd_reset_permission.py

@@ -0,0 +1,53 @@
+"""Add user reset password history permission to Domain Admins role.
+
+Revision ID: a99f866a7e3a
+Revises: 6c858cc05da7
+Create Date: 2025-12-23 10:20:29.147813
+
+"""
+
+from alembic import op
+from dishka import AsyncContainer
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession
+
+from entities import Role
+from enums import AuthorizationRules, RoleConstants
+
+# revision identifiers, used by Alembic.
+revision: None | str = "a99f866a7e3a"
+down_revision: None | str = "6c858cc05da7"
+branch_labels: None | list[str] = None
+depends_on: None | list[str] = None
+
+
+def upgrade(container: AsyncContainer) -> None:  # noqa: ARG001
+    """Upgrade."""
+
+    async def _add_api_permission(connection: AsyncConnection) -> None:
+        session = AsyncSession(bind=connection)
+        query = (
+            select(Role)
+            .filter_by(name=RoleConstants.DOMAIN_ADMINS_ROLE_NAME)
+        )  # fmt: skip
+        role = (await session.scalars(query)).first()
+        if role:
+            role.permissions |= AuthorizationRules.USER_CLEAR_PASSWORD_HISTORY
+
+    op.run_async(_add_api_permission)
+
+
+def downgrade(container: AsyncContainer) -> None:  # noqa: ARG001
+    """Downgrade."""
+
+    async def _remove_api_permission(connection: AsyncConnection) -> None:
+        session = AsyncSession(bind=connection)
+        query = (
+            select(Role)
+            .filter_by(name=RoleConstants.DOMAIN_ADMINS_ROLE_NAME)
+        )  # fmt: skip
+        role = (await session.scalars(query)).first()
+        if role:
+            role.permissions &= ~AuthorizationRules.USER_CLEAR_PASSWORD_HISTORY
+
+    op.run_async(_remove_api_permission)

app/api/__init__.py

@@ -14,7 +14,11 @@
 from .main.krb5_router import krb5_router
 from .main.router import entry_router
 from .network.router import network_router
-from .password_policy import password_ban_word_router, password_policy_router
+from .password_policy import (
+    password_ban_word_router,
+    password_policy_router,
+    user_password_history_router,
+)
 from .shadow.router import shadow_router
 
 __all__ = [
@@ -25,6 +29,7 @@
     "mfa_router",
     "password_ban_word_router",
     "password_policy_router",
+    "user_password_history_router",
     "ldap_schema_router",
     "dns_router",
     "krb5_router",

app/api/password_policy/__init__.py

@@ -6,8 +6,10 @@
 
 from .password_ban_word_router import password_ban_word_router
 from .password_policy_router import password_policy_router
+from .user_password_history_router import user_password_history_router
 
 __all__ = [
     "password_ban_word_router",
     "password_policy_router",
+    "user_password_history_router",
 ]

app/api/password_policy/adapter.py

@@ -19,6 +19,7 @@
 from ldap_protocol.policies.password.use_cases import (
     PasswordBanWordUseCases,
     PasswordPolicyUseCases,
+    UserPasswordHistoryUseCases,
 )
 
 _convert_schema_to_dto = get_converter(PasswordPolicySchema, PasswordPolicyDTO)
@@ -28,6 +29,15 @@
 )
 
 
+class UserPasswordHistoryResetFastAPIAdapter(
+    BaseAdapter[UserPasswordHistoryUseCases],
+):
+    """Adapter for clearing user password history."""
+
+    async def clear(self, user_name: str) -> None:
+        await self._service.clear(user_name)
+
+
 class PasswordPolicyFastAPIAdapter(BaseAdapter[PasswordPolicyUseCases]):
     """Adapter for password policies."""
 
@@ -46,9 +56,7 @@ async def get_password_policy_by_dir_path_dn(
         path_dn: str,
     ) -> PasswordPolicySchema[int]:
         """Get one Password Policy for one Directory by its path."""
-        dto = await self._service.get_password_policy_by_dir_path_dn(
-            path_dn,
-        )
+        dto = await self._service.get_password_policy_by_dir_path_dn(path_dn)
         return _convert_dto_to_schema(dto)
 
     async def update(

app/api/password_policy/user_password_history_router.py

@@ -0,0 +1,51 @@
+"""Password Policy router.
+
+Copyright (c) 2024 MultiFactor
+License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
+"""
+
+from dishka import FromDishka
+from fastapi import Depends, status
+from fastapi_error_map.routing import ErrorAwareRouter
+from fastapi_error_map.rules import rule
+
+from api.auth.utils import verify_auth
+from api.error_routing import (
+    ERROR_MAP_TYPE,
+    DishkaErrorAwareRoute,
+    DomainErrorTranslator,
+)
+from api.password_policy.adapter import UserPasswordHistoryResetFastAPIAdapter
+from enums import DomainCodes
+from ldap_protocol.identity.exceptions import (
+    AuthorizationError,
+    UserNotFoundError,
+)
+
+translator = DomainErrorTranslator(DomainCodes.PASSWORD_POLICY)
+
+error_map: ERROR_MAP_TYPE = {
+    UserNotFoundError: rule(
+        status=status.HTTP_400_BAD_REQUEST,
+        translator=translator,
+    ),
+    AuthorizationError: rule(
+        status=status.HTTP_401_UNAUTHORIZED,
+        translator=translator,
+    ),
+}
+
+user_password_history_router = ErrorAwareRouter(
+    prefix="/user/password_history",
+    dependencies=[Depends(verify_auth)],
+    tags=["User Password history"],
+    route_class=DishkaErrorAwareRoute,
+)
+
+
+@user_password_history_router.post("/clear/{user_name}", error_map=error_map)
+async def clear(
+    user_name: str,
+    adapter: FromDishka[UserPasswordHistoryResetFastAPIAdapter],
+) -> None:
+    await adapter.clear(user_name)

app/enums.py

@@ -215,6 +215,8 @@ class AuthorizationRules(IntFlag):
     NETWORK_POLICY_VALIDATOR_IS_USER_GROUP_VALID = auto()
     NETWORK_POLICY_VALIDATOR_CHECK_MFA_GROUP = auto()
 
+    USER_CLEAR_PASSWORD_HISTORY = auto()
+
     @classmethod
     def get_all(cls) -> Self:
         return cls(sum(cls))

app/ioc.py

@@ -35,6 +35,7 @@
 from api.password_policy.adapter import (
     PasswordBanWordsFastAPIAdapter,
     PasswordPolicyFastAPIAdapter,
+    UserPasswordHistoryResetFastAPIAdapter,
 )
 from api.shadow.adapter import ShadowAdapter
 from authorization_provider_protocol import AuthorizationProviderProtocol
@@ -132,7 +133,10 @@
     PasswordBanWordRepository,
 )
 from ldap_protocol.policies.password.settings import PasswordValidatorSettings
-from ldap_protocol.policies.password.use_cases import PasswordBanWordUseCases
+from ldap_protocol.policies.password.use_cases import (
+    PasswordBanWordUseCases,
+    UserPasswordHistoryUseCases,
+)
 from ldap_protocol.roles.access_manager import AccessManager
 from ldap_protocol.roles.ace_dao import AccessControlEntryDAO
 from ldap_protocol.roles.role_dao import RoleDAO
@@ -438,6 +442,10 @@ async def get_dhcp_mngr(
     )
     object_class_use_case = provide(ObjectClassUseCase, scope=Scope.REQUEST)
 
+    user_password_history_use_cases = provide(
+        UserPasswordHistoryUseCases,
+        scope=Scope.REQUEST,
+    )
     password_policy_validator = provide(
         PasswordPolicyValidator,
         scope=Scope.REQUEST,
@@ -563,6 +571,10 @@ async def get_audit_monitor(
         scope=Scope.REQUEST,
     )
 
+    user_password_history_reset_adapter = provide(
+        UserPasswordHistoryResetFastAPIAdapter,
+        scope=Scope.REQUEST,
+    )
     password_policies_adapter = provide(
         PasswordPolicyFastAPIAdapter,
         scope=Scope.REQUEST,

app/ldap_protocol/policies/password/dao.py

@@ -10,7 +10,7 @@
 from adaptix.conversion import get_converter, link_function
 from sqlalchemy import Integer, String, cast, exists, func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import selectinload
+from sqlalchemy.orm import attributes, selectinload
 
 from abstract_dao import AbstractDAO
 from entities import Attribute, Group, PasswordPolicy, User
@@ -440,6 +440,7 @@ async def post_save_password_actions(self, user: User) -> None:
         await self._session.execute(query)
 
         user.password_history.append(tcast("str", user.password))
+        attributes.flag_modified(user, "password_history")
         await self._session.flush()
 
     async def is_password_change_restricted(

app/ldap_protocol/policies/password/use_cases.py

@@ -6,22 +6,52 @@
 
 from typing import ClassVar, Iterable
 
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import attributes
+
 from abstract_service import AbstractService
 from entities import User
 from enums import AuthorizationRules
+from ldap_protocol.identity.exceptions import UserNotFoundError
 from ldap_protocol.policies.password.ban_word_repository import (
     PasswordBanWordRepository,
 )
 from ldap_protocol.policies.password.constants import (
     MAX_BANWORD_LENGTH,
     MIN_LENGTH_FOR_TRGM,
 )
+from ldap_protocol.utils.queries import get_user
 
 from .dao import PasswordPolicyDAO
 from .dataclasses import PasswordPolicyDTO, PriorityT
 from .validator import PasswordPolicyValidator
 
 
+class UserPasswordHistoryUseCases(AbstractService):
+    """User Password History Use Cases."""
+
+    _session: AsyncSession
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    async def clear(self, user_name: str) -> None:
+        user = await get_user(self._session, user_name)
+
+        if not user:
+            raise UserNotFoundError(
+                f"User {user_name} not found in the database.",
+            )
+
+        user.password_history.clear()
+        attributes.flag_modified(user, "password_history")
+        await self._session.flush()
+
+    PERMISSIONS: ClassVar[dict[str, AuthorizationRules]] = {
+        clear.__name__: AuthorizationRules.USER_CLEAR_PASSWORD_HISTORY,
+    }
+
+
 class PasswordPolicyUseCases(AbstractService):
     """Password Policy Use Cases."""
 

app/multidirectory.py

@@ -34,6 +34,7 @@
     password_policy_router,
     session_router,
     shadow_router,
+    user_password_history_router,
 )
 from api.exception_handlers import handle_auth_error, handle_db_connect_error
 from api.middlewares import proc_time_header_middleware, set_key_middleware
@@ -82,6 +83,7 @@ def _create_basic_app(settings: Settings) -> FastAPI:
     app.include_router(password_policy_router)
     app.include_router(krb5_router)
     app.include_router(dns_router)
+    app.include_router(user_password_history_router)
     app.include_router(session_router)
     app.include_router(ldap_schema_router)
     app.include_router(dhcp_router)

tests/conftest.py

@@ -138,7 +138,10 @@
     PasswordBanWordRepository,
 )
 from ldap_protocol.policies.password.settings import PasswordValidatorSettings
-from ldap_protocol.policies.password.use_cases import PasswordBanWordUseCases
+from ldap_protocol.policies.password.use_cases import (
+    PasswordBanWordUseCases,
+    UserPasswordHistoryUseCases,
+)
 from ldap_protocol.roles.access_manager import AccessManager
 from ldap_protocol.roles.ace_dao import AccessControlEntryDAO
 from ldap_protocol.roles.dataclasses import RoleDTO
@@ -310,6 +313,10 @@ def get_object_class_dao(self, session: AsyncSession) -> ObjectClassDAO:
     )
     object_class_use_case = provide(ObjectClassUseCase, scope=Scope.REQUEST)
 
+    user_password_history_use_cases = provide(
+        UserPasswordHistoryUseCases,
+        scope=Scope.REQUEST,
+    )
     password_ban_word_repository = provide(
         PasswordBanWordRepository,
         scope=Scope.REQUEST,