Address review feedback: kafka_ssl, duplicate silent, exception chain, log file reload, should_reload timing

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>
Agent-Logs-Url: https://github.com/domainaware/parsedmarc/sessions/a8a43c55-23fa-4471-abe6-7ac966f381f9

parsedmarc/__init__.py

@@ -2249,11 +2249,14 @@ def watch_inbox(
         )
         callback(res)
 
-    mailbox_connection.watch(
-        check_callback=check_callback,
-        check_timeout=check_timeout,
-        should_reload=should_reload,
-    )
+    watch_kwargs: dict = {
+        "check_callback": check_callback,
+        "check_timeout": check_timeout,
+    }
+    if should_reload is not None:
+        watch_kwargs["should_reload"] = should_reload
+
+    mailbox_connection.watch(**watch_kwargs)
 
 
 def append_json(

parsedmarc/cli.py

@@ -201,8 +201,20 @@ def _parse_config_file(config_file, opts):
                 "normalize_timespan_threshold_hours"
             )
         if "index_prefix_domain_map" in general_config:
-            with open(general_config["index_prefix_domain_map"]) as f:
-                index_prefix_domain_map = yaml.safe_load(f)
+            map_path = general_config["index_prefix_domain_map"]
+            try:
+                with open(map_path) as f:
+                    index_prefix_domain_map = yaml.safe_load(f)
+            except OSError as exc:
+                raise ConfigurationError(
+                    "Failed to read index_prefix_domain_map file "
+                    "'{0}': {1}".format(map_path, exc)
+                ) from exc
+            except yaml.YAMLError as exc:
+                raise ConfigurationError(
+                    "Failed to parse YAML in index_prefix_domain_map "
+                    "file '{0}': {1}".format(map_path, exc)
+                ) from exc
         if "offline" in general_config:
             opts.offline = bool(general_config.getboolean("offline"))
         if "strip_attachment_payloads" in general_config:
@@ -242,7 +254,7 @@ def _parse_config_file(config_file, opts):
             except Exception as ns_error:
                 raise ConfigurationError(
                     "DNS pre-flight check failed: {}".format(ns_error)
-                )
+                ) from ns_error
             if not dummy_hostname:
                 raise ConfigurationError(
                     "DNS pre-flight check failed: no PTR record for {} from {}".format(
@@ -259,8 +271,6 @@ def _parse_config_file(config_file, opts):
             opts.debug = bool(general_config.getboolean("debug"))
         if "verbose" in general_config:
             opts.verbose = bool(general_config.getboolean("verbose"))
-        if "silent" in general_config:
-            opts.silent = bool(general_config.getboolean("silent"))
         if "warnings" in general_config:
             opts.warnings = bool(general_config.getboolean("warnings"))
         if "fail_on_output_error" in general_config:
@@ -588,9 +598,9 @@ def _parse_config_file(config_file, opts):
                 "index setting missing from the splunk_hec config section"
             )
         if "skip_certificate_verification" in hec_config:
-            opts.hec_skip_certificate_verification = hec_config[
-                "skip_certificate_verification"
-            ]
+            opts.hec_skip_certificate_verification = bool(
+                hec_config.getboolean("skip_certificate_verification", fallback=False)
+            )
 
     if "kafka" in config.sections():
         kafka_config = config["kafka"]
@@ -620,14 +630,14 @@ def _parse_config_file(config_file, opts):
         if "forensic_topic" in kafka_config:
             opts.kafka_forensic_topic = kafka_config["forensic_topic"]
         else:
-            logger.critical(
+            raise ConfigurationError(
                 "forensic_topic setting missing from the kafka config section"
             )
         if "smtp_tls_topic" in kafka_config:
             opts.kafka_smtp_tls_topic = kafka_config["smtp_tls_topic"]
         else:
-            logger.critical(
-                "forensic_topic setting missing from the splunk_hec config section"
+            raise ConfigurationError(
+                "smtp_tls_topic setting missing from the kafka config section"
             )
 
     if "smtp" in config.sections():
@@ -934,15 +944,17 @@ def _init_output_clients(opts):
 
     if opts.kafka_hosts:
         ssl_context = None
-        if opts.kafka_skip_certificate_verification:
-            logger.debug("Skipping Kafka certificate verification")
+        if opts.kafka_ssl:
             ssl_context = create_default_context()
-            ssl_context.check_hostname = False
-            ssl_context.verify_mode = CERT_NONE
+            if opts.kafka_skip_certificate_verification:
+                logger.debug("Skipping Kafka certificate verification")
+                ssl_context.check_hostname = False
+                ssl_context.verify_mode = CERT_NONE
         clients["kafka_client"] = kafkaclient.KafkaClient(
             opts.kafka_hosts,
             username=opts.kafka_username,
             password=opts.kafka_password,
+            ssl=opts.kafka_ssl,
             ssl_context=ssl_context,
         )
 
@@ -1578,7 +1590,6 @@ def _main():
         normalize_timespan_threshold_hours=24.0,
         fail_on_output_error=False,
     )
-    args = arg_parser.parse_args()
 
     # Snapshot opts as set from CLI args / hardcoded defaults, before any config
     # file is applied. On each SIGHUP reload we restore this baseline first so
@@ -2036,6 +2047,31 @@ def _main():
                 if opts.debug:
                     logger.setLevel(logging.DEBUG)
 
+                # Refresh FileHandler if log_file changed
+                old_log_file = getattr(opts, "active_log_file", None)
+                new_log_file = opts.log_file
+                if old_log_file != new_log_file:
+                    # Remove old FileHandlers
+                    for h in list(logger.handlers):
+                        if isinstance(h, logging.FileHandler):
+                            h.close()
+                            logger.removeHandler(h)
+                    # Add new FileHandler if configured
+                    if new_log_file:
+                        try:
+                            fh = logging.FileHandler(new_log_file, "a")
+                            file_formatter = logging.Formatter(
+                                "%(asctime)s - %(levelname)s"
+                                " - [%(filename)s:%(lineno)d] - %(message)s"
+                            )
+                            fh.setFormatter(file_formatter)
+                            logger.addHandler(fh)
+                        except Exception as log_error:
+                            logger.warning(
+                                "Unable to write to log file: {}".format(log_error)
+                            )
+                    opts.active_log_file = new_log_file
+
                 logger.info("Configuration reloaded successfully")
             except Exception:
                 logger.exception(

parsedmarc/mail/gmail.py

@@ -178,10 +178,12 @@ class GmailConnection(MailboxConnection):
     def watch(self, check_callback, check_timeout, should_reload=None):
         """Checks the mailbox for new messages every n seconds"""
         while True:
-            sleep(check_timeout)
-            check_callback(self)
             if should_reload and should_reload():
                 return
+            sleep(check_timeout)
+            if should_reload and should_reload():
+                return
+            check_callback(self)
 
     @lru_cache(maxsize=10)
     def _find_label_id_for_label(self, label_name: str) -> str:

parsedmarc/mail/graph.py

@@ -281,10 +281,10 @@ class MSGraphConnection(MailboxConnection):
     def watch(self, check_callback, check_timeout, should_reload=None):
         """Checks the mailbox for new messages every n seconds"""
         while True:
-            sleep(check_timeout)
-            check_callback(self)
             if should_reload and should_reload():
                 return
+            sleep(check_timeout)
+            check_callback(self)
 
     @lru_cache(maxsize=10)
     def _find_folder_id_from_folder_path(self, folder_name: str) -> str:

tests.py

@@ -4,6 +4,7 @@
 from __future__ import absolute_import, print_function, unicode_literals
 
 import os
+import signal
 import sys
 import tempfile
 import unittest
@@ -1925,6 +1926,7 @@ password = pass
 watch = true
 """
 
+    @unittest.skipUnless(hasattr(signal, "SIGHUP"), "SIGHUP not available on this platform")
     @patch("parsedmarc.cli._init_output_clients")
     @patch("parsedmarc.cli._parse_config_file")
     @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
@@ -1990,6 +1992,7 @@ watch = true
         # _parse_config_file called for initial load + reload
         self.assertGreaterEqual(mock_parse_config.call_count, 2)
 
+    @unittest.skipUnless(hasattr(signal, "SIGHUP"), "SIGHUP not available on this platform")
     @patch("parsedmarc.cli._init_output_clients")
     @patch("parsedmarc.cli._parse_config_file")
     @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
@@ -2062,6 +2065,7 @@ watch = true
         # The failed reload must not have closed the original clients
         initial_clients["s3_client"].close.assert_not_called()
 
+    @unittest.skipUnless(hasattr(signal, "SIGHUP"), "SIGHUP not available on this platform")
     @patch("parsedmarc.cli._init_output_clients")
     @patch("parsedmarc.cli._parse_config_file")
     @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
@@ -2133,6 +2137,7 @@ watch = true
         # Old client must have been closed when reload succeeded
         old_client.close.assert_called_once()
 
+    @unittest.skipUnless(hasattr(signal, "SIGHUP"), "SIGHUP not available on this platform")
     @patch("parsedmarc.cli._init_output_clients")
     @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
     @patch("parsedmarc.cli.watch_inbox")