"""Миксин для API endpoint'ов, работающих с паролями в LDAP.""" import logging from django.conf import settings from employees.ldap import UserService from employees.ldap.errors import ( DirectoryDbError, DirectoryLdapError, DirectoryServiceError, ) from employees.models import LdapSyncState logger = logging.getLogger(__name__) def _is_ldap_enabled(): """Проверяет, включена ли интеграция с LDAP.""" return getattr(settings, "LDAP_ENABLED", False) class LdapPasswordMixin: """Миксин для endpoint'ов, которые работают с паролями. Автоматически синхронизирует изменения паролей с LDAP. Использование: class ChangePasswordAPIView(LdapPasswordMixin, APIView): def post(self, request): user = request.user # Безопасный доступ: new_password = request.data.get('new_password') if not new_password: return Response({'error': 'Password required'}, 400) # Меняем пароль в БД user.set_password(new_password) user.save() # Синхронизируем с LDAP (автоматически через миксин) self.sync_password_to_ldap(user, new_password) return Response({'ok': True}) """ def sync_password_to_ldap(self, employee_instance, new_password): """Синхронизирует пароль с LDAP для данного Employee. Args: employee_instance: инстанс Employee new_password: новый пароль (plaintext) Returns: tuple: (success: bool, error: str or None) """ if not _is_ldap_enabled(): return True, None if not employee_instance.is_ldap_managed: return True, None # Проверяем наличие LDAP пользователя sync_state = LdapSyncState.objects.filter( model="employee", object_pk=str(employee_instance.pk) ).first() if not sync_state or not (sync_state.ldap_dn or sync_state.ldap_guid): logger.warning( "LDAP password synchronization skipped", extra={ "event": "ldap.password_sync.skipped", "employee_id": employee_instance.id, "reason": "missing_sync_state", }, ) return False, "no_ldap_user" try: svc = UserService() svc.update_user( emp=employee_instance, changes={"password": new_password}, group_cns=None, move_to_department_dn=None, ) logger.info( "LDAP password synchronized", extra={ "event": "ldap.password_sync.completed", "employee_id": employee_instance.id, }, ) return True, None except DirectoryLdapError as error: logger.exception( "LDAP password synchronization failed", extra={ "event": "ldap.password_sync.failed", "employee_id": employee_instance.id, "error_type": type(error).__name__, }, ) return False, f"ldap_error: {str(error)}" except DirectoryServiceError as error: logger.exception( "LDAP password synchronization failed", extra={ "event": "ldap.password_sync.failed", "employee_id": employee_instance.id, "error_type": type(error).__name__, }, ) return False, f"service_error: {str(error)}" except DirectoryDbError as error: logger.exception( "LDAP password synchronization failed", extra={ "event": "ldap.password_sync.failed", "employee_id": employee_instance.id, "error_type": type(error).__name__, }, ) return False, f"db_error: {str(error)}" except Exception as error: logger.exception( "LDAP password synchronization failed", extra={ "event": "ldap.password_sync.failed", "employee_id": employee_instance.id, "error_type": type(error).__name__, }, ) return False, f"unexpected_error: {str(error)}"