使用 API 管理 IP 位址範圍
您可以使用資料擴充 API 來管理 IP 位址範圍。
使用管理IP位址範圍腳本
建立具有下列預期欄位的 CSV 檔案：Name、IP_Address_Ranges、Category、Tag (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更新下列腳本變數的值： OPTION_DELETE_ENABLED、 IP_RANGES_BASE_URL、 CSV_ABSOLUTE_PATH、 YOUR_TOKEN
重要事項
如果您將 OPTION_DELETE_ENABLED 設為 True，腳本會從租用戶中刪除租用戶中定義但不存在於 CSV 檔案中的任何 IP 位址範圍。 如果您使用此選項，請確定 CSV 檔案會定義您想要在租使用者中使用的所有 IP 位址範圍。
執行文本以建立新的記錄，並使用相符的名稱更新現有的規則。
要求本文參數
- “filters”： Filter objects with all the search filters for the request， see Data Enrichment filters for more information. 若要避免要求受到節流，請務必在查詢中包含限制。
- “limit”： Integer。 在掃描模式中，500 到 5000 (預設為 500) 。 控制掃描所有資料的反覆項目數目。
回應參數
- “data”：傳回的數據。 每個反覆專案最多會包含「限制」記錄數目。 如果在 hasNext=true) (提取更多記錄，則會卸除最後幾筆記錄，以確保所有數據只會列出一次。
- “hasNext”： Boolean。 表示是否需要數據上的另一個反覆專案。
- “nextQueryFilters”：如果需要另一個反復專案，則會包含要執行的連續 JSON 查詢。 在下一個要求中使用此作為 「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()
後續步驟
如果您遇到任何問題，我們在這裡提供協助。 若要取得產品問題的協助或支援，請 開啟支援票證。