feat: 上云环境针对0区域限制新增主机校验提到API层级 (closed TencentBlueKing#2501)

# Reviewed, transaction id: 25959

apps/backend/components/collections/agent_new/add_or_update_hosts.py

@@ -258,7 +258,7 @@ def query_hosts(self, sub_insts: List[models.SubscriptionInstanceRecord]) -> Lis
             query_hosts_params_list.append({"host_infos": host_infos, "bk_addressing": addressing})
 
         cmdb_hosts: List[Dict] = concurrent.batch_call(
-            func=self.query_hosts_by_addressing, params_list=query_hosts_params_list, extend_result=True
+            func=tools.JobTools.query_hosts_by_addressing, params_list=query_hosts_params_list, extend_result=True
         )
 
         bk_host_ids: List[int] = [cmdb_host["bk_host_id"] for cmdb_host in cmdb_hosts]
@@ -504,7 +504,9 @@ def _execute(self, data, parent_data, common_data: CommonData):
 
         # 按 IpKey 聚合主机信息
         # IpKey：ip（v4 or v6）+ bk_addressing（寻值方式）+ bk_cloud_id（管控区域）
-        cmdb_host_infos_gby_ip_key: Dict[str, List[Dict[str, Any]]] = self.get_host_infos_gby_ip_key(exist_cmdb_hosts)
+        cmdb_host_infos_gby_ip_key: Dict[str, List[Dict[str, Any]]] = tools.JobTools.get_host_infos_gby_ip_key(
+            exist_cmdb_hosts
+        )
         for sub_inst in subscription_instances:
             id__sub_inst_obj_map[sub_inst.id] = sub_inst
             host_info: Dict[str, Any] = sub_inst.instance_info["host"]

apps/node_man/exceptions.py

@@ -220,3 +220,9 @@ class YunTiPolicyConfigNotExistsError(NodeManBaseException):
     MESSAGE = _("云梯策略配置不存在")
     MESSAGE_TPL = _("云梯策略配置不存在")
     ERROR_CODE = 43
+
+
+class LimitAddHostError(NodeManBaseException):
+    MESSAGE = _("管控区域已被管理员限制新增主机")
+    MESSAGE_TPL = _("管控区域【ID：{id}】已被管理员限制新增主机")
+    ERROR_CODE = 44

apps/node_man/tools/job.py

@@ -10,24 +10,25 @@
 """
 import itertools
 import operator
-from collections import Counter
+from collections import Counter, defaultdict
 from functools import reduce
-from typing import Any, Dict, Iterable, List, Optional, Union
+from typing import Any, Dict, Iterable, List, Optional, Set, Union
 
 from django.conf import settings
 from django.db.models import Q
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
 
 from apps.backend.subscription import tools
-from apps.node_man import constants, models
-from apps.utils import basic
+from apps.node_man import constants, exceptions, models
+from apps.node_man.tools import HostV2Tools
+from apps.utils import basic, batch_request
 from apps.utils.basic import filter_values
 from apps.utils.local import (
     get_request_app_code_or_local_app_code,
     get_request_username,
 )
-from common.api import NodeApi
+from common.api import CCApi, NodeApi
 
 
 class JobTools:
@@ -458,3 +459,138 @@ def get_job_queryset_with_biz_scope(cls, all_biz_info, biz_info, biz_permission,
         job_result = job_result.filter(~Q(bk_biz_scope__isnull=True) & ~Q(bk_biz_scope={}))
 
         return job_result
+
+    @classmethod
+    def query_hosts_by_addressing(cls, host_infos: List[Dict[str, Any]], bk_addressing: str) -> List[Dict[str, Any]]:
+        """
+        按寻址方式查询主机
+        :param host_infos: 主机信息列表
+        :param bk_addressing: 寻址方式
+        :return: 主机信息列表
+        """
+        bk_cloud_id_set: Set[int] = set()
+        bk_host_innerip_set: Set[str] = set()
+        bk_host_innerip_v6_set: Set[str] = set()
+
+        for host_info in host_infos:
+            bk_cloud_id: Optional[int] = host_info.get("bk_cloud_id")
+            # IPv6 / IPv6 可能存在其中一个为空的现象
+            bk_host_innerip: Optional[str] = host_info.get("bk_host_innerip") or host_info.get("inner_ip")
+            bk_host_innerip_v6: Optional[str] = host_info.get("bk_host_innerip_v6") or host_info.get("inner_ipv6")
+
+            if bk_cloud_id is not None:
+                bk_cloud_id_set.add(bk_cloud_id)
+            if bk_host_innerip:
+                bk_host_innerip_set.add(bk_host_innerip)
+            if bk_host_innerip_v6:
+                bk_host_innerip_v6_set.add(bk_host_innerip_v6)
+
+        # 查询空列表对 cc 来说是一个非常耗时的行为，此处进行过滤
+        ip_query_rules: List[Dict[str, Any]] = []
+        if bk_host_innerip_set:
+            ip_query_rules.append({"field": "bk_host_innerip", "operator": "in", "value": list(bk_host_innerip_set)})
+        if bk_host_innerip_v6_set:
+            ip_query_rules.append(
+                {"field": "bk_host_innerip_v6", "operator": "in", "value": list(bk_host_innerip_v6_set)}
+            )
+
+        query_hosts_params: Dict[str, Any] = {
+            "fields": constants.CC_HOST_FIELDS,
+            "host_property_filter": {
+                "condition": "AND",
+                "rules": [
+                    {"field": "bk_addressing", "operator": "equal", "value": bk_addressing},
+                    {"field": "bk_cloud_id", "operator": "in", "value": list(bk_cloud_id_set)},
+                    {"condition": "OR", "rules": ip_query_rules},
+                ],
+            },
+        }
+
+        cmdb_host_infos: List[Dict[str, Any]] = batch_request.batch_request(
+            func=CCApi.list_hosts_without_biz, params=query_hosts_params
+        )
+
+        processed_cmdb_host_infos: List[Dict[str, Any]] = []
+        JobTools.mark_and_replace_multi_inner_ip(exist_cmdb_hosts=cmdb_host_infos)
+
+        host_infos_gby_ip_key: Dict[str, List[Dict[str, Any]]] = JobTools.get_host_infos_gby_ip_key(host_infos)
+        cmdb_host_infos_gby_ip_key: Dict[str, List[Dict[str, Any]]] = JobTools.get_host_infos_gby_ip_key(
+            cmdb_host_infos
+        )
+        # 模糊查询所得的主机信息列表可能出现：同 IP + 不同管控区域的冗余主机
+        # 仅选择原数据中存在的 IP + 管控区域组合
+        for ip_key, partial_cmdb_host_infos in cmdb_host_infos_gby_ip_key.items():
+            if not host_infos_gby_ip_key.get(ip_key):
+                continue
+            processed_cmdb_host_infos.extend(partial_cmdb_host_infos)
+        # 数据按 IP 协议版本进行再聚合，同时存在 v4/v6 的情况下会生成重复项，需要按 主机ID 去重
+        processed_cmdb_host_infos = HostV2Tools.host_infos_deduplication(processed_cmdb_host_infos)
+        return processed_cmdb_host_infos
+
+    @classmethod
+    def mark_and_replace_multi_inner_ip(cls, exist_cmdb_hosts: List[Dict[str, Any]]) -> None:
+        for cmdb_host in exist_cmdb_hosts:
+            bk_host_inner_ip = (cmdb_host.get("bk_host_innerip") or "").split(",")
+            if len(bk_host_inner_ip) > 1:
+                cmdb_host["is_multi_inner_ip"] = True
+                cmdb_host["bk_host_innerip"] = bk_host_inner_ip[0]
+
+    @classmethod
+    def get_host_infos_gby_ip_key(cls, host_infos: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
+        """
+        获取 主机信息根据 IP 等关键信息聚合的结果
+        :param host_infos: 主机信息列表
+        :return: 主机信息根据 IP 等关键信息聚合的结果
+        """
+        host_infos_gby_ip_key: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
+        for host_info in host_infos:
+            bk_cloud_id: int = host_info["bk_cloud_id"]
+            bk_addressing: str = host_info.get("bk_addressing", constants.CmdbAddressingType.STATIC.value)
+            optional_ips: List[Optional[str]] = [
+                host_info.get("bk_host_innerip") or host_info.get("inner_ip"),
+                host_info.get("bk_host_innerip_v6") or host_info.get("inner_ipv6"),
+            ]
+
+            for ip in optional_ips:
+                if not ip:
+                    continue
+                ip_key: str = f"{bk_addressing}:{bk_cloud_id}:{ip}"
+                host_infos_gby_ip_key[ip_key].append(host_info)
+        return host_infos_gby_ip_key
+
+    @classmethod
+    def add_host_cloud_blacklist(cls) -> List[int]:
+        """
+         管控区域新增主机黑名单，用于限制指定管控区域通过安装 Agent 新增主机
+        :return:
+        """
+        add_host_cloud_blacklist: List[int] = models.GlobalSettings.get_config(
+            models.GlobalSettings.KeyEnum.ADD_HOST_CLOUD_BLACKLIST.value, default=[]
+        )
+        return add_host_cloud_blacklist or []
+
+    @classmethod
+    def check_limit_add_host(cls, hosts):
+        host_infos_gby_addressing: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
+        for host in hosts:
+            bk_addressing: str = host.get("bk_addressing", constants.CmdbAddressingType.STATIC.value)
+            host_infos_gby_addressing[bk_addressing].append(host)
+        for addressing, host_infos in host_infos_gby_addressing.items():
+            cmdb_hosts = JobTools.query_hosts_by_addressing(host_infos, addressing)
+            cmdb_host_ips_set = {
+                ip for host in cmdb_hosts for ip in [host.get("bk_host_innerip"), host.get("bk_host_innerip_v6")] if ip
+            }
+            host_ips = {
+                ip for host in host_infos for ip in [host.get("inner_ip"), host.get("inner_ipv6")] if ip is not None
+            }
+            ip_host_map = {host["inner_ip"]: host for host in host_infos if host.get("inner_ip") is not None}
+            ip_host_map.update({host["inner_ipv6"]: host for host in host_infos if host.get("inner_ipv6") is not None})
+            for ip in host_ips:
+                if ip in cmdb_host_ips_set:
+                    continue
+                host = ip_host_map.get(ip)
+                if not host:
+                    continue
+                except_bk_cloud_id = host.get("bk_cloud_id")
+                if except_bk_cloud_id in JobTools.add_host_cloud_blacklist():
+                    raise exceptions.LimitAddHostError(id=except_bk_cloud_id)

apps/node_man/views/job.py

@@ -28,6 +28,7 @@
     OperateSerializer,
     RetrieveSerializer,
 )
+from apps.node_man.tools import JobTools
 from apps.utils.local import get_request_username
 
 JOB_VIEW_TAGS = ["job"]
@@ -128,6 +129,7 @@ def install(self, request):
             "script_hooks": validated_data.get("script_hooks", []),
         }
         extra_config = validated_data.get("agent_setup_info") or {}
+        JobTools.check_limit_add_host(hosts)
         return Response(JobHandler().install(hosts, op_type, node_type, job_type, ticket, extra_params, extra_config))
 
     @swagger_auto_schema(