139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
"""Миксин для 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)}"
|