mirror of
https://github.com/domainaware/parsedmarc.git
synced 2026-09-05 13:38:00 +00:00
Sean WhalenandClaude Fable 5
07b00a931b
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 <rcpt@example.com>' string reaches the saved document (autospec save), plus the exactly-once call, instead of only that save ran. Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
312 lines
14 KiB
Python
312 lines
14 KiB
Python
"""Tests for parsedmarc.kafkaclient"""
|
|
|
|
import json
|
|
import unittest
|
|
from typing import cast
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from kafka.errors import UnknownTopicOrPartitionError
|
|
|
|
from parsedmarc.kafkaclient import KafkaClient, KafkaError, _BootstrapError
|
|
|
|
|
|
def _producer(client: KafkaClient) -> MagicMock:
|
|
"""The patched KafkaProducer as a MagicMock, for assertion access."""
|
|
return cast(MagicMock, client.producer)
|
|
|
|
|
|
def _aggregate_report():
|
|
return {
|
|
"report_metadata": {
|
|
"org_name": "TestOrg",
|
|
"org_email": "test@example.com",
|
|
"report_id": "r-123",
|
|
"begin_date": "2024-01-01 00:00:00",
|
|
"end_date": "2024-01-02 00:00:00",
|
|
},
|
|
"policy_published": {"domain": "example.com", "p": "none"},
|
|
"records": [
|
|
{"source": {"ip_address": "192.0.2.1"}, "count": 1},
|
|
{"source": {"ip_address": "192.0.2.2"}, "count": 2},
|
|
],
|
|
}
|
|
|
|
|
|
class TestKafkaClientInit(unittest.TestCase):
|
|
"""KafkaProducer config wiring: SSL, SASL, plain — each branch has
|
|
user-facing security consequences if it's wrong."""
|
|
|
|
def test_init_plain_no_ssl(self):
|
|
"""No SSL, no auth: just bootstrap_servers and serializer."""
|
|
with patch("parsedmarc.kafkaclient.KafkaProducer") as mock_producer:
|
|
KafkaClient(kafka_hosts=["broker:9092"])
|
|
kwargs = mock_producer.call_args.kwargs
|
|
self.assertEqual(kwargs["bootstrap_servers"], ["broker:9092"])
|
|
self.assertNotIn("security_protocol", kwargs)
|
|
self.assertNotIn("sasl_plain_username", kwargs)
|
|
|
|
def test_init_ssl_enables_ssl_security_protocol(self):
|
|
with (
|
|
patch("parsedmarc.kafkaclient.KafkaProducer") as mock_producer,
|
|
patch("parsedmarc.kafkaclient.create_default_context") as mock_ctx,
|
|
):
|
|
KafkaClient(kafka_hosts=["broker:9093"], ssl=True)
|
|
kwargs = mock_producer.call_args.kwargs
|
|
self.assertEqual(kwargs["security_protocol"], "SSL")
|
|
self.assertIs(kwargs["ssl_context"], mock_ctx.return_value)
|
|
|
|
def test_init_username_implies_ssl(self):
|
|
"""Doc says ssl=True is implied when username/password supplied."""
|
|
with (
|
|
patch("parsedmarc.kafkaclient.KafkaProducer") as mock_producer,
|
|
patch("parsedmarc.kafkaclient.create_default_context"),
|
|
):
|
|
KafkaClient(kafka_hosts=["broker:9093"], username="user", password="pass")
|
|
kwargs = mock_producer.call_args.kwargs
|
|
self.assertEqual(kwargs["security_protocol"], "SSL")
|
|
self.assertEqual(kwargs["sasl_plain_username"], "user")
|
|
self.assertEqual(kwargs["sasl_plain_password"], "pass")
|
|
|
|
def test_init_uses_provided_ssl_context(self):
|
|
"""A caller-supplied SSLContext takes precedence over the
|
|
default context — this lets ops pin to a private CA."""
|
|
custom_ctx = MagicMock()
|
|
with (
|
|
patch("parsedmarc.kafkaclient.KafkaProducer") as mock_producer,
|
|
patch("parsedmarc.kafkaclient.create_default_context") as mock_default,
|
|
):
|
|
KafkaClient(kafka_hosts=["b:9093"], ssl=True, ssl_context=custom_ctx)
|
|
self.assertIs(mock_producer.call_args.kwargs["ssl_context"], custom_ctx)
|
|
mock_default.assert_not_called()
|
|
|
|
def test_init_value_serializer_emits_utf8_json(self):
|
|
"""The value_serializer turns Python objects into UTF-8 JSON
|
|
bytes. A regression here would corrupt every event sent."""
|
|
with patch("parsedmarc.kafkaclient.KafkaProducer") as mock_producer:
|
|
KafkaClient(kafka_hosts=["b"])
|
|
serializer = mock_producer.call_args.kwargs["value_serializer"]
|
|
result = serializer({"hello": "world", "n": 1})
|
|
self.assertEqual(json.loads(result.decode("utf-8")), {"hello": "world", "n": 1})
|
|
|
|
def test_init_no_brokers_available_raises_kafka_error(self):
|
|
with patch(
|
|
"parsedmarc.kafkaclient.KafkaProducer",
|
|
side_effect=_BootstrapError(),
|
|
):
|
|
with self.assertRaises(KafkaError) as ctx:
|
|
KafkaClient(kafka_hosts=["unreachable:9092"])
|
|
self.assertIn("No Kafka brokers", str(ctx.exception))
|
|
|
|
|
|
class TestKafkaClientHelpers(unittest.TestCase):
|
|
"""Static helpers used by save_aggregate."""
|
|
|
|
def test_strip_metadata_lifts_keys_to_root_and_drops_metadata(self):
|
|
report = _aggregate_report()
|
|
result = KafkaClient.strip_metadata(report)
|
|
self.assertEqual(result["org_name"], "TestOrg")
|
|
self.assertEqual(result["org_email"], "test@example.com")
|
|
self.assertEqual(result["report_id"], "r-123")
|
|
self.assertNotIn("report_metadata", result)
|
|
|
|
def test_generate_date_range_iso_format(self):
|
|
report = _aggregate_report()
|
|
date_range = KafkaClient.generate_date_range(report)
|
|
self.assertEqual(date_range, ["2024-01-01T00:00:00", "2024-01-02T00:00:00"])
|
|
|
|
|
|
class TestSaveAggregateReportsToKafka(unittest.TestCase):
|
|
"""save_aggregate sends one Kafka message per record (slice), with
|
|
the metadata + policy duplicated onto each slice for Kibana parity."""
|
|
|
|
def _client(self):
|
|
with patch("parsedmarc.kafkaclient.KafkaProducer"):
|
|
return KafkaClient(kafka_hosts=["b:9092"])
|
|
|
|
def test_sends_one_message_per_record(self):
|
|
client = self._client()
|
|
client.save_aggregate_reports_to_kafka(_aggregate_report(), "dmarc-aggregate")
|
|
# 2 records in the sample report → 2 producer.send calls.
|
|
self.assertEqual(_producer(client).send.call_count, 2)
|
|
# Topic is forwarded verbatim.
|
|
for call in _producer(client).send.call_args_list:
|
|
self.assertEqual(call.args[0], "dmarc-aggregate")
|
|
|
|
def test_each_slice_carries_metadata(self):
|
|
client = self._client()
|
|
client.save_aggregate_reports_to_kafka(_aggregate_report(), "topic")
|
|
sent = [call.args[1] for call in _producer(client).send.call_args_list]
|
|
for slice_ in sent:
|
|
self.assertEqual(slice_["org_name"], "TestOrg")
|
|
self.assertEqual(slice_["org_email"], "test@example.com")
|
|
self.assertEqual(slice_["report_id"], "r-123")
|
|
self.assertEqual(
|
|
slice_["date_range"], ["2024-01-01T00:00:00", "2024-01-02T00:00:00"]
|
|
)
|
|
self.assertEqual(
|
|
slice_["policy_published"], {"domain": "example.com", "p": "none"}
|
|
)
|
|
|
|
def test_empty_list_is_a_noop(self):
|
|
client = self._client()
|
|
client.save_aggregate_reports_to_kafka([], "topic")
|
|
_producer(client).send.assert_not_called()
|
|
|
|
def test_dict_input_normalized_to_list(self):
|
|
"""Single-report dict input is wrapped to a list."""
|
|
client = self._client()
|
|
client.save_aggregate_reports_to_kafka(_aggregate_report(), "topic")
|
|
# 2 records still sent (one report with 2 records, not multiple reports).
|
|
self.assertEqual(_producer(client).send.call_count, 2)
|
|
|
|
def test_unknown_topic_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).send.side_effect = UnknownTopicOrPartitionError()
|
|
with self.assertRaises(KafkaError) as ctx:
|
|
client.save_aggregate_reports_to_kafka(_aggregate_report(), "missing")
|
|
self.assertIn("Unknown topic or partition", str(ctx.exception))
|
|
|
|
def test_generic_send_exception_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).send.side_effect = RuntimeError("transport failure")
|
|
with self.assertRaises(KafkaError) as ctx:
|
|
client.save_aggregate_reports_to_kafka(_aggregate_report(), "topic")
|
|
self.assertIn("transport failure", str(ctx.exception))
|
|
|
|
def test_flush_exception_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).flush.side_effect = RuntimeError("flush failure")
|
|
with self.assertRaises(KafkaError) as ctx:
|
|
client.save_aggregate_reports_to_kafka(_aggregate_report(), "topic")
|
|
self.assertIn("flush failure", str(ctx.exception))
|
|
|
|
|
|
class TestSaveFailureReportsToKafka(unittest.TestCase):
|
|
def _client(self):
|
|
with patch("parsedmarc.kafkaclient.KafkaProducer"):
|
|
return KafkaClient(kafka_hosts=["b:9092"])
|
|
|
|
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 = _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")
|
|
self.assertEqual(_producer(client).send.call_count, 1)
|
|
args = _producer(client).send.call_args.args
|
|
self.assertEqual(args[1], {"id": "single"})
|
|
|
|
def test_empty_list_is_a_noop(self):
|
|
client = self._client()
|
|
client.save_failure_reports_to_kafka([], "topic")
|
|
_producer(client).send.assert_not_called()
|
|
|
|
def test_unknown_topic_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).send.side_effect = UnknownTopicOrPartitionError()
|
|
with self.assertRaises(KafkaError):
|
|
client.save_failure_reports_to_kafka([{"a": 1}], "missing")
|
|
|
|
def test_generic_send_error_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).send.side_effect = OSError("net")
|
|
with self.assertRaises(KafkaError):
|
|
client.save_failure_reports_to_kafka([{"a": 1}], "topic")
|
|
|
|
def test_flush_error_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).flush.side_effect = OSError("flush")
|
|
with self.assertRaises(KafkaError):
|
|
client.save_failure_reports_to_kafka([{"a": 1}], "topic")
|
|
|
|
|
|
class TestSaveSmtpTlsReportsToKafka(unittest.TestCase):
|
|
def _client(self):
|
|
with patch("parsedmarc.kafkaclient.KafkaProducer"):
|
|
return KafkaClient(kafka_hosts=["b:9092"])
|
|
|
|
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"}, {"organization_name": "y"}]
|
|
client.save_smtp_tls_reports_to_kafka(reports, "smtp-tls")
|
|
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"})
|
|
|
|
def test_empty_list_is_a_noop(self):
|
|
client = self._client()
|
|
client.save_smtp_tls_reports_to_kafka([], "topic")
|
|
_producer(client).send.assert_not_called()
|
|
|
|
def test_unknown_topic_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).send.side_effect = UnknownTopicOrPartitionError()
|
|
with self.assertRaises(KafkaError):
|
|
client.save_smtp_tls_reports_to_kafka([{"a": 1}], "missing")
|
|
|
|
def test_generic_send_error_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).send.side_effect = RuntimeError("oops")
|
|
with self.assertRaises(KafkaError):
|
|
client.save_smtp_tls_reports_to_kafka([{"a": 1}], "topic")
|
|
|
|
def test_flush_error_translates_to_kafka_error(self):
|
|
client = self._client()
|
|
_producer(client).flush.side_effect = RuntimeError("flush")
|
|
with self.assertRaises(KafkaError):
|
|
client.save_smtp_tls_reports_to_kafka([{"a": 1}], "topic")
|
|
|
|
|
|
class TestKafkaClientClose(unittest.TestCase):
|
|
def test_close_calls_underlying_producer_close(self):
|
|
with patch("parsedmarc.kafkaclient.KafkaProducer"):
|
|
client = KafkaClient(kafka_hosts=["b"])
|
|
client.close()
|
|
_producer(client).close.assert_called_once()
|
|
|
|
|
|
class TestKafkaBackwardCompatAlias(unittest.TestCase):
|
|
def test_forensic_alias_points_to_failure_method(self):
|
|
self.assertIs(
|
|
KafkaClient.save_forensic_reports_to_kafka,
|
|
KafkaClient.save_failure_reports_to_kafka,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|