Skip to main content

Bulk Change to apply SPG Workflow


In this section we present a workflow example for bulk change apply Security Profile group to all allowed security rules in a PAN-OS configuration. The sample script is stored in /policies folder of the Expedition-API script container.

Below flowchart demo the workflow and the related API calls in each of the steps:


Step 1. Obtain the API Keys

Refer to Obtaining the API Keys section to obtain a valid API key stored in the hed variable.

Step 2. Add PAN-OS Device

Making a POST call to the Device route, we can create a Device with a desired name. Notice that we attach the credentials hed in the CURL headers to present our credentials and verify we have permission to create a device.

API syntax for creating a new device :

MethodRouteRequest Body Parameters
POSThttps://localhost/api/v1/device{"name": devicename, "serial": serialnumber,"hostname": panosip,"type": devicetype,"description": pandescription,
}
Examplehttps://localhost/api/v1/device{"name":"mypanorama", "serial":"123456789","hostname":"pan1","type":"vm-panorama""description":"Project for testing"}
info

The supported device "type" are below:

        "m100",
"m500",
"m600",
"vm-panorama",
"pa200",
"pa220",
"pa500",
"pa800",
"pa3000",
"pa3200",
"pa4000",
"pa5000",
"pa5200",
"pa7000",
"vm-series"
print("*****Add a new PAN-OS device\n")
panosip = '1.1.1.1'
serialnumber = '123412'
devicetype = "pa220"
pandescription = 'test'
url = "https://localhost/api/v1/device"
data = {
"name": devicename,
"serial": serialnumber,
"hostname": panosip,
"type": devicetype,
"description": pandescription,
"port":device_port
}
r = requests.post(url, data=data, verify=False, headers=hed)
response = r.json()
success = json.dumps(response["success"])
if success == "true":
print("New device created successfully" + " \n")
DeviceId = json.dumps(response['data']['id'])
print("Your Device-ID is " + str(DeviceId) + " \n")
else:
print("Unable to create the device")
print("*****Upload PAN-OS config into device*****\n")

Step 3. Create an Expedition Project

In the large amount of automation cases, we will require having an Expedition project. Making a POST call to the project route, we can create a project with a desired name.

API syntax for creating a new project:

MethodRouteParameters
POSThttps://localhost/api/v1/projectin url
{ "project":"project1", "description":"Project for testing" }
examplehttps://localhost/api/v1/project{"project":"MyLittleProject", "description":"A migration project"}
print("***** Create new project *****\n")

url = "https://localhost/api/v1/project"
data = {
"name": PROJECT_NAME,
"description": PROJECT_DESCRIPTION,
"device_id[0]": DeviceId,
}
r = requests.post(url, data=data, verify=False, headers=hed)
response = r.json()
success = json.dumps(response["success"])
if success == "true":
print("New project created successfully" + " \n")
ProjectId = int(json.dumps(response['data']['id']))
print("Your Project-ID is " + str(ProjectId) + " \n")
else:
print("Unable to create the project")

Step 4. Upload PAN-OS config into device

After device has been created , the next step will be uploading your pan-os config to associate with the device.

API syntax for upload PAN-OS config into device :

MethodRoutePath Parameters
POST"https://localhost/api/v1/device/{device_id}/upload_config""device_id": device_Id
Example"https://localhost/api/v1/device/23/upload_config"
print("*****Upload PAN-OS config into device*****\n")
Vendorfile = open(PANOS_CONFIG_PATH, 'rb')
files = {'config': Vendorfile}
url= 'https://localhost/api/v1/device/'+str(DeviceId)+'/upload_config'
r = requests.post(url, files=files, data=data, verify=False, headers=hed)
response = r.json()
success = json.dumps(response["success"])
if success == "true":
print("Upload configuration successfully" + " \n")
else:
print("Unable to upload the configuration")
print(response)

Step 5. Import the PAN-OS configuration of your device to the project

This step will associate the device with project. The API response will contain a job ID , you can then use API call to check job status. Please refer to checking job status Checking Job Status section .

API syntax for the step:

MethodRouteParameters
POSThttps://localhost/api/v1/project/{project_id}/import/devicein url
"project_id":project_Id
in_body
{"device_id":device_Id }
examplehttps://localhost/api/v1/project/22/import/device{"device_id":"23""}
print("*****Import the PAN-OS configuration of your device to the project\n")
url = 'https://localhost/api/v1/project/'+str(ProjectId)+'/import/device'
data = {
"device_id": DeviceId,
}
r = requests.post(url, verify=False, headers=hed, data=data)
response = r.json()
success = json.dumps(response["success"])
if success == "true":
jobId = json.dumps(response['data']['job_id'])
print("Job id: "+jobId)
print("***** Wait for job to finish *****")
wait_for_job(EXPEDITION_URL+"job/" + jobId + "?complete=true", jobId, hed)

else:
print(response)
print("Unable to import configuration ")

Step 6. Get Source ID of the config file

In this step, we will make a API call to get source_id of the config file that's been imported to the project. After this API call, you will parse the response that contains source_id. The source_id represent the pan-os config file that you would like to work on, and it will be used in the subsequent API calls.

API syntax for the step:

MethodRouteParameters
GEThttps://localhost/api/v1/project/{project_id}/sourcein url
"project_id":project_Id
in_body
{"device_id":device_Id}
examplehttps://localhost/api/v1/project/22/source{"device_id": 23}
print("Get Source_ID of the config file")
url = 'https://localhost/api/v1/project/'+str(ProjectId)+'/source'
r = requests.get(url, verify=False, headers=hed)
response = r.json()
success = json.dumps(response["success"])
if success == "true":
sourceId = json.dumps(response['data']['source'][0]['id'])
print("Source id: "+sourceId)

else:
print("Unable to get sources ")

Step 7. Create a filter for all allowed security rules

In this step, we will create a filter for all security rules that have action "allowed" . Please refer to the Expedition-API Filters section for details on filters. In this specific example, we are sending the request body contains below data, In the json response, you will get a filter_id , this filter_id will be used in the subsequent steps.

data = {
"name": "all allowed rules",
"reference": "all allowed rules",
"filter": "[security_rule] action equals \"allow\"",
"source": sourceId
}

API syntax for the step:

MethodRouteParameters
POSThttps://localhost/api/v1/project/{project_id}/tools/filterin url
"project_id":project_Id
in_body
{"name":filter_name,"filter":filter, "reference":filter_reference,"source":sourceId, }
examplehttps://localhost/api/v1/project/22/tools/filter{"name": "all allowed rules", "filter" : "[security_rule] action equals \"allow\"","reference": "all allowed rules", "source" :4486 }
print("***** create a filter *****")
#Filter Variables
FILTER_NAME="allallowedrules"
FILTER_DESC="All Security Policy with allowed action"
FILTER_OBJECT="[security_rule] action equals \"allow\""
url='https://localhost/api/v1/project/'+str(ProjectId)+'/tools/filter'
data = {
"name": FILTER_NAME,
"reference": FILTER_NAME,
"description" : FILTER_DESC,
"filter": FILTER_OBJECT,
"source":sourceId
}
r = requests.post(url, data=data, verify=False, headers=hed)
response = r.json()
print(response)
FILTER_ID = json.dumps(response['data']['last_history_entry']['filter_id'])
FILTER_ID=FILTER_ID.replace('"', '')
print('Your Filter ID is :'+FILTER_ID)

Step 8. Execute the filter

After create a filter, we will execute the filter based on filter_Id , in the request body, you will need to provide "source_id" obtained from the previous step as required parameter.

API syntax for the step:

MethodRouteParameters
POSThttps://localhost/api/v1/project/{project_id}/tools/filter/{filter_id}/executein url
"project_id":project_Id, "filter_id":filter_Id
in_body
{"source_id": source_id of the PAN-OS config file}
examplehttps://localhost/api/v1/project/22/tools/filter/1/execute{"source_id": "23564"}
print("***** Execute a filter *****")
url='https://localhost/api/v1/project/'+str(ProjectId)+'/tools/filter/'+FILTER_ID+'/execute'
data={"source_id": sourceId}
r = requests.post(url, data=data, verify=False, headers=hed)
response = r.json()
jobId = json.dumps(response["data"]["job_id"])
success = json.dumps(response["success"])
if success == "true":
jobId = json.dumps(response['data']['job_id'])
print("Job id: "+jobId)
print("***** Wait for job to finish *****")
wait_for_job(EXPEDITION_URL+"job/" + jobId + "?complete=true", jobId, hed)

else:
print(response)
print("Unable to execute filter ")

Step 9. Print the filter execution result

After the filter is executed , we can view the result using below API call. The result should contain all the objects that matches with the filter. From the response, you will parse the collection_Id to be used in the subsequent API call.

API syntax for the step:

MethodRouteParameters
GEThttps://localhost/api/v1/project/{project_id}/tools/filter/{filter_id}/successin url
"project_id":project_Id, "filter_id":filter_Id
examplehttps://localhost/api/v1/project/22/tools/filter/1/success
print("***** Print the Filter Execution Result *****")
url = 'https://localhost/api/v1/project/'+'project/'+str(ProjectId)+'/tools/filter/'+FILTER_ID+'/success'
r = requests.get(url, verify=False, headers=hed)
response = r.json()
print(response)
#Print the Collection ID
Collection_ID = json.dumps(response["data"]["id"])
print('Your Collection ID is :'+Collection_ID)

Step 10. Print the Collection Content

After the filter is executed , we can print the collection content using below API call.

API syntax for the step:

MethodRouteParameters
GEThttps://localhost/api/v1/project/{project_id}/collection/{Collection_Id}/contentin url
"project_id":project_Id, "collection_id":collection_Id
examplehttps://localhost/api/v1/project/22/collection/20793/content
print("***** Print the Collection that contain unused objects *****")
url = 'https://localhost/api/v1/project/'+str(ProjectId) + '/collection/'+Collection_ID+'/content'
r = requests.get(url, verify=False, headers=hed)
response=r.json()
print(response)

Step 11. Get Security Profile Group ID

In order to apply the security profile group to the security policy, we will need to find out the object ID of the security profile group first. In the example, we will parse the first object ID from the response.

API syntax for the step:

MethodRouteParameters
GEThttps://localhost/api/v1/project/{project_id}/object/profile_groupin url
"project_id":project_Id
examplehttps://localhost/api/v1/project/22/object/profile_group
print("***** Get Security Profile group ID *****")
url = 'https://localhost/api/v1/project/'+str(ProjectId)+ "/object/profile_group"
r = requests.get(url,verify=False, headers=hed)
response=r.json()
SPG_ID=json.dumps(response["data"]["profile_group"][0]["id"])
print(SPG_ID)

Step 12. Bulk Change Apply SPG to all allowed rules

The final step we perform a bulk change to apply the security profile group to all allowed rules.

API syntax for the step:

MethodRouteParameters
PUThttps://localhost/api/v1/project/{project_id}/policy/securityin url
"project_id":project_Id
in_body
{"add[profile][0]": object_id of the SPG, "id": collection_id}
examplehttps://localhost/api/v1/project/22/policy/security{"add[profile][0]": "11714", "id":"20793"}
print("Bulk Change Apply SPG to all allowed rules")
url = 'https://localhost/api/v1/project/'+str(ProjectId)+ "/policy/security"
data = {"add[profile][0]": int(SPG_ID), "id": int(Collection_ID)}
r = requests.put(url, data=data, verify=False, headers=hed)
response = r.json()
print(response)