276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""Миксины для API views с LDAP-специфичной логикой."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import TYPE_CHECKING, Optional
|
||
|
||
from django.conf import settings
|
||
from rest_framework.response import Response
|
||
|
||
from employees.ldap import UserService
|
||
from employees.ldap.domain.dtos import DirectoryUserDTO
|
||
from employees.ldap.errors import DirectoryDbError, DirectoryLdapError
|
||
|
||
if TYPE_CHECKING:
|
||
from employees.models import Employee
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _is_ldap_enabled() -> bool:
|
||
"""Проверяет, включена ли интеграция с LDAP."""
|
||
return getattr(settings, "LDAP_ENABLED", False)
|
||
|
||
|
||
class LdapUserCreationMixin:
|
||
"""Миксин для создания пользователей в LDAP при регистрации.
|
||
|
||
Используется в RegisterAPIView для вынесения LDAP-специфичной логики
|
||
из основного метода регистрации.
|
||
|
||
Методы:
|
||
create_user(): Создаёт пользователя (выбирает LDAP или БД автоматически)
|
||
create_ldap_user(): Создаёт пользователя в LDAP через DirectoryService
|
||
create_db_user(): Создаёт пользователя напрямую в БД (fallback без LDAP)
|
||
"""
|
||
|
||
def is_ldap_enabled(self) -> bool:
|
||
return _is_ldap_enabled()
|
||
|
||
def create_user(
|
||
self,
|
||
*,
|
||
first_name: str,
|
||
last_name: str,
|
||
email: str,
|
||
phone: str,
|
||
password: str,
|
||
avatar_bytes: Optional[bytes] = None,
|
||
is_active: bool = False,
|
||
) -> tuple[Optional["Employee"], Optional[Response]]:
|
||
"""Создаёт пользователя - автоматически выбирает LDAP или БД режим.
|
||
|
||
View не должен знать о режиме LDAP - вся логика инкапсулирована здесь.
|
||
|
||
Args:
|
||
first_name: Имя
|
||
last_name: Фамилия
|
||
email: Email (нормализованный, lowercase)
|
||
phone: Телефон в E.164 формате
|
||
password: Пароль
|
||
avatar_bytes: Байты аватара (опционально)
|
||
is_active: False до верификации email
|
||
|
||
Returns:
|
||
tuple[Employee | None, Response | None]:
|
||
- (Employee, None) при успехе
|
||
- (None, Response) при ошибке (Response для возврата клиенту)
|
||
"""
|
||
if self.is_ldap_enabled():
|
||
# Режим с LDAP: создаём disabled учётку в LDAP + пароль
|
||
return self.create_ldap_user(
|
||
first_name=first_name,
|
||
last_name=last_name,
|
||
email=email,
|
||
phone_e164=phone,
|
||
password=password,
|
||
avatar_bytes=avatar_bytes,
|
||
is_active=is_active,
|
||
)
|
||
else:
|
||
# Режим без LDAP: создаём пользователя напрямую в БД
|
||
emp = self.create_db_user(
|
||
first_name=first_name,
|
||
last_name=last_name,
|
||
email=email,
|
||
phone_number=phone,
|
||
password=password,
|
||
is_active=is_active,
|
||
)
|
||
return emp, None
|
||
|
||
def create_ldap_user(
|
||
self,
|
||
*,
|
||
first_name: str,
|
||
last_name: str,
|
||
email: str,
|
||
phone_e164: str,
|
||
password: str,
|
||
avatar_bytes: Optional[bytes] = None,
|
||
is_active: bool = False,
|
||
) -> tuple[Optional["Employee"], Optional[Response]]:
|
||
"""Создаёт disabled пользователя в LDAP с паролем.
|
||
|
||
Args:
|
||
first_name: Имя
|
||
last_name: Фамилия
|
||
email: Email (нормализованный, lowercase)
|
||
phone_e164: Телефон в E.164 формате
|
||
password: Пароль (идёт только в LDAP, в БД будет unusable)
|
||
avatar_bytes: Байты аватара (опционально)
|
||
is_active: False до верификации email
|
||
|
||
Returns:
|
||
tuple[Employee | None, Response | None]:
|
||
- (Employee, None) при успехе
|
||
- (None, Response) при ошибке (Response для возврата клиенту)
|
||
"""
|
||
svc = UserService()
|
||
dto = DirectoryUserDTO(
|
||
first_name=first_name,
|
||
last_name=last_name,
|
||
email=email,
|
||
phone_e164=phone_e164,
|
||
department_dn=None,
|
||
group_cns=[],
|
||
initial_password=password, # пароль идёт только в LDAP
|
||
avatar_bytes=avatar_bytes,
|
||
is_active=is_active, # disabled до верификации email
|
||
)
|
||
|
||
try:
|
||
emp = svc.create_user(dto)
|
||
return emp, None
|
||
except DirectoryLdapError as e:
|
||
logger.exception(
|
||
"LDAP user creation failed",
|
||
extra={
|
||
"event": "ldap.user.creation_failed",
|
||
"error_type": type(e).__name__,
|
||
},
|
||
)
|
||
return None, Response(
|
||
{"ok": False, "error": "ldap_error", "detail": str(e)},
|
||
status=502,
|
||
)
|
||
except DirectoryDbError as e:
|
||
logger.exception(
|
||
"Local database operation failed during LDAP user creation",
|
||
extra={
|
||
"event": "ldap.user.database_creation_failed",
|
||
"error_type": type(e).__name__,
|
||
},
|
||
)
|
||
return None, Response(
|
||
{"ok": False, "error": "db_error", "detail": str(e)}, status=500
|
||
)
|
||
|
||
def create_db_user(
|
||
self,
|
||
*,
|
||
first_name: str,
|
||
last_name: str,
|
||
email: str,
|
||
phone_number: str,
|
||
password: str,
|
||
is_active: bool = False,
|
||
) -> "Employee":
|
||
"""Создаёт пользователя напрямую в БД (режим без LDAP).
|
||
|
||
Args:
|
||
first_name: Имя
|
||
last_name: Фамилия
|
||
email: Email (нормализованный, lowercase)
|
||
phone_number: Телефон в E.164 формате
|
||
password: Пароль (устанавливается через set_password)
|
||
is_active: False до верификации email
|
||
|
||
Returns:
|
||
Employee: Созданный пользователь
|
||
"""
|
||
from employees.models import Employee
|
||
|
||
emp = Employee.objects.create(
|
||
first_name=first_name,
|
||
last_name=last_name,
|
||
email=email,
|
||
phone_number=phone_number,
|
||
is_active=is_active,
|
||
is_ldap_managed=False,
|
||
)
|
||
# Устанавливаем пароль в БД
|
||
emp.set_password(password)
|
||
emp.save(update_fields=["password"])
|
||
|
||
return emp
|
||
|
||
|
||
class LdapPasswordMixin:
|
||
"""Миксин для endpoints, работающих с паролями LDAP пользователей.
|
||
|
||
Используется в PasswordChangeView, PasswordResetView и других
|
||
endpoints, которым нужно обновлять пароли в LDAP.
|
||
|
||
Методы:
|
||
update_ldap_password(): Обновляет пароль пользователя в LDAP
|
||
"""
|
||
|
||
def update_ldap_password(
|
||
self,
|
||
employee: "Employee",
|
||
new_password: str,
|
||
) -> tuple[bool, Optional[Response]]:
|
||
"""Обновляет пароль пользователя в LDAP.
|
||
|
||
Args:
|
||
employee: Employee instance
|
||
new_password: Новый пароль
|
||
|
||
Returns:
|
||
tuple[bool, Response | None]:
|
||
- (True, None) при успехе
|
||
- (False, Response) при ошибке (Response для возврата клиенту)
|
||
"""
|
||
if not employee.is_ldap_managed:
|
||
# Не LDAP пользователь - устанавливаем пароль в БД
|
||
employee.set_password(new_password)
|
||
employee.save(update_fields=["password"])
|
||
return True, None
|
||
|
||
# LDAP пользователь - обновляем через UserService
|
||
try:
|
||
svc = UserService()
|
||
svc.update_user(
|
||
emp=employee,
|
||
changes={"password": new_password},
|
||
group_cns=None,
|
||
move_to_department_dn=None,
|
||
)
|
||
# Сохраняем локальный пароль в unusable-состоянии заново,
|
||
# чтобы инвалидировать любые reset-токены Django.
|
||
employee.set_unusable_password()
|
||
employee.save(update_fields=["password"])
|
||
return True, None
|
||
except DirectoryLdapError as error:
|
||
logger.exception(
|
||
"LDAP password update failed",
|
||
extra={
|
||
"event": "ldap.password_update.failed",
|
||
"employee_id": employee.id,
|
||
"error_type": type(error).__name__,
|
||
},
|
||
)
|
||
return False, Response(
|
||
{"ok": False, "error": "ldap_error", "detail": str(error)},
|
||
status=502,
|
||
)
|
||
except Exception as error:
|
||
logger.exception(
|
||
"LDAP password update failed",
|
||
extra={
|
||
"event": "ldap.password_update.failed",
|
||
"employee_id": employee.id,
|
||
"error_type": type(error).__name__,
|
||
},
|
||
)
|
||
return False, Response(
|
||
{
|
||
"ok": False,
|
||
"error": "password_update_failed",
|
||
"detail": str(error),
|
||
},
|
||
status=500,
|
||
)
|