feat: Check for April PAN-OS Certificate Advisory

docs/panos-upgrade-assurance/api/check_firewall.md

@@ -686,6 +686,32 @@ __Returns__
 
 `dict`: Results of all configured checks.
 
+### `CheckFirewall.check_version_against_version_match_dict`
+
+```python
+@staticmethod
+def check_version_against_version_match_dict(version: Version,
+                                             match_dict: dict)
+```
+
+Compare the given software version against the match dict.
+
+# Parameters
+version (str): The software version to compare. Example: "10.1.11"
+match_dict (dict): A dictionary of tuples mapping major/minor versions to match criteria
+example
+
+
+Returns
+
+bool: `True` If the given software version matches the provided match criteria
+    ```python
+    {
+        "81": [("==", "8.1.21.2"), (">=", "8.1.25.1")],
+        "90": [(">=", "9.0.16.5")],
+    }
+    ```
+
 ### `CheckFirewall.check_device_root_certificate_issue`
 
 ```python
@@ -712,3 +738,20 @@ __Parameters__
     fail if the software version is affected by the root certificate issue, AND the device is used for data
     redistribution OR it's using an out-of-date content DB version.
 
+### `CheckFirewall.check_cdss_and_panorama_certificate_issue`
+
+```python
+def check_cdss_and_panorama_certificate_issue()
+```
+
+Checks whether the device is affected by the following advisory;
+
+https://live.paloaltonetworks.com/t5/customer-advisories/additional-pan-os-certificate-expirations-and-new-comprehensive/ta-p/572158
+
+Check will fail in either of following scenarios:
+
+ * Device is running an affected software version
+ * Device is running an affected content version
+ * Device is running the fixed content version or higher but has not been rebooted - note this is best effort,
+    and is based on when the content version was released and the device was rebooted
+

docs/panos-upgrade-assurance/api/firewall_proxy.md

@@ -1255,3 +1255,22 @@ __Returns__
 }
 ```
 
+### `FirewallProxy.get_system_time_rebooted`
+
+```python
+def get_system_time_rebooted() -> datetime
+```
+
+Returns the date and time the system last rebooted using the system uptime.
+
+The actual API command is `show system info`.
+
+__Returns__
+
+
+`datetime`: Time system was last rebooted based on current time - system uptime string
+
+```python showLineNumbers title="Sample output"
+datetime(2024, 01, 01, 00, 00, 00)
+```
+

examples/readiness_checks/run_health_checks.py

@@ -66,11 +66,12 @@
     vsys = args.vsys
 
     if serial:
+        print(address)
         panorama = Panorama(
             hostname=address, api_password=password, api_username=username
         )
         firewall = FirewallProxy(serial=serial)
-        panorama.add(firewall)
+        panorama.add(firewall._fw)
     else:
         firewall = FirewallProxy(
             hostname=address, api_password=password, api_username=username, vsys=vsys
@@ -79,7 +80,8 @@
     check_node = CheckFirewall(firewall)
 
     checks = [
-        "device_root_certificate_issue"
+        "device_root_certificate_issue",
+        "check_cdss_and_panorama_certificate_issue"
     ]
 
     check_health = check_node.run_health_checks(

panos_upgrade_assurance/check_firewall.py

@@ -6,6 +6,7 @@
 
 import panos.errors
 from packaging.version import parse as parse_version
+from packaging.version import Version
 
 from panos_upgrade_assurance.utils import (
     CheckResult,
@@ -104,7 +105,10 @@ def __init__(self, node: FirewallProxy, skip_force_locale: Optional[bool] = Fals
             CheckType.JOBS: self.check_non_finished_jobs,
         }
 
-        self._health_check_method_mapping = {HealthType.DEVICE_ROOT_CERTIFICATE_ISSUE: self.check_device_root_certificate_issue}
+        self._health_check_method_mapping = {
+            HealthType.DEVICE_ROOT_CERTIFICATE_ISSUE: self.check_device_root_certificate_issue,
+            HealthType.DEVICE_CDSS_AND_PANORAMA_CERTIFICATE_ISSUE: self.check_cdss_and_panorama_certificate_issue,
+        }
 
         if not skip_force_locale:
             locale.setlocale(
@@ -1253,6 +1257,37 @@ def run_health_checks(
 
         return result
 
+    @staticmethod
+    def check_version_against_version_match_dict(version: Version, match_dict: dict):
+        """Compare the given software version against the match dict.
+
+        # Parameters
+        version (str): The software version to compare. Example: "10.1.11"
+        match_dict (dict): A dictionary of tuples mapping major/minor versions to match criteria
+            example
+
+            ```python
+            {
+                "81": [("==", "8.1.21.2"), (">=", "8.1.25.1")],
+                "90": [(">=", "9.0.16.5")],
+            }
+            ```
+
+        Returns
+
+        bool: `True` If the given software version matches the provided match criteria
+        """
+        match_versions = match_dict.get(f"{version.major}{version.minor}")
+        if match_versions:
+            for operator, match_version in match_versions:
+                match_version = parse_version(match_version)
+                if operator == "==":
+                    if version == match_version:
+                        return True
+                elif operator == ">=":
+                    if version >= match_version:
+                        return True
+
     def check_device_root_certificate_issue(self, fail_when_affected_version_only: bool = True) -> CheckResult:
         """Checks whether the target device is affected by the Root Certificate Expiration issue;
 
@@ -1316,16 +1351,8 @@ def check_device_root_certificate_issue(self, fail_when_affected_version_only: b
         }
         fixed_content_version = 8776.8390
 
-        fixed_versions = fixed_version_map.get(f"{software_version.major}{software_version.minor}")
-        if fixed_versions:
-            for operator, fixed_version in fixed_versions:
-                fixed_version = parse_version(fixed_version)
-                if operator == "==":
-                    if software_version == fixed_version:
-                        result.status = CheckStatus.SUCCESS
-                elif operator == ">=":
-                    if software_version >= fixed_version:
-                        result.status = CheckStatus.SUCCESS
+        if self.check_version_against_version_match_dict(software_version, fixed_version_map):
+            result.status = CheckStatus.SUCCESS
 
         # If the device is already running fixed software, we can return immediately
         if result.status == CheckStatus.SUCCESS:
@@ -1372,3 +1399,89 @@ def check_device_root_certificate_issue(self, fail_when_affected_version_only: b
             "expire December 31st, 2023."
         )
         return result
+
+    def check_cdss_and_panorama_certificate_issue(self):
+        """Checks whether the device is affected by the following advisory;
+
+        https://live.paloaltonetworks.com/t5/customer-advisories/additional-pan-os-certificate-expirations-and-new-comprehensive/ta-p/572158
+
+        Check will fail in either of following scenarios:
+
+         * Device is running an affected software version
+         * Device is running an affected content version
+         * Device is running the fixed content version or higher but has not been rebooted - note this is best effort,
+            and is based on when the content version was released and the device was rebooted
+
+        """
+        fixed_version_map = {
+            "81": [("==", "8.1.21.3"), ("==", "8.1.25.3"), (">=", "8.1.26")],
+            "90": [("==", "9.0.16.7"), ("==", "9.0.17.5")],
+            "91": [
+                ("==", "9.1.11.5"),
+                ("==", "9.1.12.7"),
+                ("==", "9.1.13.5"),
+                ("==", "9.1.14.8"),
+                ("==", "9.1.16.5"),
+                (">=", "9.1.17"),
+            ],
+            "100": [("==", "10.0.8.11"), ("==", "10.0.11.4"), ("==", "10.0.12.5")],
+            "101": [
+                ("==", "10.1.3.3"),
+                ("==", "10.1.4.6"),
+                ("==", "10.1.5.4"),
+                ("==", "10.1.6.8"),
+                ("==", "10.1.7.1"),
+                ("==", "10.1.8.7"),
+                ("==", "10.1.9.8"),
+                ("==", "10.1.10.5"),
+                ("==", "10.1.11.4"),
+                (">=", "10.1.12"),
+            ],
+            "102": [
+                ("==", "10.2.0.2"),
+                ("==", "10.2.1.1"),
+                ("==", "10.2.2.4"),
+                ("==", "10.2.3.11"),
+                ("==", "10.2.4.10"),
+                ("==", "10.2.5.4"),
+                ("==", "10.2.6.1"),
+                ("==", "10.2.7.3"),
+                (">=", "10.2.8"),
+            ],
+            "110": [("==", "11.0.0.2"), ("==", "11.0.1.3"), ("==", "11.0.2.3"), (">=", "11.0.3.3"), (">=", "11.0.4")],
+            "111": [("==", "11.1.0.2"), (">=", "11.1.1")],
+        }
+
+        # Release date and fixed version are both static
+        fixed_content_version = 8795.8489
+        fixed_content_version_release_date = datetime(2024, 1, 8, 19, 26, 43)
+
+        result = CheckResult()
+
+        software_version = self._node.get_device_software_version()
+
+        if self.check_version_against_version_match_dict(software_version, fixed_version_map):
+            # Fixed software means we can return immediately, no need to further check
+            result.status = CheckStatus.SUCCESS
+            return result
+
+        content_version = float(self._node.get_content_db_version().replace("-", "."))
+
+        if content_version >= fixed_content_version:
+            # Check the device has been rebooted since the release of the fixed content version
+            # This is not a perfect test - if the customer reboots without installing the content update, then
+            # later installs it, it will pass even though one further restart is required.
+            reboot_time = self._node.get_system_time_rebooted()
+            if reboot_time < fixed_content_version_release_date:
+                result.reason = "Device is running fixed Content but still requires a restart for the fix to take " "effect."
+                return result
+            else:
+                result.status = CheckStatus.SUCCESS
+                return result
+
+        result.reason = (
+            "Device is running a software version, and a content version, that is affected by the 2024 certificate"
+            " expiration, the first of which will occur on the 7th of April, 2024."
+        )
+
+        return result

panos_upgrade_assurance/firewall_proxy.py

@@ -7,7 +7,7 @@
 from pan.xapi import PanXapiError
 from panos_upgrade_assurance import exceptions
 from math import floor
-from datetime import datetime
+from datetime import datetime, timedelta
 from packaging import version
 
 
@@ -1461,3 +1461,32 @@ def get_fib(self) -> dict:
                         results[key] = result_entry
 
         return results
+
+    def get_system_time_rebooted(self) -> datetime:
+        """Returns the date and time the system last rebooted using the system uptime.
+
+        The actual API command is `show system info`.
+
+        # Returns
+
+        datetime: Time system was last rebooted based on current time - system uptime string
+
+        ```python showLineNumbers title="Sample output"
+        datetime(2024, 01, 01, 00, 00, 00)
+        ```
+
+        """
+        response = self.op_parser(cmd="show system info", return_xml=True)
+        uptime_string = response.findtext("./system/uptime")
+        current_time = datetime.now()
+
+        time_re_match = re.search(r"(\d+) days, (\d+):(\d+):(\d+)", uptime_string)
+
+        rebooted_time = current_time - timedelta(
+            days=int(time_re_match.group(1)),
+            hours=int(time_re_match.group(2)),
+            minutes=int(time_re_match.group(3)),
+            seconds=int(time_re_match.group(4)),
+        )
+
+        return rebooted_time

panos_upgrade_assurance/utils.py

@@ -63,6 +63,7 @@ class HealthType:
     """
 
     DEVICE_ROOT_CERTIFICATE_ISSUE = "device_root_certificate_issue"
+    DEVICE_CDSS_AND_PANORAMA_CERTIFICATE_ISSUE = "check_cdss_and_panorama_certificate_issue"
 
 
 class CheckStatus(Enum):

tests/test_check_firewall.py

@@ -1307,3 +1307,45 @@ def test_run_health_checks(self, check_firewall_mock):
 
         check_firewall_mock._health_check_method_mapping["check1"].assert_called_once_with()
         check_firewall_mock._health_check_method_mapping["check2"].assert_called_once_with(param1=123)
+
+    @pytest.mark.parametrize(
+        "running_software, expected_status",
+        [
+            ("10.1.2", CheckStatus.FAIL),  # Device running broken version
+            ("10.1.13", CheckStatus.SUCCESS),  # Device running fixed version
+        ],
+    )
+    def test_check_cdss_and_panorama_certificate_issue(self, check_firewall_mock, running_software, expected_status):
+        """This test validates the behavior when the test is only checking the software version is affected by
+        the issue."""
+
+        from packaging import version
+
+        check_firewall_mock._node.get_device_software_version = MagicMock(return_value=version.parse(running_software))
+        assert check_firewall_mock.check_cdss_and_panorama_certificate_issue().status == expected_status
+
+    @pytest.mark.parametrize(
+        "running_content_version, last_reboot, expected_status",
+        [
+            ("8000-8391", datetime(2022, 1, 1, 0, 0, 0), CheckStatus.FAIL),  # Device running older content version and no reboot
+            ("8795-8489", datetime(2022, 1, 1, 0, 0, 0), CheckStatus.FAIL),  # Device running fixed version without reboot
+            ("8795-8489", datetime(2024, 1, 10, 0, 0, 0), CheckStatus.SUCCESS),  # Device running fixed version and rebooted
+        ],
+    )
+    def test_check_cdss_and_panorama_certificate_issue_by_content_version(
+        self, check_firewall_mock, running_content_version, expected_status, last_reboot
+    ):
+        """Tests that we check the content version and use a best effort approach for seeing if the device has been
+        rebooted in the time since it was released/installed"""
+        from packaging import version
+
+        check_firewall_mock._node.get_device_software_version = MagicMock(
+            return_value=version.parse("10.1.0")  # Affected Version
+        )
+
+        check_firewall_mock._node.get_content_db_version = MagicMock(return_value=running_content_version)
+
+        # Device hasn't been rebooted
+        check_firewall_mock._node.get_system_time_rebooted = MagicMock(return_value=last_reboot)
+
+        assert check_firewall_mock.check_cdss_and_panorama_certificate_issue().status == expected_status