自动导入基于威胁情报源的IP黑名单

操作步骤

1. 已购买云防火墙，防火墙实例处于开启状态，且已开启流量封堵功能。
   1. 参见购买云防火墙选购标准版或专业版防火墙。
   2. 开启流量封堵。
      1. 在云防火墙控制台的左侧导航栏中，选择“访问控制 > 流量过滤”，进入流量封堵页面。
      2. 单击按钮，并在弹出的确认框中，单击“确定”，开启流量封堵功能。
2. 获取必要的认证和参数信息。
   • 访问密钥（AK/SK）：获取方式请参见获取AK/SK。
   • 项目ID（PROJECT_ID）：
     1. 登录华为云控制台，鼠标移动至右上方的用户名，在下拉列表中选择“我的凭证”。
     2. 在“API凭证”页面的“项目列表”中，获取“项目ID”。
   • 防火墙实例ID（FW_INSTANCE_ID）：
     1. 登录云防火墙控制台。
     2. 单击管理控制台左上角的，选择区域。
     3. （可选）切换防火墙实例：如果有多个防火墙实例，可以在页面左上角的下拉框中进行切换。
     4. 在“总览”页面中，单击右上角的“防火墙列表”，并在弹出的防火墙列表页面中获取目标防火墙实例的实例ID。
3. 使用自动导入基于威胁情报源的IP黑名单接口，批量导入IP黑名单。

   调用代码示例如下：

   # 依赖安装: pip install requests huaweicloudsdkcfw huaweicloudsdkcore
   import requests
   import re
   # 配置参数（AK/SK等信息建议通过环境变量读取）
   AK = "your-access-key"
   SK = "your-secret-key"
   PROJECT_ID = "your-project-id"
   FW_INSTANCE_ID = "your-fw-id"
   REGION = "cn-south-1"
   EFFECT_SCOPE = [1]              # 生效范围：[1]互联网边界, [2]VPC边界
   URL = "https://{your-threat-source}/blacklist.txt"   # 威胁情报URL
   BATCH_SIZE = 2000               # 每批导入的IP数量
   # 初始化认证和客户端
   credentials = BasicCredentials(AK, SK, PROJECT_ID)
   client = CfwClient.new_builder() \
       .with_credentials(credentials) \
       .with_region(CfwRegion.value_of(REGION)) \
       .build()
   def import_ip_batch(client, fw_instance_id, ip_set, effect_scope, add_type=0):
       """批量导入IP到CFW黑名单"""
       if not ip_set:
           return 0
       ip_list_str = "\n".join(ip_set)
       request = ImportIpBlacklistRequest()
       request.fw_instance_id = fw_instance_id
       body = ImportIpBlacklistRequestBody()
       body.add_type = add_type  # 0=增量导入, 1=全量覆盖
       body.ip_blacklist = ip_list_str
       body.effect_scope = effect_scope  # [1]互联网边界, [2]VPC边界, [1,2]同时生效
       request.body = body
       try:
           response = client.import_ip_blacklist(request)
           return len(ip_set)
       except exceptions.ClientRequestException as e:
           print(f"Error: status={e.status_code}, msg={e.error_msg}")
           return 0
   # 下载并提取IP
   batch_ips = set()
   all_unique_ips = set()
   IP_PATTERN = re.compile(
       r'\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}'
       r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(?:/(?:[12]?[0-9]|3[0-2]))?\b'
   )
   with requests.get(URL, timeout=60, stream=True) as response:
       response.raise_for_status()
       for line in response.iter_lines(decode_unicode=True):
           if not line:
               continue
           line = line.strip().lstrip('#')
           if not line:
               continue
           for ip in IP_PATTERN.findall(line):
               if ip not in all_unique_ips:
                   all_unique_ips.add(ip)
                   batch_ips.add(ip)
           if len(batch_ips) >= BATCH_SIZE:
               import_ip_batch(client, FW_INSTANCE_ID, batch_ips, EFFECT_SCOPE)
               batch_ips.clear()
   # 处理剩余IP
   if batch_ips:
       import_ip_batch(client, FW_INSTANCE_ID, batch_ips, EFFECT_SCOPE)
   print(f"Total imported: {len(all_unique_ips)} unique IPs")

操作验证

1. 确认执行输出中导入的IP数量符合预期。
2. 检查新增的IP黑名单条目是否添加成功。
   1. 登录云防火墙控制台。
   2. 单击管理控制台左上角的，选择区域。
   3. （可选）切换防火墙实例：如果有多个防火墙实例，可以在页面左上角的下拉框中进行切换。
   4. 在左侧导航栏选择“访问控制 > 流量过滤”，进入流量封堵页面。
   5. 检查新增的黑名单信息是否添加成功。

API查询示例

以下示例展示如何调用云防火墙API查询当前IP黑名单列表：

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkcfw.v1.region.cfw_region import CfwRegion
from huaweicloudsdkcfw.v1 import *
from huaweicloudsdkcfw.v1.model import *

# 初始化认证和客户端
credentials = BasicCredentials(AK, SK, PROJECT_ID)
client = CfwClient.new_builder() \
    .with_credentials(credentials) \
    .with_region(CfwRegion.value_of(REGION)) \
    .build()

# 查询IP黑名单列表
request = ListIpBlacklistRequest()
request.fw_instance_id = FW_INSTANCE_ID
request.offset = 0
request.limit = 100

try:
    response = client.list_ip_blacklist(request)
    print(f"Total blacklisted IPs: {response.data.total}")
    for item in response.data.records:
        print(f"  IP: {item.ip}, Type: {item.type}, Created: {item.create_time}")
except Exception as e:
    print(f"Query failed: {str(e)}")

API调用关键接口说明