Corp/backend/eusrr_backend/auth_backends.py

# eusrr_backend/auth_backends.py
from __future__ import annotations

import logging
import re
from typing import Any, List, Optional, Tuple

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 employees.ldap.utils.text_utils import esc_filter
from employees.models import Department, Employee, EmployeeDepartment
from employees.utils import _normalize_phone
from ldap3 import SUBTREE, Connection, Server
from ldap3.core.exceptions import LDAPBindError, LDAPSocketOpenError

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(
            "EmailOrPhoneBackend.authenticate: login=%s, password_provided=%s",
            bool(login), bool(password)
        )
        
        if not login or not password:
            logger.debug("EmailOrPhoneBackend: 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("EmailOrPhoneBackend: email lookup=%s, found=%s", login, 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("EmailOrPhoneBackend: phone lookup, found=%s", bool(user))

        if not user:
            logger.debug("EmailOrPhoneBackend: user not found")
            return None

        logger.debug(
            "EmailOrPhoneBackend: found user id=%s email=%s is_active=%s has_usable_password=%s",
            user.id, user.email, user.is_active, user.has_usable_password()
        )

        # допускаем логин только активных (email подтверждён → is_active=True)
        if not user.is_active:
            logger.debug("EmailOrPhoneBackend: user not active")
            return None

        password_valid = user.check_password(password)
        logger.debug("EmailOrPhoneBackend: password_valid=%s", password_valid)
        
        if password_valid:
            logger.info("EmailOrPhoneBackend: AUTH SUCCESS for %s", user.email)
            return user
        
        logger.warning(
            "EmailOrPhoneBackend: password check FAILED for user=%s (id=%s). "
            "Password hash starts with: %s...",
            user.email, user.id, user.password[:20] if user.password else "NO_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.

        Логика:
            1) Валидирует входные данные и настройки.
            2) Service bind -> поиск записи пользователя (DN) по идентификатору/телефону.
            3) User bind -> проверка пароля.
            4) Поиск/создание локального пользователя и лёгкая синхронизация профиля.

        Args:
            request: Django HttpRequest или None.
            username: Идентификатор (может быть email/телефон/uid).
            password: Пароль пользователя.
            **kwargs: Могут содержать email/login/USERNAME_FIELD.

        Returns:
            User | None: Аутентифицированный пользователь или None при неудаче.

        Raises:
            None: Метод перехватывает ошибки LDAP/сети и возвращает None при сбое.
        """
        identifier = (
            (username or "")
            or (kwargs.get("email") or "")
            or (kwargs.get("login") or "")
            or (kwargs.get(getattr(UserModel, "USERNAME_FIELD", "email")) or "")
        ).strip()

        logger.debug(
            "LDAP3Backend.authenticate called: identifier_provided=%s, password_provided=%s, LDAP_ENABLED=%s",
            bool(identifier),
            bool(password),
            getattr(settings, "LDAP_ENABLED", False),
        )

        if (
            not identifier
            or not password
            or not getattr(settings, "LDAP_ENABLED", False)
        ):
            logger.debug("Fast-exit: identifier/password missing or LDAP disabled")
            return None

        uri = getattr(settings, "LDAP_URI", "")
        user_base = getattr(settings, "LDAP_USERS_AUTH", None) or getattr(
            settings, "LDAP_USER_BASE", None
        )
        bind_dn = getattr(settings, "LDAP_BIND_DN", "")
        bind_pw = getattr(settings, "LDAP_BIND_PASSWORD", "")

        base_tpl = getattr(
            settings,
            "LDAP_USER_FILTER",
            "(|(uid={username})(mail={username}))",
        )
        phone_attrs: Tuple[str, ...] = tuple(
            getattr(settings, "LDAP_PHONE_ATTRS", ("mobile", "telephoneNumber"))
        )

        if not uri or not user_base:
            logger.warning("Fast-exit: LDAP_URI or LDAP_USERS_AUTH not configured")
            return None

        raw = identifier
        is_email = _looks_like_email(raw)
        e164 = _normalize_phone(raw)
        logger.debug(
            "Identifier parse: is_email=%s, normalized_phone=%s", is_email, e164
        )

        # --- Service bind & search ---
        # Важно: используем get_info=NONE для скорости/минимума утечек.
        server = Server(uri)  # get_info=NONE по умолчанию
        svc_conn: Optional[Connection] = None
        try:
            logger.debug("Service bind: connecting...")
            if bind_dn and bind_pw:
                svc_conn = Connection(
                    server,
                    user=bind_dn,
                    password=bind_pw,
                    auto_bind=True,
                    receive_timeout=30,
                )
                logger.debug("Service bind OK (bind_dn)")
            else:
                svc_conn = Connection(server, auto_bind=True, receive_timeout=30)
                logger.debug("Service bind OK (anonymous/simple)")

            # Базовый объектный фильтр AD
            base_object_filter = getattr(
                settings,
                "LDAP_OBJECT_FILTER",
                "(&(objectCategory=person)(objectClass=user))",
            )

            # Фильтр по идентификатору
            if "{username}" in base_tpl:
                id_filter = base_tpl.format(username=esc_filter(raw))
                logger.debug("Using configured ID filter: %s", id_filter)
            else:
                base_object_filter = base_tpl  # админ дал базовый объектный фильтр
                if is_email:
                    id_attrs = getattr(
                        settings,
                        "LDAP_LOGIN_MAIL_ATTRS",
                        ("mail", "userPrincipalName"),
                    )
                else:
                    id_attrs = getattr(
                        settings,
                        "LDAP_LOGIN_UID_ATTRS",
                        ("uid", "sAMAccountName"),
                    )
                id_filter = (
                    "(|" + "".join(f"({a}={esc_filter(raw)})" for a in id_attrs) + ")"
                )
                logger.debug("Built default ID filter: %s", id_filter)

            phone_filter = ""
            if e164:
                phone_filter = (
                    "(|"
                    + "".join(f"({attr}={esc_filter(e164)})" for attr in phone_attrs)
                    + ")"
                )
                logger.debug("Built phone filter: %s", phone_filter)

            # (& base_object (| id_filter phone_filter?))
            if phone_filter:
                ldap_filter = f"(&{base_object_filter}(|{id_filter}{phone_filter}))"
            else:
                ldap_filter = f"(&{base_object_filter}{id_filter})"
            logger.debug("FINAL LDAP search filter=%s", ldap_filter)

            # AD атрибуты для поиска
            ok = svc_conn.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(svc_conn.entries) if svc_conn else 0
            logger.debug("Search ok=%s entries=%s", ok, entries_count)
            if not ok or not svc_conn.entries:
                logger.info("No entries found -> auth failed")
                return None

            entry = svc_conn.entries[0]
            user_dn: str = str(entry.entry_dn)
            logger.debug("Found DN: %s", user_dn)

        except (LDAPSocketOpenError, LDAPBindError) as e:
            logger.error("Service bind/search LDAP error: %s: %s", type(e).__name__, e)
            return None
        except Exception as e:
            logger.exception("Service bind/search unexpected error: %s", e)
            return None
        finally:
            try:
                if svc_conn is not None:
                    svc_conn.unbind()
                    logger.debug("Service connection unbound")
            except Exception as e:
                logger.warning("Service unbind error: %s", e)

        # --- User bind (password check) ---
        try:
            logger.debug("User bind check for DN=%s ...", user_dn)
            user_conn = Connection(
                server,
                user=user_dn,
                password=password,
                auto_bind=True,
                receive_timeout=30,
            )
            user_conn.unbind()
            logger.debug("User bind OK (password valid)")
        except LDAPBindError as e:
            logger.info("User bind FAILED (invalid credentials): %s", e)
            return None
        except Exception as e:
            logger.error("User bind FAILED: %s: %s", type(e).__name__, e)
            return None

        # --- Read attributes ---
        get = self._attr
        first_name = (
            get(entry, getattr(settings, "LDAP_ATTR_GIVENNAME", "givenName")) or ""
        )
        last_name = get(entry, getattr(settings, "LDAP_ATTR_SN", "sn")) or ""
        email = (get(entry, getattr(settings, "LDAP_ATTR_MAIL", "mail")) or "").lower()
        logger.debug(
            "Attrs: first_name=%r last_name=%r email=%r", first_name, last_name, email
        )

        phone_e164: Optional[str] = None
        for attr in list(phone_attrs) + [
            getattr(settings, "LDAP_ATTR_PHONE", "telephoneNumber")
        ]:
            val = get(entry, attr)
            if val:
                phone_e164 = _normalize_phone(val)
                logger.debug(
                    "Phone candidate attr=%s raw=%r -> norm=%r", attr, val, phone_e164
                )
                if phone_e164:
                    break

        # --- Find or create local user ---
        user = None
        if email and _model_has_field(UserModel, "email"):
            logger.debug("Lookup local user by email=%r", email)
            user = UserModel.objects.filter(email__iexact=email).first()
        if user is None and PHONE_FIELD and phone_e164:
            logger.debug("Lookup local user by %s=%r", PHONE_FIELD, phone_e164)
            user = UserModel.objects.filter(**{PHONE_FIELD: phone_e164}).first()

        if user and not self.user_can_authenticate(user):
            logger.info("user_can_authenticate=False -> auth denied")
            return None

        # Auto-create отключен для LDAP3Backend (создание через API)
        if user is None:
            logger.info("User not found in DB, auto-create disabled")
            return None

        # --- Light profile sync ---
        if user:
            changed: List[str] = []

            if (
                _model_has_field(UserModel, "first_name")
                and user.first_name != first_name
            ):
                user.first_name = first_name
                changed.append("first_name")
            if _model_has_field(UserModel, "last_name") and user.last_name != last_name:
                user.last_name = last_name
                changed.append("last_name")
            if (
                email
                and _model_has_field(UserModel, "email")
                and (user.email or "").lower() != email
            ):
                user.email = email
                changed.append("email")
            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:
                setattr(user, "ldap_dn", user_dn)
                changed.append("ldap_dn")

            guid = get(entry, "objectGUID")
            if (
                hasattr(user, "ldap_guid")
                and guid
                and getattr(user, "ldap_guid") != str(guid)
            ):
                # При необходимости замените str(guid) на корректную конверсию GUID.
                setattr(user, "ldap_guid", str(guid))
                changed.append("ldap_guid")

            if hasattr(user, "is_ldap_managed") and not getattr(
                user, "is_ldap_managed", False
            ):
                setattr(user, "is_ldap_managed", True)
                changed.append("is_ldap_managed")

            if hasattr(user, "last_ldap_sync_at"):
                setattr(user, "last_ldap_sync_at", timezone.now())
                changed.append("last_ldap_sync_at")

            if changed:
                user.save(update_fields=list(set(changed)))
                logger.debug("Local user updated fields: %s", sorted(set(changed)))
            else:
                logger.debug("Local user has no changes to save")

            if not self.user_can_authenticate(user):
                logger.info("user_can_authenticate=False after sync -> auth denied")
                return None

            logger.info(
                "AUTH SUCCESS: user_id=%s email=%s",
                getattr(user, "id", None),
                getattr(user, "email", None),
            )
            return user

        logger.info(
            "AUTH FAILED: no local user (auto_create disabled or constraints not met)"
        )
        return 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 e:
            logger.debug("LDAP3Backend._attr('%s') error: %s", name, e)
            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(
            "SuperuserOnlyBackend.authenticate: username=%s, password_provided=%s",
            bool(username), bool(password)
        )
        
        user = super().authenticate(
            request, username=username, password=password, **kwargs
        )
        
        logger.debug("SuperuserOnlyBackend: parent auth returned user=%s", bool(user))
        
        if user:
            is_super = getattr(user, "is_superuser", False)
            logger.debug(
                "SuperuserOnlyBackend: user=%s, is_superuser=%s",
                getattr(user, "email", "?"), is_super
            )
            if is_super:
                logger.info("SuperuserOnlyBackend: AUTH SUCCESS for superuser %s", user.email)
                return user
            else:
                logger.debug("SuperuserOnlyBackend: user is not superuser, rejecting")
        
        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) Глобальные права от должности — оставить
        pos = getattr(user_obj, "position", None)
        if pos:
            pos_perms = (
                Permission.objects.filter(group__positions=pos)
                .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