Gérer les plages d’adresses IP à l’aide de l’API

Utilisez les API d’enrichissement des données dans Microsoft Defender for Cloud Apps pour créer, mettre à jour et supprimer des plages d’adresses IP par programmation. Cet article décrit les paramètres de requête et de réponse de l’API et fournit un script Python qui lit les plages d’adresses IP à partir d’un fichier CSV et les synchronise avec votre locataire.

Gérer les plages d’adresses IP avec le script

Effectuez les étapes suivantes pour utiliser le script Python afin de gérer les plages d’adresses IP dans votre locataire :

  1. Créez un fichier CSV avec les champs attendus suivants : Nom, IP_Address_Ranges, Catégorie, Balise(id) et Override_ISP_Name.
    Voici un exemple du contenu du fichier 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. Mettez à jour les valeurs des variables de script suivantes : OPTION_DELETE_ENABLED, IP_RANGES_BASE_URL, CSV_ABSOLUTE_PATH, YOUR_TOKEN

  3. Exécutez le script pour créer de nouveaux enregistrements et mettre à jour les règles existantes avec le nom correspondant.

    Important

    Si vous définissez OPTION_DELETE_ENABLED sur True, toutes les plages d’adresses IP définies dans votre locataire mais qui n’existent pas dans les fichiers CSV sont supprimées du locataire par le script. Si vous mettez OPTION_DELETE_ENABLED sur Vrai, assurez-vous que le fichier CSV définit toutes les plages d’adresses IP que vous souhaitez dans votre locataire.

Paramètres du corps de la requête

Le corps de la requête prend en charge les paramètres suivants :

  • « filtres » : filtrez les objets avec tous les filtres de recherche pour la requête. Pour plus d’informations, consultez : Filtres d’enrichissement des données pour plus d’informations. Pour éviter que vos demandes soient limitées, veillez à inclure une limitation sur votre requête.
  • « limit » : entier. En mode analyse, entre 500 et 5 000 (la valeur par défaut est 500). Contrôle le nombre d’itérations utilisées pour analyser toutes les données.

Paramètres de réponse

La réponse inclut les paramètres suivants :

  • « data » : les données renvoyées. Contient jusqu'au nombre d'enregistrements spécifié par « limit » à chaque itération. S’il y a plus d’enregistrements à extraire (hasNext=true), les derniers enregistrements sont supprimés pour garantir que toutes les données ne sont répertoriées qu’une seule fois.
  • « hasNext » : booléen. Indique si une autre itération sur les données est nécessaire.
  • « nextQueryFilters » : si une autre itération est nécessaire, elle contient la requête JSON consécutive à exécuter. Utilisez la valeur « nextQueryFilters » comme paramètre « filters » dans la requête suivante.

Cet exemple Python utilise le contenu d’un fichier CSV pour gérer (créer, mettre à jour ou supprimer) des plages d’adresses IP dans votre environnement 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()

Étapes suivantes

Pour obtenir des conseils sur la sécurisation de votre déploiement de Defender for Cloud Apps, consultez les meilleures pratiques pour protéger votre organisation.

Si vous rencontrez des problèmes, nous sommes là pour vous aider. Pour obtenir de l’aide ou un support concernant votre problème de produit, veuillez contacter le support Defender for Cloud Apps.