使用 API 管理 IP 位址範圍

使用 Microsoft Defender for Cloud Apps 中的資料豐富 API,以程式方式建立、更新及刪除 IP 位址範圍。 本文說明了 API 的請求與回應參數,並提供一個 Python 腳本,可以從 CSV 檔案讀取 IP 位址範圍,並與你的租戶同步。

用腳本管理 IP 位址範圍

請執行以下步驟,使用 Python 腳本管理租戶中的 IP 位址範圍:

  1. 建立包含下列預期欄位的CSV檔案:名稱、IP_Address_Ranges、類別、標籤 (ID) 和Override_ISP_Name。
    以下是 CSV 檔案內容的範例:

    name, ip_address_ranges, category, tag(id), isp_name
    Test 1,200.200.200.200/24 201.201.201.201/24,1, ,
    Test 2,1.1.1.5/32 31.31.31.4/24,1,000000290000000000000000 0000002d0000000000000000  ,
    Test 3,2.2.2.2/32,1,0000002d0000000000000000,
    Test 4,5.6.5.4/32,2,,test
    
  2. 更新下列指令碼變數的值: OPTION_DELETE_ENABLEDIP_RANGES_BASE_URLCSV_ABSOLUTE_PATHYOUR_TOKEN

  3. 執行指令碼以建立新記錄,並使用相符的名稱更新現有規則。

    重要事項

    如果您將 OPTION_DELETE_ENABLED 設定為 True,則腳本會從租用戶中刪除租用戶中定義但不存在於 CSV 檔案中的任何 IP 位址範圍。 如果你把 OPTION_DELETE_ENABLED 設為 True,請確保 CSV 檔定義了你租戶想要的所有 IP 位址範圍。

要求本文參數

請求體支援以下參數:

  • "filters":使用請求的所有搜尋篩選器來篩選物件。 如需詳細資訊,請參閱:資料擴充篩選器。 若要避免您的請求受到節流,請務必在查詢中包含限制。
  • “limit”:整數。 在掃描模式下,500 到 5000 之間 (默認為 500) 。 控制用於掃描所有資料的疊代次數。

回應參數

回應包含以下參數:

  • “data”:傳回的資料。 每次迭代最多包含「限制」數量的記錄。 如果有更多記錄要提取 (hasNext=true) ,則會捨棄最後幾筆記錄,以確保所有資料只列出一次。
  • 「hasNext」:布林值。 表示是否需要對資料進行另一次疊代。
  • 「nextQueryFilters」:如果需要進行下一次迭代,則包含接續要執行的 JSON 查詢。 在下一個請求中使用「nextQueryFilters」值作為「filters」參數。

這個 Python 範例使用 CSV 檔案的內容來管理(建立、更新或刪除)Defender for Cloud Apps 環境中的 IP 位址範圍。

import csv
import http
import requests
import json
 
OPTION_DELETE_ENABLED = False
IP_RANGES_BASE_URL = 'https://<tenant_id>.<tenant_region>.portal.cloudappsecurity.com/api/v1/subnet/'
IP_RANGES_UPDATE_SUFFIX = 'update_rule/'
IP_RANGES_CREATE_SUFFIX = 'create_rule/'
CSV_ABSOLUTE_PATH = 'rules.csv'
YOUR_TOKEN = '<your_token>'
HEADERS = {
  'Authorization': 'Token {}'.format(YOUR_TOKEN),
  'Content-Type': 'application/json'
}
 
# Get all records.
def get_records():
  list_request_data = {
    # Optionally, edit to match your filters
    'filters': {},
    "skip": 0,
    "limit": 20
  }
  records = []
  has_next = True
  while has_next:
    response = requests.post(IP_RANGES_BASE_URL, json=list_request_data, headers=HEADERS)
    if response.status_code != http.HTTPStatus.OK:
      raise Exception(f'Error getting existing subnets from tenant. Stopping script run. Error: {response.content}')
    content = json.loads(response.content)
    response_data = content.get('data', [])
    records += response_data
    has_next = content.get('hasNext', False)
    list_request_data["skip"] += len(response_data)
  return records
 
# Rule fields are compared to the CSV row.
def rule_matching(record, ip_address_ranges, category, tag, isp_name, ):
  new_tags = sorted([new_tag for new_tag in tag.split(' ') if new_tag != ''])
  existing_tags = sorted([existing_tag.get('id') for existing_tag in record.get('tags', [])])
  rule_exists_conditions = [sorted([subnet.get('originalString', False) for subnet in record.get('subnets', [])]) !=
  sorted(ip_address_ranges.split(' ')),
  str(record.get('category', False)) != category,
  existing_tags != new_tags,
  bool(record.get('organization', False)) != bool(isp_name) or
  (record.get('organization', False) is not None and not isp_name)]
  if any(rule_exists_conditions):
    return False
  return True
 
def create_update_rule(name, ip_address_ranges, category, tag, isp_name, records, request_data):
  for record in records:
    # Records are compared by name(unique).
    # This can be changed to id (to update the name), it will include adding id to the CSV and changing row shape.
    if record["name"] == name:
    # request_data["_tid"] = record["_tid"]
      if not rule_matching(record, ip_address_ranges, category, tag, isp_name):
        # Update existing rule
        request_data['_id'] = record['_id']
        response = requests.post(f"{IP_RANGES_BASE_URL}{record['_id']}/{IP_RANGES_UPDATE_SUFFIX}", json=request_data, headers=HEADERS)
        if response.status_code == http.HTTPStatus.OK:
          print('Rule updated', request_data)
        else:
          print('Error updating rule. Request data:', request_data, ' Response:', response.content)
          json.loads(response.content)
        return record
      else:
        # The exact same rule exists. no need for change.
        print('The exact same rule exists. no need for change. Rule name: ', name)
        return record
  # Create new rule.
  response = requests.post(f"{IP_RANGES_BASE_URL}{IP_RANGES_CREATE_SUFFIX}", json=request_data, headers=HEADERS)
  if response.status_code == http.HTTPStatus.OK:
    print('Rule created:', request_data)
  else:
    print('Error creating rule. Request data:', request_data, ' Response:', response.content)
    # added record
    return record
 
# Request data creation.
def create_request_data(name, ip_address_ranges, category, tag, isp_name):
  tags = [new_tag for new_tag in tag.split(' ') if new_tag != '']
  request_data = {"name": name, "subnets": ip_address_ranges.split(' '), "category": category, "tags": tags}
  if isp_name:
    request_data["overrideOrganization"] = True
    request_data["organization"] = isp_name
  return request_data
 
def main():
  # CSV fields are: Name,IP_Address_Ranges,Category,Tag(id),Override_ISP_Name
  # Multiple values (eg: multiple subnets) will be space-separated. (eg: value1 value2)
  records = get_records()
  with open(CSV_ABSOLUTE_PATH, newline='\n') as your_file:
    reader = csv.reader(your_file, delimiter=',')
    # move the reader object to point on the next row, headers are not needed
    next(reader)
    for row in reader:
      name, ip_address_ranges, category, tag, isp_name = row
      request_data = create_request_data(name, ip_address_ranges, category, tag, isp_name)
      if records:
      # Existing records were retrieved from your tenant
        record = create_update_rule(name, ip_address_ranges, category, tag, isp_name, records, request_data)
        record_id = record['_id']
      else:
        # No existing records were retrieved from your tenant
        response = requests.post(f"{IP_RANGES_BASE_URL}{IP_RANGES_CREATE_SUFFIX}", json=request_data, headers=HEADERS)
        if response.status_code == http.HTTPStatus.OK:
          record_id = json.loads(response.content)
          print('Rule created:', request_data)
        else:
          print('Error creating rule. Request data:', request_data, ' Response:', response.content)
      if OPTION_DELETE_ENABLED:
        # Remove CSV file record from tenant records.
        if record_id:
          for record in records:
            if record['_id'] == record_id:
              records.remove(record)
  if OPTION_DELETE_ENABLED:
    # Delete remaining tenant records, i.e. records that aren't in the CSV file.
    for record in records:
      requests.delete(f"{IP_RANGES_BASE_URL}{record['_id']}/", headers=HEADERS)
 
if __name__ == '__main__':
  main()

後續步驟

關於如何保護您的 Defender for Cloud Apps 部署,請參閱保護組織的最佳實務

如果您遇到任何問題,我們隨時為您提供協助。 如需產品問題的協助或支援,請聯絡 Defender for Cloud Apps 客服