panos upgrade assurance

.gitignore

@@ -62,6 +62,9 @@ target/
 # pyenv
 .python-version
 
+# poetry
+poetry.lock
+
 # dotenv
 .env
 

plugins/module_utils/panos.py

@@ -106,6 +106,7 @@ def __init__(
         self,
         min_pandevice_version,
         min_panos_version,
+        min_panos_upgrade_assurance_version,
         error_on_firewall_shared,
         panorama_error,
         firewall_error,
@@ -127,6 +128,7 @@ def __init__(
         self.vsys_shared = None
         self.min_pandevice_version = min_pandevice_version
         self.min_panos_version = min_panos_version
+        self.min_panos_upgrade_assurance_version = min_panos_upgrade_assurance_version
         self.error_on_firewall_shared = error_on_firewall_shared
         self.panorama_error = panorama_error
         self.firewall_error = firewall_error
@@ -214,6 +216,29 @@ def get_pandevice_parent(self, module, timeout=0):
                     sdk_package_path=panos.__file__.rsplit("/", 1)[0],
                 )
 
+        if self.min_panos_upgrade_assurance_version is not None:
+            try:
+                import panos_upgrade_assurance
+            except ImportError:
+                module.fail_json(
+                    msg='Missing required library "panos_upgrade_assurance".',
+                    syspath=sys.path,
+                )
+            # This code assumes both panos_upgrade_assurance.version and self.min_panos_upgrade_assurance
+            # are a tuple of 3 ints.  If panos_upgrade_assurance.version is a string, then you'll have
+            # to turn it into a 3 element tuple of ints to do the comparison.
+            pua_ver = tuple(
+                int(x) for x in panos_upgrade_assurance.__version__.split(".")
+            )
+            if pua_ver < self.min_panos_upgrade_assurance_version:
+                module.fail_json(
+                    msg=_MIN_VERSION_ERROR.format(
+                        "panos_upgrade_assurance",
+                        _vstr(pua_ver),
+                        _vstr(self.min_panos_upgrade_assurance_version),
+                    )
+                )
+
         pan_device_auth, serial_number = None, None
         if module.params["provider"] and module.params["provider"]["ip_address"]:
             pan_device_auth = (
@@ -1333,6 +1358,7 @@ def get_connection(
     required_one_of=None,
     min_pandevice_version=None,
     min_panos_version=None,
+    min_panos_upgrade_assurance_version=None,
     error_on_firewall_shared=False,
     panorama_error=None,
     firewall_error=None,
@@ -1406,6 +1432,7 @@ def get_connection(
         required_one_of(list): List of lists to extend into required_one_of.
         min_pandevice_version(tuple): Minimum pandevice version allowed.
         min_panos_version(tuple): Minimum PAN-OS version allowed.
+        min_panos_upgrade_assurance_version(tuple): Minimum panos-upgrade-assurance package version.
         error_on_firewall_shared(bool): Don't allow "shared" vsys.
         panorama_error(str): The error message if the device is Panorama.
         firewall_error(str): The error message if the device is a firewall.
@@ -1464,6 +1491,7 @@ class in the package (e.g. - "VirtualRouter").  If the class is a singleton
     helper = helper_cls(
         min_pandevice_version,
         min_panos_version,
+        min_panos_upgrade_assurance_version,
         error_on_firewall_shared,
         panorama_error,
         firewall_error,

plugins/modules/panos_active_in_ha.py

@@ -0,0 +1,143 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+#  Copyright 2023 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_active_in_ha
+short_description: Checks if node is an active node in HA pair.
+description:
+    - A wrapper around the PAN-OS Upgrade Assurance package.
+    - A simple boolean check, verifies if a node is an active (B(true)) or passive (B(false)) node in an HA pair.
+    - If node does not belong to an HA pair or the pair is no configured correctly the module will fail.
+author: "Łukasz Pawlęga (@fosix)"
+version_added: '2.18.0'
+requirements:
+    - pan-python can be obtained from PyPI U(https://pypi.python.org/pypi/pan-python)
+    - pandevice can be obtained from PyPI U(https://pypi.python.org/pypi/pandevice)
+    - pan-os-upgrade-assurance can be obtained from PyPI U(https://pypi.org/project/panos-upgrade-assurance)
+notes:
+    - Only Firewalls are supported.
+    - Check mode is not supported.
+extends_documentation_fragment:
+    - paloaltonetworks.panos.fragments.transitional_provider
+    - paloaltonetworks.panos.fragments.vsys
+options:
+    force_fail:
+        description:
+            - When set to B(true) will make the module fail also when node is passive.
+              This option is useful when we want to skip using M(ansible.builtin.assert).
+        type: bool
+        default: false
+    skip_config_sync:
+        description:
+            - When set to B(true) will skip configuration synchronization state between nodes before trying to retrieve
+              node's current state in an HA pair. Can be useful when working with partially upgraded nodes. Use with caution.
+        type: bool
+        default: false
+# """
+
+EXAMPLES = """
+- name: Check if a node is active in HA pair
+  panos_active_in_ha:
+    provider: '{{ provider }}'
+  register: active_ha
+- name: Run tasks dedicated to active node
+  ansible.builtin.include_tasks: active_dedicated.yml
+  when: active_ha.response.active
+"""
+
+RETURN = """
+# Default return values
+response:
+    description:
+        - Information on test results.
+        - This dict is available also when module is failed.
+    returned: always
+    type: dict
+    sample:
+        active: true
+        reason: '[SUCCESS]'
+    contains:
+        active:
+            description: Information if the device is active or not.
+            returned: always
+            type: bool
+        reason:
+            description: Meaningful if the device is not active.
+            returned: always
+            type: str
+"""
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.paloaltonetworks.panos.plugins.module_utils.panos import (
+    get_connection,
+)
+
+try:
+    from panos_upgrade_assurance.check_firewall import CheckFirewall
+    from panos_upgrade_assurance.firewall_proxy import FirewallProxy
+    from panos_upgrade_assurance.utils import CheckStatus
+except ImportError:
+    pass
+
+MIN_PUA_VER = (0, 3, 0)
+
+
+def main():
+    results = dict()
+
+    helper = get_connection(
+        vsys=True,
+        with_classic_provider_spec=True,
+        min_panos_upgrade_assurance_version=MIN_PUA_VER,
+        argument_spec=dict(
+            force_fail=dict(type="bool", default=False),
+            skip_config_sync=dict(type="bool", default=False),
+        ),
+    )
+
+    module = AnsibleModule(
+        argument_spec=helper.argument_spec, supports_check_mode=False
+    )
+
+    firewall = FirewallProxy(firewall=helper.get_pandevice_parent(module))
+
+    is_active = CheckFirewall(firewall).check_is_ha_active(
+        skip_config_sync=module.params["skip_config_sync"]
+    )
+
+    if module.params["force_fail"]:
+        response = str(is_active)
+        module_failed = not bool(is_active)
+    else:
+        response = {"active": bool(is_active), "reason": str(is_active)}
+        module_failed = (
+            True
+            if is_active.status in [CheckStatus.ERROR, CheckStatus.SKIPPED]
+            else False
+        )
+
+    module.exit_json(changed=False, response=response, failed=module_failed)
+
+
+if __name__ == "__main__":
+    main()