Enhance resource management: add close methods for S3Client and HECClient, and improve IMAP connection handling during IDLE. Update CHANGELOG.md for config reload improvements and bug fixes.

* Initial plan

* Address review feedback: Kafka SSL context, SIGHUP handler safety, test formatting

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>
Agent-Logs-Url: https://github.com/domainaware/parsedmarc/sessions/8f2fd48f-32a4-4258-9a89-06f7c7ac29bf

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

.claude/settings.json

@@ -7,7 +7,8 @@
       "Bash(GITHUB_ACTIONS=true pytest --cov tests.py)",
       "Bash(ls tests*)",
       "Bash(GITHUB_ACTIONS=true python -m pytest --cov tests.py -x)",
-      "Bash(GITHUB_ACTIONS=true python -m pytest tests.py -x -v)"
+      "Bash(GITHUB_ACTIONS=true python -m pytest tests.py -x -v)",
+      "Bash(python -m pytest tests.py --no-header -q)"
     ],
     "additionalDirectories": [
       "/tmp"

CHANGELOG.md

@@ -8,11 +8,38 @@
   destinations, DNS/GeoIP settings, processing flags, and log level
   without restarting the service or interrupting in-progress report
   processing. Use `systemctl reload parsedmarc` when running under
-  systemd.
+  systemd. On a successful reload, old output clients are closed and
+  recreated. On a failed reload, the previous configuration remains
+  fully active.
+- `close()` methods on GelfClient, KafkaClient, SyslogClient,
+  WebhookClient, HECClient, and S3Client for clean resource teardown
+  on reload.
+- `should_reload` parameter on all `MailboxConnection.watch()`
+  implementations and `watch_inbox()` to ensure SIGHUP never triggers
+  a new email batch mid-reload.
+- Elasticsearch and OpenSearch connections are now tracked and cleaned
+  up on reload via `_close_output_clients()`.
 - Extracted `_parse_config_file()` and `_init_output_clients()` from
   `_main()` in `cli.py` to support config reload and reduce code
   duplication.
 
+### Fixed
+
+- `get_index_prefix()` crashed on forensic reports with `TypeError`
+  due to `report()` instead of `report[]` dict access.
+- Missing `exit(1)` after IMAP user/password validation failure
+  allowed execution to continue with `None` credentials.
+- IMAP `watch()` leaked a connection on every IDLE cycle by not
+  closing the old `IMAPClient` before replacing it.
+- Resource leak in `_init_output_clients()` when Splunk HEC
+  configuration is invalid — the partially-constructed HEC client
+  is now cleaned up on error.
+- Elasticsearch/OpenSearch `set_hosts()` global state was not
+  rollback-safe on reload failure — init now runs last so other
+  client failures don't leave stale global connections.
+- `active_log_file` was not initialized at startup, causing the
+  first reload to unnecessarily remove and re-add the FileHandler.
+
 ## 9.2.1
 
 ### Added

docs/source/usage.md

@@ -666,8 +666,15 @@ Mailbox connection settings (IMAP host/credentials, Microsoft Graph,
 Gmail API, Maildir path) are **not** reloaded — changing those still
 requires a full restart.
 
-If the new configuration file contains errors, the reload is aborted
-and the previous configuration remains active. Check the logs for
+On a **successful** reload, existing output client connections are
+closed and new ones are created from the updated configuration. The
+service then resumes watching with the new settings.
+
+If the new configuration file contains errors (missing required
+settings, unreachable output destinations, etc.), the **entire reload
+is aborted** — no output clients are replaced and the previous
+configuration remains fully active. This means a typo in one section
+will not take down an otherwise working setup. Check the logs for
 details:
 
 ```bash

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

@@ -242,7 +242,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 +259,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 +586,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 +618,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():
@@ -749,15 +747,15 @@ def _parse_config_file(config_file, opts):
         if "oauth2_port" in gmail_api_config:
             opts.gmail_api_oauth2_port = gmail_api_config.getint("oauth2_port", 8080)
         if "auth_mode" in gmail_api_config:
-            opts.gmail_api_auth_mode = gmail_api_config.get("auth_mode").strip()
+            opts.gmail_api_auth_mode = gmail_api_config["auth_mode"].strip()
         if "service_account_user" in gmail_api_config:
-            opts.gmail_api_service_account_user = gmail_api_config.get(
+            opts.gmail_api_service_account_user = gmail_api_config[
                 "service_account_user"
-            ).strip()
+            ].strip()
         elif "delegated_user" in gmail_api_config:
-            opts.gmail_api_service_account_user = gmail_api_config.get(
+            opts.gmail_api_service_account_user = gmail_api_config[
                 "delegated_user"
-            ).strip()
+            ].strip()
 
     if "maildir" in config.sections():
         maildir_api_config = config["maildir"]
@@ -812,6 +810,26 @@ def _parse_config_file(config_file, opts):
     return index_prefix_domain_map
 
 
+class _ElasticsearchHandle:
+    """Sentinel so Elasticsearch participates in _close_output_clients."""
+
+    def close(self):
+        try:
+            elastic.connections.remove_connection("default")
+        except Exception:
+            pass
+
+
+class _OpenSearchHandle:
+    """Sentinel so OpenSearch participates in _close_output_clients."""
+
+    def close(self):
+        try:
+            opensearch.connections.remove_connection("default")
+        except Exception:
+            pass
+
+
 def _init_output_clients(opts):
     """Create output clients based on current opts.
 
@@ -823,76 +841,6 @@ def _init_output_clients(opts):
     """
     clients = {}
 
-    if opts.save_aggregate or opts.save_forensic or opts.save_smtp_tls:
-        if opts.elasticsearch_hosts:
-            es_aggregate_index = "dmarc_aggregate"
-            es_forensic_index = "dmarc_forensic"
-            es_smtp_tls_index = "smtp_tls"
-            if opts.elasticsearch_index_suffix:
-                suffix = opts.elasticsearch_index_suffix
-                es_aggregate_index = "{0}_{1}".format(es_aggregate_index, suffix)
-                es_forensic_index = "{0}_{1}".format(es_forensic_index, suffix)
-                es_smtp_tls_index = "{0}_{1}".format(es_smtp_tls_index, suffix)
-            if opts.elasticsearch_index_prefix:
-                prefix = opts.elasticsearch_index_prefix
-                es_aggregate_index = "{0}{1}".format(prefix, es_aggregate_index)
-                es_forensic_index = "{0}{1}".format(prefix, es_forensic_index)
-                es_smtp_tls_index = "{0}{1}".format(prefix, es_smtp_tls_index)
-            elastic_timeout_value = (
-                float(opts.elasticsearch_timeout)
-                if opts.elasticsearch_timeout is not None
-                else 60.0
-            )
-            elastic.set_hosts(
-                opts.elasticsearch_hosts,
-                use_ssl=opts.elasticsearch_ssl,
-                ssl_cert_path=opts.elasticsearch_ssl_cert_path,
-                username=opts.elasticsearch_username,
-                password=opts.elasticsearch_password,
-                api_key=opts.elasticsearch_api_key,
-                timeout=elastic_timeout_value,
-            )
-            elastic.migrate_indexes(
-                aggregate_indexes=[es_aggregate_index],
-                forensic_indexes=[es_forensic_index],
-            )
-
-        if opts.opensearch_hosts:
-            os_aggregate_index = "dmarc_aggregate"
-            os_forensic_index = "dmarc_forensic"
-            os_smtp_tls_index = "smtp_tls"
-            if opts.opensearch_index_suffix:
-                suffix = opts.opensearch_index_suffix
-                os_aggregate_index = "{0}_{1}".format(os_aggregate_index, suffix)
-                os_forensic_index = "{0}_{1}".format(os_forensic_index, suffix)
-                os_smtp_tls_index = "{0}_{1}".format(os_smtp_tls_index, suffix)
-            if opts.opensearch_index_prefix:
-                prefix = opts.opensearch_index_prefix
-                os_aggregate_index = "{0}{1}".format(prefix, os_aggregate_index)
-                os_forensic_index = "{0}{1}".format(prefix, os_forensic_index)
-                os_smtp_tls_index = "{0}{1}".format(prefix, os_smtp_tls_index)
-            opensearch_timeout_value = (
-                float(opts.opensearch_timeout)
-                if opts.opensearch_timeout is not None
-                else 60.0
-            )
-            opensearch.set_hosts(
-                opts.opensearch_hosts,
-                use_ssl=opts.opensearch_ssl,
-                ssl_cert_path=opts.opensearch_ssl_cert_path,
-                username=opts.opensearch_username,
-                password=opts.opensearch_password,
-                api_key=opts.opensearch_api_key,
-                timeout=opensearch_timeout_value,
-                auth_type=opts.opensearch_auth_type,
-                aws_region=opts.opensearch_aws_region,
-                aws_service=opts.opensearch_aws_service,
-            )
-            opensearch.migrate_indexes(
-                aggregate_indexes=[os_aggregate_index],
-                forensic_indexes=[os_forensic_index],
-            )
-
     if opts.s3_bucket:
         clients["s3_client"] = s3.S3Client(
             bucket_name=opts.s3_bucket,
@@ -965,6 +913,83 @@ def _init_output_clients(opts):
             timeout=opts.webhook_timeout,
         )
 
+    # Elasticsearch and OpenSearch mutate module-level global state via
+    # connections.create_connection(), which cannot be rolled back if a later
+    # step fails.  Initialise them last so that all other clients are created
+    # successfully first; this minimises the window for partial-init problems
+    # during config reload.
+    if opts.save_aggregate or opts.save_forensic or opts.save_smtp_tls:
+        if opts.elasticsearch_hosts:
+            es_aggregate_index = "dmarc_aggregate"
+            es_forensic_index = "dmarc_forensic"
+            es_smtp_tls_index = "smtp_tls"
+            if opts.elasticsearch_index_suffix:
+                suffix = opts.elasticsearch_index_suffix
+                es_aggregate_index = "{0}_{1}".format(es_aggregate_index, suffix)
+                es_forensic_index = "{0}_{1}".format(es_forensic_index, suffix)
+                es_smtp_tls_index = "{0}_{1}".format(es_smtp_tls_index, suffix)
+            if opts.elasticsearch_index_prefix:
+                prefix = opts.elasticsearch_index_prefix
+                es_aggregate_index = "{0}{1}".format(prefix, es_aggregate_index)
+                es_forensic_index = "{0}{1}".format(prefix, es_forensic_index)
+                es_smtp_tls_index = "{0}{1}".format(prefix, es_smtp_tls_index)
+            elastic_timeout_value = (
+                float(opts.elasticsearch_timeout)
+                if opts.elasticsearch_timeout is not None
+                else 60.0
+            )
+            elastic.set_hosts(
+                opts.elasticsearch_hosts,
+                use_ssl=opts.elasticsearch_ssl,
+                ssl_cert_path=opts.elasticsearch_ssl_cert_path,
+                username=opts.elasticsearch_username,
+                password=opts.elasticsearch_password,
+                api_key=opts.elasticsearch_api_key,
+                timeout=elastic_timeout_value,
+            )
+            elastic.migrate_indexes(
+                aggregate_indexes=[es_aggregate_index],
+                forensic_indexes=[es_forensic_index],
+            )
+            clients["elasticsearch"] = _ElasticsearchHandle()
+
+        if opts.opensearch_hosts:
+            os_aggregate_index = "dmarc_aggregate"
+            os_forensic_index = "dmarc_forensic"
+            os_smtp_tls_index = "smtp_tls"
+            if opts.opensearch_index_suffix:
+                suffix = opts.opensearch_index_suffix
+                os_aggregate_index = "{0}_{1}".format(os_aggregate_index, suffix)
+                os_forensic_index = "{0}_{1}".format(os_forensic_index, suffix)
+                os_smtp_tls_index = "{0}_{1}".format(os_smtp_tls_index, suffix)
+            if opts.opensearch_index_prefix:
+                prefix = opts.opensearch_index_prefix
+                os_aggregate_index = "{0}{1}".format(prefix, os_aggregate_index)
+                os_forensic_index = "{0}{1}".format(prefix, os_forensic_index)
+                os_smtp_tls_index = "{0}{1}".format(prefix, os_smtp_tls_index)
+            opensearch_timeout_value = (
+                float(opts.opensearch_timeout)
+                if opts.opensearch_timeout is not None
+                else 60.0
+            )
+            opensearch.set_hosts(
+                opts.opensearch_hosts,
+                use_ssl=opts.opensearch_ssl,
+                ssl_cert_path=opts.opensearch_ssl_cert_path,
+                username=opts.opensearch_username,
+                password=opts.opensearch_password,
+                api_key=opts.opensearch_api_key,
+                timeout=opensearch_timeout_value,
+                auth_type=opts.opensearch_auth_type,
+                aws_region=opts.opensearch_aws_region,
+                aws_service=opts.opensearch_aws_service,
+            )
+            opensearch.migrate_indexes(
+                aggregate_indexes=[os_aggregate_index],
+                forensic_indexes=[os_forensic_index],
+            )
+            clients["opensearch"] = _OpenSearchHandle()
+
     return clients
 
 
@@ -995,7 +1020,7 @@ def _main():
         if "policy_published" in report:
             domain = report["policy_published"]["domain"]
         elif "reported_domain" in report:
-            domain = report("reported_domain")
+            domain = report["reported_domain"]
         elif "policies" in report:
             domain = report["policies"][0]["domain"]
         if domain:
@@ -1578,7 +1603,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
@@ -1613,6 +1637,8 @@ def _main():
         except Exception as error:
             logger.warning("Unable to write to log file: {}".format(error))
 
+    opts.active_log_file = opts.log_file
+
     if (
         opts.imap_host is None
         and opts.graph_client_id is None
@@ -1771,8 +1797,9 @@ def _main():
         try:
             if opts.imap_user is None or opts.imap_password is None:
                 logger.error(
-                    "IMAP user and password must be specified ifhost is specified"
+                    "IMAP user and password must be specified if host is specified"
                 )
+                exit(1)
 
             ssl = True
             verify = True
@@ -1949,8 +1976,9 @@ def _main():
 
     def _handle_sighup(signum, frame):
         nonlocal _reload_requested
+        # Logging is not async-signal-safe; only set the flag here.
+        # The log message is emitted from the main loop when the flag is read.
         _reload_requested = True
-        logger.info("SIGHUP received, config will reload after current batch")
 
     if hasattr(signal, "SIGHUP"):
         signal.signal(signal.SIGHUP, _handle_sighup)
@@ -1959,7 +1987,12 @@ def _main():
         logger.info("Watching for email - Quit with ctrl-c")
 
         while True:
-            _reload_requested = False
+            # Re-check mailbox_watch in case a config reload disabled watch mode
+            if not opts.mailbox_watch:
+                logger.info(
+                    "Mailbox watch disabled in reloaded configuration, stopping watcher"
+                )
+                break
             try:
                 watch_inbox(
                     mailbox_connection=mailbox_connection,
@@ -1992,7 +2025,12 @@ def _main():
             if not _reload_requested:
                 break
 
-            # Reload configuration
+            # Reload configuration — emit the log message here (not in the
+            # signal handler, which is not async-signal-safe), then clear the
+            # flag so that any new SIGHUP arriving while we reload will be
+            # captured for the next iteration rather than being silently dropped.
+            logger.info("SIGHUP received, config will reload after current batch")
+            _reload_requested = False
             logger.info("Reloading configuration...")
             try:
                 # Build a fresh opts starting from CLI-only defaults so that
@@ -2036,6 +2074,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,12 @@ 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)
+            if should_reload and should_reload():
+                return
+            check_callback(self)
 
     @lru_cache(maxsize=10)
     def _find_folder_id_from_folder_path(self, folder_name: str) -> str:

parsedmarc/mail/imap.py

@@ -90,10 +90,17 @@ class IMAPConnection(MailboxConnection):
         # IDLE callback sends IMAPClient object,
         # send back the imap connection object instead
         def idle_callback_wrapper(client: IMAPClient):
+            old_client = self._client
             self._client = client
+            try:
+                old_client.logout()
+            except Exception:
+                pass
             check_callback(self)
 
         while True:
+            if should_reload and should_reload():
+                return
             try:
                 IMAPClient(
                     host=self._client.host,

parsedmarc/mail/maildir.py

@@ -65,6 +65,8 @@ class MaildirConnection(MailboxConnection):
 
     def watch(self, check_callback, check_timeout, should_reload=None):
         while True:
+            if should_reload and should_reload():
+                return
             try:
                 check_callback(self)
             except Exception as e:

parsedmarc/s3.py

@@ -93,3 +93,11 @@ class S3Client(object):
         self.bucket.put_object(
             Body=json.dumps(report), Key=object_path, Metadata=object_metadata
         )
+
+    def close(self):
+        """Clean up the boto3 resource."""
+        try:
+            if self.s3.meta is not None:
+                self.s3.meta.client.close()
+        except Exception:
+            pass

parsedmarc/splunk.py

@@ -207,3 +207,7 @@ class HECClient(object):
             raise SplunkError(e.__str__())
         if response["code"] != 0:
             raise SplunkError(response["text"])
+
+    def close(self):
+        """Close the underlying HTTP session."""
+        self.session.close()

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,10 @@ 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 +1995,10 @@ 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 +2071,10 @@ 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 +2146,10 @@ 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")