# eusrr_backend/auth_backends.py from __future__ import annotations import logging import re from typing import Any, List, Optional from django.conf import settings from django.contrib.auth import get_user_model from django.contrib.auth.backends import ModelBackend from django.contrib.auth.models import Permission from django.http import HttpRequest from django.utils import timezone from ldap3 import SUBTREE, Connection, Server from ldap3.core.exceptions import LDAPBindError, LDAPSocketOpenError from employees.ldap.utils.text_utils import esc_filter from employees.models import Department, Employee, EmployeeDepartment from employees.utils import _normalize_phone logger = logging.getLogger(__name__) UserModel = get_user_model() EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") def _looks_like_email(s: str) -> bool: return bool(EMAIL_RE.match(s or "")) def _model_has_field(model, field_name: str) -> bool: """Проверяет наличие поля модели по имени (по _meta).""" return any(getattr(f, "name", None) == field_name for f in model._meta.get_fields()) _DEFAULT_PHONE_FIELD = getattr(settings, "AUTH_PHONE_FIELD", "phone_number") PHONE_FIELD: Optional[str] = ( _DEFAULT_PHONE_FIELD if _model_has_field(UserModel, _DEFAULT_PHONE_FIELD) else None ) class EmailOrPhoneBackend(ModelBackend): """ Аутентификация по email ИЛИ по телефону (E.164) + стандартная проверка пароля. Пользователь должен быть активен (is_active=True) — активируйте после verify_email(). """ def authenticate(self, request, username=None, password=None, **kwargs): login = username or kwargs.get("email") or kwargs.get("phone") logger.debug( "Local authentication attempted", extra={ "event": "authentication.local.attempted", "identifier_provided": bool(login), "password_provided": bool(password), }, ) if not login or not password: logger.debug( "Local authentication skipped", extra={ "event": "authentication.local.skipped", "reason": "missing_credentials", }, ) return None user: Optional[Employee] = None # 1) Email if _looks_like_email(login): user = Employee.objects.filter(email__iexact=login).first() logger.debug( "Local authentication lookup completed", extra={ "event": "authentication.local.lookup", "identifier_type": "email", "user_found": bool(user), }, ) else: # 2) Phone if PHONE_FIELD: normalized = _normalize_phone(login) # сначала пробуем нормализованный E.164 if normalized: user = Employee.objects.filter(**{PHONE_FIELD: normalized}).first() # если не нашли — пробуем как ввёл пользователь # (на случай старых # данных) if not user: raw = str(login).strip() user = Employee.objects.filter(**{PHONE_FIELD: raw}).first() logger.debug( "Local authentication lookup completed", extra={ "event": "authentication.local.lookup", "identifier_type": "phone", "user_found": bool(user), }, ) if not user: logger.info( "Local authentication denied", extra={ "event": "authentication.local.denied", "reason": "user_not_found", }, ) return None logger.debug( "Local authentication candidate found", extra={ "event": "authentication.local.candidate_found", "user_id": user.id, "is_active": user.is_active, "has_usable_password": user.has_usable_password(), }, ) # допускаем логин только активных (email подтверждён → is_active=True) if not user.is_active: logger.info( "Local authentication denied", extra={ "event": "authentication.local.denied", "user_id": user.id, "reason": "inactive_user", }, ) return None password_valid = user.check_password(password) if password_valid: logger.info( "Local authentication succeeded", extra={ "event": "authentication.local.succeeded", "user_id": user.id, }, ) return user logger.warning( "Local authentication denied", extra={ "event": "authentication.local.denied", "user_id": user.id, "reason": "invalid_password", }, ) return None class LDAP3Backend(ModelBackend): """LDAP-аутентификация через ldap3 с логином по email/телефону/uid. Особенности: * Поиск DN сервисной учёткой (service bind), затем проверка пароля user bind. * Автосоздание локального пользователя (опционально). * Лёгкая синхронизация профиля (имя/фамилия/email/телефон/DN/GUID). Безопасность: * Пароли не логируются. * Рекомендуется LDAPS/StartTLS с проверкой сертификата. """ def authenticate( self, request, username: Optional[str] = None, password: Optional[str] = None, **kwargs: Any, ): """Аутентифицирует существующего локального пользователя через LDAP/AD.""" identifier = self._authentication_identifier(username, kwargs) ldap_enabled = getattr(settings, "LDAP_ENABLED", False) logger.debug( "LDAP authentication attempt received", extra={ "event": "authentication.ldap.attempted", "identifier_provided": bool(identifier), "password_provided": bool(password), "ldap_enabled": ldap_enabled, }, ) if not identifier or not password or not ldap_enabled: logger.debug( "LDAP authentication skipped because prerequisites are missing", extra={ "event": "authentication.ldap.skipped", "identifier_provided": bool(identifier), "password_provided": bool(password), "ldap_enabled": ldap_enabled, "reason": "prerequisite_missing", }, ) return None search_result = self._search_ldap_user(identifier) if search_result is None: return None server, entry, user_dn, phone_attrs = search_result if not self._validate_ldap_password(server, user_dn, password): return None profile = self._read_ldap_profile(entry, phone_attrs) user = self._find_local_ldap_user(profile["email"], profile["phone_e164"]) if user is None: logger.info( "LDAP authentication denied because no local account exists", extra={ "event": "authentication.ldap.denied", "reason": "local_account_missing", "auto_create_enabled": False, }, ) return None if not self.user_can_authenticate(user): logger.info( "LDAP authentication denied by local account policy", extra={ "event": "authentication.ldap.denied", "user_id": getattr(user, "id", None), "reason": "local_account_policy", }, ) return None self._sync_local_ldap_profile(user, entry, user_dn, profile) if not self.user_can_authenticate(user): logger.info( "LDAP authentication denied by local policy after profile sync", extra={ "event": "authentication.ldap.denied", "user_id": getattr(user, "id", None), "reason": "post_sync_local_account_policy", }, ) return None logger.info( "LDAP authentication succeeded", extra={ "event": "authentication.ldap.succeeded", "user_id": getattr(user, "id", None), }, ) return user @staticmethod def _authentication_identifier(username: Optional[str], kwargs: dict) -> str: return ( (username or "") or (kwargs.get("email") or "") or (kwargs.get("login") or "") or (kwargs.get(getattr(UserModel, "USERNAME_FIELD", "email")) or "") ).strip() def _search_ldap_user( self, identifier: str ) -> tuple[Server, Any, str, tuple[str, ...]] | None: uri = getattr(settings, "LDAP_URI", "") user_base = getattr(settings, "LDAP_USERS_AUTH", None) or getattr( settings, "LDAP_USER_BASE", None ) if not uri or not user_base: logger.warning( "LDAP authentication unavailable because configuration is incomplete", extra={ "event": "authentication.ldap.configuration_incomplete", "uri_configured": bool(uri), "user_base_configured": bool(user_base), }, ) return None phone_attrs = tuple( getattr(settings, "LDAP_PHONE_ATTRS", ("mobile", "telephoneNumber")) ) ldap_filter = self._authentication_filter(identifier, phone_attrs) server = Server(uri) connection: Optional[Connection] = None try: connection = self._service_connection(server) ok = connection.search( search_base=user_base, search_filter=ldap_filter, search_scope=SUBTREE, attributes=[ getattr(settings, "LDAP_ATTR_GIVENNAME", "givenName"), getattr(settings, "LDAP_ATTR_SN", "sn"), getattr(settings, "LDAP_ATTR_MAIL", "mail"), "telephoneNumber", "mobile", "sAMAccountName", "userPrincipalName", "userAccountControl", "objectGUID", "displayName", ], size_limit=1, ) entries_count = len(connection.entries) logger.debug( "LDAP authentication search completed", extra={ "event": "authentication.ldap.search_completed", "search_succeeded": ok, "entry_count": entries_count, }, ) if not ok or not connection.entries: logger.info( "LDAP authentication denied because directory entry was not found", extra={ "event": "authentication.ldap.denied", "reason": "entry_not_found", "search_succeeded": ok, }, ) return None entry = connection.entries[0] user_dn = str(entry.entry_dn) logger.debug( "LDAP user entry found", extra={"event": "authentication.ldap.entry_found"}, ) return server, entry, user_dn, phone_attrs except (LDAPSocketOpenError, LDAPBindError) as error: logger.error( "LDAP service bind or search failed", extra={ "event": "authentication.ldap.service_failed", "error_type": type(error).__name__, }, ) return None except Exception as error: logger.exception( "Unexpected LDAP service error", extra={ "event": "authentication.ldap.service_failed", "error_type": type(error).__name__, }, ) return None finally: self._close_service_connection(connection) @staticmethod def _service_connection(server: Server) -> Connection: logger.debug( "LDAP service bind started", extra={"event": "authentication.ldap.service_bind_started"}, ) bind_dn = getattr(settings, "LDAP_BIND_DN", "") bind_password = getattr(settings, "LDAP_BIND_PASSWORD", "") if bind_dn and bind_password: connection = Connection( server, user=bind_dn, password=bind_password, auto_bind=True, receive_timeout=30, ) bind_mode = "configured_dn" else: connection = Connection(server, auto_bind=True, receive_timeout=30) bind_mode = "anonymous_or_simple" logger.debug( "LDAP service bind completed", extra={ "event": "authentication.ldap.service_bind_completed", "bind_mode": bind_mode, }, ) return connection @staticmethod def _close_service_connection(connection: Optional[Connection]) -> None: if connection is None: return try: connection.unbind() logger.debug( "LDAP service connection closed", extra={"event": "authentication.ldap.service_connection_closed"}, ) except Exception as error: logger.warning( "LDAP service unbind failed", extra={ "event": "authentication.ldap.unbind_failed", "error_type": type(error).__name__, }, ) @staticmethod def _authentication_filter(identifier: str, phone_attrs: tuple[str, ...]) -> str: base_filter = getattr( settings, "LDAP_OBJECT_FILTER", "(&(objectCategory=person)(objectClass=user))", ) configured_filter = getattr( settings, "LDAP_USER_FILTER", "(|(uid={username})(mail={username}))" ) if "{username}" in configured_filter: id_filter = configured_filter.format(username=esc_filter(identifier)) filter_type = "configured" else: base_filter = configured_filter attrs_setting = ( "LDAP_LOGIN_MAIL_ATTRS" if _looks_like_email(identifier) else "LDAP_LOGIN_UID_ATTRS" ) default_attrs = ( ("mail", "userPrincipalName") if _looks_like_email(identifier) else ("uid", "sAMAccountName") ) identifier_attrs = getattr(settings, attrs_setting, default_attrs) id_filter = ( "(|" + "".join( f"({attr}={esc_filter(identifier)})" for attr in identifier_attrs ) + ")" ) filter_type = "default" logger.debug( "LDAP identifier filter selected", extra={ "event": "authentication.ldap.filter_selected", "filter_type": filter_type, }, ) phone = _normalize_phone(identifier) if not phone: return f"(&{base_filter}{id_filter})" phone_filter = ( "(|" + "".join(f"({attr}={esc_filter(phone)})" for attr in phone_attrs) + ")" ) logger.debug( "LDAP phone filter enabled", extra={"event": "authentication.ldap.phone_filter_enabled"}, ) return f"(&{base_filter}(|{id_filter}{phone_filter}))" @staticmethod def _validate_ldap_password(server: Server, user_dn: str, password: str) -> bool: try: logger.debug( "LDAP user bind started", extra={"event": "authentication.ldap.user_bind_started"}, ) connection = Connection( server, user=user_dn, password=password, auto_bind=True, receive_timeout=30, ) connection.unbind() logger.debug( "LDAP credentials validated", extra={"event": "authentication.ldap.credentials_validated"}, ) return True except LDAPBindError: logger.info( "LDAP authentication denied", extra={ "event": "authentication.ldap.denied", "reason": "invalid_credentials", }, ) return False except Exception as error: logger.error( "LDAP user bind failed", extra={ "event": "authentication.ldap.user_bind_failed", "error_type": type(error).__name__, }, ) return False def _read_ldap_profile( self, entry: Any, phone_attrs: tuple[str, ...] ) -> dict[str, Optional[str]]: first_name = ( self._attr(entry, getattr(settings, "LDAP_ATTR_GIVENNAME", "givenName")) or "" ) last_name = self._attr(entry, getattr(settings, "LDAP_ATTR_SN", "sn")) or "" email = ( self._attr(entry, getattr(settings, "LDAP_ATTR_MAIL", "mail")) or "" ).lower() ldap_username = (self._attr(entry, "sAMAccountName") or "").strip() phone_e164 = None for attr in ( *phone_attrs, getattr(settings, "LDAP_ATTR_PHONE", "telephoneNumber"), ): value = self._attr(entry, attr) if value and (phone_e164 := _normalize_phone(value)): break logger.debug( "LDAP attributes read", extra={ "event": "authentication.ldap.attributes_read", "has_first_name": bool(first_name), "has_last_name": bool(last_name), "has_email": bool(email), "has_username": bool(ldap_username), "phone_normalized": bool(phone_e164), }, ) return { "first_name": first_name, "last_name": last_name, "email": email, "ldap_username": ldap_username, "phone_e164": phone_e164, } @staticmethod def _find_local_ldap_user(email: str, phone_e164: Optional[str]): user = None if email and _model_has_field(UserModel, "email"): user = UserModel.objects.filter(email__iexact=email).first() if user is None and PHONE_FIELD and phone_e164: user = UserModel.objects.filter(**{PHONE_FIELD: phone_e164}).first() return user def _sync_local_ldap_profile( self, user, entry: Any, user_dn: str, profile: dict[str, Optional[str]] ) -> None: changed: List[str] = [] values = { "username": profile["ldap_username"], "first_name": profile["first_name"], "last_name": profile["last_name"], "email": profile["email"], } for field_name, value in values.items(): if not value or not _model_has_field(UserModel, field_name): continue current = getattr(user, field_name, "") or "" if field_name == "email": current = current.lower() if current != value: setattr(user, field_name, value) changed.append(field_name) if ( profile["email"] and _model_has_field(UserModel, "email_verified") and not getattr(user, "email_verified", False) ): user.email_verified = True changed.append("email_verified") for field_name in ("email_activation_code", "email_activation_sent_at"): if _model_has_field(UserModel, field_name): setattr(user, field_name, None) changed.append(field_name) phone_e164 = profile["phone_e164"] if ( PHONE_FIELD and phone_e164 and getattr(user, PHONE_FIELD, None) != phone_e164 ): setattr(user, PHONE_FIELD, phone_e164) changed.append(PHONE_FIELD) if hasattr(user, "ldap_dn") and getattr(user, "ldap_dn") != user_dn: user.ldap_dn = user_dn changed.append("ldap_dn") guid = self._attr(entry, "objectGUID") if ( hasattr(user, "ldap_guid") and guid and getattr(user, "ldap_guid") != str(guid) ): user.ldap_guid = str(guid) changed.append("ldap_guid") if hasattr(user, "is_ldap_managed") and not getattr( user, "is_ldap_managed", False ): user.is_ldap_managed = True changed.append("is_ldap_managed") if hasattr(user, "last_ldap_sync_at"): user.last_ldap_sync_at = timezone.now() changed.append("last_ldap_sync_at") if changed: updated_fields = sorted(set(changed)) user.save(update_fields=updated_fields) logger.debug( "Local LDAP user profile updated", extra={ "event": "authentication.ldap.profile_updated", "user_id": getattr(user, "id", None), "updated_fields": updated_fields, "updated_field_count": len(updated_fields), }, ) else: logger.debug( "Local LDAP user profile is current", extra={ "event": "authentication.ldap.profile_unchanged", "user_id": getattr(user, "id", None), }, ) @staticmethod def _attr(entry: Any, name: str) -> Optional[str]: """Возвращает первое строковое значение LDAP-атрибута. Args: entry: ldap3 Entry. name: Имя атрибута (например, 'mail'). Returns: str | None: Нормализованное строковое значение или None. Raises: None: Ошибки приводят к возврату None и логируются на debug. """ try: val = getattr(entry, name, None) raw = getattr(val, "value", None) if raw is None: return None if isinstance(raw, (list, tuple)): return str(raw[0]).strip() if raw else None return str(raw).strip() except Exception as error: logger.debug( "LDAP attribute read failed", extra={ "event": "authentication.ldap.attribute_failed", "attribute": name, "error_type": type(error).__name__, }, ) return None class SuperuserOnlyBackend(ModelBackend): """Локальный фолбэк-бэкенд: пускает только суперюзера. Используется как «break-glass» при отказе LDAP. Любые не-суперюзеры не проходят локальную аутентификацию, даже при корректном пароле. Это устраняет локальный фолбэк для рядовых сотрудников (все они идут через LDAP), не ломая админ-доступ на случай недоступности каталога. """ def authenticate( self, request: HttpRequest | None, username: str | None = None, password: str | None = None, **kwargs, ) -> Optional[UserModel]: """Аутентифицирует только суперюзера через локальную БД. Args: request: HttpRequest или None. username: Идентификатор (для вашей модели это email). password: Пароль. Returns: Optional[UserModel]: Пользователь, если это суперюзер и пароль верен; иначе None. Raises: Ничего не выбрасывает наружу — при неуспехе возвращает None. """ logger.debug( "Local break-glass authentication attempt received", extra={ "event": "authentication.break_glass.attempted", "identifier_provided": bool(username), "password_provided": bool(password), }, ) user = super().authenticate( request, username=username, password=password, **kwargs ) logger.debug( "Local break-glass credential check completed", extra={ "event": "authentication.break_glass.credentials_checked", "user_found": bool(user), }, ) if user: is_super = getattr(user, "is_superuser", False) logger.debug( "Local break-glass candidate evaluated", extra={ "event": "authentication.break_glass.evaluated", "user_id": getattr(user, "id", None), "is_superuser": is_super, }, ) if is_super: logger.info( "Local break-glass authentication succeeded", extra={ "event": "authentication.break_glass.succeeded", "user_id": getattr(user, "id", None), }, ) return user else: logger.debug( "Local break-glass authentication denied", extra={ "event": "authentication.break_glass.denied", "user_id": getattr(user, "id", None), "reason": "not_superuser", }, ) return None class PositionRoleBackend(ModelBackend): def authenticate(self, request, username=None, password=None, **kwargs): """Этот бэкенд не участвует в аутентификации, только в расчёте прав.""" return None def get_all_permissions(self, user_obj, obj=None): perms = super().get_all_permissions(user_obj, obj) # Проверяем, что пользователь authenticated и active if not getattr(user_obj, "is_authenticated", False) or obj is not None: return perms if not getattr(user_obj, "is_active", True): return perms # 1) Глобальные права от должности — оставить position_ids = set() pos = getattr(user_obj, "position", None) if pos: position_ids.add(pos.pk) if getattr(user_obj, "pk", None): position_ids.update( user_obj.additional_positions.values_list("pk", flat=True) ) if position_ids: pos_perms = ( Permission.objects.filter(group__positions__id__in=position_ids) .select_related("content_type") .distinct() ) perms |= {f"{p.content_type.app_label}.{p.codename}" for p in pos_perms} # 2) ⚠️ НЕ добавляем сюда пермишены из DepartmentRole. # Они должны проверяться только на уровне конкретного Department в # has_perm(..., obj=dept) return perms def has_perm(self, user_obj, perm, obj=None): # стандартная ветка + суперюзер/группы/пользовательские if super().has_perm(user_obj, perm, obj=obj): return True if not getattr(user_obj, "is_authenticated", False): return False if obj is None: # добавить глобальные (от должности) return perm in self.get_all_permissions(user_obj) # Объектные права на Department — как у вас if isinstance(obj, Department): if obj.head_id == user_obj.id: return True link = ( EmployeeDepartment.objects.select_related("role") .filter(employee=user_obj, department=obj, is_active=True) .first() ) if not (link and link.role_id): return False app_label, sep, codename = perm.partition(".") if not sep: return False return link.role.scoped_permissions.filter( content_type__app_label=app_label, codename=codename, ).exists() return False