diff --git a/app/alembic/versions/a99f866a7e3a_add_user_pwd_reset_permission.py b/app/alembic/versions/a99f866a7e3a_add_user_pwd_reset_permission.py new file mode 100644 index 000000000..e517afe15 --- /dev/null +++ b/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) diff --git a/app/api/__init__.py b/app/api/__init__.py index 468cea5c5..69f1e8f37 100644 --- a/app/api/__init__.py +++ b/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", diff --git a/app/api/password_policy/__init__.py b/app/api/password_policy/__init__.py index fbb1affbf..c66ab654a 100644 --- a/app/api/password_policy/__init__.py +++ b/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", ] diff --git a/app/api/password_policy/adapter.py b/app/api/password_policy/adapter.py index 63ce9ce3d..fb6f0da02 100644 --- a/app/api/password_policy/adapter.py +++ b/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( diff --git a/app/api/password_policy/user_password_history_router.py b/app/api/password_policy/user_password_history_router.py new file mode 100644 index 000000000..ca9c53629 --- /dev/null +++ b/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) diff --git a/app/enums.py b/app/enums.py index 5258c3a0d..f0bec8c21 100644 --- a/app/enums.py +++ b/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)) diff --git a/app/ioc.py b/app/ioc.py index 3ad06c6b4..c21449450 100644 --- a/app/ioc.py +++ b/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, diff --git a/app/ldap_protocol/policies/password/dao.py b/app/ldap_protocol/policies/password/dao.py index 9a85d2ad7..5c818ca0a 100644 --- a/app/ldap_protocol/policies/password/dao.py +++ b/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( diff --git a/app/ldap_protocol/policies/password/use_cases.py b/app/ldap_protocol/policies/password/use_cases.py index fe22e9b45..67cc27e0a 100644 --- a/app/ldap_protocol/policies/password/use_cases.py +++ b/app/ldap_protocol/policies/password/use_cases.py @@ -6,9 +6,13 @@ 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, ) @@ -16,12 +20,38 @@ 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.""" diff --git a/app/multidirectory.py b/app/multidirectory.py index 613dce878..a0fc40d56 100644 --- a/app/multidirectory.py +++ b/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) diff --git a/interface b/interface index f31962020..017b7f344 160000 --- a/interface +++ b/interface @@ -1 +1 @@ -Subproject commit f31962020a6689e6a4c61fb3349db5b5c7895f92 +Subproject commit 017b7f344e290814e3af5ca0f210a592afaf08ed diff --git a/tests/conftest.py b/tests/conftest.py index b97a8ce4a..0ccf261a2 100644 --- a/tests/conftest.py +++ b/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, diff --git a/tests/test_api/test_password_policy/conftest.py b/tests/test_api/test_password_policy/conftest.py index f5a7b3841..157a94725 100644 --- a/tests/test_api/test_password_policy/conftest.py +++ b/tests/test_api/test_password_policy/conftest.py @@ -16,10 +16,16 @@ provide, ) -from api.password_policy.adapter import PasswordPolicyFastAPIAdapter +from api.password_policy.adapter import ( + PasswordPolicyFastAPIAdapter, + UserPasswordHistoryResetFastAPIAdapter, +) from config import Settings from ldap_protocol.policies.password import PasswordPolicyUseCases from ldap_protocol.policies.password.dataclasses import PasswordPolicyDTO +from ldap_protocol.policies.password.use_cases import ( + UserPasswordHistoryUseCases, +) from tests.conftest import TestProvider @@ -35,11 +41,18 @@ class TestLocalProvider(Provider): """Test provider for local scope.""" _cached_policy_use_cases: PasswordPolicyUseCases | None = None + _cached_user_password_history_use_cases: ( + UserPasswordHistoryUseCases | None + ) = None password_policies_adapter = provide( PasswordPolicyFastAPIAdapter, scope=Scope.REQUEST, ) + user_password_history_reset_adapter = provide( + UserPasswordHistoryResetFastAPIAdapter, + scope=Scope.REQUEST, + ) @provide(scope=Scope.REQUEST, provides=PasswordPolicyUseCases) async def get_password_use_cases( @@ -120,6 +133,22 @@ async def get_password_use_cases( yield self._cached_policy_use_cases self._cached_policy_use_cases = None + @provide( + scope=Scope.REQUEST, + provides=UserPasswordHistoryUseCases, + ) + async def get_user_password_history_use_cases( + self, + ) -> AsyncIterator[UserPasswordHistoryUseCases]: + if self._cached_user_password_history_use_cases is None: + session = Mock() + use_cases = UserPasswordHistoryUseCases(session) + use_cases.clear = make_mock("clear") # type: ignore + self._cached_user_password_history_use_cases = use_cases + + yield self._cached_user_password_history_use_cases + self._cached_user_password_history_use_cases = None + @pytest_asyncio.fixture(scope="session") async def container(settings: Settings) -> AsyncIterator[AsyncContainer]: @@ -141,3 +170,12 @@ async def password_use_cases( """Get di password_use_cases.""" async with container(scope=Scope.REQUEST) as container: yield await container.get(PasswordPolicyUseCases) + + +@pytest_asyncio.fixture +async def user_password_history_use_cases( + container: AsyncContainer, +) -> AsyncIterator[UserPasswordHistoryUseCases]: + """Get di user_password_history_use_cases.""" + async with container(scope=Scope.REQUEST) as container: + yield await container.get(UserPasswordHistoryUseCases) diff --git a/tests/test_api/test_password_policy/test_user_password_history_router.py b/tests/test_api/test_password_policy/test_user_password_history_router.py new file mode 100644 index 000000000..c04dfde49 --- /dev/null +++ b/tests/test_api/test_password_policy/test_user_password_history_router.py @@ -0,0 +1,45 @@ +"""Test User Password History router. + +Copyright (c) 2025 MultiFactor +License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE +""" + +from unittest.mock import Mock + +import pytest +from fastapi import status +from httpx import AsyncClient + + +@pytest.mark.asyncio +async def test_clear_success( + http_client: AsyncClient, + user_password_history_use_cases: Mock, +) -> None: + """Test clear user password history endpoint.""" + user_name = "testuser" + response = await http_client.post( + f"/user/password_history/clear/{user_name}", + ) + + # NOTE to user_password_history_use_cases.reset returned Mock, not wrapper # noqa: E501 + user_password_history_use_cases._perm_checker = None # noqa: SLF001 + user_password_history_use_cases.clear.assert_called_once() + assert response.status_code == status.HTTP_200_OK + + +@pytest.mark.asyncio +async def test_clear_unauthorized( + http_client_with_login_perm: AsyncClient, + user_password_history_use_cases: Mock, +) -> None: + """Test clear user password history endpoint without permissions.""" + user_name = "testuser" + response = await http_client_with_login_perm.post( + f"/user/password_history/clear/{user_name}", + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + # NOTE to user_password_history_use_cases.reset returned Mock, not wrapper # noqa: E501 + user_password_history_use_cases._perm_checker = None # noqa: SLF001 + user_password_history_use_cases.clear.assert_not_called()