feat: Add panos_config_element

Fixes #219 

Co-authored-by: Michael Richardson <[email protected]>

Pipfile

@@ -7,6 +7,8 @@ name = "pypi"
 pan-os-python = "*"
 xmltodict = "==0.12.0"
 requests = "==2.22.0"
+pytest-xdist = "*"
+pytest-mock = "*"
 
 [dev-packages]
 ansible = "*"
@@ -16,6 +18,8 @@ pytest = "*"
 coverage = "==4.5.4"
 pycodestyle = "*"
 pylint = "*"
+pytest-xdist = "*"
+pytest-mock = "*"
 voluptuous = "*"
 yamllint = "*"
 sphinx = "*"

fix-pytest-ini.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+
+# Taken from: https://raw.githubusercontent.com/sensu/sensu-go-ansible/master/fix-pytest-ini.py
+
+from __future__ import absolute_import, division, print_function
+
+import os.path
+
+from ansible_test._internal.util import ANSIBLE_TEST_DATA_ROOT
+
+__metaclass__ = type
+
+
+with open(os.path.join(ANSIBLE_TEST_DATA_ROOT, "pytest.ini")) as fd:
+    lines = fd.readlines(True)
+
+with open(os.path.join(ANSIBLE_TEST_DATA_ROOT, "pytest.ini"), "w") as fd:
+    fd.writelines(line for line in lines if line.strip()[0] != "#")

plugins/module_utils/panos.py

@@ -35,14 +35,13 @@
 _MIN_VERSION_ERROR = "{0} version ({1}) < minimum version ({2})"
 HAS_PANDEVICE = True
 try:
+    import panos
     from panos.base import PanDevice
     from panos.device import Vsys
     from panos.errors import PanCommitNotNeeded, PanDeviceError
     from panos.firewall import Firewall
     from panos.panorama import DeviceGroup, Template, TemplateStack
     from panos.policies import PostRulebase, PreRulebase, Rulebase
-
-    import panos
 except ImportError:
     try:
         import pandevice as panos

plugins/modules/panos_config_element.py

@@ -0,0 +1,337 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+#  Copyright 2019 Palo Alto Networks, Inc
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+DOCUMENTATION = """
+---
+module: panos_config_element
+short_description: Modifies an element in the PAN-OS configuration.
+description:
+    - This module allows the user to modify an element in the PAN-OS configuration
+      by specifying an element and its location in the configuration (xpath).
+author:
+    - 'Nathan Embery (@nembery)'
+    - 'Michael Richardson (@mrichardson03)'
+version_added: '2.7.0'
+requirements:
+    - pan-os-python
+notes:
+    - Checkmode is supported.
+    - Panorama is supported.
+extends_documentation_fragment:
+    - paloaltonetworks.panos.fragments.provider
+    - paloaltonetworks.panos.fragments.state
+options:
+    xpath:
+        description:
+            - Location of the specified element in the XML configuration.
+        type: str
+        required: true
+    element:
+        description:
+            - The element, in XML format.
+        type: str
+    edit:
+        description:
+            - If **true**, replace any existing configuration at the specified
+              location with the contents of *element*.
+            - If **false**, merge the contents of *element* with any existing
+              configuration at the specified location.
+        type: bool
+        default: False
+        required: false
+"""
+
+EXAMPLES = """
+- name: Configure login banner
+  vars:
+    banner_text: 'Authorized Personnel Only!'
+  panos_config_element:
+    provider: '{{ provider }}'
+    xpath: '/config/devices/entry[@name="localhost.localdomain"]/deviceconfig/system'
+    element: '<login-banner>{{ banner_text }}</login-banner>'
+
+- name: Create address object
+  panos_config_element:
+    provider: '{{ provider }}'
+    xpath: "/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='vsys1']/address"
+    element: |
+      <entry name="Test-One">
+        <ip-netmask>1.1.1.1</ip-netmask>
+      </entry>
+
+- name: Delete address object 'Test-One'
+  panos_config_element:
+    provider: '{{ provider }}'
+    xpath: "/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='vsys1']/address/entry[@name='Test-One']"
+    state: 'absent'
+"""
+
+RETURN = """
+changed:
+    description: A boolean value indicating if the task had to make changes.
+    returned: always
+    type: bool
+msg:
+    description: A string with an error message, if any.
+    returned: failure, always
+    type: str
+diff:
+    description:
+        - Information about the differences between the previous and current
+          state.
+        - Contains 'before' and 'after' keys.
+    returned: success, when needed
+    type: dict
+    elements: str
+"""
+
+import xml.etree.ElementTree
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.paloaltonetworks.panos.plugins.module_utils.panos import (
+    get_connection,
+)
+
+try:
+    from panos.errors import PanDeviceError
+except ImportError:
+    try:
+        from pandevice.errors import PanDeviceError
+    except ImportError:
+        pass
+
+
+def xml_compare(one, two, excludes=None):
+    """
+    Compares the contents of two xml.etree.ElementTrees for equality.
+
+    :param one: First ElementTree.
+    :param two: Second ElementTree.
+    :param excludes: List of tag attributes to disregard.
+    """
+    if excludes is None:
+        excludes = ["admin", "dirtyId", "time", "uuid"]
+
+    if one is None or two is None:
+        return False
+
+    if one.tag != two.tag:
+        # Tag does not match.
+        return False
+
+    # Compare attributes.
+    for name, value in one.attrib.items():
+        if name not in excludes:
+            if two.attrib.get(name) != value:
+                return False
+
+    for name, value in two.attrib.items():
+        if name not in excludes:
+            if one.attrib.get(name) != value:
+                return False
+
+    if not text_compare(one.text, two.text):
+        # Text differs at this node.
+        return False
+
+    # Sort children by tag name to make sure they're compared in order.
+    children_one = sorted(one, key=lambda e: e.tag)
+    children_two = sorted(two, key=lambda e: e.tag)
+
+    if len(children_one) != len(children_two):
+        # Number of children differs.
+        return False
+
+    for child_one, child_two in zip(children_one, children_two):
+        if not xml_compare(child_one, child_two, excludes):
+            # Child documents do not match.
+            return False
+
+    return True
+
+
+def text_compare(one, two):
+    """Compares the contents of two XML text attributes."""
+    if not one and not two:
+        return True
+    return (one or "").strip() == (two or "").strip()
+
+
+def iterpath(node, tag=None, path="."):
+    """
+    Similar to Element.iter(), but the iterator gives each element's path along
+    with the element itself.
+
+    Reference: https://docs.python.org/3/library/xml.etree.elementtree.html#xml.etree.ElementTree.Element.iter
+
+    Taken from: https://stackoverflow.com/questions/13136334/get-xpath-dynamically-using-elementtree-getpath
+    """
+    if tag == "*":
+        tag = None
+
+    if tag is None or node.tag == tag:
+        yield node, path
+
+    for child in node:
+        if child.tag == "entry":
+            _child_path = "{0}/{1}[@name='{2}']".format(
+                path, child.tag, child.attrib["name"]
+            )
+        else:
+            _child_path = "{0}/{1}".format(path, child.tag)
+
+        for child, child_path in iterpath(child, tag, path=_child_path):
+            yield child, child_path
+
+
+def xml_contained(big, small):
+    """
+    Check to see if all the XML elements with no children in "small" are
+    present in "big", at the same locations in the tree.
+
+    This ensures all the configuration in "small" is contained in "big", but
+    "big" can have configuration not contained in "small".
+
+    :param big: Big document ElementTree.
+    :param small: Small document ElementTree.
+    """
+
+    if big is None or small is None:
+        return False
+
+    for element, path in iterpath(small):
+
+        if element.tag == "wrapped":
+            continue
+
+        # Elements with "member" children must have all their children be equal.
+        if element.find("*/member/..") is not None:
+            big_element = big.find(path)
+
+            if not xml_compare(big_element, element):
+                return False
+
+        # Elements with no children at the same point in the tree must match
+        # exactly.
+        elif len(element) == 0 and (element.tag != "member"):
+            if path != ".":
+                big_element = big.find(path)
+            else:
+                # handle case where small is only a single tag, thus the path ends up as '.'
+                big_element = big.find("./{0}".format(element.tag))
+
+            if not xml_compare(big_element, element):
+                return False
+
+    return True
+
+
+def main():
+    helper = get_connection(
+        with_classic_provider_spec=False,
+        argument_spec=dict(
+            xpath=dict(required=True),
+            element=dict(required=False),
+            edit=dict(type="bool", default=False, required=False),
+        ),
+        with_state=True,
+    )
+
+    module = AnsibleModule(
+        argument_spec=helper.argument_spec,
+        supports_check_mode=True,
+        required_one_of=helper.required_one_of,
+    )
+
+    parent = helper.get_pandevice_parent(module)
+
+    xpath = module.params["xpath"]
+    element_xml = module.params["element"]
+    edit = module.params["edit"]
+    state = module.params["state"]
+
+    try:
+        existing_element = parent.xapi.get(xpath)
+        existing_xml = parent.xapi.xml_document
+        existing = existing_element.find("./result/")
+
+        changed = False
+        diff = {}
+
+        if state == "present":
+            if element_xml is None:
+                module.fail_json(msg="'element' is required when state is 'present'.")
+
+            if edit:
+                element = xml.etree.ElementTree.fromstring(element_xml)
+
+                # Edit action is a regular comparison between the two
+                # XML documents for equality.
+                if not xml_compare(existing, element):
+                    changed = True
+
+                    if not module.check_mode:  # pragma: no cover
+                        parent.xapi.edit(xpath, element_xml)
+
+            else:
+                # When using set action, element can be an invalid XML document.
+                # Wrap it in a dummy tag if so.
+                try:
+                    element = xml.etree.ElementTree.fromstring(element_xml)
+                except xml.etree.ElementTree.ParseError:
+                    element = xml.etree.ElementTree.fromstring(
+                        "<wrapped>" + element_xml + "</wrapped>"
+                    )
+
+                if not xml_contained(existing, element):
+                    changed = True
+
+                    if not module.check_mode:  # pragma: no cover
+                        parent.xapi.set(xpath, element_xml)
+
+            diff = {
+                "before": existing_xml,
+                "after": element_xml,
+            }
+
+        # state == "absent"
+        else:
+            # Element exists, delete it.
+            if existing is not None:
+                changed = True
+
+                if not module.check_mode:  # pragma: no cover
+                    parent.xapi.delete(xpath)
+
+                diff = {"before": existing_xml, "after": ""}
+
+            # Element doesn't exist, nothing needs to be done.
+            else:
+                diff = {"before": "", "after": ""}
+
+        module.exit_json(changed=changed, diff=diff)
+
+    except PanDeviceError as e:  # pragma: no cover
+        module.fail_json(msg="{0}".format(e))
+
+
+if __name__ == "__main__":
+    main()