822 lines
30 KiB
Python
822 lines
30 KiB
Python
# 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
|