Source code for panos.firewall

#!/usr/bin/env python

# Copyright (c) 2014, 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.

"""Palo Alto Networks Firewall object"""

import itertools
import logging
import re
import xml.etree.ElementTree as ET
from decimal import Decimal

import panos.errors as err
from panos import device, getlogger, yesno
from panos.base import ENTRY, PanDevice, Root
from panos.base import VarPath as Var

logger = getlogger(__name__)

[docs]class Firewall(PanDevice): """A Palo Alto Networks Firewall This object can represent a firewall physical chassis, virtual firewall, or individual vsys. Args: hostname: Hostname or IP of device for API connections api_username: Username of administrator to access API api_password: Password of administrator to access API api_key: The API Key for connecting to the device's API serial: The serial number of this firewall port: Port of device for API connections vsys: The vsys of this firewall (eg. "vsys1", "vsys2", etc.) is_virtual (bool): Physical or Virtual firewall timeout: The timeout for asynchronous jobs interval: The interval to check asynchronous jobs """ XPATH = "/devices" ROOT = Root.MGTCONFIG SUFFIX = ENTRY NAME = "serial" DEFAULT_VSYS = "vsys1" CHILDTYPES = ( "device.AuthenticationProfile", "device.AuthenticationSequence", "device.Vsys", "device.VsysResources", "device.SystemSettings", "device.LogSettingsSystem", "device.LogSettingsConfig", "device.PasswordProfile", "device.Administrator", "device.Telemetry", "device.SnmpServerProfile", "device.EmailServerProfile", "device.LdapServerProfile", "device.SyslogServerProfile", "device.HttpServerProfile", "device.CertificateProfile", "device.SslDecrypt", "device.LocalUserDatabaseUser", "device.LocalUserDatabaseGroup", "ha.HighAvailability", "objects.AddressObject", "objects.AddressGroup", "objects.ServiceObject", "objects.ServiceGroup", "objects.Tag", "objects.ApplicationObject", "objects.ApplicationGroup", "objects.ApplicationTag", "objects.ApplicationFilter", "objects.ApplicationContainer", "objects.ScheduleObject", "objects.SecurityProfileGroup", "objects.CustomUrlCategory", "objects.LogForwardingProfile", "objects.DynamicUserGroup", "objects.Region", "objects.Edl", "policies.Rulebase", "network.EthernetInterface", "network.AggregateInterface", "network.LoopbackInterface", "network.TunnelInterface", "network.VlanInterface", "network.Layer2Subinterface", "network.Layer3Subinterface", "network.Vlan", "network.VirtualRouter", "network.ManagementProfile", "network.VirtualWire", "network.IkeGateway", "network.IpsecTunnel", "network.IpsecCryptoProfile", "network.IkeCryptoProfile", "network.GreTunnel", "network.Dhcp", "network.Zone", ) def __init__( self, hostname=None, api_username=None, api_password=None, api_key=None, serial=None, port=443, vsys=None, # 'vsys#', 'shared', or None is_virtual=None, multi_vsys=None, *args, **kwargs ): """Initialize PanDevice""" vsys_name = kwargs.pop("vsys_name", None) serial_ha_peer = kwargs.pop("serial_ha_peer", None) management_ip = kwargs.pop("management_ip", None) super(Firewall, self).__init__( hostname, api_username, api_password, api_key, port=port, is_virtual=is_virtual, *args, **kwargs ) # create a class logger self._logger = logging.getLogger(__name__ + "." + self.__class__.__name__) self.serial = serial self._vsys = vsys self.vsys_name = vsys_name self.multi_vsys = multi_vsys self.serial_ha_peer = serial_ha_peer self.management_ip = management_ip self.shared = False """Set to True to act on the shared part of this firewall""" self.state = FirewallState() """Panorama state variables refreshed by Panorama""" def __repr__(self): return "<%s %s %s at 0x%x>" % ( type(self).__name__, repr(, repr(self.vsys), id(self), ) @property def id(self): return self.serial or self.hostname or "<no-id>" @property def vsys(self): # Check if attribute exists because this could be called during # init of the object before 'shared' exists. if hasattr(self, "shared") and self.shared: return None else: return self._vsys @vsys.setter def vsys(self, value): self._vsys = value # Check if attribute exists because this could be called during # init of the object before _ha_peer exists. if hasattr(self, "_ha_peer") and self.ha_peer is not None: self.ha_peer._vsys = value def xpath_vsys(self): return self._root_xpath_vsys(self.vsys) def xpath_panorama(self): raise err.PanDeviceError( "Attempt to modify Panorama configuration on non-Panorama device" )
[docs] def op( self, cmd=None, vsys=None, xml=False, cmd_xml=True, extra_qs=None, retry_on_peer=False, quote='"', ): """Perform operational command on this Firewall Operational commands are most any command that is not a debug or config command. These include many 'show' commands such as ``show system info``. When passing the cmd as a command string (not XML) you must include any non-keyword strings in the command inside double quotes (``"``). Here's some examples:: # The string "facebook-base" must be in quotes because it is not a keyword fw.op('clear session all filter application "facebook-base"') # The string "ethernet1/1" must be in quotes because it is not a keyword fw.op('show interface "ethernet1/1"') # Using an alternative quote character to get DHCP info on ethernet1/1 fw.op('show dhcp client state `ethernet1/1`', quote='`') Args: cmd (str): The operational command to execute vsys (str): Vsys id. Defaults to the vsys of the firewall or the Vsys object in the parent tree. xml (bool): Return value should be a string (Default: False) cmd_xml (bool): True: cmd is not XML, False: cmd is XML (Default: True) extra_qs: Extra parameters for API call retry_on_peer (bool): Try on active Firewall first, then try on passive Firewall quote (str): The quote character when the supplied `cmd` is a string and `cmd_xml=True` Returns: xml.etree.ElementTree: The result of the operational command. May also return a string of XML if xml=True """ if vsys is None: vsys = self.vsys return super(Firewall, self).op( cmd, vsys, xml, cmd_xml, extra_qs, retry_on_peer )
def generate_xapi(self): # Override super class to connect to Panorama # # Connect to this firewall via Panorama with 'target' argument set # to this firewall's serial number. This happens when panorama and serial # variables are set in this firewall prior to the first connection. try: self.panorama() except err.PanDeviceNotSet: return super(Firewall, self).generate_xapi() if self.serial is not None and self.hostname is None: xapi_constructor = PanDevice.XapiWrapper kwargs = { "pan_device": self, "api_key": self.panorama().api_key, "hostname": self.panorama().hostname, "port": self.panorama().port, "timeout": self.timeout, "serial": self.serial, } return xapi_constructor(**kwargs) else: return super(Firewall, self).generate_xapi() def _save_system_info(self, system_info): """Save all the shared system info, plus firewall specific info. Invoked during "refresh_system_info()" """ super(Firewall, self)._save_system_info(system_info) self.content_version = system_info["system"]["app-version"] self.multi_vsys = system_info["system"]["multi-vsys"] == "on"
[docs] def element(self): if self.serial is None: raise ValueError("Serial number must be set to generate element") entry = ET.Element("entry", {"name": self.serial}) if self.parent == self.panorama() and self.serial is not None: # This is a firewall under a panorama if not self.multi_vsys: vsys = ET.SubElement(entry, "vsys") ET.SubElement(vsys, "entry", {"name": "vsys1"}) elif self.parent == self.devicegroup() and self.multi_vsys: # This is a firewall under a device group if self.vsys.startswith("vsys"): vsys = ET.SubElement(entry, "vsys") ET.SubElement(vsys, "entry", {"name": self.vsys}) else: vsys = ET.SubElement(entry, "vsys") all_vsys = self.findall(device.Vsys) for a_vsys in all_vsys: ET.SubElement(vsys, "entry", {"name": a_vsys}) return entry
[docs] def apply(self): return
[docs] def create(self): if self.parent is None: self.create_vsys() return # This is a firewall under a panorama or devicegroup panorama = self.panorama() logger.debug( panorama.hostname + ': create called on %s object "%s"' % (type(self), self.uid) ) panorama.set_config_changed() element = self.element_str() panorama.xapi.set(self.xpath_short(), element)
[docs] def delete(self): if self.parent is None: self.delete_vsys() return panorama = self.panorama() logger.debug( panorama.hostname + ': delete called on %s object "%s"' % (type(self), self.serial) ) if self.parent == self.devicegroup() and self.multi_vsys: # This is a firewall under a devicegroup # Refresh device-group first to see if this is the only vsys devices_xpath = self.devicegroup().xpath() + self.XPATH devices_xml = panorama.xapi.get(devices_xpath) dg_vsys = devices_xml.findall( "result/devices/entry[@name='%s']/vsys/entry" % self.serial ) if dg_vsys: if len(dg_vsys) == 1: # Only vsys, so delete whole entry panorama.set_config_changed() panorama.xapi.delete(self.xpath()) else: # It's not the only vsys, just delete the vsys panorama.set_config_changed() panorama.xapi.delete( self.xpath() + "/vsys/entry[@name='%s']" % self.vsys ) else: # This is a firewall under a panorama panorama.set_config_changed() panorama.xapi.delete(self.xpath()) if self.parent is not None: self.parent.remove_by_name(self.uid, type(self))
[docs] def create_vsys(self): """Create the vsys on the live device that this Firewall object represents""" if self.vsys.startswith("vsys"): element = ET.Element("entry", {"name": self.vsys}) if self.vsys_name is not None: ET.SubElement(element, "display-name").text = self.vsys_name self.set_config_changed() path = self._root_xpath_vsys(None).rsplit("/", 1)[0] self.xapi.set( path, ET.tostring(element, encoding="utf-8"), retry_on_peer=True )
[docs] def delete_vsys(self): """Delete the vsys on the live device that this Firewall object represents""" if self.vsys.startswith("vsys"): self.set_config_changed() self.xapi.delete(self._root_xpath_vsys(self.vsys), retry_on_peer=True)
[docs] def refreshall_from_xml(self, xml, refresh_children=False, variables=None): if len(xml) == 0: return [] if variables is not None: return super(Firewall, self).refreshall_from_xml( xml, refresh_children, variables ) op_vars = ( Var("serial"), Var("multi-vsys", vartype="bool"), Var("vsys_id", "vsys", default="vsys1"), Var("vsys_name"), Var("ha/state/peer/serial", "serial_ha_peer"), ) if len(xml[0]) > 1: # This is a 'show devices' op command firewall_instances = super(Firewall, self).refreshall_from_xml( xml, refresh_children=False, variables=op_vars ) # Add system settings to firewall instances for fw in firewall_instances: entry = xml.find("entry[@name='%s']" % fw.serial) system = fw.find_or_create(None, device.SystemSettings) system.hostname = entry.findtext("hostname") system.ip_address = entry.findtext("ip-address") if entry.findtext("ipv6-address") != "unknown": system.ipv6_address = entry.findtext("ipv6-address") # Add state fw.state.connected = yesno(entry.findtext("connected")) fw.state.unsupported_version = yesno( entry.findtext("unsupported-version") ) fw._set_version_and_version_info(entry.findtext("sw-version")) fw.content_version = entry.findtext("app-version") else: # This is a config command # For each vsys, instantiate a new firewall firewall_instances = [] all_serial = xml.findall("entry") for entry in all_serial: all_vsys = entry.findall("vsys/entry") if all_vsys: for vsys in all_vsys: firewall_instances.append( Firewall(serial=entry.get("name"), vsys=vsys.get("name")) ) else: firewall_instances.append(Firewall(serial=entry.get("name"))) return firewall_instances
def show_system_resources(self): self.xapi.op(cmd="show system resources", cmd_xml=True) result = self.xapi.xml_root() if self._version_info >= (9, 0, 0): regex = re.compile( r"load average: ([\d\.]+).*? ([\d\.]+) id,.*KiB Mem :\s+(\d+) total,.*? (\d+) free", re.DOTALL, ) else: regex = re.compile( r"load average: ([\d.]+).* ([\d.]+)%id.*Mem:.*?([\d.]+)k total.*?([\d]+)k free", re.DOTALL, ) match = if match: """ return cpu, mem_free, load """ return { "load": Decimal(, "cpu": 100 - Decimal(, "mem_total": int(, "mem_free": int(, } else: raise err.PanDeviceError( "Problem parsing show system resources", pan_device=self ) def commit_device_and_network(self, sync=False, exception=False): return self._commit( sync=sync, exclude="policy-and-objects", exception=exception ) def commit_policy_and_objects(self, sync=False, exception=False): return self._commit( sync=sync, exclude="device-and-network", exception=exception )
[docs] def organize_into_vsys(self, create_vsys_objects=True, refresh_vsys=True): """Organizes all imported objects under the appropriate Vsys object. Args: create_vsys_objects (bool): Create the vsys objects (True) or use the ones already connected to this firewall (False). refresh_vsys (bool): Refresh all vsys objects' parameters before doing the reorganization or not. This is assumed True if create_vsys_objects is True. """ from panos import network # Mapping of device.Vsys params to pan-os-python classes. mapping = { "interface": network.Interface, "vlans": network.Vlan, "virtual_wires": network.VirtualWire, "virtual_routers": network.VirtualRouter, } # Optional: create the vsys objects. if create_vsys_objects: device.Vsys.refreshall(self, name_only=True) # Vsys to put objects into. available_vsys = [x for x in self.children if isinstance(x, device.Vsys)] # Optional: refresh the vsys params. if create_vsys_objects or refresh_vsys: for x in available_vsys: x.refresh(refresh_children=False) # List of objects we need to iterate over. parents = self.children[:] # Reorganize into vsys. for x in itertools.chain(parents): # Skip device.Vsys children. if isinstance(x, device.Vsys): continue # Add children for later processing. parents.extend(x.children) # Check this class against the importable classes. for param, importable_class in mapping.items(): if isinstance(x, importable_class): # Importable class found, check if it should be moved. for vsys in available_vsys: if getattr(vsys, param) is not None and x.uid in getattr( vsys, param ): # If its vsys isn't right, move it. if x.vsys != vsys.uid: x.parent.remove(x) vsys.add(x) break else: # Checked every vsys, this importable isn't in any of # them (vsys is None), so move this node to be a child # of the firewall. if x.parent != self: x.parent.remove(x) self.add(x) break
class FirewallState(object): def __init__(self): self.connected = None self.shared_policy_synced = None self.unsupported_version = None def set_shared_policy_synced(self, sync_status): if sync_status == "In Sync": self.shared_policy_synced = True elif sync_status == "Out of Sync": self.shared_policy_synced = False elif not sync_status: self.shared_policy_synced = None else: raise err.PanDeviceError( "Unknown shared policy status: %s" % str(sync_status) )
[docs]class FirewallCommit(object): """Normalization of a firewall commit. Instances of this class can be passed in to ``Firewall.commit()`` (inherited from :meth:`panos.base.PanDevice.commit()`) as the ``cmd`` parameter. Args: description (str): The commit message. admins (list): (PAN-OS 8.0+) List of admins whose changes are to be committed. exclude_device_and_network (bool): Set to True to exclude device and network changes. exclude_shared_objects (bool): Set to True to exclude shared objects changes. exclude_policy_and_objects (bool): Set to True to exclude policy and objects changes. force (bool): Set to True to force a commit even if one is not needed. """ def __init__( self, description=None, admins=None, exclude_device_and_network=False, exclude_shared_objects=False, exclude_policy_and_objects=False, force=False, ): self.description = description self.admins = admins if admins and not isinstance(admins, list): raise ValueError("admins must be a list") self.exclude_device_and_network = exclude_device_and_network self.exclude_shared_objects = exclude_shared_objects self.exclude_policy_and_objects = exclude_policy_and_objects self.force = force @property def commit_action(self): return None def is_partial(self): pp_list = [ self.admins, self.exclude_device_and_network, self.exclude_shared_objects, self.exclude_policy_and_objects, self.force, ] return any(x for x in pp_list) def element_str(self): return ET.tostring(self.element(), encoding="utf-8")
[docs] def element(self): """Returns an xml representation of the commit requested. Returns: xml.etree.ElementTree """ root = ET.Element("commit") if self.description: ET.SubElement(root, "description").text = self.description if self.is_partial(): partial = ET.Element("partial") if self.admins: e = ET.SubElement(partial, "admin") for name in self.admins: ET.SubElement(e, "member").text = name if self.exclude_device_and_network: ET.SubElement(partial, "device-and-network").text = "excluded" if self.exclude_shared_objects: ET.SubElement(partial, "shared-object").text = "excluded" if self.exclude_policy_and_objects: ET.SubElement(partial, "policy-and-objects").text = "excluded" if self.force: fe = ET.SubElement(root, "force") fe.append(partial) else: root.append(partial) return root