From 2f89507e5dcf8384f1a6eed8bca19a601a973807 Mon Sep 17 00:00:00 2001 From: liangzaihuanyv <144539158+liangzaihuanyv@users.noreply.github.com> Date: Sat, 31 Oct 2026 10:46:52 +0800 Subject: [PATCH] Fix aggregate report deduplication across policy domains (#934) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Fix aggregate report deduplication across policy domains Scope aggregate-report identity to the reporting organization, Report-ID, and policy domain, matching RFC 9990 section 3.5.1. Encode the fields as a JSON array and compare ASCII domain letters without changing returned report data. Preserve accepted empty-domain reports and save-confirmed cache semantics. Add real MIME/mbox and dry-run Maildir regressions for distinct domains, same-domain repeats, shared configurations, boundary collisions, parallel parsing, and false/raising save callbacks. The implementation and tests were prepared with AI assistance, including a separate review of the scoped changes and test evidence. * Share the domain-scoped dedup key with the CLI direct-file path Responds to Copilot review on #934: the CLI's direct-file path still wrote `{org_name}_{report_id}` keys into the same seen-aggregate-report cache that the mbox/mailbox classifier now reads with a JSON [org, report_id, policy_domain] key, so a report passed both as a file and inside an mbox in one run was saved twice, and direct files from one reporter sharing a Report-ID for different domains (RFC 9990 §3.5.1) were still collapsed. Extract the key into `_aggregate_report_dedup_key()` and use it from both paths; the direct-file path now writes through `parser_config.seen_aggregate_report_ids` like the mbox path. Adds end-to-end `_main()` regression tests for mixed file+mbox input and for domain-scoped direct-file dedup. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Sean Whalen <44679+seanthegeek@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 --- CHANGELOG.md | 2 + parsedmarc/__init__.py | 46 +++- parsedmarc/cli.py | 8 +- parsedmarc/config.py | 5 +- tests/test_cli.py | 132 +++++++++++ tests/test_init.py | 492 ++++++++++++++++++++++++++++++++++++++++- 6 files changed, 670 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 530ac992..9b5be7d2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,8 @@ - **Failure reports are now saved to S3.** `save_report_to_s3()` read `report["report_metadata"]` for every report type except SMTP TLS, but failure reports have no `report_metadata` (see `FailureReport` in `parsedmarc/types.py`), so every failure report raised `KeyError` since S3 output was added in 7.0.0. The error was counted as an S3 output error, so the batch holding the report was held back. A failure report is now saved under `/failure/year=YYYY/month=MM/day=DD/_.json`, dated by its UTC arrival date, with `reported_domain` and `arrival_date` metadata. Its hash covers the report type, the arrival date, reported domain, original mail from, original recipient, original envelope ID, source IP address, and the message sample. +- **Keep distinct aggregate reports when one reporter reuses a Report-ID for different policy domains.** RFC 9990 §3.5.1 scopes Report-ID uniqueness to the same domain. Direct report files, mbox files, and mailbox ingestion now all include the policy domain in one shared dedup key, using a JSON array to avoid ambiguous field separators. ASCII domain case differences still count as the same domain (RFC 4343 §3), without changing the returned report. Same-domain repeats remain suppressed — including a report passed both as a file and inside an mbox in one CLI run — and mailbox keys are still committed only after a successful save callback. + ## 11.1.0 ### Security diff --git a/parsedmarc/__init__.py b/parsedmarc/__init__.py index 85251238..b5c31a49 100644 --- a/parsedmarc/__init__.py +++ b/parsedmarc/__init__.py @@ -2551,6 +2551,35 @@ def parse_report_file( return results +_ASCII_UPPER_TO_LOWER = str.maketrans( + "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz" +) + + +def _aggregate_report_dedup_key(report: AggregateReport) -> str: + """Return the ``seen_aggregate_report_ids`` cache key for ``report``. + + RFC 9990 §3.5.1 defines a Report-ID as unique only "among reports to the + same domain", so the key is the reporter, the Report-ID, and the policy + domain. ASCII domain letters are compared without case per RFC 4343 §3, + without changing the domain in the report itself. A JSON array encodes + the three fields without the ambiguity of a delimiter that a field could + contain. + + Every writer of the shared cache must build its key here -- the mbox and + mailbox classifier and the CLI's direct-file path -- or a report read + through one path is not recognized as a duplicate of the same report + read through another. + """ + report_org = report["report_metadata"]["org_name"] + report_id = report["report_metadata"]["report_id"] + report_domain = report["policy_published"]["domain"] + # The existing XML parser returns None for an empty . + if isinstance(report_domain, str): + report_domain = report_domain.translate(_ASCII_UPPER_TO_LOWER) + return json.dumps((report_org, report_id, report_domain)) + + def _classify_parsed_email( parsed_email: ParsedReport, aggregate_reports: list[AggregateReport], @@ -2564,12 +2593,15 @@ def _classify_parsed_email( Owns the seen-aggregate-report-ID dedup check against ``seen_aggregate_report_ids``: an aggregate report already seen (keyed - on ``{org_name}_{report_id}``) is logged and dropped instead of - appended. Shared, unmodified, by the sequential and parallel branches - of ``get_dmarc_reports_from_mbox`` and ``get_dmarc_reports_from_mailbox`` - so both dedup identically -- callers pass the config's - ``seen_aggregate_report_ids`` cache so dedup state stays scoped to - whichever ``ParserConfig`` (explicit or module-default) is in effect. + on the reporter, the Report-ID, and the reported policy domain) is + logged and dropped instead of appended. Shared, unmodified, by the + sequential and parallel branches of ``get_dmarc_reports_from_mbox`` and + ``get_dmarc_reports_from_mailbox`` so both dedup identically -- callers + pass the config's ``seen_aggregate_report_ids`` cache so dedup state + stays scoped to whichever ``ParserConfig`` (explicit or module-default) + is in effect. The key comes from :func:`_aggregate_report_dedup_key`, + which the CLI's direct-file path also uses, so a report seen in a file + and again in an mbox in the same run is recognized as a duplicate. ``pending_aggregate_keys`` lets a caller *defer* the dedup-cache write: when a set is supplied, a newly seen key is staged in that set instead @@ -2590,7 +2622,7 @@ def _classify_parsed_email( if parsed_email["report_type"] == "aggregate": report_org = parsed_email["report"]["report_metadata"]["org_name"] report_id = parsed_email["report"]["report_metadata"]["report_id"] - report_key = f"{report_org}_{report_id}" + report_key = _aggregate_report_dedup_key(parsed_email["report"]) already_seen = report_key in seen_aggregate_report_ids or ( pending_aggregate_keys is not None and report_key in pending_aggregate_keys ) diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index 8a4de398..5ca01d89 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -31,6 +31,7 @@ from parsedmarc import ( ParserConfig, ParserError, __version__, + _aggregate_report_dedup_key, email_results, email_results_via_msgraph, get_dmarc_reports_from_mailbox, @@ -3249,9 +3250,10 @@ def _main(): if result["report_type"] == "aggregate": report_org = result["report"]["report_metadata"]["org_name"] report_id = result["report"]["report_metadata"]["report_id"] - report_key = f"{report_org}_{report_id}" - if report_key not in SEEN_AGGREGATE_REPORT_IDS: - SEEN_AGGREGATE_REPORT_IDS[report_key] = True + report_key = _aggregate_report_dedup_key(result["report"]) + seen_ids = parser_config.seen_aggregate_report_ids + if report_key not in seen_ids: + seen_ids[report_key] = True aggregate_reports.append(result["report"]) else: logger.debug( diff --git a/parsedmarc/config.py b/parsedmarc/config.py index 0a171f2f..110419f3 100644 --- a/parsedmarc/config.py +++ b/parsedmarc/config.py @@ -121,8 +121,9 @@ class ParserConfig: normalize_timespan_threshold_hours: Aggregate reports whose timespan exceeds this many hours are normalized/split. ip_address_cache: Cache of IP address enrichment results. - seen_aggregate_report_ids: Cache of already-seen aggregate report - IDs, used for de-duplication. + seen_aggregate_report_ids: Cache of already-seen aggregate reports, + keyed on reporter, Report-ID, and policy domain, used for + de-duplication. reverse_dns_map: The reverse DNS map used to classify base domains found via reverse DNS. """ diff --git a/tests/test_cli.py b/tests/test_cli.py index 6fb792f2..b6a4af52 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -4,9 +4,11 @@ env-var overrides, mailbox watch wiring, and SIGHUP reload.""" import importlib import importlib.abc import importlib.machinery +import gzip import io import json import logging +import mailbox import os import shutil import signal @@ -18,6 +20,8 @@ import zipfile from collections.abc import Iterator, Sequence from configparser import ConfigParser from contextlib import contextmanager +from email.mime.application import MIMEApplication +from email.mime.multipart import MIMEMultipart from tempfile import NamedTemporaryFile from types import ModuleType, SimpleNamespace from typing import cast @@ -1201,6 +1205,134 @@ class TestDirectoryFilePaths(unittest.TestCase): self.assertEqual(report_ids, {top_level_id, nested_id}) +class TestMixedInputAggregateDedup(unittest.TestCase): + """The direct-file and mbox paths of one ``_main()`` run share the + ``SEEN_AGGREGATE_REPORT_IDS`` cache, so they must key it identically. + + If the direct-file path in ``_main()`` wrote ``{org_name}_{report_id}`` + while ``get_dmarc_reports_from_mbox`` looked up a JSON + ``[org_name, report_id, policy_domain]`` key, a report passed both as a + file and inside an mbox in the same run would be saved twice. RFC 9990 + §3.5.1 scopes Report-ID uniqueness to "reports to the same domain", so + the shared key also keeps same-ID reports for different domains apart in + the direct-file path. Runs the real ``_main()`` with real parsing and + asserts on the written ``aggregate.json``. + """ + + REPORT_ID = "1721054318-shared.example@reporter.example" + + def setUp(self): + parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear() + self.addCleanup(parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear) + env_patcher = patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}, clear=False) + env_patcher.start() + self.addCleanup(env_patcher.stop) + self._tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self._tmp, ignore_errors=True) + + def _xml(self, policy_domain: str, record_count: int) -> bytes: + return ( + '\n' + "" + "reporter.example" + "dmarc@reporter.example" + f"{self.REPORT_ID}" + "16800000001680086400" + "" + f"{policy_domain}

none

" + "" + "192.0.2.1" + f"{record_count}" + "nonepass" + "pass" + f"{policy_domain}" + f"{policy_domain}" + "pass" + ).encode() + + def _xml_file(self, name: str, policy_domain: str, record_count: int) -> str: + path = os.path.join(self._tmp, name) + with open(path, "wb") as f: + f.write(self._xml(policy_domain, record_count)) + return path + + def _mbox_file(self, name: str, policy_domain: str, record_count: int) -> str: + message = MIMEMultipart() + message["From"] = "dmarc@reporter.example" + message["To"] = "postmaster@example.net" + message["Subject"] = f"Report domain: {policy_domain}" + message["Date"] = "Mon, 1 Jan 2024 00:00:00 +0000" + attachment = MIMEApplication( + gzip.compress(self._xml(policy_domain, record_count)), "gzip" + ) + attachment.add_header( + "Content-Disposition", "attachment", filename="report.xml.gz" + ) + message.attach(attachment) + path = os.path.join(self._tmp, name) + box = mailbox.mbox(path) + try: + box.add(mailbox.mboxMessage(message.as_bytes())) + box.flush() + finally: + box.close() + return path + + def _run(self, *paths: str) -> list[tuple[str, int]]: + output_dir = os.path.join(self._tmp, "output") + cfg_path = os.path.join(self._tmp, "parsedmarc.ini") + with open(cfg_path, "w") as f: + f.write( + f"[general]\noffline = True\nsilent = True\noutput = {output_dir}\n" + ) + with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, *paths]): + parsedmarc.cli._main() + with open(os.path.join(output_dir, "aggregate.json")) as f: + reports = json.load(f) + return sorted( + (report["policy_published"]["domain"], report["records"][0]["count"]) + for report in reports + ) + + def test_same_report_in_file_and_mbox_is_saved_once(self): + """A report given as a file and again inside an mbox is one report.""" + xml_path = self._xml_file("report.xml", "a.example", 5) + mbox_path = self._mbox_file("reports.mbox", "a.example", 7) + + kept = self._run(xml_path, mbox_path) + + # Direct files are parsed before mbox files, so the file's copy wins. + self.assertEqual(kept, [("a.example", 5)]) + + def test_same_id_for_other_domain_in_file_and_mbox_is_kept(self): + """The shared key keeps the domain, so a same-ID report for another + domain in the mbox is not mistaken for the file's report.""" + xml_path = self._xml_file("report.xml", "a.example", 5) + mbox_path = self._mbox_file("reports.mbox", "b.example", 7) + + kept = self._run(xml_path, mbox_path) + + self.assertEqual(kept, [("a.example", 5), ("b.example", 7)]) + + def test_direct_files_with_same_id_for_different_domains_are_kept(self): + """Direct-file dedup is domain-scoped too: two files from one + reporter sharing a Report-ID for different domains are both kept, + while an ASCII-case variant of the same domain is still a duplicate. + """ + a_path = self._xml_file("a.xml", "a.example", 5) + b_path = self._xml_file("b.xml", "b.example", 7) + a_upper_path = self._xml_file("a-upper.xml", "A.EXAMPLE", 9) + + kept = self._run(a_path, b_path, a_upper_path) + + # file_paths is deduplicated through a set, so which case variant of + # a.example survives is unordered; compare domains case-folded. + # tests/test_init.py covers that the kept report's case is unchanged. + domains = sorted(domain.lower() for domain, _ in kept) + self.assertEqual(domains, ["a.example", "b.example"]) + self.assertIn(("b.example", 7), kept) + + class TestArchiveDirectory(unittest.TestCase): """End-to-end coverage of issue #570: ``[general] archive_directory`` moves successfully processed local report files into a dated diff --git a/tests/test_init.py b/tests/test_init.py index be03d740..eec7f60a 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -21,6 +21,7 @@ import time import unittest from collections.abc import Callable from datetime import datetime, timedelta, timezone +from email.mime.application import MIMEApplication from email.mime.multipart import MIMEMultipart from email.mime.nonmultipart import MIMENonMultipart from email.mime.text import MIMEText @@ -4106,9 +4107,7 @@ class TestCentralizedConfig(unittest.TestCase): self.assertEqual(len(first["aggregate_reports"]), 1) self.assertEqual(len(second["aggregate_reports"]), 0) - report_metadata = first["aggregate_reports"][0]["report_metadata"] - report_key = f"{report_metadata['org_name']}_{report_metadata['report_id']}" - self.assertIn(report_key, parsedmarc.SEEN_AGGREGATE_REPORT_IDS) + self.assertEqual(len(parsedmarc.SEEN_AGGREGATE_REPORT_IDS), 1) def test_kwargs_and_config_equivalence(self): """parse_report_file must produce identical results whether called @@ -6320,5 +6319,492 @@ class TestSaveOutput(unittest.TestCase): ) +DOMAIN_SCOPE_REPORTER = "reporter.example" +DOMAIN_SCOPE_REPORT_ID = "1721054318-shared.example@reporter.example" + + +def _domain_scope_aggregate_xml( + policy_domain: str | None, + record_count: int, + *, + reporter: str = DOMAIN_SCOPE_REPORTER, + report_id: str = DOMAIN_SCOPE_REPORT_ID, +) -> bytes: + """Builds one well-formed aggregate report document. + + ``policy_domain`` fills ``policy_published/domain`` -- the field RFC 9990 + scopes Report-ID uniqueness to -- and ``record_count`` fills the single + ``record``'s ``count``, so a caller can tell which document a batch kept. + ``reporter`` and ``report_id`` default to one fixed reporter/ID so that a + caller varying only the domain is varying only the identity component + under test. ``None`` emits an empty policy-domain element, which the + existing parser accepts as ``None``; record domains remain populated. + All domain names use the reserved ``.example`` TLD. + """ + record_domain = policy_domain or "a.example" + return ( + '\n' + "\n" + " \n" + f" {reporter}\n" + " dmarc@reporter.example\n" + f" {report_id}\n" + " \n" + " 1680000000\n" + " 1680086400\n" + " \n" + " \n" + " \n" + f" {policy_domain or ''}\n" + "

none

\n" + " \n" + " \n" + " \n" + " 192.0.2.1\n" + f" {record_count}\n" + " \n" + " none\n" + " pass\n" + " pass\n" + " \n" + " \n" + f" {record_domain}\n" + " \n" + f" {record_domain}pass\n" + " \n" + " \n" + "" + ).encode() + + +def _domain_scope_email( + policy_domain: str | None, + record_count: int, + *, + reporter: str = DOMAIN_SCOPE_REPORTER, + report_id: str = DOMAIN_SCOPE_REPORT_ID, +) -> bytes: + """Wraps that report in a minimal, well-formed MIME aggregate-report email.""" + message = MIMEMultipart() + message["From"] = "dmarc@reporter.example" + message["To"] = "postmaster@example.net" + message["Subject"] = f"Report domain: {policy_domain}" + message["Date"] = "Mon, 1 Jan 2024 00:00:00 +0000" + attachment = MIMEApplication( + gzip.compress( + _domain_scope_aggregate_xml( + policy_domain, + record_count, + reporter=reporter, + report_id=report_id, + ) + ), + "gzip", + ) + attachment.add_header("Content-Disposition", "attachment", filename="report.xml.gz") + message.attach(attachment) + return message.as_bytes() + + +class TestAggregateReportIdentityIsDomainScoped(unittest.TestCase): + """Aggregate-report dedup must follow RFC 9990 §3.5.1's identity scope. + + RFC 9990 §3.5.1 ("Definition of Report-ID") says: "This identifier MUST + be unique among reports to the same domain to aid receivers in + identifying duplicate reports should they happen." Uniqueness is + therefore per (reporting domain), not global: one reporter legitimately + reusing the same Report-ID for ``a.example`` and ``b.example`` is + sending two distinct, non-duplicate reports. + + These tests drive the public consumer -- ``get_dmarc_reports_from_mbox`` + (real temporary mbox, ``n_procs=1`` and ``n_procs=2``) and the companion + class's ``get_dmarc_reports_from_mailbox`` (real on-disk Maildir) -- and assert on + the returned reports' own ``policy_published.domain`` and record + ``count`` values, never on internal helpers or mock call counts. Configs + are isolated except where a test explicitly shares one across calls. + """ + + def setUp(self): + self._tmp = mkdtemp() + self.addCleanup(rmtree, self._tmp, ignore_errors=True) + + def _mbox(self, messages: list[bytes]) -> str: + with NamedTemporaryFile(dir=self._tmp, suffix=".mbox", delete=False) as tmp: + path = tmp.name + box = mailbox.mbox(path) + try: + box.lock() + try: + for message in messages: + box.add(mailbox.mboxMessage(message)) + box.flush() + finally: + box.unlock() + finally: + box.close() + return path + + def _parse_mbox(self, path: str, *, n_procs: int = 1): + return parsedmarc.get_dmarc_reports_from_mbox( + path, + n_procs=n_procs, + config=parsedmarc.ParserConfig(offline=True, always_use_local_files=True), + ) + + @staticmethod + def _domains(result) -> list[str | None]: + return [ + report["policy_published"]["domain"] + for report in result["aggregate_reports"] + ] + + @staticmethod + def _counts(result) -> list[int]: + return [ + sum(record["count"] for record in report["records"]) + for report in result["aggregate_reports"] + ] + + def test_same_report_id_for_different_domains_keeps_both_reports(self): + """One reporter, one Report-ID, two policy domains in one batch: + both reports survive, with their own content intact. + + Fails on a dedup key that is only ``org_name + report_id``: the + second report is then classified as a duplicate of the first and + silently dropped, even though RFC 9990 §3.5.1 only forbids reusing + an ID *within* one domain. + """ + path = self._mbox( + [ + _domain_scope_email("a.example", 3), + _domain_scope_email("b.example", 5), + ] + ) + + result = self._parse_mbox(path) + + self.assertEqual(self._domains(result), ["a.example", "b.example"]) + self.assertEqual(self._counts(result), [3, 5]) + + def test_different_domains_are_kept_in_either_delivery_order(self): + """Delivery order must not decide which of the two domains survives: + with the same batch reversed, the roles swap and both are still + returned, in message order.""" + path = self._mbox( + [ + _domain_scope_email("b.example", 5), + _domain_scope_email("a.example", 3), + ] + ) + + result = self._parse_mbox(path) + + self.assertEqual(self._domains(result), ["b.example", "a.example"]) + self.assertEqual(self._counts(result), [5, 3]) + + def test_same_domain_duplicate_report_is_still_suppressed(self): + """The negative half of the contract: the same reporter re-sending + the same Report-ID for the *same* domain is a genuine duplicate and + is still dropped, leaving one report.""" + path = self._mbox( + [ + _domain_scope_email("a.example", 3), + _domain_scope_email("a.example", 3), + ] + ) + + result = self._parse_mbox(path) + + self.assertEqual(self._domains(result), ["a.example"]) + self.assertEqual(self._counts(result), [3]) + + def test_dedup_is_per_domain_within_one_batch(self): + """Three messages, one Report-ID: a.example, a.example, b.example. + + Exactly one report per domain is kept (the first of each, so the + content assertion pins *which* document survived rather than just + how many), which is what per-domain identity means: duplicates are + suppressed inside a domain and never across domains. + """ + path = self._mbox( + [ + _domain_scope_email("a.example", 1), + _domain_scope_email("a.example", 2), + _domain_scope_email("b.example", 3), + ] + ) + + result = self._parse_mbox(path) + + self.assertEqual(self._domains(result), ["a.example", "b.example"]) + self.assertEqual(self._counts(result), [1, 3]) + + def test_domain_ascii_case_does_not_disable_duplicate_suppression(self): + """RFC 4343 §3: ASCII domain letters compare without case. + + Keep the original spelling in the returned report while suppressing + a repeated report for the same domain with a different spelling. + """ + path = self._mbox( + [ + _domain_scope_email("A.Example", 3), + _domain_scope_email("a.example", 5), + _domain_scope_email("b.example", 7), + ] + ) + result = self._parse_mbox(path) + self.assertEqual(self._domains(result), ["A.Example", "b.example"]) + self.assertEqual(self._counts(result), [3, 7]) + + def test_empty_policy_domain_keeps_existing_parser_behavior(self): + """The XML parser accepts an empty policy domain as None. + + Adding the domain to report identity must not make classification + crash on this previously accepted input, or merge it with a named + domain. Repeated empty-domain reports still count as duplicates. + """ + path = self._mbox( + [ + _domain_scope_email(None, 3), + _domain_scope_email(None, 5), + _domain_scope_email("b.example", 7), + ] + ) + result = self._parse_mbox(path) + self.assertEqual(self._domains(result), [None, "b.example"]) + self.assertEqual(self._counts(result), [3, 7]) + + def test_parallel_mbox_keeps_both_domains_and_suppresses_duplicates(self): + """Parallel parsing leaves domain-scoped dedup in the parent process.""" + path = self._mbox( + [ + _domain_scope_email("a.example", 3), + _domain_scope_email("a.example", 5), + _domain_scope_email("b.example", 7), + ] + ) + result = self._parse_mbox(path, n_procs=2) + self.assertEqual(self._domains(result), ["a.example", "b.example"]) + self.assertEqual(self._counts(result), [3, 7]) + + def test_one_config_keeps_each_domain_and_still_suppresses_repeats(self): + """Across two calls sharing one ``ParserConfig`` -- i.e. one live dedup + cache, the way a long-running watch loop uses it -- the domain is + still part of the identity. + + Run one parses ``a.example``; run two carries a *different domain* + with the same reporter and Report-ID and must be retained, then a + repeat of that same domain must be suppressed. A key without the + domain fails the middle assertion: run two's report is dropped as a + duplicate of run one's. + """ + config = parsedmarc.ParserConfig(offline=True, always_use_local_files=True) + + def parse(messages): + return parsedmarc.get_dmarc_reports_from_mbox( + self._mbox(messages), n_procs=1, config=config + ) + + first = parse([_domain_scope_email("a.example", 3)]) + second = parse([_domain_scope_email("b.example", 5)]) + third = parse([_domain_scope_email("b.example", 5)]) + + self.assertEqual(self._domains(first), ["a.example"]) + self.assertEqual(self._domains(second), ["b.example"]) + self.assertEqual(self._counts(second), [5]) + self.assertEqual(self._domains(third), []) + + def test_key_encoding_does_not_collide_on_field_boundaries(self): + """Two different reports must not share one dedup key just because a + reporter's name and a Report-ID contain the same characters. + + ``("a_b", "c")`` and ``("a", "b_c")`` both flatten to ``a_b_c`` under + a key built by joining the fields with underscores; a JSON + array of the fields cannot be ambiguous. Both messages below report + the same policy domain, so the only thing telling them apart is the + reporter/ID split: a colliding key drops the second one. + """ + messages = [ + _domain_scope_email("a.example", 1, reporter="a_b", report_id="c"), + _domain_scope_email("a.example", 2, reporter="a", report_id="b_c"), + ] + + result = self._parse_mbox(self._mbox(messages)) + + self.assertEqual(len(result["aggregate_reports"]), 2) + self.assertEqual(self._counts(result), [1, 2]) + self.assertEqual( + [ + report["report_metadata"]["org_name"] + for report in result["aggregate_reports"] + ], + ["a_b", "a"], + ) + + +class TestAggregateReportIdentityDomainScopePendingSave(unittest.TestCase): + """The same domain-scoped identity contract on the mailbox path that + defers its dedup-cache writes until the batch is known to be saved. + + ``get_dmarc_reports_from_mailbox`` stages new aggregate keys in a + per-batch pending set and folds them into the config's cache only after + the caller's ``save_callback`` reports success (see + ``_classify_parsed_email``'s ``pending_aggregate_keys`` and + ``get_dmarc_reports_from_mailbox``). The pending set has to make exactly + the same domain-scoped decision as the immediate-write path; otherwise a + fix to one path leaves the other silently dropping reports. + + Real on-disk Maildir through mailsuite's ``MaildirConnection``, using + the public ``save_callback`` and dry-run ``test=True`` options. These + tests cover classification and cache commit/retry decisions, not + message marking, archive/delete, or failed-save retry counters. + """ + + def setUp(self): + self._tmp = mkdtemp() + self.addCleanup(rmtree, self._tmp, ignore_errors=True) + self._maildir = os.path.join(self._tmp, "Maildir") + self._inbox = mailbox.Maildir(self._maildir, create=True) + self.addCleanup(self._inbox.close) + self._config = parsedmarc.ParserConfig( + offline=True, always_use_local_files=True + ) + + def _deliver(self, policy_domain: str, record_count: int) -> None: + self._inbox.add( + mailbox.MaildirMessage(_domain_scope_email(policy_domain, record_count)) + ) + self._inbox.flush() + + def _run(self, save_callback=None): + connection = MaildirConnection(self._maildir, maildir_create=True) + # Dry-run leaves messages in place for the next call, while still + # classifying reports, invoking save_callback, and folding pending + # keys into the cache only after a successful save. + return parsedmarc.get_dmarc_reports_from_mailbox( + connection=connection, + config=self._config, + save_callback=save_callback, + test=True, + ) + + @staticmethod + def _domains(result) -> list[str]: + return [ + report["policy_published"]["domain"] + for report in result["aggregate_reports"] + ] + + def _count_by_domain(self, result) -> dict[str, int]: + """Maps each returned report's policy domain to its total record count. + + Order-free on purpose: a Maildir's message iteration order is not + part of any contract, so these tests pin *which* domains came back + and what they contain, not the sequence they arrived in. + """ + reports = result["aggregate_reports"] + counts = { + report["policy_published"]["domain"]: sum( + record["count"] for record in report["records"] + ) + for report in reports + } + self.assertEqual(len(reports), len(counts), "duplicate domain report retained") + return counts + + @staticmethod + def _saved(batch) -> bool: + del batch + return True + + @staticmethod + def _not_saved(batch) -> bool: + del batch + return False + + def test_committed_batch_keeps_reports_for_both_domains(self): + """Two domains, one Report-ID, batch saved: both reports are returned.""" + self._deliver("a.example", 3) + self._deliver("b.example", 5) + + result = self._run(save_callback=self._saved) + + self.assertEqual( + self._count_by_domain(result), {"a.example": 3, "b.example": 5} + ) + + def test_committed_batch_still_suppresses_same_domain_duplicate(self): + """Same domain twice, batch saved: still a duplicate, so one report + is returned -- the pending path must not weaken same-domain dedup + either.""" + self._deliver("a.example", 3) + self._deliver("a.example", 3) + + result = self._run(save_callback=self._saved) + + self.assertEqual(len(result["aggregate_reports"]), 1) + self.assertEqual(self._count_by_domain(result), {"a.example": 3}) + + def test_failed_save_drops_pending_keys_so_the_retry_reparses(self): + """A failed save must not commit the pending keys: the retry has to + reparse the same message rather than skip it as a duplicate.""" + self._deliver("a.example", 3) + + first = self._run(save_callback=self._not_saved) + retry = self._run(save_callback=self._saved) + + self.assertEqual(self._count_by_domain(first), {"a.example": 3}) + self.assertEqual(self._count_by_domain(retry), {"a.example": 3}) + + def test_successful_save_then_rerun_returns_nothing(self): + """After a batch is saved, the same two messages are deduplicated on + the next run: the pending keys were folded into the cache, so the + committed batch is not reprocessed. Both domains are recorded, so + neither one is left out of the commit.""" + self._deliver("a.example", 3) + self._deliver("b.example", 5) + + committed = self._run(save_callback=self._saved) + rerun = self._run(save_callback=self._saved) + + self.assertEqual( + self._count_by_domain(committed), {"a.example": 3, "b.example": 5} + ) + self.assertEqual(self._domains(rerun), []) + + def test_false_callback_retry_still_returns_both_domains(self): + """A callback returning ``False`` must not leave the two domains + poisoned against each other: the retry reparses both messages and + returns both reports, rather than suppressing one as a duplicate of + the other.""" + self._deliver("a.example", 3) + self._deliver("b.example", 5) + + unsaved = self._run(save_callback=self._not_saved) + retry = self._run(save_callback=self._saved) + + expected = {"a.example": 3, "b.example": 5} + self.assertEqual(self._count_by_domain(unsaved), expected) + self.assertEqual(self._count_by_domain(retry), expected) + + def test_raising_callback_retry_still_returns_both_domains(self): + """A callback that *raises* fails the save the same way: the + exception propagates to the caller, and the retry must still return + both domains.""" + self._deliver("a.example", 3) + self._deliver("b.example", 5) + + def boom(batch): + del batch + raise RuntimeError("output destination down") + + with self.assertRaises(RuntimeError): + self._run(save_callback=boom) + + retry = self._run(save_callback=self._saved) + + self.assertEqual(self._count_by_domain(retry), {"a.example": 3, "b.example": 5}) + + if __name__ == "__main__": unittest.main(verbosity=2)