Upgrade Assurance new incident layout dynamic inputs for checks (#45)

Enable dynamic check list and optional inputs in Assurance Test incident
displaying optional inputs if selected in check list in the new incident layout.

Packs/PAN_OS_Upgrade_Services/IncidentFields/incidentfield-DP_MP_Clock_Diff.json

@@ -0,0 +1,56 @@
+{
+	"aliasTo": "",
+	"aliases": null,
+	"associatedToAll": false,
+	"associatedTypes": [
+		"PAN-OS Network Operations - Upgrade Assurance",
+		"PAN-OS Network Operations - Device Upgrade"
+	],
+	"autoCompleteTags": null,
+	"breachScript": "",
+	"cacheVersn": 0,
+	"caseInsensitive": true,
+	"cliName": "dpmpclockdiff",
+	"closeForm": false,
+	"columns": null,
+	"content": false,
+	"defaultRows": null,
+	"definitionId": "",
+	"description": "Max. clock difference in seconds between data plane and management plane.",
+	"editForm": true,
+	"fieldCalcScript": "",
+	"fromServerVersion": "",
+	"group": 0,
+	"hidden": false,
+	"id": "incident_dpmpclockdiff",
+	"isReadOnly": false,
+	"itemVersion": "",
+	"locked": false,
+	"mergeStrategy": "",
+	"name": "DP/MP Clock Diff",
+	"neverSetAsRequired": false,
+	"openEnded": false,
+	"orgType": "number",
+	"ownerOnly": false,
+	"packID": "",
+	"packName": "",
+	"placeholder": "0",
+	"required": false,
+	"runScriptAfterUpdate": false,
+	"script": "",
+	"selectValues": null,
+	"sla": 0,
+	"system": false,
+	"systemAssociatedTypes": null,
+	"template": "",
+	"threshold": 72,
+	"toServerVersion": "",
+	"type": "number",
+	"unmapped": false,
+	"unsearchable": true,
+	"useAsKpi": false,
+	"validatedError": "",
+	"validationRegex": "",
+	"version": -1,
+	"x2_fields": ""
+}

Packs/PAN_OS_Upgrade_Services/IncidentFields/incidentfield-IPsec_Tunnel.json

@@ -0,0 +1,56 @@
+{
+	"aliasTo": "",
+	"aliases": null,
+	"associatedToAll": false,
+	"associatedTypes": [
+		"PAN-OS Network Operations - Upgrade Assurance",
+		"PAN-OS Network Operations - Device Upgrade"
+	],
+	"autoCompleteTags": null,
+	"breachScript": "",
+	"cacheVersn": 0,
+	"caseInsensitive": false,
+	"cliName": "ipsectunnel",
+	"closeForm": false,
+	"columns": null,
+	"content": false,
+	"defaultRows": null,
+	"definitionId": "",
+	"description": "IPsec tunnel name to check if it's in active state.",
+	"editForm": true,
+	"fieldCalcScript": "",
+	"fromServerVersion": "",
+	"group": 0,
+	"hidden": false,
+	"id": "incident_ipsectunnel",
+	"isReadOnly": false,
+	"itemVersion": "",
+	"locked": false,
+	"mergeStrategy": "",
+	"name": "IPsec Tunnel",
+	"neverSetAsRequired": false,
+	"openEnded": false,
+	"orgType": "shortText",
+	"ownerOnly": false,
+	"packID": "",
+	"packName": "",
+	"placeholder": "",
+	"required": false,
+	"runScriptAfterUpdate": false,
+	"script": "",
+	"selectValues": null,
+	"sla": 0,
+	"system": false,
+	"systemAssociatedTypes": null,
+	"template": "",
+	"threshold": 72,
+	"toServerVersion": "",
+	"type": "shortText",
+	"unmapped": false,
+	"unsearchable": false,
+	"useAsKpi": false,
+	"validatedError": "",
+	"validationRegex": "",
+	"version": -1,
+	"x2_fields": ""
+}

Packs/PAN_OS_Upgrade_Services/IncidentFields/incidentfield-Minimum_Content_Version.json

@@ -16,7 +16,7 @@
 	"content": false,
 	"defaultRows": null,
 	"definitionId": "",
-	"description": "The minimum content version to test against",
+	"description": "The minimum content version to test against. Checks for latest content version if not provided.",
 	"editForm": true,
 	"fieldCalcScript": "",
 	"fromServerVersion": "",

Packs/PAN_OS_Upgrade_Services/IncidentFields/incidentfield-Readiness_Check_List.json

@@ -39,12 +39,17 @@
 	"runScriptAfterUpdate": false,
 	"script": "",
 	"selectValues": [
-		"arp",
 		"panorama",
 		"ntp_sync",
 		"ha",
 		"candidate_config",
-		"expired_licenses"
+		"expired_licenses",
+		"active_support",
+		"content_version",
+		"session_exist",
+		"arp",
+		"ipsec_tunnel",
+		"dp_mp_clock_diff"
 	],
 	"sla": 0,
 	"system": false,

Packs/PAN_OS_Upgrade_Services/Integrations/PAN_OS_Upgrade_Assurance/PAN_OS_Upgrade_Assurance.py

@@ -88,16 +88,18 @@ def run_readiness_checks(
     Run all the readiness checks and return an xsoar-compatible result.
 
     :arg firewall: Firewall object
-    :arg check_list: List of basic checks
-    :arg min_content_version: The minimum content version to check for, enables "content_version" check
+    :arg check_list: List of basic checks. Must match with Upgrade Assurance check types.
+    :arg min_content_version: The minimum content version to check for, otherwise latest content version is
+        checked if "content_version" is provided in `check_list`.
     :arg candidate_version: The candidate version to runchecks against. Enables "free_disk_space" check
-    :arg dp_mp_clock_diff: The drift allowed between DP clock and MP clock. Enabled "planes_clock_sync" check.
+    :arg dp_mp_clock_diff: The drift allowed between DP clock and MP clock. Enables "planes_clock_sync" check.
     :arg ipsec_tunnel_status: Check a specific IPsec - by tunnel name. Tunnel must be up for this check to pass.
     :arg check_session_exists: Check for the presence of a specific connection.
         Session check format is <source>/destination/destination-port
         example: 10.10.10.10/8.8.8.8/443
     :arg arp_entry_exists: Check for the prescence of a specific ARP entry.
         example: 10.0.0.6
+
     """
 
     if not check_list:
@@ -114,49 +116,61 @@ def run_readiness_checks(
     custom_checks = []
 
     # Add the custom checks
-    if min_content_version:
-        custom_checks.append({"content_version": {'version': min_content_version}})
+
+    if 'content_version' in check_list:
+        if min_content_version:
+            custom_checks.append({'content_version': {'version': min_content_version}})
+            check_list.remove('content_version')
+        # else it will check for latest content version
 
     if candidate_version:
         custom_checks.append({
-            "free_disk_space": {
-                "image_version": candidate_version
+            'free_disk_space': {
+                'image_version': candidate_version
             }
         })
     else:
         check_list.append('free_disk_space')
 
-    if isinstance(dp_mp_clock_diff, int) and dp_mp_clock_diff >= 0:
-        custom_checks.append({
-            "planes_clock_sync": {
-                'diff_threshold': dp_mp_clock_diff
-            }
-        })
-
-    if ipsec_tunnel_status:
-        custom_checks.append({
-            'ip_sec_tunnel_status': {
-                'tunnel_name': ipsec_tunnel_status
-            }
-        })
-
-    if check_session_exists:
-        try:
-            check_value = parse_session(check_session_exists)
-        except ValueError:
-            raise ValueError(
-                f"{check_session_exists} is not a valid session string. Must be 'source/destination/port'."
-            )
-        custom_checks.append({
-            "session_exist": check_value
-        })
-
-    if arp_entry_exists:
-        custom_checks.append({
-            'arp_entry_exist': {
-                'ip': arp_entry_exists
-            }
-        })
+    if 'planes_clock_sync' in check_list:
+        if isinstance(dp_mp_clock_diff, int) and dp_mp_clock_diff >= 0:
+            custom_checks.append({
+                'planes_clock_sync': {
+                    'diff_threshold': dp_mp_clock_diff
+                }
+            })
+        check_list.remove('planes_clock_sync')
+
+    if 'ip_sec_tunnel_status' in check_list:
+        if ipsec_tunnel_status:
+            custom_checks.append({
+                'ip_sec_tunnel_status': {
+                    'tunnel_name': ipsec_tunnel_status
+                }
+            })
+        check_list.remove('ip_sec_tunnel_status')
+
+    if 'session_exist' in check_list:
+        if check_session_exists:
+            try:
+                check_value = parse_session(check_session_exists)
+            except ValueError:
+                raise ValueError(
+                    f"{check_session_exists} is not a valid session string. Must be 'source/destination/port'."
+                )
+            custom_checks.append({
+                'session_exist': check_value
+            })
+        check_list.remove('session_exist')
+
+    if 'arp_entry_exists' in check_list:
+        if arp_entry_exists:
+            custom_checks.append({
+                'arp_entry_exist': {
+                    'ip': arp_entry_exists
+                }
+            })
+        check_list.remove('arp_entry_exists')
 
     check_config = check_list + custom_checks
 
@@ -222,48 +236,83 @@ def convert_snapshot_result_to_table(results: dict):
 
 
 def command_run_readiness_checks(panorama: Panorama):
+    """Parse readiness check command to run corresponding checks
+
+    Implemented checks:
+
+    - panorama
+    - ha
+    - ntp_sync
+    - candidate_config
+    - expired_licenses
+    - active_support
+    - content_version
+    - session_exist
+    - arp (arp_entry_exist)
+    - ipsec_tunnel (ip_sec_tunnel_status)
+    - free_disk_space (hardcoded check)
+    - dp_mp_clock_diff (planes_clock_sync)
+
+    """
     args = demisto.args()
-    firewall = get_firewall_object(panorama, args.get("firewall_serial"))
-    del args["firewall_serial"]
+    firewall = get_firewall_object(panorama, args.get('firewall_serial'))
+    del args['firewall_serial']
 
     # this will set it to [] if emptry string or not provided - meaning check all
     check_list = argToList(args.get('check_list'))
     if args.get('check_list') is not None:
-        del args["check_list"]
+        del args['check_list']
+
+    if 'dp_mp_clock_diff' in check_list:
+        check_list.remove('dp_mp_clock_diff')
+        check_list.append('planes_clock_sync')  # to match with upgrade assurance check type
 
     # this will set it to None if emptry string or not provided
     dp_mp_clock_diff = arg_to_number(args.get('dp_mp_clock_diff'), required=False)
     if args.get('dp_mp_clock_diff') is not None:
-        del args["dp_mp_clock_diff"]
+        del args['dp_mp_clock_diff']
+
+    if 'ipsec_tunnel' in check_list:
+        check_list.remove('ipsec_tunnel')
+        check_list.append('ip_sec_tunnel_status')
+
+    if 'arp' in check_list:
+        check_list.remove('arp')
+        check_list.append('arp_entry_exists')
+
+    if (arp_entry := args.get('arp_entry_exists')) is not None and not is_ip_valid(arp_entry):
+        raise ValueError(
+            f"{arp_entry} is not a valid IPv4 address."
+        )
 
     results = run_readiness_checks(firewall, check_list,
                                    dp_mp_clock_diff=dp_mp_clock_diff, **args)
 
     return CommandResults(
         outputs={
-            "ReadinessCheckResults": convert_readiness_results_to_table(results),
-            "Firewall": firewall.serial
+            'ReadinessCheckResults': convert_readiness_results_to_table(results),
+            'Firewall': firewall.serial
         },
-        readable_output=tableToMarkdown("Readiness Check Results",
+        readable_output=tableToMarkdown('Readiness Check Results',
                                         convert_readiness_results_to_table(results),
-                                        headers=["Test", "state", "reason"]),
-        outputs_prefix="FirewallAssurance"
+                                        headers=['Test', 'state', 'reason']),
+        outputs_prefix='FirewallAssurance'
     )
 
 
 def command_run_snapshot(panorama: Panorama):
     """Runs a single snapshot and returns it as a file."""
     args = demisto.args()
-    firewall = get_firewall_object(panorama, args.get("firewall_serial"))
-    snapshot_name = args.get("snapshot_name", "fw_snapshot")
-    del args["firewall_serial"]
-    if args.get("snapshot_name"):
-        del args["snapshot_name"]
+    firewall = get_firewall_object(panorama, args.get('firewall_serial'))
+    snapshot_name = args.get('snapshot_name', 'fw_snapshot')
+    del args['firewall_serial']
+    if args.get('snapshot_name'):
+        del args['snapshot_name']
 
     # this will set it to [] if emptry string or not provided - meaning all snapshots
     snapshot_list = argToList(args.get('check_list'))
     if args.get('check_list') is not None:
-        del args["check_list"]
+        del args['check_list']
 
     snapshot = run_snapshot(firewall, snapshot_list, **args)
 
@@ -276,17 +325,17 @@ def command_run_snapshot(panorama: Panorama):
 def command_compare_snapshots():
     """Compare two snapshot files, accepting th left and right snapshots as arguments."""
     args = demisto.args()
-    left_snapshot = read_file_by_id(args.get("left_snapshot_id"))
-    right_snapshot = read_file_by_id(args.get("right_snapshot_id"))
+    left_snapshot = read_file_by_id(args.get('left_snapshot_id'))
+    right_snapshot = read_file_by_id(args.get('right_snapshot_id'))
     result = compare_snapshots(left_snapshot, right_snapshot)
     return CommandResults(
         outputs={
-            "SnapshotComparisonResult": convert_snapshot_result_to_table(result),
-            "SnapshotComparisonRawResult": result
+            'SnapshotComparisonResult': convert_snapshot_result_to_table(result),
+            'SnapshotComparisonRawResult': result
         },
         readable_output=tableToMarkdown(
-            "Snapshot Comparison Results", convert_snapshot_result_to_table(result), headers=["test", "passed"]),
-        outputs_prefix="FirewallAssurance"
+            'Snapshot Comparison Results', convert_snapshot_result_to_table(result), headers=['test', 'passed']),
+        outputs_prefix='FirewallAssurance'
     )