feat: Add Cloud Services plugin support

Authored-by: Bastien Migette <[email protected]>

This adds support for Prisma Access / Cloud Services Panorama plugin.

.gitignore

@@ -6,6 +6,9 @@ __pycache__/
 # C extensions
 *.so
 
+#vscode file containing some env variables
+launch.json
+
 # Distribution / packaging
 .Python
 env/

.vscode/settings.json

@@ -1,31 +1,31 @@
 {
-  "files.exclude": {
-    "**/__pycache__": true
-  },
-  "editor.codeActionsOnSave": {
-    "source.organizeImports": true
-  },
-  "python.testing.unittestEnabled": false,
-  "python.testing.nosetestsEnabled": false,
-  "python.testing.pytestEnabled": true,
-  "python.testing.pytestArgs": ["--no-cov"],
-  "python.linting.enabled": true,
-  "python.linting.pylintEnabled": false,
-  "python.linting.banditEnabled": true,
-  "python.linting.banditArgs": ["-r", "--ini .bandit"],
-  "python.linting.flake8Enabled": true,
-  "python.linting.flake8CategorySeverity.E": "Warning",
-  "python.linting.mypyEnabled": false,
-  "python.formatting.provider": "black",
-  "editor.formatOnSave": true,
-  "restructuredtext.confPath": "${workspaceFolder}/docs",
-  "cSpell.words": [
-    "Vlan",
-    "etree",
-    "nosuffix",
-    "onlink",
-    "pandevice",
-    "pan-os-python",
-    "refreshall"
-  ]
-}
+    "files.exclude": {
+      "**/__pycache__": true
+    },
+    "editor.codeActionsOnSave": {
+      "source.organizeImports": true
+    },
+    "python.testing.unittestEnabled": false,
+    "python.testing.nosetestsEnabled": false,
+    "python.testing.pytestEnabled": true,
+    "python.testing.pytestArgs": ["--no-cov"],
+    "python.linting.enabled": true,
+    "python.linting.pylintEnabled": false,
+    "python.linting.banditEnabled": true,
+    "python.linting.banditArgs": ["-r", "--ini .bandit"],
+    "python.linting.flake8Enabled": true,
+    "python.linting.flake8CategorySeverity.E": "Warning",
+    "python.linting.mypyEnabled": false,
+    "python.formatting.provider": "black",
+    "editor.formatOnSave": true,
+    "restructuredtext.confPath": "${workspaceFolder}/docs",
+    "cSpell.words": [
+      "Vlan",
+      "etree",
+      "nosuffix",
+      "onlink",
+      "pandevice",
+      "pan-os-python",
+      "refreshall"
+    ]
+  }

examples/prisma_access_create_remote_network.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python
+
+# Copyright (c) 2022, Palo Alto Networks
+#
+# Permission to use, copy, modify, and/or distribute this software for any
+# purpose with or without fee is hereby granted, provided that the above
+# copyright notice and this permission notice appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+# Author: Bastien Migette <[email protected]>
+
+"""
+prisma_access_create_remote_network.py
+==========
+
+This script is an example on how to create a prisma access Remote Network,
+along with needed IPSEC Tunnel and IKEv2 Gateway.
+To use the script, you need to replace the variables below with desired values.
+
+"""
+import logging
+import os
+import sys
+
+# This is needed to import module from parent folder
+curpath = os.path.dirname(os.path.abspath(__file__))
+sys.path[:0] = [os.path.join(curpath, os.pardir)]
+
+
+from panos.panorama import Panorama
+from panos.plugins import (
+    CloudServicesPlugin,
+    RemoteNetwork,
+    RemoteNetworks,
+    Bgp,
+    AggBandwidth,
+)
+from panos.network import IkeGateway, IpsecTunnel
+from panos.panorama import Template
+
+__author__ = "bmigette"
+
+
+HOSTNAME = os.environ["PAN_HOSTNAME"]
+USERNAME = os.environ["PAN_USERNAME"]
+PASSWORD = os.environ["PAN_PASSWORD"]
+
+IPSEC_PEER = "1.2.3.4"
+BGP_PEER = "1.2.3.4"
+BGP_PEER_AS = 65123
+IPSEC_TUNNEL_NAME = "panos-sdk-tunnel"
+IKE_GW = "panos-sdk-ikev2-gw"
+IKE_PSK = "Secret123"
+IKE_CRYPTO = "Generic-IKE-Crypto-Default"
+IPSEC_CRYPTO = "Generic-IPSEC-Crypto-Default"
+TEMPLATE = "Remote_Network_Template"
+
+REMOTE_NETWORK_NAME = "panos-sdk-rn"
+# This is the Region that you put in the RN. A compute region can have multiple Regions
+REMOTE_NETWORK_REGION = "eu-central-1"
+# This is the Compute Region, used to get SPN list. You can use Panorama CLI to get available options
+REMOTE_NETWORK_COMPUTEREGION = "europe-central"
+
+
+def get_region_spn(remote_networks, region):
+    """This function will return first SPN from a given region name.
+    You should implement some logic here to get the correct SPN.
+    The script will break if the region has no SPN / BW allocated
+
+    Args:
+        remote_networks (RemoteNetworks): RemoteNetworks Object
+        region (str): The region to get SPN from
+
+    Returns:
+        str: spn name
+    """
+    agg_bw = remote_networks.findall(AggBandwidth)
+    region_obj = agg_bw[0].find(region)
+    print(f"SPN for region {region}: {region_obj.spn_name_list[0]}")
+    return region_obj.spn_name_list[0]
+
+
+def main():
+    # Setting logging to debug the PanOS SDK
+    logging_format = "%(levelname)s:%(name)s:%(message)s"
+    # logging.basicConfig(format=logging_format, level=logging.DEBUG - 2) #Use this to be even more verbose
+    logging.basicConfig(format=logging_format, level=logging.DEBUG)
+    # 1 - let's create the panorama  object that we want to modify.
+    pan = Panorama(HOSTNAME, USERNAME, PASSWORD)
+
+    # 2 - Refreshing Prisma Access config
+    csp = pan.add(CloudServicesPlugin())
+    csp.refresh()
+
+    rn_template = pan.add(Template(name=TEMPLATE))
+    rn_template.refresh()
+    # 3 - Getting the remote_networks object
+    remote_networks = csp.findall(RemoteNetworks)[0]
+
+    # 4 - Creating IKEv2 GW and IPSEC Tunnels
+    # 4.1 - IKEv2 GW
+    gw = IkeGateway(
+        name=IKE_GW,
+        version="ikev2",
+        peer_ip_type="ip",
+        peer_ip_value=IPSEC_PEER,
+        peer_id_type="ipaddr",
+        peer_id_value=IPSEC_PEER,
+        auth_type="pre-shared-key",
+        pre_shared_key=IKE_PSK,
+        ikev2_crypto_profile=IKE_CRYPTO,
+        enable_liveness_check=True,
+    )
+    rn_template.add(gw).create()
+
+    # 4.2 - IPSEC Tunnel
+    ipsec_tun = IpsecTunnel(
+        name=IPSEC_TUNNEL_NAME,
+        ak_ike_gateway=IKE_GW,
+        ak_ipsec_crypto_profile=IPSEC_CRYPTO,
+        mk_remote_address=IPSEC_PEER,
+    )
+
+    rn_template.add(ipsec_tun).create()
+
+    # 5 - Creating Remote Network
+    rn = RemoteNetwork(
+        name=REMOTE_NETWORK_NAME,
+        static_routes=["10.11.12.0/24"],
+        region=REMOTE_NETWORK_REGION,
+        spn_name=get_region_spn(remote_networks, REMOTE_NETWORK_COMPUTEREGION),
+        ipsec_tunnel=IPSEC_TUNNEL_NAME,
+    )
+    bgp = Bgp(enable=True, peer_as=BGP_PEER_AS, peer_ip_address=BGP_PEER)
+
+    rn.add(bgp)
+    remote_networks.add(rn).create()
+    # 6 - Commit + Push
+    # pan.commit_all(devicegroup="Remote_Network_Device_Group") #commit + push
+    pan.commit()  # commit only
+
+
+if __name__ == "__main__":
+    main()

examples/prisma_access_list_RN_regions_bw.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python
+
+# Copyright (c) 2022, Palo Alto Networks
+#
+# Permission to use, copy, modify, and/or distribute this software for any
+# purpose with or without fee is hereby granted, provided that the above
+# copyright notice and this permission notice appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+# ACTION OF CONTRACT, NEGLIGENCE OR OTpHER TORTIOUS ACTION, ARISING OUT OF
+# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+# Author: Bastien Migette <[email protected]>
+
+"""
+prisma_access_list_RN_regions_bw.py
+==========
+
+This script is an example on how to retrieve list of prisma access 
+remote networks locations and bandwidth allocation and print it.
+
+"""
+__author__ = "bmigette"
+
+
+import logging
+import os
+import sys
+
+
+# This is needed to import module from parent folder
+curpath = os.path.dirname(os.path.abspath(__file__))
+sys.path[:0] = [os.path.join(curpath, os.pardir)]
+
+
+from panos.plugins import (
+    CloudServicesPlugin,
+    RemoteNetwork,
+    RemoteNetworks,
+    AggBandwidth,
+    Region,
+)
+from panos.panorama import Panorama
+from panos.base import PanDevice
+
+
+curpath = os.path.dirname(os.path.abspath(__file__))
+sys.path[:0] = [os.path.join(curpath, os.pardir)]
+
+
+HOSTNAME = os.environ["PAN_HOSTNAME"]
+USERNAME = os.environ["PAN_USERNAME"]
+PASSWORD = os.environ["PAN_PASSWORD"]
+
+
+def main():
+    # Setting logging to debug the PanOS SDK
+    logging_format = "%(levelname)s:%(name)s:%(message)s"
+    # logging.basicConfig(format=logging_format, level=logging.DEBUG - 2) #Use this to be even more verbose
+    logging.basicConfig(format=logging_format, level=logging.DEBUG)
+    # First, let's create the panorama  object that we want to modify.
+    pan = Panorama(HOSTNAME, USERNAME, PASSWORD)
+    csp = pan.add(CloudServicesPlugin())
+
+    csp.refresh()
+
+    rn = csp.findall(RemoteNetworks)
+    rnes = rn[0].findall(RemoteNetwork)
+    agg_bw = rn[0].findall(AggBandwidth)
+
+    regions = agg_bw[0].findall(Region)
+    ### Print XML Dump of Prisma Config ###
+    print(csp.element_str())
+    print(csp.about())
+
+    ### Print Remote networks name ###
+    print(" -- Remote Networks --")
+    for rne in rnes:
+        print(
+            f"{rne.name} - spn: {rne.spn_name}, region: {rne.region}, tunnel {rne.ipsec_tunnel}, subnets: {rne.subnets}"
+        )
+        print(
+            f"{rne.name} - secondary_wan: {rne.secondary_wan_enabled}, secondary ipsec tunnel: {rne.secondary_ipsec_tunnel}"
+        )
+
+    ### Print Regions BW ###
+    print(f"Agg BW Enabled: {agg_bw[0].enabled}")
+    print(" -- Regions --")
+    print(regions)
+    for region in regions:
+        print(
+            f"Region:  {region}, allocated_bw: {region.allocated_bw}, spns: {region.spn_name_list}"
+        )
+
+
+if __name__ == "__main__":
+    main()

panos/base.py

@@ -1002,6 +1002,8 @@ def _refresh_children(self, running_config=False, xml=None):
                 import panos.panorama
             elif module_name == "policies":
                 import panos.policies
+            elif module_name == "plugins":
+                import panos.plugins
             child = getattr(getattr(panos, module_name), class_name)()
 
             # Versioned objects need a PanDevice to get the version from, so

panos/panorama.py

@@ -396,6 +396,7 @@ class Panorama(base.PanDevice):
         "panorama.DeviceGroup",
         "panorama.Template",
         "panorama.TemplateStack",
+        "plugins.CloudServicesPlugin"
     )
 
     def __init__(