Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Handle existing LDAP users during Django sync

Browse Source
This commit is contained in:
lizerk 2026-05-14 15:06:47 +03:00
parent b14fd0f04a
commit da0cadc50c
4 changed files with 229 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
backend/api/v1/directory/services.py
View File

@ -20,6 +20,26 @@ def _find_ldap_user(employee):
from employees.ldap.orm_models import LdapUser
employee_number = str(employee.pk)
employee_email = (employee.email or "").strip().lower()
def find_by_email():
if not employee_email:
return None
email_candidates = list(
LdapUser.objects.filter(mail=employee_email)
)
if len(email_candidates) == 1:
return email_candidates[0]
upn_candidates = list(
LdapUser.objects.filter(user_principal_name=employee_email)
)
if len(upn_candidates) == 1:
return upn_candidates[0]
return None
sync_state = LdapSyncState.objects.filter(
model="employee", object_pk=employee_number
).first()
@ -29,15 +49,19 @@ def _find_ldap_user(employee):
except LdapUser.DoesNotExist:
pass
if not employee.is_ldap_managed:
email_match = find_by_email()
if email_match is not None:
return email_match
candidates = list(
LdapUser.objects.filter(employee_number=employee_number)
)
if not candidates:
return None
return find_by_email()
if len(candidates) == 1:
return candidates[0]
employee_email = (employee.email or "").strip().lower()
if employee_email:
for candidate in candidates:
if (getattr(candidate, "mail", "") or "").strip().lower() == employee_email:

85
backend/employees/admin.py
View File

@ -632,26 +632,29 @@ class EmployeeAdmin(DjangoUserAdmin):
}: is_ldap_managed={employee.is_ldap_managed}"
)
if employee.is_ldap_managed:
# Пробуем найти по employee_number
# Пробуем найти существующий объект даже для сотрудников,
# которые еще не помечены как LDAP-managed: при ручной
# синхронизации Django считается источником истины, поэтому
# найденную AD-запись нужно привязать и обновить, а не
# пытаться создать повторно.
logger.info(
f"[sync_from_django_to_ldap] Employee {
employee.pk
}: поиск в LDAP по sync-state/employeeNumber/email"
)
ldap_user = _resolve_ldap_user_for_employee(employee)
if ldap_user:
logger.info(
f"[sync_from_django_to_ldap] Employee {
employee.pk
}: поиск в LDAP по employee_number={employee.pk}"
}: найден LDAP user с DN={ldap_user.dn}"
)
else:
logger.info(
f"[sync_from_django_to_ldap] Employee {
employee.pk
}: LDAP пользователь не найден"
)
ldap_user = _resolve_ldap_user_for_employee(employee)
if ldap_user:
logger.info(
f"[sync_from_django_to_ldap] Employee {
employee.pk
}: найден LDAP user с DN={ldap_user.dn}"
)
else:
logger.info(
f"[sync_from_django_to_ldap] Employee {
employee.pk
}: не найден по employee_number"
)
if ldap_user:
# ОБНОВЛЕНИЕ существующего LDAP пользователя
@ -661,6 +664,56 @@ class EmployeeAdmin(DjangoUserAdmin):
"изменений"
)
try:
from django.utils import timezone
from ldap3 import MODIFY_REPLACE
from employees.ldap.infrastructure.connections import (
_ldap,
)
LdapSyncState.objects.update_or_create(
model="employee",
object_pk=str(employee.pk),
defaults={
"ldap_dn": ldap_user.dn,
"last_sync_dir": "django",
"last_django_modify_ts": timezone.now(),
},
)
current_employee_number = get_ldap_str(
getattr(ldap_user, "employee_number", "")
)
if current_employee_number != str(employee.pk):
with _ldap() as conn:
ok = conn.modify(
ldap_user.dn,
{
"employeeNumber": [
(
MODIFY_REPLACE,
[str(employee.pk)],
)
]
},
)
if not ok:
logger.warning(
"[sync_from_django_to_ldap] Employee "
"%s: не удалось записать "
"employeeNumber в LDAP: %s",
employee.pk,
conn.result,
)
else:
ldap_user.employee_number = str(
employee.pk
)
if not employee.is_ldap_managed:
employee.is_ldap_managed = True
employee._skip_ldap_sync = True
employee.save(update_fields=["is_ldap_managed"])
changes = {}
ldap_first = get_ldap_str(ldap_user.given_name)

28
backend/tests/api/v1/test_directory_login.py
View File

@ -271,3 +271,31 @@ def test_find_ldap_user_prefers_email_match_among_multiple_candidates(settings):
resolved = _find_ldap_user(user)
assert resolved is preferred
@pytest.mark.django_db
def test_find_ldap_user_falls_back_to_email_without_employee_number(settings):
settings.LDAP_ENABLED = True
user = make_user(
username="",
is_ldap_managed=False,
email="existing@example.com",
)
preferred = make_ldap_user(
dn="CN=Existing,OU=Users,OU=company,DC=robotail,DC=local",
sam_account_name="existing.login",
mail="existing@example.com",
user_principal_name="existing@robotail.local",
)
with patch(
"employees.ldap.orm_models.LdapUser.objects.filter"
) as filter_mock:
filter_mock.side_effect = [[], [preferred]]
from api.v1.directory.services import _find_ldap_user
resolved = _find_ldap_user(user)
assert resolved is preferred

107
backend/tests/manual/ldap_integration/test_entities_roundtrip.py
View File

@ -1,18 +1,33 @@
from __future__ import annotations
import uuid
import pytest
from django.contrib import admin
from django.contrib.auth import get_user_model
from django.conf import settings
from django.contrib.auth.models import Group
from django.test import RequestFactory
from rest_framework.test import APIClient
from employees.admin import EmployeeAdmin
from employees.ldap import (
DepartmentService,
LdapGroup,
LdapOrganizationalUnit,
LdapOrganizationalUnitGroup,
LdapUser,
_ldap,
)
from employees.ldap.repositories.ldap_repository import ensure_container_exists
from employees.ldap.services.constants import UserAccountControl
from employees.models import (
Department,
DepartmentRole,
EmployeeDepartment,
LdapSyncState,
Position,
)
from employees.models import Department, DepartmentRole, LdapSyncState, Position
"""
Живой LDAP integration-suite.
@ -35,6 +50,8 @@ TINY_PNG_BASE64 = (
"+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
)
User = get_user_model()
def _employee_dn(employee) -> str:
return LdapSyncState.objects.get(
@ -88,6 +105,94 @@ def test_employee_profile_patch_updates_live_ldap(
assert bool(ldap_user.thumbnail_photo) is True
def test_django_to_ldap_sync_reuses_existing_unlinked_user(
ensure_live_ldap,
ldap_cleanup,
unique_name,
):
dept_name = unique_name("qa-link-dept")
department = Department.objects.create(
name=dept_name,
description="Target department for existing LDAP user",
)
ldap_cleanup(_department_dn(department))
suffix = unique_name("existing")
email = f"{suffix}@example.com"
phone = f"+79992{uuid.uuid4().int % 10**6:06d}"
cn = f"Existing User {suffix}"
sam = suffix.replace("-", "")[:20]
user_base = settings.LDAP_USERS_BASE
existing_dn = f"CN={cn},{user_base}"
ldap_cleanup(existing_dn)
with _ldap() as conn:
ensure_container_exists(conn, user_base)
ok = conn.add(
existing_dn,
["top", "person", "organizationalPerson", "user"],
{
"cn": cn,
"sAMAccountName": sam,
"userPrincipalName": email,
"userAccountControl": UserAccountControl.DISABLED,
"givenName": "Existing",
"sn": f"User {suffix}",
"displayName": cn,
"mail": email,
"telephoneNumber": phone,
},
)
assert ok, conn.result
employee = User.objects.create_user(
email=email,
password="TestPass123!",
first_name="Existing",
last_name=f"User {suffix}",
phone_number=phone,
send_activation_email=False,
)
employee.email_verified = True
employee.is_active = True
employee.is_ldap_managed = False
employee.save(
update_fields=["email_verified", "is_active", "is_ldap_managed"]
)
link = EmployeeDepartment(
employee=employee,
department=department,
is_active=True,
)
link._skip_ldap_sync = True
link.save()
model_admin = EmployeeAdmin(User, admin.site)
model_admin.message_user = lambda *args, **kwargs: None
request = RequestFactory().post("/admin/employees/employee/")
model_admin.sync_from_django_to_ldap(
request,
User.objects.filter(pk=employee.pk),
)
employee.refresh_from_db()
sync_state = LdapSyncState.objects.get(
model="employee",
object_pk=str(employee.pk),
)
ldap_cleanup(sync_state.ldap_dn)
assert employee.is_ldap_managed is True
assert _norm_dn(sync_state.ldap_dn).endswith(_norm_dn(_department_dn(department)))
assert not LdapUser.objects.filter(dn=existing_dn).exists()
ldap_user = LdapUser.objects.get(dn=sync_state.ldap_dn)
assert ldap_user.mail == email
assert ldap_user.employee_number == str(employee.pk)
def test_department_api_roundtrip_syncs_ou_head_and_membership(
ensure_live_ldap,
superuser_client,