From 07b00a931bc2a8b4984e2c975b2736507baccaf4 Mon Sep 17 00:00:00 2001 From: Sean Whalen <44679+seanthegeek@users.noreply.github.com> Date: Fri, 28 Aug 2026 18:23:07 -0400 Subject: [PATCH] Warn on unmatched CLI paths; send Kafka failure/SMTP TLS reports per-report (#890) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes the two remaining bugs and the nits uncovered during the review session: - The CLI logs a warning for each file_path argument that matches no files, instead of silently succeeding with empty results — a typo'd path in a cron job previously 'worked' forever while processing nothing (_expand_file_path_args dropped it as a zero-match glob). Deliberately a warning, not an error: a glob legitimately matching nothing must not break existing workflows, and mailbox-only runs stay silent. - save_failure_reports_to_kafka and save_smtp_tls_reports_to_kafka now send one message per report, mirroring the aggregate saver's per-slice sends. Every released version documented per-record sends, but the code sent the whole list as one message, which a large batch — failure reports carry message samples — could push past Kafka's default 1MB message limit. Consumer-visible: these topics now carry individual report objects, not one JSON array per batch. - The CLI module docstring and argparse description now mention SMTP TLS reports; usage.md's CLI-help mirror regenerated. - The elastic/opensearch to_header display-name tests now assert the joined 'RT ' string reaches the saved document (autospec save), plus the exactly-once call, instead of only that save ran. Co-authored-by: Claude Fable 5 --- CHANGELOG.md | 2 ++ docs/source/usage.md | 2 +- parsedmarc/cli.py | 14 ++++++-- parsedmarc/kafkaclient.py | 64 +++++++++++++++++++++------------- tests/test_cli.py | 72 +++++++++++++++++++++++++++++++++++++++ tests/test_elastic.py | 11 ++++-- tests/test_kafkaclient.py | 47 ++++++++++++++++++------- tests/test_opensearch.py | 11 ++++-- 8 files changed, 180 insertions(+), 43 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 91c2e049..0d175d30 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,10 +6,12 @@ - **Breaking: the output and mailbox integrations are now optional extras, so `pip install parsedmarc` installs the parsing core and a working core CLI instead of every SDK** ([#883](https://github.com/domainaware/parsedmarc/issues/883)). A 10.x install pulled in the Elasticsearch, OpenSearch, Kafka, AWS (boto3), Azure, Gmail, and Microsoft Graph client libraries whether or not a deployment used any of them — roughly 1 GB of site-packages against roughly 300 MB without — which is a lot to ask of the motivating case, running parsedmarc as a parsing library on a small mail host. The base install now covers the parsing library plus the CLI reading from files, IMAP, Maildir, and mbox, and writing CSV/JSON, Splunk HEC, webhook, and syslog output; those outputs need only `httpx` and the standard library, so they deliberately have no extra of their own. Everything else moves behind an extra: `elastic` (Elasticsearch), `opensearch` (OpenSearch, including the boto3 SigV4 signer), `kafka`, `s3`, `gelf`, `loganalytics` (Azure Monitor), `msgraph` (Microsoft 365 mailboxes), and `gmail` (Gmail API mailboxes), joining the `postgresql` extra that already existed. **CLI users upgrading from 10.x should switch their upgrade command to `pip install -U "parsedmarc[all]"`** — the new umbrella extra — to keep every integration available; add `postgresql` to the list (`parsedmarc[all,postgresql]`) if the PostgreSQL backend is in use. `all` deliberately excludes `postgresql` because `psycopg`'s prebuilt binary wheels do not exist for every platform, and `pip install parsedmarc[all]` must not fail on a platform they do not cover. Users of the prebuilt Docker image (`ghcr.io/domainaware/parsedmarc`) see no change at all: the image now installs `[all,postgresql]`, so it still bundles every integration. A configuration section whose extra is missing no longer fails with an `ImportError` traceback at startup; it fails fast with a `ConfigurationError` naming both the section and the command that fixes it, e.g. `The [elasticsearch] configuration section requires the elastic extra: pip install parsedmarc[elastic]`. Finally, the never-imported `dateparser` dependency is dropped in favor of declaring `python-dateutil`, which `parsedmarc.utils` actually imports and which used to arrive only transitively through `dateparser`. - The CLI now accepts `--dns-timeout` as an alias of `--dns_timeout`, which is kept for backward compatibility (public since 6.0.0); `--dns-retries` already used the hyphenated form. +- **Failure and SMTP TLS reports are now sent to Kafka as one message per report, matching the aggregate saver's long-documented per-record behavior.** `save_failure_reports_to_kafka` and `save_smtp_tls_reports_to_kafka` documented per-record sends in every released version, but the code actually sent the entire report list as a single Kafka message (an unreleased docstring pass in [#888](https://github.com/domainaware/parsedmarc/pull/888) had briefly aligned the wording to the buggy code); a large batch could exceed Kafka's default 1MB message limit, and failure reports in particular carry message samples that make that more likely. Both savers now send/flush one message per report, mirroring `save_aggregate_reports_to_kafka`'s existing per-slice shape. This is consumer-visible: consumers now receive individual report objects on these topics rather than one JSON array per batch. ### Bug fixes - **`[elasticsearch]`/`[opensearch]` `number_of_replicas` is no longer ignored when `number_of_shards` is not also set** — the parser only read `number_of_replicas` inside the `number_of_shards` branch (accidental nesting dating to the 6.4.0-era code), while the documentation lists the two options independently and the client code accepts them independently. +- **The CLI now logs a warning for each file path argument that matches no files**, instead of silently succeeding with empty results. `_expand_file_path_args` glob-expands each `file_path` argument, so a non-existent plain path (e.g. a typo in a cron job) previously vanished as a zero-match glob with no message of any kind; a mailbox-only run that passes no file arguments still logs nothing. ## 10.5.0 diff --git a/docs/source/usage.md b/docs/source/usage.md index b43effd6..978dade1 100644 --- a/docs/source/usage.md +++ b/docs/source/usage.md @@ -11,7 +11,7 @@ usage: parsedmarc [-h] [-c CONFIG_FILE] [-r] [--strip-attachment-payloads] [-o O [-w] [--verbose] [--debug] [--log-file LOG_FILE] [--no-prettify-json] [-v] [file_path ...] -Parses DMARC reports +Parses DMARC and SMTP TLS reports positional arguments: file_path one or more paths to aggregate, failure, or SMTP TLS report files, emails, mbox files, or diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index f7f09978..5a670e7d 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""A CLI for parsing DMARC reports""" +"""A CLI for parsing DMARC and SMTP TLS reports""" import atexit import functools @@ -286,9 +286,17 @@ def _expand_file_path_args(paths: list[str], recursive: bool = False) -> list[st ``recursive`` also enables ``**`` to match any number of directories (including none) in glob patterns supplied directly as arguments, per the same stdlib glob semantics. + + A ``path`` argument that matches nothing (a plain path that does + not exist and is not a glob match, or a directory/glob pattern + with no matches) is logged as a WARNING naming that argument, + so a typo'd path in a cron job doesn't silently "succeed" while + processing nothing forever. An empty ``paths`` list (e.g. a + mailbox-only run with no file arguments) logs nothing. """ expanded: list[str] = [] for path in paths: + before = len(expanded) if os.path.isdir(path): pattern = os.path.join(glob_escape(path), "**" if recursive else "*") for match in sorted(glob(pattern, recursive=recursive)): @@ -303,6 +311,8 @@ def _expand_file_path_args(paths: list[str], recursive: bool = False) -> list[st expanded.append(path) else: expanded += glob(path, recursive=recursive) + if len(expanded) == before: + logger.warning("No files matched %s", path) return expanded @@ -2449,7 +2459,7 @@ def _main(): return output_errors - arg_parser = ArgumentParser(description="Parses DMARC reports") + arg_parser = ArgumentParser(description="Parses DMARC and SMTP TLS reports") arg_parser.add_argument( "-c", "--config-file", diff --git a/parsedmarc/kafkaclient.py b/parsedmarc/kafkaclient.py index 2338f506..94b0eb65 100644 --- a/parsedmarc/kafkaclient.py +++ b/parsedmarc/kafkaclient.py @@ -160,7 +160,15 @@ class KafkaClient(object): failure_topic: str, ): """ - Saves failure DMARC reports to Kafka as a single message + Saves failure DMARC reports to Kafka, one message per report, so a + large batch is far less likely to exceed Kafka's default 1MB + message limit + + Mirrors ``save_aggregate_reports_to_kafka``'s per-slice send/flush + shape (minus the aggregate-only ``strip_metadata``/date-range + logic): failure reports carry message samples, so a batch sent as + a single JSON array could easily exceed the broker's default max + message size. Args: failure_reports (list): A list of failure report dicts @@ -174,17 +182,18 @@ class KafkaClient(object): if len(failure_reports) < 1: return - try: - logger.debug("Saving failure reports to Kafka") - self.producer.send(failure_topic, failure_reports) - except UnknownTopicOrPartitionError: - raise KafkaError("Kafka error: Unknown topic or partition on broker") - except Exception as e: - raise KafkaError(f"Kafka error: {e.__str__()}") - try: - self.producer.flush() - except Exception as e: - raise KafkaError(f"Kafka error: {e.__str__()}") + for report in failure_reports: + try: + logger.debug("Saving failure report to Kafka") + self.producer.send(failure_topic, report) + except UnknownTopicOrPartitionError: + raise KafkaError("Kafka error: Unknown topic or partition on broker") + except Exception as e: + raise KafkaError(f"Kafka error: {e.__str__()}") + try: + self.producer.flush() + except Exception as e: + raise KafkaError(f"Kafka error: {e.__str__()}") # Backward-compatible alias save_forensic_reports_to_kafka = save_failure_reports_to_kafka @@ -195,7 +204,13 @@ class KafkaClient(object): smtp_tls_topic: str, ): """ - Saves SMTP TLS reports to Kafka as a single message + Saves SMTP TLS reports to Kafka, one message per report, so a + large batch is far less likely to exceed Kafka's default 1MB + message limit + + Mirrors ``save_aggregate_reports_to_kafka``'s per-slice send/flush + shape (minus the aggregate-only ``strip_metadata``/date-range + logic). Args: smtp_tls_reports (list): A list of SMTP TLS report dicts @@ -209,14 +224,15 @@ class KafkaClient(object): if len(smtp_tls_reports) < 1: return - try: - logger.debug("Saving SMTP TLS reports to Kafka") - self.producer.send(smtp_tls_topic, smtp_tls_reports) - except UnknownTopicOrPartitionError: - raise KafkaError("Kafka error: Unknown topic or partition on broker") - except Exception as e: - raise KafkaError(f"Kafka error: {e.__str__()}") - try: - self.producer.flush() - except Exception as e: - raise KafkaError(f"Kafka error: {e.__str__()}") + for report in smtp_tls_reports: + try: + logger.debug("Saving SMTP TLS report to Kafka") + self.producer.send(smtp_tls_topic, report) + except UnknownTopicOrPartitionError: + raise KafkaError("Kafka error: Unknown topic or partition on broker") + except Exception as e: + raise KafkaError(f"Kafka error: {e.__str__()}") + try: + self.producer.flush() + except Exception as e: + raise KafkaError(f"Kafka error: {e.__str__()}") diff --git a/tests/test_cli.py b/tests/test_cli.py index 37041edb..787aaebd 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -570,6 +570,78 @@ hosts = localhost [nested_xml], ) + def test_expand_file_path_args_warns_on_nonexistent_path(self): + """A file_path argument that matches no files at all -- a plain + path that doesn't exist and isn't a glob match -- logs a WARNING + naming that argument, instead of silently vanishing as a + zero-match glob. The argument still contributes no paths to the + returned list. + + Regression test: ``parsedmarc --offline -o out /nonexistent.xml`` + used to exit 0 with empty results and no message of any kind, so + a typo'd path in a cron job "worked" forever while processing + nothing. + """ + from parsedmarc.cli import _expand_file_path_args + + with tempfile.TemporaryDirectory() as d: + missing = os.path.join(d, "nonexistent.xml") + with self.assertLogs("parsedmarc.log", level="WARNING") as cm: + result = _expand_file_path_args([missing]) + self.assertEqual(result, []) + self.assertTrue( + any(missing in line for line in cm.output), + f"expected a warning naming {missing}, got: {cm.output}", + ) + + def test_expand_file_path_args_no_warning_for_matching_path(self): + """A file_path argument that matches at least one file logs no + warning at all.""" + from parsedmarc.cli import _expand_file_path_args + + with tempfile.TemporaryDirectory() as d: + report = os.path.join(d, "report.xml") + with open(report, "w") as f: + f.write("x") + + with self.assertNoLogs("parsedmarc.log", level="WARNING"): + result = _expand_file_path_args([report]) + self.assertEqual(result, [report]) + + wildcard = os.path.join(d, "*.xml") + with self.assertNoLogs("parsedmarc.log", level="WARNING"): + result = _expand_file_path_args([wildcard]) + self.assertEqual(result, [report]) + + def test_expand_file_path_args_empty_list_is_silent(self): + """An empty ``paths`` list (e.g. a mailbox-only run with no file + arguments) logs nothing.""" + from parsedmarc.cli import _expand_file_path_args + + with self.assertNoLogs("parsedmarc.log", level="WARNING"): + result = _expand_file_path_args([]) + self.assertEqual(result, []) + + def test_expand_file_path_args_mixed_warns_only_for_unmatched(self): + """When some arguments match and others don't, only the + unmatched argument produces a warning.""" + from parsedmarc.cli import _expand_file_path_args + + with tempfile.TemporaryDirectory() as d: + report = os.path.join(d, "report.xml") + with open(report, "w") as f: + f.write("x") + missing = os.path.join(d, "nonexistent.xml") + + with self.assertLogs("parsedmarc.log", level="WARNING") as cm: + result = _expand_file_path_args([report, missing]) + + self.assertEqual(result, [report]) + warning_lines = [line for line in cm.output if "WARNING" in line] + self.assertEqual(len(warning_lines), 1) + self.assertIn(missing, warning_lines[0]) + self.assertNotIn(report, warning_lines[0]) + def test_apply_env_overrides_injects_values(self): """Env vars are injected into an existing ConfigParser.""" from configparser import ConfigParser diff --git a/tests/test_elastic.py b/tests/test_elastic.py index ba2ffa68..a6da2a5f 100644 --- a/tests/test_elastic.py +++ b/tests/test_elastic.py @@ -1200,16 +1200,23 @@ class TestSaveFailureReport(unittest.TestCase): def test_to_header_with_non_empty_display_joins_with_brackets(self): """The other branch: non-empty display joins display+addr - with " <" and appends ">", e.g. 'RT '.""" + with " <" and appends ">", e.g. 'RT '. + Asserts the joined string actually reaches the saved document + (see test_reply_to_header_flattened_and_indexed for the + autospec-save pattern), not merely that save ran.""" report = _failure_report() report["parsed_sample"]["headers"]["To"] = [["RT", "rcpt@example.com"]] with ( patch("parsedmarc.elastic.Search", return_value=_empty_search()), patch("parsedmarc.elastic.Index"), - patch.object(elastic_module._FailureReportDoc, "save") as mock_save, + patch.object( + elastic_module._FailureReportDoc, "save", autospec=True + ) as mock_save, ): save_failure_report_to_elasticsearch(report) mock_save.assert_called_once() + doc = mock_save.call_args.args[0] + self.assertEqual(doc.sample.headers["to"], "RT ") def test_sample_address_lists_indexed_for_reply_to_cc_bcc_attachments(self): """A failure report sample can carry reply_to / cc / bcc / diff --git a/tests/test_kafkaclient.py b/tests/test_kafkaclient.py index 5385b9d4..c1bc8504 100644 --- a/tests/test_kafkaclient.py +++ b/tests/test_kafkaclient.py @@ -186,22 +186,33 @@ class TestSaveFailureReportsToKafka(unittest.TestCase): with patch("parsedmarc.kafkaclient.KafkaProducer"): return KafkaClient(kafka_hosts=["b:9092"]) - def test_sends_full_list_in_one_message(self): - """Failure reports are sent as one Kafka message carrying the - whole list — unlike aggregate records, which are sent as - individual slices to stay under Kafka's default 1MB message - cap.""" + def test_sends_one_message_per_report(self): + """Failure reports are sent as one Kafka message per report, + mirroring the aggregate saver's per-slice sends, so a large batch + is far less likely to exceed Kafka's default 1MB message limit + — failure + reports carry message samples, so a whole-list send is + particularly likely to blow past that cap.""" client = self._client() reports = [{"id": "f1"}, {"id": "f2"}] client.save_failure_reports_to_kafka(reports, "dmarc-failure") - _producer(client).send.assert_called_once_with("dmarc-failure", reports) + producer = _producer(client) + self.assertEqual(producer.send.call_count, 2) + sent = [call.args for call in producer.send.call_args_list] + self.assertEqual( + sent, [("dmarc-failure", reports[0]), ("dmarc-failure", reports[1])] + ) + self.assertEqual(producer.flush.call_count, 2) def test_dict_input_normalized_to_list(self): + """A single-report dict is wrapped to a one-element list + internally, then sent as that one unwrapped report (not the + list itself).""" client = self._client() client.save_failure_reports_to_kafka({"id": "single"}, "topic") - # The send payload is wrapped to a single-element list. + self.assertEqual(_producer(client).send.call_count, 1) args = _producer(client).send.call_args.args - self.assertEqual(args[1], [{"id": "single"}]) + self.assertEqual(args[1], {"id": "single"}) def test_empty_list_is_a_noop(self): client = self._client() @@ -232,17 +243,29 @@ class TestSaveSmtpTlsReportsToKafka(unittest.TestCase): with patch("parsedmarc.kafkaclient.KafkaProducer"): return KafkaClient(kafka_hosts=["b:9092"]) - def test_sends_full_list_in_one_message(self): + def test_sends_one_message_per_report(self): + """SMTP TLS reports are sent as one Kafka message per report, + mirroring the aggregate saver's per-slice sends, so a large batch + is far less likely to exceed Kafka's default 1MB message + limit.""" client = self._client() - reports = [{"organization_name": "x"}] + reports = [{"organization_name": "x"}, {"organization_name": "y"}] client.save_smtp_tls_reports_to_kafka(reports, "smtp-tls") - _producer(client).send.assert_called_once_with("smtp-tls", reports) + producer = _producer(client) + self.assertEqual(producer.send.call_count, 2) + sent = [call.args for call in producer.send.call_args_list] + self.assertEqual(sent, [("smtp-tls", reports[0]), ("smtp-tls", reports[1])]) + self.assertEqual(producer.flush.call_count, 2) def test_dict_input_normalized_to_list(self): + """A single-report dict is wrapped to a one-element list + internally, then sent as that one unwrapped report (not the + list itself).""" client = self._client() client.save_smtp_tls_reports_to_kafka({"organization_name": "x"}, "topic") + self.assertEqual(_producer(client).send.call_count, 1) args = _producer(client).send.call_args.args - self.assertEqual(args[1], [{"organization_name": "x"}]) + self.assertEqual(args[1], {"organization_name": "x"}) def test_empty_list_is_a_noop(self): client = self._client() diff --git a/tests/test_opensearch.py b/tests/test_opensearch.py index 281d7ae8..41218f4b 100644 --- a/tests/test_opensearch.py +++ b/tests/test_opensearch.py @@ -1193,16 +1193,23 @@ class TestSaveFailureReport(unittest.TestCase): def test_to_header_with_non_empty_display_joins_with_brackets(self): """The other branch: non-empty display joins display+addr - with " <" and appends ">", e.g. 'RT '.""" + with " <" and appends ">", e.g. 'RT '. + Asserts the joined string actually reaches the saved document + (see test_reply_to_header_flattened_and_indexed for the + autospec-save pattern), not merely that save ran.""" report = _failure_report() report["parsed_sample"]["headers"]["To"] = [["RT", "rcpt@example.com"]] with ( patch("parsedmarc.opensearch.Search", return_value=_empty_search()), patch("parsedmarc.opensearch.Index"), - patch.object(opensearch_module._FailureReportDoc, "save") as mock_save, + patch.object( + opensearch_module._FailureReportDoc, "save", autospec=True + ) as mock_save, ): save_failure_report_to_opensearch(report) mock_save.assert_called_once() + doc = mock_save.call_args.args[0] + self.assertEqual(doc.sample.headers["to"], "RT ") def test_sample_address_lists_indexed_for_reply_to_cc_bcc_attachments(self): """A failure report sample can carry reply_to / cc / bcc /