Improved logging (#13)

Reworked logging and added new options:

- Ability to output JSON logs to stdout (-js/--json-stdout)
- Ability to save JSON logs to a specified path (-jo/--json-output)

---------

Co-authored-by: Mathieu <35560225+0xFustang@users.noreply.github.com>

src/droid/__main__.py

@@ -46,6 +46,8 @@ def init_argparse() -> argparse.ArgumentParser:
     parser.add_argument("-m", "--mssp", help="Enable MSSP mode", action="store_true")
     parser.add_argument("-mo", "--module", help="Module mode to return converted rules as a list", action="store_true")
     parser.add_argument("-j", "--json", help="Drop a JSON log file", action="store_true")
+    parser.add_argument("-jo", "--json-output", help="Optional path for JSON log file")
+    parser.add_argument("-js", "--json-stdout", help="Enable logging to stdout in JSON", action="store_true")
     parser.add_argument("-i", "--integrity", help="Perform an integrity check on platforms", action="store_true")
     return parser
 
@@ -253,18 +255,20 @@ def main(argv=None) -> None:
     parser = init_argparse()
     args = parser.parse_args(argv)
 
-    logger = ColorLogger("droid")
+    logger_param = {
+        "debug_mode": args.debug,
+        "json_enabled": args.json,
+        "json_stdout": args.json_stdout,
+        "log_file": args.json_output
+    }
+    logger = ColorLogger("droid", **logger_param)
 
     # Set logger level based on debug flag
     if args.debug:
         logger.setLevel(logging.DEBUG)
     else:
         logger.setLevel(logging.INFO)
 
-    # Configure JSON logging if requested
-    if args.json:
-        logger.enable_json_logging()
-
     logging.setLoggerClass(ColorLogger)
 
     parameters = args
@@ -285,7 +289,7 @@ def main(argv=None) -> None:
             logger.info("Raw rules are not subject to Sigma validation.")
             exit(0)
 
-        rule_error = validate_rules(parameters, False, base_config)
+        rule_error = validate_rules(parameters, False, base_config, logger_param)
 
         if rule_error:
             logger.error("Validation issues found")
@@ -308,7 +312,7 @@ def main(argv=None) -> None:
             logger.error("Please select a platform.")
             exit(1)
 
-        conversion_error, search_warning = convert_rules(parameters, droid_platform_config(args, config_path), base_config)
+        conversion_error, search_warning = convert_rules(parameters, droid_platform_config(args, config_path), base_config, logger_param)
 
         if conversion_error:
             logger.error("Conversion errors found")
@@ -325,10 +329,10 @@ def main(argv=None) -> None:
 
         if is_raw_rule(args, base_config):
             logger.info(f"Searching raw rule for platform {args.platform} selected")
-            search_error, search_warning = search_rule_raw(parameters, droid_platform_config(args, config_path))
+            search_error, search_warning = search_rule_raw(parameters, droid_platform_config(args, config_path), logger_param)
         else:
             logger.info(f"Searching Sigma rule for platform {args.platform} selected")
-            search_error, search_warning = convert_rules(parameters, droid_platform_config(args, config_path), base_config)
+            search_error, search_warning = convert_rules(parameters, droid_platform_config(args, config_path), base_config, logger_param)
 
         if search_error and search_warning:
             logger.warning("Hits found while search one or multiple rules")
@@ -355,38 +359,38 @@ def main(argv=None) -> None:
         if args.platform == 'splunk':
             if is_raw_rule(args, base_config):
                 logger.info("Splunk raw rule selected")
-                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path))
+                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path), logger_param)
             else:
-                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config)
+                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config, logger_param)
 
         elif args.platform == 'azure':
             if is_raw_rule(args, base_config):
                 logger.info("Azure Sentinel raw rule selected")
-                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path))
+                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path), logger_param)
             else:
-                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config)
+                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config, logger_param)
 
         elif args.platform == 'microsoft_defender' and args.sentinel_mde:
             if is_raw_rule(args, base_config):
                 logger.info("Microsoft Defender for Endpoint raw rule selected")
-                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path))
+                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path), logger_param)
             else:
-                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config)
+                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config, logger_param)
 
         elif args.platform == "microsoft_defender":
             if is_raw_rule(args, base_config):
                 logger.info("Microsoft XDR raw rule selected")
-                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path))
+                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path), logger_param)
             else:
-                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config)
+                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config, logger_param)
 
         elif args.platform == 'esql' or args.platform == 'eql':
             args.platform == 'elastic'
             if is_raw_rule(args, base_config):
                 logger.info("Elastic Security raw rule selected")
-                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path))
+                export_error = export_rule_raw(parameters, droid_platform_config(args, config_path), logger_param)
             else:
-                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config)
+                export_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config, logger_param)
 
         else:
             logger.error("Please select one platform. See option -p in --help")
@@ -407,7 +411,7 @@ def main(argv=None) -> None:
     elif args.list:
 
         logger.info(f"List mode was selected - path selected: {args.rules}")
-        list_keys_errors = list_keys(parameters)
+        list_keys_errors = list_keys(parameters, logger_param)
 
     elif args.integrity:
 
@@ -417,10 +421,10 @@ def main(argv=None) -> None:
 
         if is_raw_rule(args, base_config):
             logger.info(f"Integrity check for platform {args.platform} selected")
-            integrity_error = integrity_rule_raw(parameters, droid_platform_config(args, config_path))
+            integrity_error = integrity_rule_raw(parameters, droid_platform_config(args, config_path), logger_param)
         else:
             logger.info(f"Integrity check for platform {args.platform} selected")
-            integrity_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config)
+            integrity_error = convert_rules(parameters, droid_platform_config(args, config_path), base_config, logger_param)
 
         if integrity_error:
             logger.error("Integrity error")

src/droid/color.py

@@ -36,38 +36,47 @@ def format(self, record):
         return logging.Formatter.format(self, record)
 
 class ColorLogger(logging.Logger):
-    def __init__(self, name, debug_mode=False):
-        super().__init__(name, logging.DEBUG)
-        self.json_enabled = False
+    def __init__(self, name, debug_mode=False, json_enabled=False, json_stdout=False, log_file=None):
+        if debug_mode:
+            super().__init__(name, logging.DEBUG)
+        else:
+            super().__init__(name, logging.WARNING)
+        if log_file:
+            json_enabled = True
+        if not log_file:
+            log_file = "droid.log"
+        self.json_enabled = json_enabled
+        self.json_stdout = json_stdout
+        self.log_file = log_file
         self.debug_mode = debug_mode
 
+
         self.setup_handlers()
 
     def setup_handlers(self):
+        """Setup the handlers for logging depending on the options passed."""
         format_str = "%(asctime)s:%(levelname)s:%(name)s:%(message)s"
-        self.handlers = []
+        self.handlers = []  # Clear existing handlers
 
         if self.json_enabled:
+            # Set up JSON file out
             json_formatter = jsonlogger.JsonFormatter(format_str)
-            json_handler = logging.FileHandler('droid.log')
-            json_handler.setFormatter(json_formatter)
-            self.addHandler(json_handler)
-            console = logging.StreamHandler()
-            color_formatter = ColorFormatter(format_str)
-            console.setFormatter(color_formatter)
-            self.addHandler(console)
+            # Log JSON output to a file
+            json_file_handler = logging.FileHandler(self.log_file)
+            json_file_handler.setFormatter(json_formatter)
+            self.addHandler(json_file_handler)
+
+        if self.json_stdout:
+            stdout_formater = jsonlogger.JsonFormatter(format_str)
         else:
-            color_formatter = ColorFormatter(format_str)
-            console = logging.StreamHandler()
-            console.setFormatter(color_formatter)
+            # Console logging with color formatting if JSON is not enabled for stdout
+            stdout_formater = ColorFormatter(format_str)
+        console = logging.StreamHandler()
+        console.setFormatter(stdout_formater)
 
-            # Add AzureLogFilter only when not in debug mode
-            if not self.debug_mode:
-                azure_filter = AzureLogFilter(debug_mode=self.debug_mode)
-                console.addFilter(azure_filter)
+        # Add AzureLogFilter only when not in debug mode
+        if not self.debug_mode:
+            azure_filter = AzureLogFilter(debug_mode=self.debug_mode)
+            console.addFilter(azure_filter)
 
-            self.addHandler(console)
-
-    def enable_json_logging(self):
-        self.json_enabled = True
-        self.setup_handlers()
+        self.addHandler(console)

src/droid/convert.py

@@ -23,19 +23,11 @@ class Conversion:
     Args:
         parameters(dict)
     """
-    def __init__(self, parameters: dict, base_config, platform_name, debug, json) -> None:
-        self.logger = ColorLogger("droid.convert.Conversion")
+    def __init__(self, parameters: dict, base_config, platform_name, logger_param) -> None:
+        self.logger = ColorLogger(__name__, **logger_param)
         self._parameters = parameters["pipelines"]
         self._filters_directory = base_config.get('sigma_filters_directory', None)
         self._platform_name = platform_name
-        self._debug = debug
-        self._json = json
-
-        if self._json:
-            self.logger.enable_json_logging()
-
-        if self._debug:
-            self.logger.info("Initializing droid.convert.Conversion")
 
     def get_pipeline_config_group(self, rule_content):
         """Retrieve the logsource config group name
@@ -148,22 +140,18 @@ def load_rule(rule_file):
             error = True
             return error
 
-def convert_sigma_rule(rule_file, parameters, logger, sigma_objects, target, platform, error, search_warning, rules):
+def convert_sigma_rule(rule_file, parameters, logger, sigma_objects, target, platform, error, search_warning, rules, logger_param):
 
-    if parameters.debug:
-        logger.debug("processing rule {0}".format(rule_file))
+    logger.debug("processing rule {0}".format(rule_file))
 
     rule_content = load_rule(rule_file)
     sigma_objects[rule_content['title']] = rule_content
-    error, search_warning = convert_sigma(parameters, logger, rule_content, rule_file, target, platform, error, search_warning, rules)
+    error, search_warning = convert_sigma(parameters, logger, rule_content, rule_file, target, platform, error, search_warning, rules, logger_param)
     return error, search_warning
 
-def convert_rules(parameters, droid_config, base_config):
-
-    logger = ColorLogger("droid.convert")
+def convert_rules(parameters, droid_config, base_config, logger_param):
 
-    if parameters.json:
-        logger.enable_json_logging()
+    logger = ColorLogger(__name__, **logger_param)
 
     error = False
     search_warning = False
@@ -176,28 +164,28 @@ def convert_rules(parameters, droid_config, base_config):
 
     if parameters.platform and parameters.convert:
         platform_name = parameters.platform
-        target = Conversion(droid_config, base_config, platform_name, parameters.debug, parameters.json)
+        target = Conversion(droid_config, base_config, platform_name, logger_param)
         platform = None
 
     if parameters.platform and (parameters.search or parameters.export or parameters.integrity):
         platform_name = parameters.platform
-        target = Conversion(droid_config, base_config, platform_name, parameters.debug, parameters.json)
+        target = Conversion(droid_config, base_config, platform_name, logger_param)
         if platform_name == 'splunk':
-            platform = SplunkPlatform(droid_config, parameters.debug, parameters.json)
+            platform = SplunkPlatform(droid_config, logger_param)
         elif 'esql' in platform_name:
-            platform = ElasticPlatform(droid_config, parameters.debug, parameters.json, "esql", raw=False)
+            platform = ElasticPlatform(droid_config, logger_param, "esql", raw=False)
         elif 'eql' in platform_name:
-            platform = ElasticPlatform(droid_config, parameters.debug, parameters.json, "eql", raw=False)
+            platform = ElasticPlatform(droid_config, logger_param, "eql", raw=False)
         elif 'azure' in platform_name:
-            platform = SentinelPlatform(droid_config, parameters.debug, parameters.json)
+            platform = SentinelPlatform(droid_config, logger_param)
         elif parameters.platform == 'microsoft_defender':
-            platform = MicrosoftXDRPlatform(droid_config, parameters.debug, parameters.json)
+            platform = MicrosoftXDRPlatform(droid_config, logger_param)
 
     if path.is_dir():
         error_i = False
         search_warning_i = False
         for rule_file in path.rglob("*.y*ml"):
-            error, search_warning = convert_sigma_rule(rule_file, parameters, logger, sigma_objects, target, platform, error, search_warning, rules)
+            error, search_warning = convert_sigma_rule(rule_file, parameters, logger, sigma_objects, target, platform, error, search_warning, rules, logger_param)
             if parameters.module:
                 rules.append(error)
             if error:
@@ -212,7 +200,7 @@ def convert_rules(parameters, droid_config, base_config):
             return error, search_warning
 
     elif path.is_file():
-        error, search_warning = convert_sigma_rule(path, parameters, logger, sigma_objects, target, platform, error, search_warning, rules)
+        error, search_warning = convert_sigma_rule(path, parameters, logger, sigma_objects, target, platform, error, search_warning, rules, logger_param)
         if parameters.module:
             rules.append(error)
     else:
@@ -237,13 +225,16 @@ def convert_rules(parameters, droid_config, base_config):
         return error, search_warning
 
 
-def convert_sigma(parameters, logger, rule_content, rule_file, target, platform, error, search_warning, rules):
+def convert_sigma(
+        parameters, logger, rule_content,
+        rule_file, target, platform,
+        error, search_warning, rules,
+        logger_param):
 
     try:
         rule_converted = target.convert_rule(rule_content, rule_file, platform)
 
-        if parameters.debug:
-            logger.debug(f"Rule {rule_file} converted into: {rule_converted}", extra={"rule_file": rule_file, "rule_converted": rule_converted, "rule_content": rule_content})
+        logger.debug(f"Rule {rule_file} converted into: {rule_converted}", extra={"rule_file": rule_file, "rule_converted": rule_converted, "rule_content": rule_content})
 
     except SigmaFeatureNotSupportedByBackendError as e:
         logger.warning(f"Sigma Backend Error: {rule_file} - error: {e}", extra={"rule_file": rule_file, "error": e, "rule_content": rule_content})
@@ -265,26 +256,26 @@ def convert_sigma(parameters, logger, rule_content, rule_file, target, platform,
 
     if parameters.export and parameters.search and rule_converted:
         try:
-            error, search_warning = search_rule(parameters, rule_content, rule_converted, platform, rule_file, error, search_warning)
+            error, search_warning = search_rule(parameters, rule_content, rule_converted, platform, rule_file, error, search_warning, logger_param)
         except:
             logger.error(f"Could not export the rule {rule_file} since the search ran into error.", extra={"rule_file": rule_file, "error": e, "rule_content": rule_content})
 
         if not error:
             error = False
-            error = export_rule(parameters, rule_content, rule_converted, platform, rule_file, error)
+            error = export_rule(parameters, rule_content, rule_converted, platform, rule_file, error, logger_param)
 
         return error, search_warning
 
     elif parameters.search and rule_converted:
-        error, search_warning = search_rule(parameters, rule_content, rule_converted, platform, rule_file, error, search_warning)
+        error, search_warning = search_rule(parameters, rule_content, rule_converted, platform, rule_file, error, search_warning, logger_param)
         return error, search_warning
 
     elif parameters.export and rule_converted:
-        error = export_rule(parameters, rule_content, rule_converted, platform, rule_file, error)
+        error = export_rule(parameters, rule_content, rule_converted, platform, rule_file, error, logger_param)
         return error, search_warning
 
     elif parameters.integrity and rule_converted:
-        error = integrity_rule(parameters, rule_converted, rule_content, platform, rule_file, error)
+        error = integrity_rule(parameters, rule_converted, rule_content, platform, rule_file, error, logger_param)
         return error, search_warning
 
     elif parameters.module: