138 lines
4.1 KiB
Python
138 lines
4.1 KiB
Python
"""Сервисы для directory identity lookup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from django.conf import settings
|
|
|
|
from employees.models import LdapSyncState
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DirectoryLoginResult:
|
|
username: str | None
|
|
source: str
|
|
is_cached: bool
|
|
is_ldap_managed: bool
|
|
|
|
|
|
def _find_unique_by_email(ldap_user_model, email: str):
|
|
if not email:
|
|
return None
|
|
for lookup in ("mail", "user_principal_name"):
|
|
candidates = list(ldap_user_model.objects.filter(**{lookup: email}))
|
|
if len(candidates) == 1:
|
|
return candidates[0]
|
|
return None
|
|
|
|
|
|
def _find_by_sync_state(ldap_user_model, employee_number: str):
|
|
sync_state = LdapSyncState.objects.filter(
|
|
model="employee",
|
|
object_pk=employee_number,
|
|
).first()
|
|
if not sync_state or not sync_state.ldap_dn:
|
|
return None
|
|
try:
|
|
return ldap_user_model.objects.get(dn=sync_state.ldap_dn)
|
|
except ldap_user_model.DoesNotExist:
|
|
return None
|
|
|
|
|
|
def _normalized_ldap_email(ldap_user, attribute: str) -> str:
|
|
return (getattr(ldap_user, attribute, "") or "").strip().lower()
|
|
|
|
|
|
def _matching_email_candidate(candidates, email: str):
|
|
for candidate in candidates:
|
|
if email in {
|
|
_normalized_ldap_email(candidate, "mail"),
|
|
_normalized_ldap_email(candidate, "user_principal_name"),
|
|
}:
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def _candidate_sort_key(candidate, email: str):
|
|
return (
|
|
0 if _normalized_ldap_email(candidate, "mail") == email else 1,
|
|
0 if _normalized_ldap_email(candidate, "user_principal_name") == email else 1,
|
|
str(getattr(candidate, "dn", "")).lower(),
|
|
)
|
|
|
|
|
|
def _find_ldap_user(employee):
|
|
from employees.ldap.orm_models import LdapUser
|
|
|
|
employee_number = str(employee.pk)
|
|
employee_email = (employee.email or "").strip().lower()
|
|
sync_state_match = _find_by_sync_state(LdapUser, employee_number)
|
|
if sync_state_match:
|
|
return sync_state_match
|
|
|
|
if not employee.is_ldap_managed:
|
|
email_match = _find_unique_by_email(LdapUser, employee_email)
|
|
if email_match is not None:
|
|
return email_match
|
|
|
|
candidates = list(LdapUser.objects.filter(employee_number=employee_number))
|
|
if not candidates:
|
|
return _find_unique_by_email(LdapUser, employee_email)
|
|
if len(candidates) == 1:
|
|
return candidates[0]
|
|
|
|
email_match = _matching_email_candidate(candidates, employee_email)
|
|
if email_match:
|
|
return email_match
|
|
candidates.sort(key=lambda item: _candidate_sort_key(item, employee_email))
|
|
return candidates[0]
|
|
|
|
|
|
def resolve_directory_login(employee, *, force_refresh: bool = False):
|
|
cached_username = (employee.username or "").strip()
|
|
if cached_username and not force_refresh:
|
|
return DirectoryLoginResult(
|
|
username=cached_username,
|
|
source="db",
|
|
is_cached=True,
|
|
is_ldap_managed=employee.is_ldap_managed,
|
|
)
|
|
|
|
if not employee.is_ldap_managed:
|
|
return DirectoryLoginResult(
|
|
username=cached_username or None,
|
|
source="db" if cached_username else "none",
|
|
is_cached=bool(cached_username),
|
|
is_ldap_managed=False,
|
|
)
|
|
|
|
if not getattr(settings, "LDAP_ENABLED", False):
|
|
return DirectoryLoginResult(
|
|
username=cached_username or None,
|
|
source="db" if cached_username else "none",
|
|
is_cached=bool(cached_username),
|
|
is_ldap_managed=True,
|
|
)
|
|
|
|
ldap_user = _find_ldap_user(employee)
|
|
if not ldap_user or not (ldap_user.sam_account_name or "").strip():
|
|
return DirectoryLoginResult(
|
|
username=None,
|
|
source="ldap_not_found",
|
|
is_cached=False,
|
|
is_ldap_managed=True,
|
|
)
|
|
|
|
username = ldap_user.sam_account_name.strip()
|
|
if employee.username != username:
|
|
employee.username = username
|
|
employee.save(update_fields=["username"])
|
|
|
|
return DirectoryLoginResult(
|
|
username=username,
|
|
source="ldap",
|
|
is_cached=False,
|
|
is_ldap_managed=True,
|
|
)
|