feat: add subject group search api

saas/backend/api/authorization/migrations/0014_auto_20231109_1046.py

@@ -0,0 +1,42 @@
+# Generated by Django 3.2.16 on 2023-11-09 02:46
+
+from django.db import migrations
+
+from backend.api.authorization.constants import AuthorizationAPIEnum
+from backend.api.constants import ALLOW_ANY
+
+
+def init_allow_list(apps, schema_editor):
+    """初始化授权API白名单"""
+    AuthAPIAllowListConfig = apps.get_model("authorization", "AuthAPIAllowListConfig")
+    # 查询已存在白名单，避免重复
+    all_allow_list = AuthAPIAllowListConfig.objects.all()
+    allow_set = set([(a.type, a.system_id, a.object_id) for a in all_allow_list])
+    # 新建关联实例授权API 白名单
+    system_resource_types = {
+        "bk_cmdb": [ALLOW_ANY],
+    }
+    auth_api_allow_list_config = []
+    for system_id, resource_types in system_resource_types.items():
+        for resource_type_id in resource_types:
+            # 已存在，则直接忽略
+            if (AuthorizationAPIEnum.CREATOR_AUTHORIZATION_INSTANCE.value, system_id, resource_type_id) in allow_set:
+                continue
+            auth_api_allow_list_config.append(
+                AuthAPIAllowListConfig(
+                    type=AuthorizationAPIEnum.CREATOR_AUTHORIZATION_INSTANCE.value,
+                    system_id=system_id,
+                    object_id=resource_type_id,
+                )
+            )
+    if len(auth_api_allow_list_config) != 0:
+        AuthAPIAllowListConfig.objects.bulk_create(auth_api_allow_list_config)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('authorization', '0013_auto_20221101_1216'),
+    ]
+
+    operations = [migrations.RunPython(init_allow_list)]

saas/backend/apps/subject/urls.py

@@ -60,6 +60,16 @@
                     views.SubjectTemporaryPolicySystemViewSet.as_view({"get": "list"}),
                     name="subject.temporary_policies_systems",
                 ),
+                path(
+                    "groups/search/",
+                    views.SubjectGroupSearchViewSet.as_view({"get": "list"}),
+                    name="subject.group_search",
+                ),
+                path(
+                    "departments/-/groups/search/",
+                    views.SubjectDepartmentGroupSearchViewSet.as_view({"get": "list"}),
+                    name="subject.department.group_search",
+                ),
             ]
         ),
     )

saas/backend/apps/subject/views.py

@@ -17,8 +17,10 @@
 from backend.account.serializers import AccountRoleSLZ
 from backend.apps.group.audit import GroupMemberDeleteAuditProvider
 from backend.apps.group.models import Group
+from backend.apps.group.serializers import GroupSearchSLZ
 from backend.apps.policy.serializers import PolicyDeleteSLZ, PolicyPartDeleteSLZ, PolicySLZ, PolicySystemSLZ
 from backend.apps.user.serializers import GroupSLZ
+from backend.apps.user.views import SubjectGroupSearchMixin
 from backend.audit.audit import audit_context_setter, view_audit_decorator
 from backend.biz.group import GroupBiz
 from backend.biz.policy import ConditionBean, PolicyOperationBiz, PolicyQueryBiz
@@ -337,3 +339,40 @@ def list(self, request, *args, **kwargs):
         data = self.biz.list_temporary_system_counter_by_subject(subject)
 
         return Response([one.dict() for one in data])
+
+
+class SubjectGroupSearchViewSet(SubjectGroupSearchMixin):
+    @swagger_auto_schema(
+        operation_description="搜索subject用户组列表",
+        request_body=GroupSearchSLZ(label="用户组搜索"),
+        responses={status.HTTP_200_OK: SubjectGroupSLZ(label="用户组", many=True)},
+        tags=["subject"],
+    )
+    def search(self, request, *args, **kwargs):
+        return super().search(request, *args, **kwargs)
+
+    def get_subject(self, request, kwargs):
+        subject = Subject(type=kwargs["subject_type"], id=kwargs["subject_id"])
+        return subject
+
+
+class SubjectDepartmentGroupSearchViewSet(SubjectGroupSearchMixin):
+    @swagger_auto_schema(
+        operation_description="搜索Subject部门用户组列表",
+        request_body=GroupSearchSLZ(label="用户组搜索"),
+        responses={status.HTTP_200_OK: SubjectGroupSLZ(label="用户组", many=True)},
+        tags=["subject"],
+    )
+    def search(self, request, *args, **kwargs):
+        return super().search(request, *args, **kwargs)
+
+    def get_subject(self, request, kwargs):
+        subject = Subject(type=kwargs["subject_type"], id=kwargs["subject_id"])
+        return subject
+
+    def get_group_dict(self, subject: Subject):
+        groups = self.biz.list_all_user_department_group(subject)
+        return {one.id: one for one in groups}
+
+    def get_page_result(self, group_dict, page):
+        return [group_dict[one.id] for one in page]

saas/backend/apps/user/views.py

@@ -228,19 +228,13 @@ def list(self, request, *args, **kwargs):
         return Response([one.dict() for one in user_roles])
 
 
-class UserGroupSearchViewSet(mixins.ListModelMixin, GenericViewSet):
+class SubjectGroupSearchMixin(mixins.ListModelMixin, GenericViewSet):
 
     queryset = Group.objects.all()
     serializer_class = GroupSLZ
 
     biz = GroupBiz()
 
-    @swagger_auto_schema(
-        operation_description="搜索用户用户组列表",
-        request_body=GroupSearchSLZ(label="用户组搜索"),
-        responses={status.HTTP_200_OK: SubjectGroupSLZ(label="用户组", many=True)},
-        tags=["user"],
-    )
     def search(self, request, *args, **kwargs):
         slz = GroupSearchSLZ(data=request.data)
         slz.is_valid(raise_exception=True)
@@ -259,10 +253,10 @@ def search(self, request, *args, **kwargs):
         )
         queryset = f.qs
 
-        subject = Subject.from_username(request.user.username)
+        subject = self.get_subject(request, kwargs)
         # 查询用户加入的所有用户组
-        groups = list_all_subject_groups(subject.type, subject.id)
-        ids = sorted([int(g["id"]) for g in groups])
+        group_dict = self.get_group_dict(subject)
+        ids = sorted(group_dict.keys())
 
         if data["system_id"] and data["action_id"]:
             # 通过实例或操作查询用户组
@@ -281,77 +275,53 @@ def search(self, request, *args, **kwargs):
 
         page = self.paginate_queryset(queryset)
         if page is not None:
-            group_dict = {int(one["id"]): one for one in groups}
-            relations = [SubjectGroup(**group_dict[one.id]) for one in page]
-            results = self.biz._convert_to_subject_group_beans(relations)
+            results = self.get_page_result(group_dict, page)
 
             slz = GroupSLZ(instance=results, many=True)
             return Response({"count": queryset.count(), "results": slz.data})
 
         return Response({"count": 0, "results": []})
 
+    def get_subject(self, request, kwargs):
+        subject = Subject.from_username(request.user.username)
+        return subject
 
-class UserDepartmentGroupSearchViewSet(mixins.ListModelMixin, GenericViewSet):
+    def get_group_dict(self, subject: Subject):
+        groups = list_all_subject_groups(subject.type, subject.id)
+        return {int(one["id"]): one for one in groups}
 
-    queryset = Group.objects.all()
-    serializer_class = GroupSLZ
+    def get_page_result(self, group_dict, page):
+        relations = [SubjectGroup(**group_dict[one.id]) for one in page]
+        return self.biz._convert_to_subject_group_beans(relations)
 
-    biz = GroupBiz()
 
+class UserGroupSearchViewSet(SubjectGroupSearchMixin):
     @swagger_auto_schema(
-        operation_description="搜索用户部门用户组列表",
+        operation_description="搜索用户用户组列表",
         request_body=GroupSearchSLZ(label="用户组搜索"),
         responses={status.HTTP_200_OK: SubjectGroupSLZ(label="用户组", many=True)},
         tags=["user"],
     )
     def search(self, request, *args, **kwargs):
-        slz = GroupSearchSLZ(data=request.data)
-        slz.is_valid(raise_exception=True)
+        return super().search(request, *args, **kwargs)
 
-        data = slz.validated_data
 
-        # 筛选
-        f = GroupFilter(
-            data={
-                k: v
-                for k, v in data.items()
-                if k in ["id", "name", "description", "hidden"]
-                if isinstance(v, bool) or v
-            },
-            queryset=self.get_queryset(),
-        )
-        queryset = f.qs
+class UserDepartmentGroupSearchViewSet(SubjectGroupSearchMixin):
+    @swagger_auto_schema(
+        operation_description="搜索用户部门用户组列表",
+        request_body=GroupSearchSLZ(label="用户组搜索"),
+        responses={status.HTTP_200_OK: SubjectGroupSLZ(label="用户组", many=True)},
+        tags=["user"],
+    )
+    def search(self, request, *args, **kwargs):
+        return super().search(request, *args, **kwargs)
 
-        subject = Subject.from_username(request.user.username)
+    def get_group_dict(self, subject: Subject):
         groups = self.biz.list_all_user_department_group(subject)
+        return {one.id: one for one in groups}
 
-        # 查询用户加入的所有用户组
-        ids = sorted([g.id for g in groups])
-
-        if data["system_id"] and data["action_id"]:
-            # 通过实例或操作查询用户组
-            data["permission_type"] = PermissionTypeEnum.RESOURCE_INSTANCE.value
-            data["limit"] = 1000
-            subjects = QueryAuthorizedSubjects(data).query_by_resource_instance(subject_type="group")
-            subject_id_set = {int(s["id"]) for s in subjects}
-
-            # 筛选同时有权限并且用户加入的用户组
-            ids = [_id for _id in ids if _id in subject_id_set]
-
-        if not ids:
-            return Response({"count": 0, "results": []})
-
-        queryset = queryset.filter(id__in=ids)
-
-        page = self.paginate_queryset(queryset)
-        if page is not None:
-            group_dict = {one.id: one for one in groups}
-            relations = [group_dict[one.id] for one in page]
-
-            slz = GroupSLZ(instance=relations, many=True)
-            return Response({"count": queryset.count(), "results": slz.data})
-
-        return Response({"count": 0, "results": []})
+    def get_page_result(self, group_dict, page):
+        return [group_dict[one.id] for one in page]
 
 
 class UserPolicySearchViewSet(mixins.ListModelMixin, GenericViewSet):

saas/backend/biz/role.py

@@ -825,11 +825,19 @@ def _list_relation_role_id(self):
         return role_ids
 
     def _check_object(self, obj_type: str, obj_id: int) -> bool:
+        # 如果是超级管理员, 直接返回True
+        if self.role.type == RoleType.SUPER_MANAGER.value:
+            return True
+
         return RoleRelatedObject.objects.filter(
             role_id__in=self._list_relation_role_id(), object_type=obj_type, object_id=obj_id
         ).exists()
 
     def _check_object_ids(self, obj_type: str, obj_ids: List[int]) -> bool:
+        # 如果是超级管理员, 直接返回True
+        if self.role.type == RoleType.SUPER_MANAGER.value:
+            return True
+
         count = RoleRelatedObject.objects.filter(
             role_id__in=self._list_relation_role_id(), object_type=obj_type, object_id__in=obj_ids
         ).count()

saas/config/default.py

@@ -171,6 +171,8 @@
     "DEFAULT_AUTO_SCHEMA_CLASS": "backend.common.swagger.ResponseSwaggerAutoSchema",
 }
 
+ENABLE_SWAGGER = env.bool("BKAPP_ENABLE_SWAGGER", default=False)
+
 # CELERY 开关，使用时请改为 True，否则请保持为False。启动方式为以下两行命令：
 # worker: python manage.py celery worker -l info
 # beat: python manage.py celery beat -l info

saas/urls.py

@@ -79,7 +79,7 @@
 ]
 
 # add swagger api document
-if settings.IS_LOCAL:
+if settings.IS_LOCAL or settings.ENABLE_SWAGGER:
     urlpatterns += [
         url(r"^swagger/$", schema_view.with_ui("swagger", cache_timeout=0), name="schema-swagger-ui"),
     ]