Data segmentation in JSON file

README.md

@@ -33,6 +33,9 @@ and will prompt you for credentials and save the retrieved API key to "API_KEY.t
 * Run all validators on an XML configuration file downloaded with "Export Panorama configuration version":
 `pan_analyzer --xml 12345.xml`
 
+* Run all validators on an XML configuration file downloaded with "Export Panorama configuration version" and choose type output file (formats support txt (text) and json (json)):
+`pan_analyzer --xml 12345.xml --output text`
+`pan_analyzer --xml 12345.xml --output json`
 
 If you're not sure where to start, I recommend downloading an XML file from:
 `Panorama -> Setup -> Operations -> Export Panorama configuration version` and running: `pan_analyzer.py --xml 12345.xml`

src/palo_alto_firewall_analyzer/core.py

@@ -8,7 +8,6 @@
 import socket
 import typing
 import xml.etree.ElementTree
-
 import xmltodict
 
 from palo_alto_firewall_analyzer.pan_config import PanConfig
@@ -140,9 +139,16 @@ class ProfilePackage:
     devicegroup_exclusive_objects: typing.Dict
     rule_limit_enabled: bool
 
-
-BadEntry = collections.namedtuple('BadEntry', ['data', 'text', 'device_group', 'entry_type'])
-
+Detail = collections.namedtuple('Detail',['policy_type','policy_name','device_group',
+                                          'entry_type','entry_name','entry_value',
+                                          'rule_type','rule_name','protocol',
+                                          'port','allowed_group_profiles','group_profile_setting','address',
+                                          'fqdn','ip','ip_mask','loc','mandated_log_profile','log_setting',
+                                          'object_entry_name','policy_entry_name','shadowing_address_name',
+                                          'zone_type','zones','extra'
+                                          ]
+                                )
+BadEntry = collections.namedtuple('BadEntry', ['data', 'text', 'device_group', 'entry_type','Detail'])
 
 @functools.lru_cache(maxsize=None)
 def cached_dns_lookup(domain):
@@ -178,11 +184,34 @@ def cached_fqdn_lookup(domain):
 
 
 @functools.lru_cache(maxsize=None)
-def xml_object_to_dict(xml_obj):
+def xml_object_to_dict1(xml_obj):
     obj_xml_string = xml.etree.ElementTree.tostring(xml_obj)
     obj_dict = xmltodict.parse(obj_xml_string)
     return obj_dict
 
+def xml_object_to_dict(xml_obj):
+    obj_xml_string = xml.etree.ElementTree.tostring(xml_obj) 
+
+    root = xml.etree.ElementTree.fromstring(obj_xml_string)    
+
+    list_atr_remove = ['loc']
+
+    def remove_loc(elements,atr):
+        if atr in elements.attrib:
+            del elements.attrib[atr]
+        for elem in elements:
+            remove_loc(elem,atr)
+
+    for atr in list_atr_remove:    
+        for entry in root:        
+            remove_loc(entry,atr)    
+
+    xml_atr_remove = xml.etree.ElementTree.tostring(root)
+
+    obj_dict = xmltodict.parse(xml_atr_remove)    
+
+    return obj_dict
+
 
 @functools.lru_cache(maxsize=None)
 def get_single_ip_from_address(address_entry):

src/palo_alto_firewall_analyzer/pan_config.py

@@ -29,6 +29,7 @@ def __init__(self, configdata: str, from_file=False):
             fake_result.append(conf)
             # fake_response.append(fake_result)
             self.configroot = fake_result
+            self.config_xml = {"version": conf.get("version"),"urldb": conf.get("urldb"),"detail-version":conf.get("detail-version")}                        
         else:
             self.configroot = xml.etree.ElementTree.fromstring(configdata).find('./result')
 
@@ -40,10 +41,9 @@ def get_device_groups(self):
 
     @functools.lru_cache(maxsize=None)
     def get_device_groups_hierarchy(self):
-        xpath = "./config/readonly/devices/entry[@name='localhost.localdomain']/device-group/entry"
-
+        xpath = "./config/readonly/devices/entry[@name='localhost.localdomain']/device-group/entry"        
         device_group_hierarchy_children = collections.defaultdict(list)
-        device_group_hierarchy_parent = {}
+        device_group_hierarchy_parent = {}        
         for devicegroup_elem in self.configroot.findall(xpath):
             name = devicegroup_elem.get('name')
             parent = devicegroup_elem.find('parent-dg')

src/palo_alto_firewall_analyzer/scripts/pan_analyzer.py

@@ -5,6 +5,7 @@
 import os.path
 import sys
 import time
+import json
 
 # Used to trigger loading the validators and fixers
 import palo_alto_firewall_analyzer.validators
@@ -13,11 +14,13 @@
 from palo_alto_firewall_analyzer.core import get_policy_validators, get_policy_fixers, ConfigurationSettings
 from palo_alto_firewall_analyzer.pan_helpers import load_config_package, load_API_key
 
+from palo_alto_firewall_analyzer.scripts.pan_details import get_json_detail
+
 DEFAULT_CONFIG_DIR = os.path.expanduser("~" + os.sep + ".pan_policy_analyzer" + os.sep)
 DEFAULT_CONFIGFILE = DEFAULT_CONFIG_DIR + "PAN_CONFIG.cfg"
 DEFAULT_API_KEYFILE = DEFAULT_CONFIG_DIR + "API_KEY.txt"
 EXECUTION_START_TIME = datetime.datetime.today().strftime('%Y%m%d_%H%M%S')
-
+RUNTIME_START = time.time()
 logger = logging.getLogger('palo_alto_firewall_analyzer')
 
 
@@ -59,19 +62,25 @@ def run_policy_fixers(fixers, profilepackage, output_fname):
 def run_policy_validators(validators, profilepackage, output_fname):
     problems = {}
     total_problems = 0
+    total_checks = 0
     logger.info("Running validators")
 
     for name, validator_values in validators.items():
         validator_name, validator_description, validator_function = validator_values
-        validator_problems = validator_function(profilepackage)
-        problems[(validator_name, validator_description)] = validator_problems
-        total_problems += len(validator_problems)
-
-    return problems, total_problems
-
-
-def write_analyzer_output(problems, fname, out_format):
-    supported_output_formats = ["text"]
+        validator_problems, count_checks = validator_function(profilepackage)
+        problems[(validator_name, validator_description),count_checks] = validator_problems
+        total_problems += len(validator_problems)        
+        total_checks += count_checks
+
+    return problems, total_problems, total_checks
+
+
+def write_analyzer_output(problems, fname, profilepackage, sum_total_checks, out_format = 'text'):
+
+    supported_output_formats = ["text", "json"]    
+    if out_format is None:
+        out_format = 'text'
+
     if out_format not in supported_output_formats:
         raise Exception(
             f"Unsupported output format of {out_format}! Output format must be one of {supported_output_formats}")
@@ -82,7 +91,7 @@ def write_analyzer_output(problems, fname, out_format):
                 validator_name, validator_description = validator_info
 
                 fh.write("#" * 80 + '\n')
-                fh.write(f"{validator_name}: {validator_description} ({len(problem_entries)})\n")
+                fh.write(f"{validator_name}: {validator_description} ({len(problem_entries)}/{sum_total_checks})\n")
                 fh.write("#" * 80 + '\n')
                 for problem_entry in problem_entries:
                     # fh.write(f"Output for config name: {config_name} \n\n")
@@ -91,7 +100,43 @@ def write_analyzer_output(problems, fname, out_format):
                     # else:
                     #    fh.write('(none)\n')
                 fh.write('\n')
-
+    elif out_format == 'json':
+        #build json
+        total_problems = 0        
+        entries = [] 
+        #print(problems)       
+        for validator_info, problem_entries in problems.items():
+            #TODO: Can I doing better?
+            validator_name, validator_description = validator_info[0]
+            total_checks = validator_info[1]            
+
+            problems = []
+            for problem_entry in problem_entries:
+                if problem_entry.Detail is not None:                   
+                    problem = {"desc":problem_entry.text,"detail":get_json_detail(problem_entry.Detail)}
+                else:
+                    problem = {"desc":problem_entry.text}
+
+                problems.append(problem)
+                total_problems+=1
+
+            entry = {"validator_name":validator_name, "total_checks": total_checks,"problems":problems}                        
+            entries.append(entry)
+
+        end_time = time.time()
+
+        data = {"config_version":profilepackage.pan_config.config_xml['version'],
+                "detail-version":profilepackage.pan_config.config_xml['detail-version'],
+                "urldb":profilepackage.pan_config.config_xml['urldb'],
+                "date_execution": EXECUTION_START_TIME,
+                "runtime":round(end_time - RUNTIME_START, 2),                
+                "total_problems": total_problems,
+                "total_checks": sum_total_checks,                
+                "entries":entries
+                }  
+
+        with open(fname,'w') as fh:
+            json.dump(data,fh)
 
 def build_output_fname(parsed_args):
     # Build the name of the output file
@@ -119,7 +164,12 @@ def build_output_fname(parsed_args):
     else:
         limit_string = ""
 
-    output_fname = f'pan_analyzer_output_{EXECUTION_START_TIME}{devicegroup_string}{xml_string}{validators_string}{fixers_string}{limit_string}.txt'
+    if parsed_args.output == 'json':
+        extension = '.json'
+    else:    
+        extension = '.txt'
+
+    output_fname = f'pan_analyzer_output_{EXECUTION_START_TIME}{devicegroup_string}{xml_string}{validators_string}{fixers_string}{limit_string}'+extension
     return output_fname
 
 
@@ -152,6 +202,7 @@ def main():
 
     parser.add_argument("--debug", help="Write all debug output to pan_validator_debug_YYMMDD_HHMMSS.log", action='store_true')
     parser.add_argument("--limit", help="Limit processing to the first N rules (useful for debugging)", type=int)
+    parser.add_argument("--output", help="Type File Output (text, json), default = text", type=str)
     parsed_args = parser.parse_args()
 
     configure_logging(parsed_args.debug, not parsed_args.quiet)
@@ -182,6 +233,7 @@ def main():
     start_time = time.time()
     profilepackage = load_config_package(configuration_settings, api_key, parsed_args.device_group,
                                          parsed_args.limit, parsed_args.xml)
+
     if parsed_args.fixer:
         fixers = {parsed_args.fixer: get_policy_fixers()[parsed_args.fixer]}
         problems, total_problems = run_policy_fixers(fixers, profilepackage, output_fname)
@@ -190,9 +242,9 @@ def main():
             validators = {validator: get_policy_validators()[validator] for validator in parsed_args.validator}
         else:
             validators = get_policy_validators()
-        problems, total_problems = run_policy_validators(validators, profilepackage, output_fname)
-
-    write_analyzer_output(problems, output_fname, 'text')
+        problems, total_problems, total_checks = run_policy_validators(validators, profilepackage, output_fname)    
+        
+    write_analyzer_output(problems, output_fname, profilepackage, total_checks, parsed_args.output)
     end_time = time.time()
 
     logger.info(f"Full run took {round(end_time - start_time, 2)} seconds")

src/palo_alto_firewall_analyzer/scripts/pan_details.py

@@ -0,0 +1,72 @@
+import logging
+
+from palo_alto_firewall_analyzer.core import Detail
+
+def get_value(data):
+    if data is None:
+        data = ""
+
+    return data
+
+def parsed_details(details):
+
+    detail = Detail(
+        policy_type=get_value(details.get('policy_type')),
+        policy_name=get_value(details.get('policy_name')),
+        device_group=get_value(details.get('device_group')),
+        entry_type=get_value(details.get('entry_type')),
+        entry_name=get_value(details.get('entry_name')),
+        entry_value=get_value(details.get('entry_value')),
+        rule_type=get_value(details.get('rule_type')),
+        rule_name=get_value(details.get('rule_name')),
+        protocol=get_value(details.get('protocol')),
+        port=get_value(details.get('port')),
+        allowed_group_profiles=get_value(details.get('allowed_group_profiles')),
+        group_profile_setting=get_value(details.get('group_profile_setting')),
+        address=get_value(details.get('address')),
+        fqdn=get_value(details.get('fqdn')),
+        ip=get_value(details.get('ip')),
+        ip_mask=get_value(details.get('ip_mask')),
+        loc=get_value(details.get('loc')),
+        shadowing_address_name=get_value(details.get('shadowing_address_name')),        
+        mandated_log_profile=get_value(details.get('mandated_log_profile')),
+        log_setting=get_value(details.get('log_setting')),        
+        object_entry_name=get_value(details.get('object_entry_name')),
+        policy_entry_name=get_value(details.get('policy_entry_name')),
+        zone_type=get_value(details.get('zone_type')),
+        zones=get_value(details.get('zones')),
+        extra=get_value(details.get('extra'))
+    )
+
+    return detail
+
+
+def get_json_detail(detail):
+    json_detail={
+                    "policy_type":detail.policy_type,
+                    "policy_name":detail.policy_name,
+                    "device_group":detail.device_group,
+                    "entry_type":detail.entry_type,
+                    "entry_name":detail.entry_name,
+                    "rule_type":detail.rule_type,
+                    "rule_name":detail.rule_name,
+                    "protocol":detail.protocol,
+                    "port":detail.port,
+                    "allowed_group_profiles":detail.allowed_group_profiles,
+                    "group_profile_setting": detail.group_profile_setting,
+                    "address": detail.address,
+                    "fqdn": detail.fqdn,
+                    "ip":detail.ip,
+                    "ip_mask":detail.ip_mask,
+                    "loc":detail.loc,
+                    "shadowing_address_name":detail.shadowing_address_name,
+                    "mandated_log_profile":detail.mandated_log_profile,
+                    "log_setting":detail.log_setting,                    
+                    "object_entry_name":detail.object_entry_name,
+                    "policy_entry_name":detail.policy_entry_name,
+                    "zone_type":detail.zone_type,
+                    "zones":detail.zones,
+                    "extra":detail.extra
+                    }
+
+    return json_detail

src/palo_alto_firewall_analyzer/validators/__init__.py

@@ -11,7 +11,7 @@
 from . import misleading_objects
 from . import redundant_rule_members
 from . import rules_missing_security_profile
-from . import shadowing_addresses_and_groups
+from . import shadowing_addresses_and_groups 
 from . import shadowing_rules
 from . import shadowing_services_and_groups
 from . import similar_objects

src/palo_alto_firewall_analyzer/validators/bad_group_profile.py

@@ -1,25 +1,27 @@
 import logging
 
 from palo_alto_firewall_analyzer.core import register_policy_validator, BadEntry
+from palo_alto_firewall_analyzer.scripts.pan_details  import parsed_details
 
 logger = logging.getLogger(__name__)
 
 @register_policy_validator("BadGroupProfile", "Rule uses an incorrect group profile")
 def find_bad_group_profile_setting(profilepackage):
     device_groups = profilepackage.device_groups
-    devicegroup_exclusive_objects = profilepackage.devicegroup_exclusive_objects
-
+    devicegroup_exclusive_objects = profilepackage.devicegroup_exclusive_objects    
+    count_checks=0
+
     if not profilepackage.settings.get('Allowed Group Profiles'):
         logger.debug("Allowed Group Profiles are not set; skipping")
-        return []
+        return [],count_checks
 
     allowed_group_profiles = profilepackage.settings.get('Allowed Group Profiles').split(',')
 
     badentries = []
-
+    
     logger.info("*"*80)
     logger.info("Checking for incorrect group profile")
-
+    
     for i, device_group in enumerate(device_groups):
         for ruletype in ('SecurityPreRules', 'SecurityPostRules'):
             rules = devicegroup_exclusive_objects[device_group][ruletype]
@@ -37,10 +39,19 @@ def find_bad_group_profile_setting(profilepackage):
                 else:
                     group_profile_setting = ""
 
-                if group_profile_setting not in allowed_group_profiles:
+                if group_profile_setting not in allowed_group_profiles:                    
                     text = f"Device Group {device_group}'s {ruletype} '{rule_name}' doesn't use an approved group " \
                            f"profile '{allowed_group_profiles}', instead it uses '{group_profile_setting}' "
                     logger.debug(text)
-                    badentries.append( BadEntry(data=entry, text=text, device_group=device_group, entry_type=ruletype) )
-
-    return badentries
+                    detail={
+                            "entry_type":ruletype,
+                            "device_group":device_group,
+                            "rule_type":ruletype,
+                            "rule_name":rule_name,
+                            "allowed_group_profiles":allowed_group_profiles,
+                            "group_profile_setting":group_profile_setting
+                            }
+                    badentries.append( BadEntry(data=entry, text=text, device_group=device_group, entry_type=ruletype, Detail=parsed_details(detail)) )
+                count_checks+=1
+    print(count_checks)
+    return badentries, count_checks