fix(backend): 修改第三方授权逻辑 #5130

dbm-ui/backend/db_services/mysql/permission/authorize/handlers.py

@@ -157,15 +157,16 @@ def parse_domain(raw_domain):
             if not app_detail:
                 raise DBPermissionBaseException(_("无法查询app: {}相关信息，请检查app输入是否合法。").format(app))
             app_detail = app_detail[0]
-            bk_biz_id = app_detail["ccId"]
+            # bk_biz_id = app_detail["ccId"]
 
         # 域名存在，则走dbm的授权方式，否则走gcs的授权方式
         domain, __ = parse_domain(target_instance)
         cluster = Cluster.objects.filter(
             clusterentry__entry=domain, clusterentry__cluster_entry_type=ClusterEntryType.DNS.value
         )
-        bk_biz_id = int(bk_biz_id)
         if cluster.exists():
+            # 为了兼容迁移过后会一个业务拆分成多个业务，造成app与集群业务不一致的情况，这里暂时优先以集群业务为准
+            bk_biz_id = cluster.first().bk_biz_id
             if not bk_biz_id:
                 raise DBPermissionBaseException(_("授权集群: [{}]。业务信息bk_biz_id为空请检查。").format(target_instance))
             if cluster.count() > 1:
@@ -217,6 +218,10 @@ def parse_domain(raw_domain):
             )
             return data
         else:
+            if not app:
+                raise DBPermissionBaseException(_("gcs授权请保证app不为空"))
+
+            # 填充headers和参数。
             headers = {"bk_username": user}
             params = {
                 "app": app,

dbm-ui/backend/tests/conftest.py

@@ -17,8 +17,8 @@
 from django.utils.crypto import get_random_string
 
 from backend.db_meta import models
-from backend.db_meta.enums import AccessLayer, ClusterType, MachineType
-from backend.db_meta.models import AppCache, BKCity, Cluster, DBModule, LogicalCity, Machine
+from backend.db_meta.enums import AccessLayer, ClusterEntryRole, ClusterEntryType, ClusterType, MachineType
+from backend.db_meta.models import AppCache, BKCity, Cluster, ClusterEntry, DBModule, LogicalCity, Machine
 from backend.flow.models import FlowTree
 from backend.tests.constants import TEST_ADMIN_USERNAME
 from backend.tests.mock_data import constant
@@ -118,9 +118,20 @@ def init_cluster():
         immute_domain=constant.CLUSTER_IMMUTE_DOMAIN,
         cluster_type=ClusterType.TenDBHA.value,
     )
+    ClusterEntry.objects.create(
+        cluster=cluster,
+        cluster_entry_type=ClusterEntryType.DNS.value,
+        entry=constant.CLUSTER_IMMUTE_DOMAIN,
+        role=ClusterEntryRole.MASTER_ENTRY,
+    )
     yield cluster
 
 
+@pytest.fixture
+def init_cluster_entry(init_cluster):
+    pass
+
+
 @pytest.fixture
 def init_app():
     app = AppCache.objects.create(

dbm-ui/backend/tests/db_services/mysql/permission/test_authorize_handler.py

@@ -14,15 +14,19 @@
 from unittest.mock import patch
 
 import pytest
+from django.test import RequestFactory
 from openpyxl.writer.excel import save_virtual_workbook
 
 from backend.db_meta.enums import ClusterType
 from backend.db_services.mysql.permission.authorize.dataclass import MySQLAuthorizeMeta, MySQLExcelAuthorizeMeta
 from backend.db_services.mysql.permission.authorize.handlers import MySQLAuthorizeHandler
 from backend.db_services.mysql.permission.constants import AUTHORIZE_EXCEL_HEADER
 from backend.tests.mock_data import constant
+from backend.tests.mock_data.components.gcs import GCS_CLUSTER_INSTANCE, GcsApiMock, ScrApiMock
 from backend.tests.mock_data.components.mysql_priv_manager import DBPrivManagerApiMock
 from backend.tests.mock_data.db_services.mysql.permission.authorize import AUTHORIZE_DATA, EXCEL_DATA_DICT__LIST
+from backend.ticket.constants import TicketStatus
+from backend.ticket.models import Ticket
 from backend.utils.excel import ExcelHandler
 
 pytestmark = pytest.mark.django_db
@@ -60,3 +64,35 @@ def test_pre_check_excel_rules(self, query_fixture):
 
         authorize_data_list = self.handler.pre_check_excel_rules(excel_authorize)
         assert authorize_data_list["pre_check"] is True
+
+    @patch("backend.db_services.mysql.permission.authorize.handlers.GcsApi", GcsApiMock)
+    @patch("backend.db_services.mysql.permission.authorize.handlers.ScrApi", ScrApiMock)
+    def test_authorize_apply(self, init_cluster, bk_user, init_app):
+        request = RequestFactory().post("")
+        request.user = bk_user
+
+        # 测试集群在dbm的授权流程
+        task = self.handler.authorize_apply(
+            request=request,
+            user="test",
+            access_db="test_db",
+            source_ips="127.0.0.1",
+            target_instance=init_cluster.immute_domain,
+            bk_biz_id=init_cluster.bk_biz_id,
+            operator=bk_user.username,
+        )
+        task_info = self.handler.query_authorize_apply_result(task["task_id"], task["platform"])
+        assert Ticket.objects.filter(id=task["task_id"]).exists()
+        assert task_info["status"] == TicketStatus.RUNNING
+
+        # 测试集群在gcs的授权流程
+        task = self.handler.authorize_apply(
+            request=request,
+            user="test",
+            access_db="test_db",
+            source_ips="127.0.0.1",
+            target_instance=GCS_CLUSTER_INSTANCE,
+            app=init_app.db_app_abbr,
+            operator=bk_user.username,
+        )
+        assert task["task_id"] == "gcs_task"

dbm-ui/backend/tests/db_services/mysql/remote_service/test_handlers.py

@@ -8,6 +8,8 @@
 an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 specific language governing permissions and limitations under the License.
 """
+import itertools
+
 import pytest
 from mock.mock import patch
 
@@ -23,8 +25,5 @@ def test_dbsingle_show_databases(self, bk_biz_id, dbsingle_cluster, dbha_cluster
         results = RemoteServiceHandler(bk_biz_id=bk_biz_id).show_databases(
             cluster_ids=[dbsingle_cluster.id, dbha_cluster.id]
         )
-        for result in results:
-            if result["cluster_id"] == dbsingle_cluster.id:
-                assert len(result["databases"]) > 0
-            if result["cluster_id"] == dbha_cluster.id:
-                assert len(result["databases"]) == 0
+        databases = list(itertools.chain(*[result["databases"] for result in results]))
+        assert databases == ["db1", "db2"]

dbm-ui/backend/tests/mock_data/components/gcs.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-DB管理系统(BlueKing-BK-DBM) available.
+Copyright (C) 2017-2023 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at https://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""
+from backend.db_meta.models import AppCache
+
+GCS_CLUSTER_INSTANCE = "gcs.testdb.dba.db"
+
+
+class GcsApiMock(object):
+    """Gcs的mock类"""
+
+    @classmethod
+    def cloud_privileges_asyn_bydbname(cls, params, headers):
+        cluster = params["target_ip"]
+        assert cluster == GCS_CLUSTER_INSTANCE, f"{cluster} not belong to gcs"
+        return {"task_id": "gcs_task", "platform": "gcs", "job_id": "gcs_task"}
+
+
+class ScrApiMock(object):
+    """Scr的mock类"""
+
+    @classmethod
+    def common_query(cls, params):
+        try:
+            bk_biz_id = AppCache.objects.get(db_app_abbr=params["app"]).bk_biz_id
+        except AppCache.DoesNotExist:
+            bk_biz_id = 0
+        return {"detail": [{"appid": params["app"], "ccId": bk_biz_id}], "rowsNum": 1}