chg: [parsers] parsers handle log file selection for more standardisa…

…tion

parsers/olddsc.py

@@ -24,8 +24,4 @@ def get_log_files(log_root_path: str) -> dict:
 
 
 def parse_path(path: str) -> list | dict:
-    return misc.load_plist_file_as_json(path)
-
-
-def parse_olddsc_file(filepath: str) -> dict:
-    return misc.load_plist_file_as_json(filepath)
+    return misc.load_plist_file_as_json(get_log_files(path)[0])

parsers/plists.py

@@ -4,6 +4,7 @@
 import misc
 import os
 
+
 parser_description = "Parsing any pslist into json"
 
 
@@ -13,10 +14,31 @@ def get_log_files(log_root_path: str) -> list:
     ]
     log_files = []
     for log_files_glob in log_files_globs:
-        log_files.extend(glob.glob(os.path.join(log_root_path, log_files_glob)))
+        log_files.extend(glob.glob(os.path.join(log_root_path, log_files_glob), recursive=True))
 
     return log_files
 
 
-def parse_path(path: str) -> list | dict:
-    return misc.load_plist_file_as_json(path)
+def parse_path(path: str) -> dict:
+    result = {}
+    for logfile in get_log_files(path):
+        try:
+            json_data = misc.load_plist_file_as_json(logfile)
+        except Exception as e:
+            json_data = {"error": str(e)}
+        end_of_path = logfile[len(path):].lstrip(os.path.sep)  # take the path after the root path
+        result[end_of_path] = json_data
+    return result
+
+
+def parse_path_to_folder(path: str, output: str) -> bool:
+    os.makedirs(output, exist_ok=True)
+    for logfile in get_log_files(path):
+        try:
+            json_data = misc.load_plist_file_as_json(logfile)
+        except Exception as e:
+            json_data = {"error": str(e)}
+        end_of_path = logfile[len(path):].lstrip(os.path.sep)   # take the path after the root path
+        output_filename = end_of_path.replace(os.path.sep, '_') + '.json'  # replace / with _ in the path
+        with open(os.path.join(output, output_filename), 'w') as f:
+            f.write(json_data)

parsers/powerlogs.py

@@ -27,5 +27,8 @@ def get_log_files(log_root_path: str) -> list:
     return log_files
 
 
-def parse_path(path: str) -> list | dict:
-    return sqlite2json.sqlite2struct(path)
+def parse_path(path: str) -> list:
+    result = []
+    for logfile in get_log_files(path):
+        result.extend(sqlite2json.sqlite2struct(logfile))
+    return result

parsers/ps.py

@@ -32,7 +32,7 @@ def get_log_files(log_root_path: str) -> list:
 
 
 def parse_path(path: str) -> list | dict:
-    return parse_ps(path)
+    return parse_ps(get_log_files(path)[0])
 
 
 def parse_ps(filename, ios_version=16):

parsers/psthread.py

@@ -28,7 +28,7 @@ def get_log_files(log_root_path: str) -> list:
 
 
 def parse_path(path: str) -> list | dict:
-    with open(path, "r") as fd:
+    with open(get_log_files(path)[0], "r") as fd:
         input = fd.readlines()
         input_clean = []
         for line in input:

parsers/remotectl_dumpstate.py

@@ -18,4 +18,4 @@ def get_log_files(log_root_path: str) -> list:
 
 
 def parse_path(path: str) -> list | dict:
-    return parse_tab_based_hierarchal_file(path)
+    return parse_tab_based_hierarchal_file(get_log_files(path)[0])

parsers/shutdownlogs.py

@@ -30,7 +30,7 @@ def get_log_files(log_root_path: str) -> list:
 def parse_path(path: str) -> list | dict:
     # read log file content
     log_lines = ""
-    with open(path, "r") as f:
+    with open(get_log_files(path)[0], "r") as f:
         log_lines = f.readlines()
 
     json_object = {}

parsers/spindumpnosymbols.py

@@ -23,11 +23,7 @@ def get_log_files(log_root_path: str) -> list:
 
 
 def parse_path(path: str) -> list | dict:
-    return parsespindumpNS(path)
-
-
-def parsespindumpNS(file):
-    with open(file, 'r') as f_in:
+    with open(get_log_files(path)[0], 'r') as f_in:
         # init section
         headers = []
         processes_raw = []

parsers/swcutil.py

@@ -22,11 +22,7 @@ def get_log_files(log_root_path: str) -> list:
 
 
 def parse_path(path: str) -> list | dict:
-    return parseswcutil(path)
-
-
-def parseswcutil(file):
-    with open(file, 'r') as f_in:
+    with open(get_log_files(path)[0], 'r') as f_in:
         # init section
         headers = []
         db = []

parsers/sys.py

@@ -25,7 +25,7 @@ def get_log_files(log_root_path: str) -> list:
 
 
 def parse_path(path: str) -> list | dict:
-    return misc.load_plist_file_as_json(path)
+    return misc.load_plist_file_as_json(get_log_files(path)[0])
 
 
 '''

parsers/taskinfo.py

@@ -30,7 +30,7 @@ def get_log_files(log_root_path: str) -> list:
 def parse_path(path: str) -> list | dict:
     processes = []
 
-    with open(path, "r") as f:
+    with open(get_log_files(path)[0], "r") as f:
         lines = f.readlines()
 
         result = re.search(r'(num tasks: )(\d+)', lines[0])

parsers/uuid2path.py

@@ -11,18 +11,7 @@
 import glob
 import misc
 
-version_string = "sysdiagnose-uuid2path.py v2020-02-07 Version 2.0"
-
-# ----- definition for parsing.py script -----#
-# -----         DO NOT DELETE             ----#
-
 parser_description = "Parsing UUIDToBinaryLocations plist file"
-parser_input = "UUIDToBinaryLocations"
-parser_call = "getUUID2path"
-
-# --------------------------------------------#
-
-# --------------------------------------------------------------------------- #
 
 
 def get_log_files(log_root_path: str) -> list:
@@ -37,7 +26,7 @@ def get_log_files(log_root_path: str) -> list:
 
 
 def parse_path(path: str) -> list | dict:
-    return misc.load_plist_file_as_json(path)
+    return misc.load_plist_file_as_json(get_log_files(path)[0])
 
 
 def printResult(data):

parsers/wifi_known_networks.py

@@ -26,7 +26,7 @@ def get_log_files(log_root_path: str) -> list:
 
 
 def parse_path(path: str) -> list | dict:
-    return misc.load_plist_file_as_json(path)
+    return misc.load_plist_file_as_json(get_log_files(path)[0])
 
 
 '''

parsers/wifinetworks.py

@@ -24,9 +24,30 @@ def get_log_files(log_root_path: str) -> list:
     return log_files
 
 
-def parse_path(path: str) -> list | dict:
-    if path.endswith('.json'):
-        with open(path, 'r') as f:
+def parse_file(fname: str) -> dict | list:
+    if fname.endswith('.json'):
+        with open(fname, 'r') as f:
             return json.load(f)
-    if path.endswith('.plist'):
-        return misc.load_plist_file_as_json(path)
+    if fname.endswith('.plist'):
+        return misc.load_plist_file_as_json(fname)
+
+
+def parse_path(path: str) -> dict:
+    result = {}
+    for logfile in get_log_files(path):
+        end_of_path = logfile[len(path):].lstrip(os.path.sep)  # take the path after the root path
+        result[end_of_path] = parse_file(logfile)
+    return result
+
+
+def parse_path_to_folder(path: str, output: str) -> bool:
+    os.makedirs(output, exist_ok=True)
+    for logfile in get_log_files(path):
+        try:
+            json_data = parse_file(logfile)
+        except Exception as e:
+            json_data = {"error": str(e)}
+        end_of_path = logfile[len(path):].lstrip(os.path.sep)   # take the path after the root path
+        output_filename = end_of_path.replace(os.path.sep, '_') + '.json'  # replace / with _ in the path
+        with open(os.path.join(output, output_filename), 'w') as f:
+            f.write(json_data)

parsers/wifiscan.py

@@ -21,56 +21,57 @@ def get_log_files(log_root_path: str) -> list:
 
 def parse_path(path: str) -> list | dict:
     output = []
-    with open(path, 'r') as f:
-        for line in f:
-            line = line.strip()
-            # skip empty lines
-            if not line:
-                continue
-            parsed_data = {}
-            # process header
-            if line.startswith('total='):
-                items = line.split(',')
-                for item in items:
-                    key, value = item.split('=')
-                    parsed_data[key.strip()] = value.strip()
-            else:
-                # extract key-value by string
+    for logfile in get_log_files(path):
+        with open(logfile, 'r') as f:
+            for line in f:
+                line = line.strip()
+                # skip empty lines
+                if not line:
+                    continue
+                parsed_data = {}
+                # process header
+                if line.startswith('total='):
+                    items = line.split(',')
+                    for item in items:
+                        key, value = item.split('=')
+                        parsed_data[key.strip()] = value.strip()
+                else:
+                    # extract key-value by string
 
-                # first ssid and ssid_hex, but need to detect the type first
-                regexes = [
-                    # iOS 16 and later: FIRSTCONWIFI - ssid=4649525354434f4e57494649
-                    r"(?P<ssid>.+?) - ssid=(?P<ssid_hex>[^,]+)",
-                    # iOS 15: 'FIRSTCONWIFI' (4649525354434f4e57494649)
-                    r"'(?P<ssid>[^\']+)' \((?P<ssid_hex>[^\)]+)\)",
-                    # hidden:  <HIDDEN>
-                    r"(?P<ssid><HIDDEN>)(?P<ssid_hex>)",
-                ]
-                for regex in regexes:
-                    m = re.match(regex, line)
-                    if m:
-                        parsed_data['ssid'] = m.group('ssid')
-                        parsed_data['ssid_hex'] = m.group('ssid_hex')
-                        break
-                # key = first place with =
-                #  check what is after =, if normal char then value is until next ,
-                #                         if [ then value is until ]
-                #                         if { then value is until }
-                index_now = line.index(',') + 1
-                # now the rest of the line
-                while index_now < len(line):
-                    index_equals = line.index('=', index_now)
-                    key = line[index_now:index_equals].strip()
-                    if line[index_equals + 1] in ['[']:
-                        index_close = line.index(']', index_now)
-                        value = line[index_equals + 1:index_close].strip()
-                    else:
-                        try:
-                            index_close = line.index(',', index_now)
-                        except ValueError:  # catch end of line
-                            index_close = len(line)
-                        value = line[index_equals + 1:index_close].strip()
-                    index_now = index_close + 2
-                    parsed_data[key] = value
-            output.append(parsed_data)
+                    # first ssid and ssid_hex, but need to detect the type first
+                    regexes = [
+                        # iOS 16 and later: FIRSTCONWIFI - ssid=4649525354434f4e57494649
+                        r"(?P<ssid>.+?) - ssid=(?P<ssid_hex>[^,]+)",
+                        # iOS 15: 'FIRSTCONWIFI' (4649525354434f4e57494649)
+                        r"'(?P<ssid>[^\']+)' \((?P<ssid_hex>[^\)]+)\)",
+                        # hidden:  <HIDDEN>
+                        r"(?P<ssid><HIDDEN>)(?P<ssid_hex>)",
+                    ]
+                    for regex in regexes:
+                        m = re.match(regex, line)
+                        if m:
+                            parsed_data['ssid'] = m.group('ssid')
+                            parsed_data['ssid_hex'] = m.group('ssid_hex')
+                            break
+                    # key = first place with =
+                    #  check what is after =, if normal char then value is until next ,
+                    #                         if [ then value is until ]
+                    #                         if { then value is until }
+                    index_now = line.index(',') + 1
+                    # now the rest of the line
+                    while index_now < len(line):
+                        index_equals = line.index('=', index_now)
+                        key = line[index_now:index_equals].strip()
+                        if line[index_equals + 1] in ['[']:
+                            index_close = line.index(']', index_now)
+                            value = line[index_equals + 1:index_close].strip()
+                        else:
+                            try:
+                                index_close = line.index(',', index_now)
+                            except ValueError:  # catch end of line
+                                index_close = len(line)
+                            value = line[index_equals + 1:index_close].strip()
+                        index_now = index_close + 2
+                        parsed_data[key] = value
+                output.append(parsed_data)
     return output

parsers/wifisecurity.py

@@ -40,7 +40,7 @@ def parse_path(path: str) -> list | dict:
     entries = []
     element = {}
     try:
-        with open(path, "r") as f:
+        with open(get_log_files(path)[0], "r") as f:
             for line in f:
                 if ' : ' in line:
                     key, value = line.split(" : ")
@@ -51,5 +51,5 @@ def parse_path(path: str) -> list | dict:
                     # print(f"appending {element}")
                     element = {}
     except Exception as e:
-        print(f"Could not parse: {path}. Reason: {str(e)}")
+        print(f"Could not parse: {get_log_files(path)[0]}. Reason: {str(e)}")
     return entries

tests/test_parsers.py

@@ -20,7 +20,7 @@ def list_all_parsers(self):
             yield parser
 
     def test_parsers_filestructure(self):
-        required_functions = ['get_log_files', 'parse_path']
+        required_functions = ['get_log_files', 'parse_path']  # TODO add parse_path_to_folder(path: str, output: str) -> bool:
         required_variables = ['parser_description']
 
         parsers = self.list_all_parsers()

tests/test_parsers_olddsc.py

@@ -8,13 +8,12 @@ class TestParsersOlddsc(SysdiagnoseTestCase):
     def test_parse_olddsc_file(self):
         for log_root_path in self.log_root_paths:
             files = get_log_files(log_root_path)
-            for file in files:
-                print(f'Parsing {file}')
-                result = parse_path(file)
-                self.assertTrue('Unslid_Base_Address' in result)
-                self.assertTrue('Cache_UUID_String' in result)
-                self.assertTrue('Binaries' in result)
-                self.assertTrue(len(result['Binaries']) > 0)
+            print(f'Parsing {files}')
+            result = parse_path(log_root_path)
+            self.assertTrue('Unslid_Base_Address' in result)
+            self.assertTrue('Cache_UUID_String' in result)
+            self.assertTrue('Binaries' in result)
+            self.assertTrue(len(result['Binaries']) > 0)
 
 
 if __name__ == '__main__':

tests/test_parsers_plist.py

@@ -8,11 +8,14 @@ class TestParsersPlist(SysdiagnoseTestCase):
     def test_get_plists(self):
         for log_root_path in self.log_root_paths:
             files = get_log_files(log_root_path)
-            for file in files:
-                print(f'Parsing {file}')
-                parse_path(file)
-                # nothing specific to assert here as there's not always a result
-                # just catching exceptions
+            self.assertGreater(len(files), 0)
+            print(f'Parsing {files}')
+            result = parse_path(log_root_path)
+            self.assertGreater(len(result), 0)
+            print(result.keys())
+            self.assertIn('hidutil.plist', result.keys())
+            # nothing specific to assert here as there's not always a result
+            # just catching exceptions
 
 
 if __name__ == '__main__':