Crea script Python per chiamate API REST a FDM

Introduzione

Questo documento descrive un esempio di utilizzo di Python per effettuare chiamate all'API Rest.

Prerequisiti

Gli strumenti e i dispositivi utilizzati nella guida sono:

• Cisco Firepower Threat Defense (FTD)
• Cisco Firepower Device Management (FDM)
• Mac OS
• Testo Sublime

Requisiti

Cisco raccomanda la conoscenza dei seguenti argomenti:

• HTTPS
• API Rest
• Python
• Json
• Gestione dispositivi Firepower
• LDAP

Componenti usati

Le informazioni fornite in questo documento si basano sulle seguenti versioni software e hardware:

• Python 3.1.4
• API FDM versione 6
• FTDv 7.3.1 

Le informazioni discusse in questo documento fanno riferimento a dispositivi usati in uno specifico ambiente di emulazione. Su tutti i dispositivi menzionati nel documento la configurazione è stata ripristinata ai valori predefiniti. Se la rete è operativa, valutare attentamente eventuali conseguenze derivanti dall'uso dei comandi.

Premesse

Lo scopo principale di questo documento è di guidare l'utente attraverso i passaggi per la creazione di uno script Python per effettuare chiamate API.

La funzionalità specifica da regolare e creare è LDAP e mappe attributi.

Le chiamate API eseguite nella guida sono: 

GET: raccolta di informazioni dal server

POST: creazione di un nuovo oggetto sul server

PUT: aggiornare un oggetto esistente sul server

DELETE: rimuove un oggetto esistente dal server.

Configurazione

OTTIENI

Esempio GET API REST per raccogliere dati sugli oggetti esistenti:

import requests
import json
import certifi
import urllib3
from pprint import pprint
from getpass import getpass

urllib3.disable_warnings() 

#Data collection
u = input ('Input username: ')
p = getpass(prompt='Input password: ')
url = input('Input Hostname or IP address: ')
protocol = 'https://'

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
def auth():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  payload = {
        'grant_type' : 'password',
        'username' : u,
        'password': p
        }

  response = requests.post(token_url, data=payload, verify=False)
  
  #ONlY USE "verify=False" in a lab setting!

#Error Checking
  if response.status_code == 400:
    raise Exception("Error Received: {}".format(response.content))
  else:
    access_token = response.json()['access_token']
    return access_token

token = auth()
#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def getactivedirectory():
  uri = "/api/fdm/v6/object/realms"
  ad_url = protocol+url+uri

  headers = {
    "Content-Type": "application/json",
    "Accept": "application/json",
    "Authorization":"Bearer {}".format(token)
  }

  response = requests.get(ad_url, headers=headers, verify=False)
  if response.status_code == 200:
    AD = response.json()
    return AD
  else:
    print(response.status_code)
    pass
  

pprint(getactivedirectory())

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def revoke():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  headers = {
      "Content-Type": "application/json",
      "Accept": "application/json",
      "Authorization":"Bearer"
  }
  payload = {
        'grant_type' : 'revoke_token',
        "access_token": token,
        "token_to_revoke": token,
        }

  response = requests.post(token_url, data=payload, verify=False)
  if response.status_code == 200:
    print("Access token revoked")
  else:
    print(response.status_code)


revoke()

POST

In questa sezione viene descritto un esempio di richiesta POST dell'API REST per la creazione di un nuovo oggetto mappa attributi LDAP:

import requests
import json
import certifi
import urllib3
from pprint import pprint
from getpass import getpass

urllib3.disable_warnings() 

#Data collection
u = input ('Input username: ')
p = getpass(prompt='Input password: ')
url = input('Input Hostname or IP address: ')
protocol = 'https://'

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
def auth():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  payload = {
        'grant_type' : 'password',
        'username' : u,
        'password': p
        }

  response = requests.post(token_url, data=payload, verify=False)
  
  #WARNING ONLY USE "verify=False" in a lab setting!

#Error Checking
  if response.status_code == 400:
    raise Exception("Error Received: {}".format(response.content))
  else:
    access_token = response.json()['access_token']
    return access_token

token = auth()

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
def postattributemap():
  uri = "/api/fdm/v6/object/ldapattributemaps"
  ad_url = protocol+url+uri


  name = input('Object name> ')
  group_policy = input('Group-Policy> ')
  base = input('LDAP DN> ')

  headers = {
    "Content-Type": "application/json",
    "Accept": "application/json",
    "Authorization":"Bearer {}".format(token)
  }

  payload = {
  "name": name,
  "ldapAttributeMaps": [
    {
      "ldapName": "memberOf",
      "ciscoName": "GROUP_POLICY",
      "valueMappings": [
        {
          "ldapValue": base,
          "ciscoValue": group_policy,
          "type": "ldaptociscovaluemapping"
        }
      ],
      "type": "ldapattributemapping"
    }
  ],
    "type": "ldapattributemap"
  }
  data = json.dumps(payload)

  response = requests.post(ad_url, headers=headers, data=data, verify=False)

  if response.status_code == 200:
    print(response.status_code)
    print("Created LDAP attribute map")
    map = response.json
    return map
  else:
    print(response.status_code)
    pprint(response.content)
    pass



postattributemap()



#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def revoke():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  headers = {
      "Content-Type": "application/json",
      "Accept": "application/json",
      "Authorization":"Bearer"
  }
  payload = {
        'grant_type' : 'revoke_token',
        "access_token": token,
        "token_to_revoke": token,
        }

  response = requests.post(token_url, data=payload, verify=False)
  if response.status_code == 200:
    print("Access token revoked")
  else:
    print(response.status_code)


revoke()

PUT

Questa sezione mostra un esempio di script Python che effettua una chiamata all'API PUT Rest. Questa funzione aggiunge la mappa degli attributi LDAP alle configurazioni di Active Directory esistenti.

import requests
import json
import certifi
import urllib3
from pprint import pprint
from getpass import getpass

urllib3.disable_warnings() 

#Data collection
u = input ('Input username: ')
p = getpass(prompt='Input password: ')
url = input('Input Hostname or IP address: ')
protocol = 'https://'

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
def auth():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  payload = {
        'grant_type' : 'password',
        'username' : u,
        'password': p
        }

  response = requests.post(token_url, data=payload, verify=False)
  
  #WARNING ONLY USE "verify=False" in a lab setting!

#Error Checking
  if response.status_code == 400:
    raise Exception("Error Received: {}".format(response.content))
  else:
    access_token = response.json()['access_token']
    return access_token

token = auth()
#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def put():
  objId = input('object id of active directory object> ')
  uri = "/api/fdm/v6/object/realms/"
  ad_url = protocol+url+uri+objId

  headers = {
    "Content-Type": "application/json",
    "Accept": "application/json",
    "Authorization":"Bearer {}".format(token)
  }
#Place the GET response from the active directory here in "payload" along with the added "ldapAttributeMap", "id" and ""type": "ldapattributemap"" 
  payload = {
  "version": "p6ueo2w2aulkf",
  "name": "Test",
  "directoryConfigurations": [
    {
      "hostname": "{omitted}",
      "port": 389,
      "encryptionProtocol": "NONE",
      "encryptionCert": None,
      "interface": { 
        "version": "mjvylmnd52agk",
        "name": "diagnostic",
        "hardwareName": "Management0/0",
        "id": "a46ef70c-06ca-11ee-9be1-bd712e622992",
        "type": "physicalinterface"
      },
      "type": "directoryconfiguration"
    }
  ],
  "enabled": True,
  "realmId": 3,
  "dirUsername": "{omitted}",
  "dirPassword": "*********",
  "baseDN": "dc={omitted},dc=com",
  "ldapAttributeMap": {
    "id": "7fbf5798-27c9-11ee-a635-a1f4b2c2e66b",
    "type": "ldapattributemap"
  },
  "adPrimaryDomain": "{omitted}.com",
  "id": "5957a304-2662-11ee-a635-a5df7d28e8c4",
  "type": "activedirectoryrealm"
  }

  data = json.dumps(payload)


  response = requests.put(ad_url, headers=headers, data=data, verify=False)

  if response.status_code == 200:
    print("Updated Object")
  else:
    pprint("Error Received: {}".format(response.content))
    pass

put()



#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def revoke():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  headers = {
      "Content-Type": "application/json",
      "Accept": "application/json",
      "Authorization":"Bearer"
  }
  payload = {
        'grant_type' : 'revoke_token',
        "access_token": token,
        "token_to_revoke": token,
        }

  response = requests.post(token_url, data=payload, verify=False)

  if response.status_code == 200:
    print("Access token revoked")
  else:
    print(response.status_code)


revoke()

ELIMINA

In questa sezione viene descritto un esempio di eliminazione dell'API Rest per rimuovere l'oggetto Active Directory:

import requests
import json
import certifi
import urllib3
from pprint import pprint
from getpass import getpass

urllib3.disable_warnings() 

#Data collection
u = input ('Input username: ')
p = getpass(prompt='Input password: ')
url = input('Input Hostname or IP address: ')
protocol = 'https://'
 
#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
def auth():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  payload = {
        'grant_type' : 'password',
        'username' : u,
        'password': p
        }

  response = requests.post(token_url, data=payload, verify=False)
  
  #ONlY USE "verify=False" in a lab setting!

#Error Checking
  if response.status_code == 400:
    raise Exception("Error Received: {}".format(response.content))
  else:
    access_token = response.json()['access_token']
    return access_token

token = auth()
#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def delete():
  objId = input('object id to delete> ')
  uri = "/api/fdm/v6/object/realms/"
  ad_url = protocol+url+uri+objId


  headers = {
    "Content-Type": "application/json",
    "Accept": "application/json",
    "Authorization":"Bearer {}".format(token)
  }

  response = requests.delete(ad_url, headers=headers, verify=False)

  
  if response.status_code == 204:
    print('Object removed')
  else:
    print("Error Received: {}".format(response.content))
    pass

delete()

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def revoke():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  headers = {
      "Content-Type": "application/json",
      "Accept": "application/json",
      "Authorization":"Bearer"
  }
  payload = {
        'grant_type' : 'revoke_token',
        "access_token": token,
        "token_to_revoke": token,
        }

  response = requests.post(token_url, data=payload, verify=False)

  if response.status_code == 200:
    print("Access token revoked")
  else:
    print(response.status_code)


revoke()

IMPLEMENTAZIONE

In questa sezione viene descritto un esempio di distribuzione delle modifiche di configurazione tramite una chiamata API POST REST e viene confermato lo stato delle distribuzioni con una richiesta GET.

import requests
import json
import certifi
import urllib3
from pprint import pprint
from getpass import getpass

urllib3.disable_warnings() 

#Data collection
u = input ('Input username: ')
p = getpass(prompt='Input password: ')
url = input('Input Hostname or IP address: ')
protocol = 'https://'
 

x="0"
attempts=0


#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
def auth():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  payload = {
        'grant_type' : 'password',
        'username' : u,
        'password': p
        }

  response = requests.post(token_url, data=payload, verify=False)
  
  #ONlY USE "verify=False" in a lab setting!

#Error Checking
  if response.status_code == 400:
    raise Exception("Error Received: {}".format(response.content))
  else:
    access_token = response.json()['access_token']
    return access_token


token = auth()

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++


def deploy():

  uri = '/api/fdm/v6/operational/deploy'
  deploy = protocol+url+uri

  headers = headers = {
    "Content-Type": "application/json",
    "Accept": "application/json",
    "Authorization":"Bearer {}".format(token)
  }

  payload = {
  "statusMessage": "string",
  "cliErrorMessage": "string",
  "state": "QUEUED",
  "queuedTime": 0,
  "startTime": 0,
  "endTime": 0,
  "statusMessages": [
    "string"
    ],
  "id": "string",
  "name": "string",
  "modifiedObjects": {},
  "forceRefreshDeploymentData": False,
  "type": "deploymentstatus"
  }

  data = json.dumps(payload)

  response = requests.get(deploy, headers=headers, data=data,verify=False)
  if response.status_code == 200:
    print(response.status_code)
    status = response.json()['items'][0]['statusMessage']
    return status
  else:
    print(response.status_code)
    pass
    
status = deploy()

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++


def deploy2():

  uri = '/api/fdm/v6/operational/deploy'
  deploy = protocol+url+uri

  headers = headers = {
    "Content-Type": "application/json",
    "Accept": "application/json",
    "Authorization":"Bearer {}".format(token)
  }

  payload = {
  "statusMessage": "string",
  "cliErrorMessage": "string",
  "state": "QUEUED",
  "queuedTime": 0,
  "startTime": 0,
  "endTime": 0,
  "statusMessages": [
    "string"
    ],
  "id": "string",
  "name": "string",
  "modifiedObjects": {},
  "forceRefreshDeploymentData": False,
  "type": "deploymentstatus"
  }

  data = json.dumps(payload)

  response = requests.post(deploy, headers=headers, data=data,verify=False)
  if response.status_code == 200:
    print(response.status_code)
    print('Deploying')
  else:
    print(response.status_code)
    pass





#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def revoke():

  uri = '/api/fdm/v6/fdm/token'
  token_url = protocol+url+uri

  headers = {
      "Content-Type": "application/json",
      "Accept": "application/json",
      "Authorization":"Bearer"
  }
  payload = {
        'grant_type' : 'revoke_token',
        "access_token": token,
        "token_to_revoke": token,
        }

  response = requests.post(token_url, data=payload, verify=False)
  if response.status_code == 200:
    print("Access token revoked")
  else:
    print(response.status_code)
#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++






while True:
  x = input("""
Press 1 for deployment status
Press 2 to deploy
Enter Exit to exit
> """).lower()
  if x == "1":
    deploy()
    pprint('Status is: ' + status)
  elif x == "2":
    deploy2()
  elif x == "exit":
    break
    revoke()
  elif x != "1" or "2" or "exit":
    attempts += 1
    print("The options are 1, 2 or Exit only!")
    if attempts == 3:
      print('Closing the program')
      revoke()
      break
  else:
    raise Exception("Program shutting down")

Verifica

OTTIENI

Con questo script, viene ricevuta una risposta al codice HTTP "200" e viene presentato l'output, come mostrato nell'immagine seguente.

Output dello script

PUT

È pronta una nuova distribuzione nell'interfaccia utente di FDM con tentativi riusciti, come mostrato nell'immagine. 

Schermata della pagina di amministrazione di FDM

La conferma può essere confermata anche dal codice di risposta HTTP 200.

ELIMINA

È pronta una nuova distribuzione nell'interfaccia utente di FDM con tentativi riusciti. 

La conferma può essere confermata anche dal codice di risposta HTTP 204.

Risoluzione dei problemi

In questa sezione viene descritto come risolvere i problemi.

Il codice di risposta HTTP può fornire informazioni sui problemi specifici rilevati:

Errori client 4XX

400- Richiesta non valida

401 - Non autorizzato

403 - Accesso negato

404- Non trovato

408- Timeout richiesta

415- Tipo di supporto non supportato

Errori server 5xx

500- Errore interno del server

501- Non implementato

502- Gateway non valido

503- Servizio non disponibile

Informazioni correlate

RFC 2616: Protocollo HTTP/1.1 per il trasferimento ipertestuale 

Introduzione all'API REST Firepower Threat Defense - FTD-API Reference v6(FTD v7.2) - Documento - Cisco Developer