Merge branch 'hotfix' into 'master'

HOTFIX 2.1.2

See merge request arenadata/development/adcm!3812

Makefile

@@ -3,7 +3,7 @@ APP_IMAGE ?= hub.adsw.io/adcm/adcm
 APP_TAG ?= $(subst /,_,$(BRANCH_NAME))
 SELENOID_HOST ?= 10.92.2.65
 SELENOID_PORT ?= 4444
-ADCM_VERSION = "2.1.1"
+ADCM_VERSION = "2.1.2"
 
 .PHONY: help
 

python/api_v2/rbac/user/views.py

@@ -54,7 +54,11 @@
 
 class UserViewSet(PermissionListMixin, CamelCaseModelViewSet):
     queryset = (
-        User.objects.prefetch_related(Prefetch(lookup="groups", queryset=AuthGroup.objects.select_related("group")))
+        # Filtering by `group__isnull` to filter out possible `auth_group` that hasn't got `rbac_group` linked.
+        # Such groups are considered invalid.
+        User.objects.prefetch_related(
+            Prefetch(lookup="groups", queryset=AuthGroup.objects.select_related("group").filter(group__isnull=False))
+        )
         .exclude(username__in=settings.ADCM_HIDDEN_USERS)
         .order_by("username")
     )

python/rbac/ldap.py

@@ -91,15 +91,15 @@ def get_ldap_config() -> dict | None:
     return None
 
 
-def get_groups_by_user_dn(
-    user_dn: str,
+def get_groups_by_username(
+    user: User | _LDAPUser,
     user_search: LDAPSearch,
     conn: ldap.ldapobject.LDAPObject,
 ) -> tuple[list[str], list[str], str | None]:
     err_msg = None
     user_name_attr = get_ldap_config()["user_name_attribute"]
     replace = f"{user_name_attr}={USER_PLACEHOLDER}"
-    search_expr = f"distinguishedName={user_dn}"
+    search_expr = f"{user_name_attr}={user.ldap_username}"
 
     users = conn.search_s(
         base=user_search.base_dn,
@@ -109,13 +109,18 @@ def get_groups_by_user_dn(
     # Remove references from the Search Result https://www.rfc-editor.org/rfc/rfc4511#section-4.5.3
     users = [user for user in users if user[0] is not None]
 
-    if len(users) != 1:
+    if len(users) == 0:
         err_msg = f"Not one user found by `{search_expr}` search"
         return [], [], err_msg
 
+    if len(users) > 1:
+        users_dn = [user_dn for user_dn, _ in users]
+        err_msg = f"More then one user found by `{search_expr}` search: {';'.join(users_dn)}"
+        return [], [], err_msg
+
     user_dn_, user_attrs = users[0]
-    if user_dn_.strip().lower() != user_dn.strip().lower():
-        err_msg = f"Got different user dn: {(user_dn_, user_dn)}. Tune search"
+    if user_dn_.strip().lower() != user.ldap_user.dn.strip().lower():
+        err_msg = f"Got different user dn: {(user_dn_, user.ldap_user.dn)}. Tune search"
         return [], [], err_msg
 
     group_cns = []
@@ -128,7 +133,7 @@ def get_groups_by_user_dn(
         if group_name:
             group_cns.append(group_name)
 
-    logger.debug("Found %s groups by user dn `%s`: %s", len(group_cns), user_dn, group_cns)
+    logger.debug("Found %s groups by user %s `%s`: %s", len(group_cns), user_name_attr, user.ldap_username, group_cns)
     return group_cns, group_dns_lower, err_msg
 
 
@@ -228,12 +233,12 @@ def authenticate_ldap_user(self, ldap_user: User | _LDAPUser, password: str) ->
         if isinstance(user_or_none, User):
             user_or_none.type = OriginType.LDAP
             user_or_none.is_active = True
-            is_in_admin_group = self._process_groups(user_or_none, ldap_user.dn, user_local_groups)
+            user_or_none.is_superuser = False
+            user_or_none.save(update_fields=["type", "is_active", "is_superuser"])
+            is_in_admin_group = self._process_groups(user_or_none, user_local_groups)
             if is_in_admin_group:
                 user_or_none.is_superuser = True
-            elif user_or_none.is_superuser:
-                user_or_none.is_superuser = False
-            user_or_none.save(update_fields=["type", "is_active", "is_superuser"])
+                user_or_none.save(update_fields=["is_superuser"])
 
         return user_or_none
 
@@ -245,7 +250,14 @@ def _group_search_enabled(self) -> bool:
     def _get_local_groups_by_username(username: str) -> list[Group]:
         with suppress(User.DoesNotExist):
             user = User.objects.get(username__iexact=username, type=OriginType.LDAP)
-            return [g.group for g in user.groups.order_by("id") if g.group.type == OriginType.LOCAL]
+            # Filtering by `group__isnull` to filter out possible `auth_group` that hasn't got `rbac_group` linked.
+            # Such groups are considered invalid.
+            return [
+                g.group
+                for g in user.groups.select_related("group")
+                .filter(group__isnull=False, group__type=OriginType.LOCAL)
+                .order_by("id")
+            ]
 
         return []
 
@@ -270,11 +282,11 @@ def _get_groups_by_group_search(self) -> list[tuple[str, dict]]:
         logger.debug("Found %s groups: %s", len(groups), [i[0] for i in groups])
         return groups
 
-    def _process_groups(self, user: User | _LDAPUser, user_dn: str, additional_groups: list[Group] = ()) -> bool:
+    def _process_groups(self, user: User | _LDAPUser, additional_groups: list[Group] = ()) -> bool:
         is_in_admin_group = False
         with self._ldap_connection() as conn:
-            ldap_group_names, ldap_group_dns, err_msg = get_groups_by_user_dn(
-                user_dn=user_dn,
+            ldap_group_names, ldap_group_dns, err_msg = get_groups_by_username(
+                user=user,
                 user_search=self.default_settings["USER_SEARCH"],
                 conn=conn,
             )