Add management command to import user by username

aleksis/apps/ldap/management/commands/import_user_by_name.py

+from django.core.management.base import BaseCommand
+
+from ...util.ldap_sync import import_user_by_username
+
+
+class Command(BaseCommand):
+    def add_arguments(self, parser):
+        parser.add_argument("username", type=str)
+
+    def handle(self, *args, **options):
+        username = options["username"]
+        import_user_by_username(username)

aleksis/apps/ldap/util/ldap_sync.py

@@ -397,3 +397,96 @@ def mass_ldap_import():
         person.save()
 
     logger.info("Commiting transaction; this can take some time.")
+
+
+@transaction.atomic
+def import_user_by_username(username):
+    """Import user from ldap by username."""
+    from django_auth_ldap.backend import LDAPBackend, _LDAPUser  # noqa
+
+    Person = apps.get_model("core", "Person")
+
+    backend = LDAPBackend()
+    connection = _LDAPUser(backend, "").connection
+
+    uid_field = re.search(r"([a-zA-Z]+)=%\(user\)s", backend.settings.USER_SEARCH.filterstr).group(
+        1
+    )
+    # Synchronise all groups first
+    if get_site_preferences()["ldap__enable_group_sync"]:
+        member_attr = getattr(backend.settings.GROUP_TYPE, "member_attr", "memberUid")
+        owner_attr = get_site_preferences()["ldap__group_sync_owner_attr"]
+
+        ldap_groups = backend.settings.GROUP_SEARCH.execute(connection, {member_attr: username, owner_attr: username}, escape=False)
+        group_objects = ldap_sync_from_groups(ldap_groups)
+
+        # Create lookup table as cache for later code
+        group_dict = {obj.ldap_dn: obj for obj in group_objects}
+
+    ldap_users = backend.settings.USER_SEARCH.execute(connection, {"user": username}, escape=False)
+    for dn, attrs in tqdm(ldap_users, desc="Sync. user infos", **TQDM_DEFAULTS):
+        uid = attrs[uid_field][0]
+
+        # Prepare an empty LDAPUser object with the target username
+        ldap_user = _LDAPUser(backend, username=uid)
+
+        # Get existing or new User object and pre-populate
+        user, created = backend.get_or_build_user(uid, ldap_user)
+        ldap_user._user = user
+        ldap_user._attrs = attrs
+        ldap_user._dn = dn
+        ldap_user._populate_user_from_attributes()
+        user.save()
+
+        if created or get_site_preferences()["ldap__sync_on_update"]:
+            try:
+                with transaction.atomic():
+                    person = ldap_sync_from_user(user, dn, attrs)
+            except Person.DoesNotExist:
+                logger.warn(f"No matching person for user {user.username}")
+                continue
+            except Person.MultipleObjectsReturned:
+                logger.error(f"More than one matching person for user {user.username}")
+                continue
+            except (DataError, IntegrityError, ValueError) as e:
+                logger.error(f"Data error while synchronising user {user.username}:\n{e}")
+                continue
+            else:
+                logger.info(f"Successfully imported user {uid}")
+
+        # Synchronise group memberships now
+        if get_site_preferences()["ldap__enable_group_sync"]:
+            for ldap_group in tqdm(
+                ldap_groups, desc="Sync. group members", total=len(group_objects), **TQDM_DEFAULTS,
+            ):
+                dn, attrs = ldap_group
+
+                if dn not in group_dict:
+                    logger.warning(f"Skip {dn} because there are no groups with this dn.")
+                    continue
+
+                group = group_dict[dn]
+
+                ldap_members = [_.lower() for _ in attrs[member_attr]] if member_attr in attrs else []
+
+                if member_attr.lower() == "memberuid":
+                    members = Person.objects.filter(user__username__in=ldap_members)
+                else:
+                    members = Person.objects.filter(ldap_dn__in=ldap_members)
+
+                if get_site_preferences()["ldap__group_sync_owner_attr"]:
+                    ldap_owners = [_.lower() for _ in attrs[owner_attr]] if owner_attr in attrs else []
+                    if get_site_preferences()["ldap__group_sync_owner_attr_type"] == "uid":
+                        owners = Person.objects.filter(user__username__in=ldap_owners)
+                    elif get_site_preferences()["ldap__group_sync_owner_attr_type"] == "dn":
+                        owners = Person.objects.filter(ldap_dn__in=ldap_owners)
+
+                group.members.set(members)
+                if get_site_preferences()["ldap__group_sync_owner_attr"]:
+                    group.owners.set(owners)
+                group.save()
+                logger.info(f"Set group members of group {group}")
+        Person.objects.get(user__username=username).auto_select_primary_group()
+        Person.objects.get(user__username=username).save()
+
+        logger.info("Commiting transaction; this can take some time.")