feat(backend): 权限二期细化 TencentBlueKing#3860

dbm-ui/backend/bk_web/viewsets.py

@@ -9,11 +9,11 @@
 specific language governing permissions and limitations under the License.
 """
 import copy
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from blueapps.account.decorators import login_exempt
 from django.utils.decorators import classonlymethod
-from rest_framework import serializers, status, viewsets
+from rest_framework import permissions, serializers, status, viewsets
 from rest_framework.response import Response
 from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet
 
@@ -23,14 +23,24 @@
 class GenericMixin:
     queryset = ""
     # 是否支持全局豁免
-    global_login_exempt = False
+    global_login_exempt: bool = False
+    # 权限动作映射表。如果权限映射逻辑复杂，可考虑覆写_get_custom_permissions
+    action_permission_map: Dict[Union[Tuple[str], str], List[permissions.BasePermission]] = {}
+    # ⚠️为了避免权限泄露，希望默认权限是永假来兜底，所以请定义好每个视图的权限类
+    default_permission_class: List[permissions.BasePermission] = [RejectPermission()]
 
     @staticmethod
     def get_request_data(request, **kwargs) -> Dict[str, Any]:
         request_data = request.data.copy() or {}
         request_data.update(**kwargs)
         return request_data
 
+    def get_action_permission_map(self) -> dict:
+        return self.action_permission_map
+
+    def get_default_permission_class(self) -> list:
+        return self.default_permission_class
+
     @property
     def validated_data(self):
         """
@@ -98,7 +108,10 @@ def get_permissions(self):
     def _get_custom_permissions(self):
         """用户自定义的permission类,由子类继承覆写"""
         # ⚠️为了避免权限泄露，希望默认权限是永假来兜底，所以请写每一个视图的时候都覆写该方法
-        return [RejectPermission()]
+        for actions, custom_perms in self.get_action_permission_map().items():
+            if actions == self.action or self.action in actions:
+                return custom_perms
+        return self.get_default_permission_class()
 
     @classmethod
     def _get_login_exempt_view_func(cls):

dbm-ui/backend/components/mysql_priv_manager/client.py

@@ -67,6 +67,11 @@ def __init__(self):
             url="/priv/modify_account",
             description=_("修改账号的密码"),
         )
+        self.get_account = self.generate_data_api(
+            method="POST",
+            url="/priv/get_account",
+            description=_("查询账号列表"),
+        )
 
         # 授权规则相关
         self.pre_check_authorize_rules = self.generate_data_api(

dbm-ui/backend/configuration/constants.py

@@ -185,4 +185,8 @@ class BizSettingsEnum(str, StructuredEnum):
 DOMAIN_RESOLUTION_SUPPORT = "DOMAIN_RESOLUTION_SUPPORT"
 
 # DB组件和admin用户的映射
-DB_ADMIN_USER_MAP = {DBType.MySQL: MYSQL_ADMIN_USER, DBType.Sqlserver: SQLSERVER_ADMIN_USER}
+DB_ADMIN_USER_MAP = {
+    DBType.TenDBCluster: MYSQL_ADMIN_USER,
+    DBType.MySQL: MYSQL_ADMIN_USER,
+    DBType.Sqlserver: SQLSERVER_ADMIN_USER,
+}

dbm-ui/backend/configuration/serializers.py

@@ -95,15 +95,15 @@ class GetMySQLAdminPasswordSerializer(serializers.Serializer):
 
     begin_time = DBTimezoneField(help_text=_("开始时间"), required=False)
     end_time = DBTimezoneField(help_text=_("结束时间"), required=False)
-    instances = serializers.CharField(help_text=_("过滤的实例列表(通过,分割，实例格式为--ip:port)"), required=False)
+    instances = serializers.CharField(help_text=_("过滤的实例列表(通过,分割，实例格式为--cloud:ip:port)"), required=False)
 
 
 class GetMySQLAdminPasswordResponseSerializer(serializers.Serializer):
     class Meta:
         swagger_schema_fields = {"example": mock_data.MYSQL_ADMIN_PASSWORD_DATA}
 
 
-class ModifyMySQLAdminPasswordSerializer(serializers.Serializer):
+class ModifyAdminPasswordSerializer(serializers.Serializer):
     class InstanceInfoSerializer(serializers.Serializer):
         ip = serializers.CharField(help_text=_("实例ip"))
         port = serializers.CharField(help_text=_("实例port"))
@@ -116,10 +116,16 @@ class InstanceInfoSerializer(serializers.Serializer):
     instance_list = serializers.ListSerializer(help_text=_("实例信息"), child=InstanceInfoSerializer())
 
     def validate(self, attrs):
+        # 校验密码中的特殊字符
         invalid_characters = re.compile(r"[`\'\"]")
         if invalid_characters.findall(attrs["password"]):
             raise serializers.ValidationError(_("修改密码中不允许包含单引号，双引号和反引号"))
 
+        # 目前只允许一次性修改一种类型
+        cluster_type = [inst["cluster_type"] for inst in attrs["instance_list"]]
+        if len(set(cluster_type)) > 1:
+            raise serializers.ValidationError(_("一次只允许修改同种集群类型的密码"))
+
         return attrs
 
 

dbm-ui/backend/configuration/views/dba.py

@@ -27,14 +27,11 @@ class DBAdminViewSet(viewsets.SystemViewSet):
     def _get_custom_permissions(self):
         if self.action == "list_admins":
             return []
-
         if not int(self.request.data.get("bk_biz_id", 0)):
             return [ResourceActionPermission([ActionEnum.GLOBAL_DBA_ADMINISTRATOR_EDIT])]
         else:
-            instance_getter = lambda request, view: [request.data["bk_biz_id"]]  # noqa: E731
-            return [
-                ResourceActionPermission([ActionEnum.DBA_ADMINISTRATOR_EDIT], ResourceEnum.BUSINESS, instance_getter)
-            ]
+            inst_getter = lambda request, view: [request.data["bk_biz_id"]]  # noqa: E731
+            return [ResourceActionPermission([ActionEnum.DBA_ADMINISTRATOR_EDIT], ResourceEnum.BUSINESS, inst_getter)]
 
     @common_swagger_auto_schema(
         operation_summary=_("查询DBA人员列表"), query_serializer=ListDBAdminSerializer, tags=[SWAGGER_TAG]

dbm-ui/backend/configuration/views/function_controller.py

@@ -26,8 +26,7 @@ class FunctionControllerViewSet(viewsets.AuditedModelViewSet):
     serializer_class = FunctionControllerSerializer
     queryset = FunctionController.objects.all()
 
-    def _get_custom_permissions(self):
-        return []
+    default_permission_class = []
 
     @common_swagger_auto_schema(
         operation_summary=_("功能开关列表"),

dbm-ui/backend/configuration/views/ip_whitelist.py

@@ -72,7 +72,7 @@ class IPWhitelistViewSet(viewsets.AuditedModelViewSet):
     pagination_class = AuditedLimitOffsetPagination
 
     @staticmethod
-    def instance_getter(request, view):
+    def inst_getter(request, view):
         # 批量删除的白名单一定在一个业务下
         if view.action == "batch_delete":
             return [IPWhitelist.objects.filter(pk__in=request.data["ids"]).first().bk_biz_id]
@@ -84,11 +84,10 @@ def instance_getter(request, view):
     def _get_custom_permissions(self):
         if self.action in ["retrieve", "iplist"]:
             return []
-
-        bk_biz_id = self.instance_getter(self.request, self)[0]
+        bk_biz_id = self.inst_getter(self.request, self)[0]
         if int(bk_biz_id):
             return [
-                ResourceActionPermission([ActionEnum.IP_WHITELIST_MANAGE], ResourceEnum.BUSINESS, self.instance_getter)
+                ResourceActionPermission([ActionEnum.IP_WHITELIST_MANAGE], ResourceEnum.BUSINESS, self.inst_getter)
             ]
         else:
             return [ResourceActionPermission([ActionEnum.GLOBAL_IP_WHITELIST_MANAGE])]

dbm-ui/backend/configuration/views/password_policy.py

@@ -25,7 +25,7 @@
 from backend.configuration.serializers import (
     GetMySQLAdminPasswordResponseSerializer,
     GetMySQLAdminPasswordSerializer,
-    ModifyMySQLAdminPasswordSerializer,
+    ModifyAdminPasswordSerializer,
     ModifyMySQLPasswordRandomCycleSerializer,
     PasswordPolicySerializer,
     VerifyPasswordResponseSerializer,
@@ -34,22 +34,21 @@
 from backend.db_periodic_task.models import DBPeriodicTask
 from backend.iam_app.dataclass.actions import ActionEnum
 from backend.iam_app.handlers.drf_perm.base import ResourceActionPermission
+from backend.iam_app.handlers.drf_perm.cluster import ModifyActionPermission
 
 SWAGGER_TAG = _("密码安全策略")
 
 
 class PasswordPolicyViewSet(viewsets.SystemViewSet):
     pagination_class = None
 
-    def _get_custom_permissions(self):
-        if self.action in [
-            self.get_password_policy.__name__,
-            self.verify_password_strength.__name__,
-            self.get_random_password.__name__,
-            self.query_random_cycle.__name__,
-        ]:
-            return []
-        return [ResourceActionPermission([ActionEnum.PASSWORD_POLICY_SET])]
+    action_permission_map = {
+        ("get_password_policy", "verify_password_strength", "get_random_password", "query_random_cycle"): [],
+        ("modify_admin_password", "query_mysql_admin_password"): [ModifyActionPermission()],
+        ("update_password_policy", "modify_random_cycle"): [
+            ResourceActionPermission([ActionEnum.PASSWORD_POLICY_SET])
+        ],
+    }
 
     @common_swagger_auto_schema(
         operation_summary=_("查询密码安全策略"),
@@ -141,10 +140,10 @@ def query_mysql_admin_password(self, request, *args, **kwargs):
 
     @common_swagger_auto_schema(
         operation_summary=_("修改db实例密码(admin)"),
-        request_body=ModifyMySQLAdminPasswordSerializer(),
+        request_body=ModifyAdminPasswordSerializer(),
         tags=[SWAGGER_TAG],
     )
-    @action(methods=["POST"], detail=False, serializer_class=ModifyMySQLAdminPasswordSerializer)
+    @action(methods=["POST"], detail=False, serializer_class=ModifyAdminPasswordSerializer)
     def modify_admin_password(self, request, *args, **kwargs):
         validated_data = self.params_validate(self.get_serializer_class())
         validated_data["operator"] = request.user.username

dbm-ui/backend/configuration/views/profile.py

@@ -24,9 +24,7 @@
 
 class ProfileViewSet(viewsets.SystemViewSet):
     serializer_class = ProfileSerializer
-
-    def _get_custom_permissions(self):
-        return []
+    default_permission_class = []
 
     @common_swagger_auto_schema(operation_summary=_("查询个人配置列表"), tags=[SWAGGER_TAG])
     @action(methods=["GET"], detail=False)

dbm-ui/backend/configuration/views/system.py

@@ -38,16 +38,12 @@
 class SystemSettingsViewSet(viewsets.SystemViewSet):
     """系统设置视图"""
 
-    def _get_custom_permissions(self):
-        # 非超级用户拒绝访问敏感信息
-        if self.action == "sensitive_environ":
-            return [RejectPermission()]
-        if self.action == "update_duty_notice_config":
-            return [ResourceActionPermission([ActionEnum.UPDATE_DUTY_NOTICE_CONFIG])]
-        if self.action in ["disk_classes", "device_classes", "duty_notice_config", "environ"]:
-            return []
-
-        return [ResourceActionPermission([ActionEnum.GLOBAL_MANAGE])]
+    action_permission_map = {
+        ("sensitive_environ",): [RejectPermission()],
+        ("update_duty_notice_config",): [ResourceActionPermission([ActionEnum.UPDATE_DUTY_NOTICE_CONFIG])],
+        ("disk_classes", "device_classes", "duty_notice_config", "environ"): [],
+    }
+    default_permission_class = [ResourceActionPermission([ActionEnum.GLOBAL_MANAGE])]
 
     @common_swagger_auto_schema(
         operation_summary=_("查询磁盘类型"),
@@ -129,8 +125,8 @@ class BizSettingsViewSet(viewsets.AuditedModelViewSet):
     serializer_class = BizSettingsSerializer
     queryset = BizSettings.objects.all()
 
-    def _get_custom_permissions(self):
-        return [DBManagePermission()]
+    action_permission_map = {}
+    default_permission_class = [DBManagePermission()]
 
     @common_swagger_auto_schema(
         operation_summary=_("业务设置列表"),

dbm-ui/backend/core/encrypt/views.py

@@ -23,10 +23,8 @@
 
 
 class EncryptViewSet(viewsets.SystemViewSet):
-    def _get_custom_permissions(self):
-        if self.action == "fetch_public_keys":
-            return []
-        return [ResourceActionPermission([ActionEnum.GLOBAL_MANAGE])]
+    action_permission_map = {("fetch_public_keys",): []}
+    default_permission_class = [ResourceActionPermission([ActionEnum.GLOBAL_MANAGE])]
 
     @common_swagger_auto_schema(
         operation_summary=_("查询公钥列表"), request_body=FetchPublicKeysSerializer(), tags=[SWAGGER_TAG]

dbm-ui/backend/core/storages/views.py

@@ -21,8 +21,7 @@
 
 
 class StorageViewSet(viewsets.SystemViewSet):
-    def _get_custom_permissions(self):
-        return []
+    default_permission_class = []
 
     @common_swagger_auto_schema(
         operation_summary=_("批量获取文件内容"), request_body=BatchDownloadFileSerializer(), tags=[SWAGGER_TAG]

dbm-ui/backend/db_dirty/views.py

@@ -39,10 +39,8 @@ class DBDirtyMachineViewSet(viewsets.SystemViewSet):
     pagination_class = None
     filter_class = None
 
-    def _get_custom_permissions(self):
-        if self.action == "query_operation_list":
-            return []
-        return [ResourceActionPermission([ActionEnum.DIRTY_POLL_MANAGE])]
+    action_permission_map = {("query_operation_list",): []}
+    default_permission_class = [ResourceActionPermission([ActionEnum.DIRTY_POLL_MANAGE])]
 
     @common_swagger_auto_schema(
         operation_summary=_("查询污点池列表"),

dbm-ui/backend/db_event/views/dbha.py

@@ -28,16 +28,16 @@
 
 
 class DBHAEventViewSet(viewsets.SystemViewSet):
+    def get_action_permission_map(self):
+        return {("cat",): []}
+
+    def get_default_permission_class(self):
+        return [ResourceActionPermission([ActionEnum.DBHA_SWITCH_EVENT_VIEW], ResourceEnum.BUSINESS, self.inst_getter)]
+
     @staticmethod
-    def biz_getter(request, view):
+    def inst_getter(request, view):
         return [get_request_key_id(request, "app")]
 
-    def _get_custom_permissions(self):
-        if self.action == "cat":
-            # TODO: 暂时豁免，后续cat接口带上业务属性后再放开
-            return []
-        return [ResourceActionPermission([ActionEnum.DBHA_SWITCH_EVENT_VIEW], ResourceEnum.BUSINESS, self.biz_getter)]
-
     @common_swagger_auto_schema(
         operation_summary=_("DBHA切换事件列表"),
         query_serializer=QueryListSerializer,

dbm-ui/backend/db_monitor/views/duty_rule.py

@@ -10,6 +10,7 @@
 """
 from django.utils.decorators import method_decorator
 from django.utils.translation import ugettext_lazy as _
+from django_filters.rest_framework import DjangoFilterBackend
 from rest_framework import filters
 from rest_framework.decorators import action
 from rest_framework.response import Response
@@ -66,36 +67,39 @@ class MonitorDutyRuleViewSet(viewsets.AuditedModelViewSet):
     queryset = DutyRule.objects.all().order_by("-update_at")
     serializer_class = DutyRuleSerializer
     pagination_class = AuditedLimitOffsetPagination
-    filter_backends = [filters.SearchFilter]
+    filter_backends = [filters.SearchFilter, DjangoFilterBackend]
     filter_fields = {"db_type": ["exact"], "name": ["exact"]}
     search_fields = ["name"]
 
+    default_permission_class = [ResourceActionPermission([ActionEnum.GLOBAL_MANAGE])]
+
+    def get_action_permission_map(self):
+        return {
+            ("list",): [ResourceActionPermission([ActionEnum.DUTY_RULE_LIST], ResourceEnum.DBTYPE, self.inst_getter)],
+            ("create",): [
+                ResourceActionPermission([ActionEnum.DUTY_RULE_CREATE], ResourceEnum.DBTYPE, self.inst_getter)
+            ],
+            ("update", "partial_update"): [
+                ResourceActionPermission([ActionEnum.DUTY_RULE_UPDATE], ResourceEnum.DBTYPE, self.inst_getter)
+            ],
+            ("destroy",): [
+                ResourceActionPermission([ActionEnum.DUTY_RULE_DESTROY], ResourceEnum.DBTYPE, self.inst_getter)
+            ],
+            ("priority_distinct",): [],
+        }
+
     @staticmethod
     def inst_getter(request, view):
         if view.action in ["list", "create"]:
             return [get_request_key_id(request, key="db_type")]
         if view.action in ["update", "partial_update", "destroy"]:
-            return [view.kwargs.get("pk")]
-
-    def _get_custom_permissions(self):
-        if self.action == "list":
-            return [ResourceActionPermission([ActionEnum.DUTY_RULE_LIST], ResourceEnum.DBTYPE, self.inst_getter)]
-        elif self.action == "create":
-            return [ResourceActionPermission([ActionEnum.DUTY_RULE_CREATE], ResourceEnum.DBTYPE, self.inst_getter)]
-        elif self.action in ["update", "partial_update"]:
-            return [ResourceActionPermission([ActionEnum.DUTY_RULE_UPDATE], ResourceEnum.DUTY_RULE, self.inst_getter)]
-        elif self.action == "destroy":
-            return [ResourceActionPermission([ActionEnum.DUTY_RULE_DESTROY], ResourceEnum.DUTY_RULE, self.inst_getter)]
-        elif self.action in ["priority_distinct"]:
-            return []
-
-        return [ResourceActionPermission([ActionEnum.GLOBAL_MANAGE])]
+            return [DutyRule.objects.get(id=view.kwargs.get("pk")).db_type]
 
     @Permission.decorator_permission_field(
-        id_field=lambda d: d["id"],
+        id_field=lambda d: d["db_type"],
         data_field=lambda d: d["results"],
-        actions=ActionEnum.get_actions_by_resource(ResourceEnum.DUTY_RULE.id),
-        resource_meta=ResourceEnum.DUTY_RULE,
+        actions=[ActionEnum.DUTY_RULE_CREATE, ActionEnum.DUTY_RULE_UPDATE, ActionEnum.DUTY_RULE_DESTROY],
+        resource_meta=ResourceEnum.DBTYPE,
     )
     def list(self, request, *args, **kwargs):
         return super().list(request, *args, **kwargs)

dbm-ui/backend/db_monitor/views/grafana.py

@@ -32,8 +32,6 @@ def _get_custom_permissions(self):
             else:
                 return [ClusterDetailPermission()]
 
-        return super()._get_custom_permissions()
-
     @common_swagger_auto_schema(
         operation_summary=_("查询内嵌仪表盘地址"),
         query_serializer=GetDashboardSerializer,