firewall.py

#!/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.
#
# 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.


"""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__)


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.AdvancedRoutingEngine",
        "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.LogicalRouter",
        "network.Vrf",
        "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(self.id),
            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"
        )

    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"

    def element(self, with_children=True, comparable=False):
        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

    def apply(self):
        return

    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)

    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))

    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
            )

    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)

    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 :"
                r"\s+(\d+) total,.*? (\d+) free.*? (\d+) used,"
                r".*? (\d+) buff/cache",
                re.DOTALL,
            )
        else:
            regex = re.compile(
                r"load average: ([\d.]+).* ([\d.]+)%id.*"
                r"Mem:.*?([\d.]+)k total.*?"
                r"([\d]+)k free.*?([\d]+)k used.*?([\d]+)k buff/cache",
                re.DOTALL,
            )
        match = regex.search(result)
        if match:
            return {
                "load": Decimal(match.group(1)),
                "cpu": 100 - Decimal(match.group(2)),
                "mem_total": int(match.group(3)),
                "mem_free": int(match.group(4)),
                "mem_used": int(match.group(5)),
                "mem_buffer": int(match.group(6)),
            }
        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
        )

    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)
            )


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")

    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