Automatically Importing IP Blacklist Items Based on Threat Intelligence Sources

Procedure

1. Purchase a CFW instance. Start the firewall instance and enable traffic blocking.
   1. Buy the standard or professional edition. For details, see Buying CFW.
   2. Enable traffic blocking.
      1. In the navigation pane on the left, choose Protection Policies > Access Control > Traffic Filtering. The Traffic Blocking tab is displayed.
      2. Click . In the displayed dialog box, click OK to enable traffic blocking.
2. Obtain required authentication and parameter information.
   • Access keys (AK/SK): For details about how to obtain the access keys, see How Do I Obtain an Access Key (AK/SK)?
   • Project ID (PROJECT_ID)
     1. Log in to the Huawei Cloud console, hover the cursor to the username in the upper right corner, and click My Credentials.
     2. On the API Credentials page, obtain the Project ID from the Projects list.
   • Firewall instance ID (FW_INSTANCE_ID)
     1. Log in to the CFW console.
     2. Click in the upper left corner of the management console and select a region or project.
     3. (Optional) Switch to another firewall instance. If there are multiple firewall instances, you can select a desired instance from the drop-down list in the upper left corner of the page.
     4. On the Dashboard page, click Firewall List in the upper right corner. On the displayed page, obtain the ID of the firewall instance.
3. Import IP addresses to the blacklist in batches. Call the API for Importing an IP Address Blacklist for Traffic Filtering.

   Example code:

   # Install dependency: pip install requests huaweicloudsdkcfw huaweicloudsdkcore
   import requests
   import re
   # Configure parameters. (You are advised to read information such as AK/SK from environment variables.)
   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]              # Effective scope: [1] Internet border, [2] VPC border
   URL = "https://{your-threat-source}/blacklist.txt" # Threat intelligence URL
   BATCH_SIZE = 2000 # Number of IP addresses to be imported in each batch
   # Initialize the authentication and the client.
   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):
       """Import IP addresses to the CFW blacklist in batches."""
       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: incremental import; 1: full overwrite.
       body.ip_blacklist = ip_list_str
       body.effect_scope = effect_scope # [1] Internet border, [2] VPC border, [1,2] Both
       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
   # Download and extract the IP address.
   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()
   # Process remaining IP addresses.
   if batch_ips:
       import_ip_batch(client, FW_INSTANCE_ID, batch_ips, EFFECT_SCOPE)
   print(f"Total imported: {len(all_unique_ips)} unique IPs")

Verification

1. Verify that the number of imported IP addresses in the command output matches expectations.
2. Confirm that the new IP address blacklist entries have been added.
   1. Log in to the CFW console.
   2. Click in the upper left corner of the management console and select a region or project.
   3. (Optional) Switch to another firewall instance. If there are multiple firewall instances, you can select a desired instance from the drop-down list in the upper left corner of the page.
   4. In the navigation pane on the left, choose Protection Policies > Access Control > Traffic Filtering. The Traffic Blocking tab is displayed.
   5. Check whether the added blacklist IP addresses are displayed.

API Query Example

The following example shows how to call a CFW API to query the current IP address blacklist.

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

# Initialize the authentication and the client.
credentials = BasicCredentials(AK, SK, PROJECT_ID)
client = CfwClient.new_builder() \
    .with_credentials(credentials) \
    .with_region(CfwRegion.value_of(REGION)) \
    .build()

# Query the IP address blacklist.
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)}")

Main APIs