Administración de intervalos de direcciones IP mediante la API

Use las API de enriquecimiento de datos en Microsoft Defender for Cloud Apps para crear, actualizar y eliminar intervalos de direcciones IP mediante programación. En este artículo se describen los parámetros de solicitud y respuesta de la API y se proporciona un script de Python que lee intervalos de direcciones IP de un archivo CSV y los sincroniza con el inquilino.

Administración de intervalos de direcciones IP con el script

  1. Cree un archivo CSV con los siguientes campos esperados: Name, IP_Address_Ranges, Category, Tag(id) y Override_ISP_Name.
    Este es un ejemplo del contenido del archivo 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. Actualice los valores de las siguientes variables de script: OPTION_DELETE_ENABLED, IP_RANGES_BASE_URL, CSV_ABSOLUTE_PATH, YOUR_TOKEN

  3. Ejecute el script para crear nuevos registros y actualizar las reglas existentes con el nombre coincidente.

    Importante

    Si establece OPTION_DELETE_ENABLED en True, el script eliminará del tenant todos los rangos de direcciones IP que estén definidos en su tenant pero que no existan en los archivos CSV. Si usa esta opción, asegúrese de que el archivo CSV define todos los intervalos de direcciones IP que desea en el inquilino.

Parámetros del cuerpo de la solicitud

El cuerpo de la solicitud admite los parámetros siguientes:

  • "filters": filtre objetos con todos los filtros de búsqueda de la solicitud. Para obtener más información, vea: Filtros de enriquecimiento de datos para obtener más información. Para evitar que las solicitudes estén limitadas, asegúrese de incluir una limitación en la consulta.
  • "limit": entero. En el modo de escaneo, entre 500 y 5000 (el valor predeterminado es 500). Controla el número de iteraciones usadas para examinar todos los datos.

Parámetros de respuesta

La respuesta incluye los parámetros siguientes:

  • "data": los datos devueltos. Contendrá hasta "límite" de registros cada iteración. Si hay más registros que extraer (hasNext=true), los últimos registros se quitan para asegurarse de que todos los datos se muestran solo una vez.
  • "hasNext": Boolean. Indica si se necesita otra iteración en los datos.
  • "nextQueryFilters": si se necesita otra iteración, contiene la consulta JSON consecutiva que se va a ejecutar. Úselo como parámetro "filters" en la siguiente solicitud.

En este Python ejemplo se usa el contenido de un archivo CSV para administrar (crear, actualizar o eliminar) intervalos de direcciones IP en el entorno de Defender for Cloud Apps.

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()

Pasos siguientes

Para obtener instrucciones sobre cómo proteger la implementación de Defender for Cloud Apps, consulte Procedimientos recomendados para proteger su organización.

Si tiene algún problema, estamos aquí para ayudarle. Para obtener ayuda o soporte técnico para el problema del producto, abra una incidencia de soporte técnico.