Implements #11 - JSONL logging instead of printing to stdout

pyproject.toml

@@ -29,7 +29,8 @@ dependencies = [
     "pandas==2.2.2",
     "numpy==2.0.0",
     "nska-deserialize==1.5.1",
-    "yara-python==4.5.1"
+    "yara-python==4.5.1",
+    "python-json-logger==2.0.7"
 ]
 
 [project.scripts]

requirements.txt

@@ -7,4 +7,5 @@ pandas==2.2.2
 numpy==2.0.0
 nska-deserialize==1.5.1
 yara-python==4.5.1
-# pycrashreport==1.2.4
+# pycrashreport==1.2.4
+python-json-logger==2.0.7

src/sysdiagnose/__init__.py

@@ -7,7 +7,7 @@
 import re
 import tarfile
 import fcntl
-from sysdiagnose.utils.base import BaseParserInterface, BaseAnalyserInterface, SysdiagnoseConfig
+from sysdiagnose.utils.base import BaseParserInterface, BaseAnalyserInterface, SysdiagnoseConfig, logger
 
 
 class Sysdiagnose:
@@ -147,7 +147,7 @@ def create_case(self, sysdiagnose_file: str, force: bool = False, case_id: bool
             case['unique_device_id'] = remotectl_dumpstate_json['Local device']['Properties']['UniqueDeviceID']
             case['ios_version'] = remotectl_dumpstate_json['Local device']['Properties']['OSVersion']
         except (KeyError, TypeError) as e:
-            print(f"WARNING: Could not parse remotectl_dumpstate, and therefore extract serial numbers. Error {e}")
+            logger.warning(f"WARNING: Could not parse remotectl_dumpstate, and therefore extract serial numbers.", exc_info=True)
 
         try:
             case['date'] = remotectl_dumpstate_parser.sysdiagnose_creation_datetime.isoformat(timespec='microseconds')

src/sysdiagnose/analysers/apps.py

@@ -4,12 +4,15 @@
 # Author: Emiliern Le Jamtel
 
 import re
+import logging
 from sysdiagnose.utils.base import BaseAnalyserInterface
 from sysdiagnose.parsers.accessibility_tcc import AccessibilityTccParser
 from sysdiagnose.parsers.brctl import BrctlParser
 from sysdiagnose.parsers.itunesstore import iTunesStoreParser
 from sysdiagnose.parsers.logarchive import LogarchiveParser
 
+logger = logging.getLogger('sysdiagnose')
+
 
 class AppsAnalyser(BaseAnalyserInterface):
     description = 'Get list of Apps installed on the device'
@@ -88,9 +91,9 @@ def execute(self):
                         if matches:
                             new_term = matches[0]
                         else:
-                            # print(f"Skipping entry: {entry['subsystem']}")
+                            logger.debug(f"Skipping entry: {entry['subsystem']}")
                             continue
-                # print(f"New entry: {new_term} - was: {entry['subsystem']}")
+                logger.debug(f"New entry: {new_term} - was: {entry['subsystem']}")
                 entry['subsystem'] = new_term
             # add it to the list
             try:

src/sysdiagnose/analysers/demo_analyser.py

@@ -3,7 +3,7 @@
 # For Python3
 # DEMO - Skeleton
 
-from sysdiagnose.utils.base import BaseAnalyserInterface
+from sysdiagnose.utils.base import BaseAnalyserInterface, logger
 
 
 class DemoAnalyser(BaseAnalyserInterface):
@@ -20,9 +20,17 @@ def execute(self):
         Load parsers here, and use the parser.get_result() to get the data.
         By doing so you will get the parser output even if it never ran before.
         """
-        print("DO SOMETHING HERE")
-
-        # json_data = p_fooparser.get_result()
+        try:
+            print("DO SOMETHING HERE")
+            logger.info("log something here", extra={'analyser': __name__})
+            if True:
+                logger.warning("This will log a warning")
+                # logger.error("This will log an error")
+
+            # json_data = p_fooparser.get_result()
+        except Exception as e:
+            logger.exception("This will log an error with the exception information")
+            # logger.warning("This will log a warning with the exception information", exc_info=True)
 
         result = {'foo': 'bar'}
         return result

src/sysdiagnose/analysers/timesketch.py

@@ -16,7 +16,7 @@
 from sysdiagnose.parsers.wifi_known_networks import WifiKnownNetworksParser
 from sysdiagnose.parsers.crashlogs import CrashLogsParser
 from collections.abc import Generator
-from sysdiagnose.utils.base import BaseAnalyserInterface
+from sysdiagnose.utils.base import BaseAnalyserInterface, logger
 
 
 class TimesketchAnalyser(BaseAnalyserInterface):
@@ -50,7 +50,7 @@ def __extract_ts_mobileactivation(self) -> Generator[dict, None, None]:
                     pass
                 yield ts_event
         except Exception as e:
-            print(f"ERROR while extracting timestamp from mobileactivation file. Reason: {str(e)}")
+            logger.exception("ERROR while extracting timestamp from mobileactivation file.")
 
     def __extract_ts_powerlogs(self) -> Generator[dict, None, None]:
         try:
@@ -91,7 +91,7 @@ def __extract_ts_powerlogs(self) -> Generator[dict, None, None]:
                     pass
 
         except Exception as e:
-            print(f"ERROR while extracting timestamp from powerlogs. Reason: {str(e)}")
+            logger.exception("ERROR while extracting timestamp from powerlogs.")
 
     def __extract_ts_swcutil(self) -> Generator[dict, None, None]:
         try:
@@ -111,10 +111,10 @@ def __extract_ts_swcutil(self) -> Generator[dict, None, None]:
                         yield ts_event
                     except KeyError:
                         # some entries do not have a Last Checked or timestamp field
-                        # print(f"WARNING {filename} while extracting timestamp from {(service['Service'])} - {(service['App ID'])}. Record not inserted.")
+                        logger.warning(f"Error while extracting timestamp from {(service['Service'])} - {(service['App ID'])}. Record not inserted.")
                         pass
         except Exception as e:
-            print(f"ERROR while extracting timestamp from swcutil. Reason {str(e)}")
+            logger.exception("ERROR while extracting timestamp from swcutil.")
 
     def __extract_ts_accessibility_tcc(self) -> Generator[dict, None, None]:
         try:
@@ -135,7 +135,7 @@ def __extract_ts_accessibility_tcc(self) -> Generator[dict, None, None]:
                 }
                 yield ts_event
         except Exception as e:
-            print(f"ERROR while extracting timestamp from accessibility_tcc. Reason {str(e)}")
+            logger.exception("ERROR while extracting timestamp from accessibility_tcc.")
 
     def __extract_ts_shutdownlogs(self) -> Generator[dict, None, None]:
         try:
@@ -153,9 +153,9 @@ def __extract_ts_shutdownlogs(self) -> Generator[dict, None, None]:
                     }
                     yield ts_event
                 except Exception as e:
-                    print(f"WARNING: shutdownlog entry not parsed: {event}. Reason: {str(e)}")
+                    logger.warning(f"WARNING: shutdownlog entry not parsed: {event}", exc_info=True)
         except Exception as e:
-            print(f"ERROR while extracting timestamp from shutdownlog. Reason: {str(e)}")
+            logger.exception("ERROR while extracting timestamp from shutdownlog.")
 
     def __extract_ts_logarchive(self) -> Generator[dict, None, None]:
         try:
@@ -173,9 +173,9 @@ def __extract_ts_logarchive(self) -> Generator[dict, None, None]:
                     }
                     yield ts_event
                 except KeyError as e:
-                    print(f"WARNING: trace not parsed: {event}. Error {e}")
+                    logger.warning(f"WARNING: trace not parsed: {event}.", exc_info=True)
         except Exception as e:
-            print(f"ERROR while extracting timestamp from logarchive. Reason: {str(e)}")
+            logger.exception(f"ERROR while extracting timestamp from logarchive.")
 
     def __extract_ts_wifisecurity(self) -> Generator[dict, None, None]:
         try:
@@ -206,7 +206,7 @@ def __extract_ts_wifisecurity(self) -> Generator[dict, None, None]:
                 }
                 yield ts_event
         except Exception as e:
-            print(f"ERROR while extracting timestamp from wifisecurity. Reason {str(e)}")
+            logger.exception("ERROR while extracting timestamp from wifisecurity.")
 
     def __extract_ts_wifi_known_networks(self) -> Generator[dict, None, None]:
         try:
@@ -228,7 +228,7 @@ def __extract_ts_wifi_known_networks(self) -> Generator[dict, None, None]:
                     yield ts_event
                 except KeyError:
                     # some wifi networks do not have an AddedAt field
-                    # print(f"ERROR {filename} while extracting timestamp from {ssid}. Reason: {str(e)}. Record not inserted.")
+                    logger.warning(f"Error while extracting timestamp from {ssid}. Record not inserted.", exc_info=True)
                     pass
 
                 # WIFI modified
@@ -245,7 +245,7 @@ def __extract_ts_wifi_known_networks(self) -> Generator[dict, None, None]:
                     yield ts_event
                 except KeyError:
                     # some wifi networks do not have an UpdatedAt field
-                    # print(f"ERROR {filename} while extracting timestamp from {ssid}. Reason: {str(e)}. Record not inserted.")
+                    logger.warning(f"Error while extracting timestamp from {ssid}.Record not inserted.", exc_info=True)
                     pass
 
                 # Password for wifi modified
@@ -262,10 +262,10 @@ def __extract_ts_wifi_known_networks(self) -> Generator[dict, None, None]:
                     yield ts_event
                 except KeyError:
                     # some wifi networks do not have a password modification date
-                    # print(f"ERROR {filename} while extracting timestamp from {ssid}. Reason: {str(e)}. Record not inserted.")
+                    logger.warning(f"Error while extracting timestamp from {ssid}. Record not inserted.", exc_info=True)
                     pass
         except Exception as e:
-            print(f"ERROR while extracting timestamp from wifi_known_networks. Reason {str(e)}")
+            logger.exception("ERROR while extracting timestamp from wifi_known_networks.")
 
     def __extract_ts_crashlogs(self) -> Generator[dict, None, None]:
         try:
@@ -288,7 +288,7 @@ def __extract_ts_crashlogs(self) -> Generator[dict, None, None]:
                     # skip bug_type fields
                     pass
         except Exception as e:
-            print(f"ERROR while extracting timestamp from crashlog. Reason {str(e)}")
+            logger.exception("ERROR while extracting timestamp from crashlog.")
 
     def execute(self):
         # Get all the functions that start with '__extract_ts_'

src/sysdiagnose/analysers/wifi_geolocation.py

@@ -6,7 +6,7 @@
 import dateutil.parser
 import gpxpy
 import gpxpy.gpx
-from sysdiagnose.utils.base import BaseAnalyserInterface
+from sysdiagnose.utils.base import BaseAnalyserInterface, logger
 from sysdiagnose.parsers.wifi_known_networks import WifiKnownNetworksParser
 
 
@@ -45,7 +45,7 @@ def generate_gpx_from_known_networks_json(json_data: str, output_file: str):
             try:
                 timestamp = dateutil.parser.parse(timestamp_str)
             except Exception as e:
-                print(f"Error converting timestamp. Reason: {str(e)}. Timestamp was: {str(timestamp_str)}. Assuming Jan 1st 1970.")
+                logger.exception(f"Error converting timestamp. Timestamp was: {str(timestamp_str)}. Assuming Jan 1st 1970.")
                 timestamp = dateutil.parser.parse('1970-01-01')     # begin of epoch
 
             bssid = network_data.get('__OSSpecific__', {}).get('BSSID', '')

src/sysdiagnose/analysers/yarascan.py

@@ -3,7 +3,7 @@
 import glob
 import threading
 import queue
-from sysdiagnose.utils.base import BaseAnalyserInterface
+from sysdiagnose.utils.base import BaseAnalyserInterface, logger
 
 
 # These are the commonly used external variables that can be used in the YARA rules
@@ -36,7 +36,7 @@ def execute(self):
         results = {'errors': [], 'matches': []}
 
         if not os.path.isdir(self.yara_rules_path):
-            print(f"ERROR: Could not find the YARA rules (.yar) folder: {self.yara_rules_path}")
+            logger.error(f"ERROR: Could not find the YARA rules (.yar) folder: {self.yara_rules_path}")
             results['errors'].append(f"Could not find the YARA rules (.yar) folder: {self.yara_rules_path}")
             return results
 
@@ -69,7 +69,7 @@ def execute(self):
             results['matches'] = matches
 
         if len(results['errors']) > 0:
-            print("Scan finished with errors. Review the results")
+            logger.error("Scan finished with errors. Review the results")
 
         return results
 
@@ -80,17 +80,17 @@ def get_valid_yara_rule_files(self) -> tuple[list, list]:
         for rule_file in rule_files_to_test:
             if not os.path.isfile(rule_file):
                 continue
-            print(f"Loading YARA rule: {rule_file}")
+            logger.info(f"Loading YARA rule: {rule_file}")
             try:
                 yara.compile(filepath=rule_file, externals=externals)
                 # if we reach this point, the rule is valid
                 rule_files_validated.append(rule_file)
             except yara.SyntaxError as e:
-                print(f"Error compiling rule {rule_file}: {str(e)}")
+                logger.exception(f"Error compiling rule {rule_file}")
                 errors.append(f"Error compiling rule {rule_file}: {str(e)}")
                 continue
             except yara.Error as e:
-                print(f"Error compiling rule {rule_file}: {str(e)}")
+                logger.exception(f"Error compiling rule {rule_file}")
                 errors.append(f"Error loading rule {rule_file}: {str(e)}")
                 continue
 
@@ -111,7 +111,7 @@ def scan_directory(directories: list, rule_filepaths: dict, ignore_files: list,
                 for ignore_folder in ignore_folders:
                     if root.startswith(ignore_folder):
                         stop = True
-                        print(f"Skipping folder: {root}")
+                        logger.info(f"Skipping folder: {root}")
                         continue
                 if stop:
                     continue
@@ -121,7 +121,7 @@ def scan_directory(directories: list, rule_filepaths: dict, ignore_files: list,
                     for ignore_file in ignore_files:
                         if file_full_path.startswith(ignore_file):
                             stop = True
-                            print(f"Skipping file: {file_full_path}")
+                            logger.info(f"Skipping file: {file_full_path}")
                             continue
                     if stop:
                         continue
@@ -133,13 +133,13 @@ def consumer():
             rules = yara.compile(filepaths=rule_filepaths, externals=externals)
 
             while True:
-                print(f"Consumer thread seeing {file_queue.qsize()} files in queue, and taking one")
+                logger.info(f"Consumer thread seeing {file_queue.qsize()} files in queue, and taking one")
                 file_path = file_queue.get()
                 if file_path is None:
-                    print("Consumer thread exiting")
+                    logger.info("Consumer thread exiting")
                     break
 
-                print(f"Scanning file: {file_path}")
+                logger.info(f"Scanning file: {file_path}")
                 # set the externals for this file - massive slowdown
                 # externals_local = externals.copy()
                 # externals_local['filename'] = file