diff --git a/.gitignore b/.gitignore index c64b01d0..cc494a0d 100644 --- a/.gitignore +++ b/.gitignore @@ -152,3 +152,7 @@ parsedmarc/resources/maps/domain_info.tsv coverage.json junit.xml dashboard-screenshots/ + +# Claude Code agent worktrees — untracked repo snapshots; ignoring them +# keeps repo-root ruff runs and git status clean +.claude/worktrees/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 840d27bd..06fb6063 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ - **`n_procs` parallel parsing now covers messages from mbox files and mailbox connections** ([#147](https://github.com/domainaware/parsedmarc/issues/147)), not just report files passed directly as CLI arguments: IMAP, Microsoft Graph, Gmail API, and Maildir connections, including watch mode. `get_dmarc_reports_from_mbox()`, `get_dmarc_reports_from_mailbox()`, and `watch_inbox()` all gained an `n_procs` keyword argument. Parallel workers are now a reused process pool for the whole run, with a bounded submission window that keeps at most roughly `2 * n_procs` messages in flight at a time so memory stays bounded even for huge mboxes; the main process sends periodic IMAP keepalives while workers parse. - **Added `ParserConfig`, a single frozen dataclass carrying every parsing/enrichment option** ([#503](https://github.com/domainaware/parsedmarc/issues/503)): offline mode, IP database path, reverse DNS map and PSL overrides paths/URLs, DNS nameservers/timeout/retries, `strip_attachment_payloads`, `normalize_timespan_threshold_hours`, and the three caches the parser and enrichment code share across calls (IP address info, seen aggregate report IDs, and the reverse DNS map). `parse_aggregate_report_xml()`, `parse_aggregate_report_file()`, `parse_failure_report()`, `parse_report_email()`, `parse_report_file()`, `get_dmarc_reports_from_mbox()`, `get_dmarc_reports_from_mailbox()`, and `watch_inbox()` all accept a keyword-only `config=` argument; when it's provided, the individual option keyword arguments are ignored in favor of the config's values, and every existing per-option keyword argument continues to work unchanged when `config=` is omitted. Every explicitly constructed `ParserConfig` owns fresh, isolated caches; omitting `config=` falls back to the existing process-wide shared default caches, unchanged and identity-preserved (`parsedmarc.IP_ADDRESS_CACHE`, `parsedmarc.SEEN_AGGREGATE_REPORT_IDS`, `parsedmarc.REVERSE_DNS_MAP`). Caches never cross multiprocessing worker boundaries: `ParserConfig.__getstate__` drops the three cache fields when a config is pickled for a worker, and each worker accumulates its own from that point on. `keep_alive` and `n_procs` remain separate keyword arguments — they control process/worker orchestration, not parsing or enrichment behavior, so they're deliberately not `ParserConfig` fields. See the new "Using parsedmarc as a library" section of the usage docs for an example. - **Added a "Domain policy" dropdown filter to the Splunk aggregate DMARC dashboard** ([#854](https://github.com/domainaware/parsedmarc/issues/854)): filters every panel on the published DMARC policy (`published_policy.p`) or subdomain policy (`published_policy.sp`), making it easy to see which domains with a `none` policy are ready to move to `quarantine` or `reject`. It sits immediately before the "Message disposition" dropdown and offers the same choices (`any`/`none`/`quarantine`/`reject`, defaulting to `any`). +- **New `[general]` option `archive_directory`**: move successfully processed local report files into a dated archive tree (`////`); files that fail to parse as a report go to `/Invalid/`, while files that fail for other reasons (e.g. transient I/O errors) are left in place so a later run can retry them; existing files are never overwritten (numeric suffix); files already under the archive directory are excluded from later runs. Applies only to report files passed directly as local path arguments ([#570](https://github.com/domainaware/parsedmarc/issues/570)). ### Bug fixes diff --git a/docs/source/usage.md b/docs/source/usage.md index 5100e0ae..9270d25a 100644 --- a/docs/source/usage.md +++ b/docs/source/usage.md @@ -132,6 +132,24 @@ The full set of configuration options are: payloads from results - `silent` - bool: Set this to `False` to output results to STDOUT - `output` - str: Directory to place JSON and CSV files in. This is required if you set either of the JSON output file options. + - `archive_directory` - str: Optional. When set, successfully + processed report files given as local file/directory path + arguments are moved into + `////` + (year and month come from the report's own begin/arrival date, with + the month zero-padded). A successfully parsed report whose archive + date can't be determined is left in place with a logged warning. + Files that fail to parse as a report are moved to + `/Invalid/`; files that fail for other reasons, + such as transient I/O errors, are left in place so a later run can + retry them. An existing destination file is never overwritten; a + numeric suffix is appended before the extension (e.g. + `report-1.xml`). This applies only to direct local file input — + reports fetched from mailboxes (IMAP, Microsoft Graph, Gmail API, + Maildir) use `[mailbox] archive_folder` instead, and mbox files are + never moved. Files already inside `archive_directory` are excluded + from processing, so the archive may safely live inside an input + directory. A failed move is logged and does not stop the run. - `aggregate_json_filename` - str: filename for the aggregate JSON output file - `failure_json_filename` - str: filename for the failure diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index 47ddabd6..e68dffaf 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -9,6 +9,7 @@ import http.client import json import logging import os +import shutil import signal import sys import time @@ -58,12 +59,13 @@ from parsedmarc.mail import ( MSGraphConnection, ) from parsedmarc.parallel import _parse_report_file_job, parallel_map -from parsedmarc.types import ParsingResults +from parsedmarc.types import ParsedReport, ParsingResults from parsedmarc.utils import ( InvalidIPinfoAPIKey, configure_ipinfo_api, get_base_domain, get_reverse_dns, + human_timestamp_to_datetime, is_mbox, load_ip_db, load_psl_overrides, @@ -225,6 +227,192 @@ def _expand_file_path_args(paths: list[str], recursive: bool = False) -> list[st return expanded +def _exclude_archived_paths(file_paths: list[str], archive_directory: str) -> list[str]: + """Filter *file_paths* down to paths that are not already inside + *archive_directory*. + + The archive directory may live inside an input directory (e.g. + ``/archive``), so without this filter a file already moved + into the archive on a previous run would be picked up again by a + later ``file_path`` directory expansion, re-parsed, and re-archived + (colliding with itself and accumulating numeric suffixes forever). + + Paths are resolved with ``os.path.realpath`` (not just + ``os.path.abspath``) so a symlinked spelling of either the archive + directory or an input path still matches: e.g. ``archive_directory`` + configured via a ``/data`` symlink while the input directory is + passed as the real ``/mnt/...`` path would otherwise never compare + equal, and every run would re-archive the same files with a new + numeric suffix forever. + """ + archive_root = os.path.normcase(os.path.realpath(archive_directory)) + kept: list[str] = [] + for path in file_paths: + abs_path = os.path.normcase(os.path.realpath(path)) + try: + inside_archive = ( + os.path.commonpath([archive_root, abs_path]) == archive_root + ) + except ValueError: + # Paths are on different drives (Windows) or otherwise not + # comparable, so the path can't be inside the archive. + inside_archive = False + if inside_archive: + logger.debug(f"Excluding already-archived file {path}") + continue + kept.append(path) + return kept + + +def _archive_subdir_for_result(result: ParsedReport) -> str | None: + """Return the ``//`` subdirectory a parsed + report's source file should be archived under, or ``None`` when the + report type is unrecognized or its date can't be determined. + + The date comes from the parsed report itself, not the source + filename or file mtime: aggregate reports use + ``report_metadata.begin_date``, failure reports use + ``arrival_date_utc``, and SMTP TLS reports use ``begin_date``. + """ + report_type = result["report_type"] + # Only the wall-clock year/month fields are read from the parsed + # datetime, so no timezone conversion ever happens here — but tag + # the strings whose zone is known, per human_timestamp_to_datetime's + # contract. Aggregate begin_date is a local-time string + # (timestamp_to_human uses datetime.fromtimestamp) and must stay + # naive; arrival_date_utc is UTC wall-clock; SMTP TLS begin_date is + # RFC 3339 with an offset, so assume_utc would be a no-op anyway. + assume_utc = False + try: + if result["report_type"] == "aggregate": + type_folder = "Aggregate" + date_string = result["report"]["report_metadata"]["begin_date"] + elif result["report_type"] == "failure": + type_folder = "Failure" + date_string = result["report"]["arrival_date_utc"] + assume_utc = True + elif result["report_type"] == "smtp_tls": + type_folder = "SMTP-TLS" + date_string = result["report"]["begin_date"] + else: + logger.warning(f"Cannot archive unknown report type: {report_type}") + return None + dt = human_timestamp_to_datetime(date_string, assume_utc=assume_utc) + except (KeyError, TypeError, ValueError, OverflowError) as e: + logger.warning(f"Cannot determine archive date for {report_type} report: {e}") + return None + + return os.path.join(f"{dt.year:04d}", f"{dt.month:02d}", type_folder) + + +def _move_file_to_archive(file_path: str, dest_dir: str) -> str: + """Move *file_path* into *dest_dir*, creating it if needed, and return + the final destination path. + + An existing file at the destination is never overwritten: a numeric + suffix is appended before the extension (``name-1.xml``, + ``name-2.xml``, ...) until a free name is found. For multi-suffix + names like ``report.xml.gz`` the numeric suffix lands before the + last suffix only (``report.xml-1.gz``); this is acceptable. + + The free-name claim is atomic (``os.open`` with + ``O_CREAT | O_EXCL``) rather than an exists-check-then-move: a plain + ``os.path.exists()`` check followed by ``shutil.move()`` is a + TOCTOU race between concurrent ``parsedmarc`` invocations sharing an + archive directory, and ``shutil.move()`` silently overwrites an + existing destination on POSIX, which would violate the + never-overwrite guarantee. Instead, each candidate name is staked + out with a zero-byte placeholder file before the real move happens; + ``shutil.move()`` then replaces that placeholder with the real file + — atomically via ``os.rename`` when source and destination are on + the same POSIX filesystem, otherwise (Windows, or a cross-device + move) via a ``copy2``-and-overwrite that is not atomic but still + cannot collide with a concurrent invocation, since the placeholder + already claimed the name. + """ + os.makedirs(dest_dir, exist_ok=True) + base, ext = os.path.splitext(os.path.basename(file_path)) + candidate = os.path.basename(file_path) + n = 1 + while True: + dest_path = os.path.join(dest_dir, candidate) + try: + # 0o600 (not os.open's 0o777 default) so a placeholder that + # outlives a failed move+cleanup is never executable or + # group/other-accessible. The mode never reaches the real + # archived file: os.rename replaces the placeholder's inode + # outright, and the copy2 fallback's copystat overwrites the + # mode with the source file's. + fd = os.open(dest_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + candidate = f"{base}-{n}{ext}" + n += 1 + continue + os.close(fd) + break + + try: + shutil.move(file_path, dest_path) + except Exception: + try: + os.remove(dest_path) + except OSError: + # Best-effort cleanup of the just-created placeholder; the + # move failure re-raised below is the error that matters. + pass + raise + return dest_path + + +def _archive_processed_file( + file_path: str, archive_directory: str, result: ParsedReport | Exception +) -> None: + """Move *file_path* into *archive_directory* after processing. + + Files that failed to parse as a report (*result* is a + ``ParserError`` — every parse-failure exception, including + ``InvalidSMTPTLSReport``, subclasses it) go to + ``/Invalid/``. Files that failed for some other + reason (a transient ``OSError``/``PermissionError`` from the parse + job's broad catch, or an unexpected parser bug) are left in place so + a later run can retry them — renaming a valid-but-currently-unreadable + report into ``Invalid/`` would permanently sideline it, since moving + a file needs no read permission on its contents, and + ``_exclude_archived_paths`` would then hide it from every future run + too. The parse loop already logged the error either way. + + Successfully parsed files go to the dated ``//`` + subdirectory returned by ``_archive_subdir_for_result``; if that + returns ``None`` (unknown report type or unparseable date), the file + is left in place — a warning was already logged by that helper. + + A move failure is logged and never allowed to abort the run: the + file has already been successfully parsed (or definitively failed + to parse), so a filesystem error while archiving it should not + cause the caller to lose that work. + """ + if isinstance(result, ParserError): + subdir = "Invalid" + elif isinstance(result, Exception): + logger.debug( + f"Leaving {file_path} in place: {result.__class__.__name__} is not " + "a report-parsing failure, so it may be retryable" + ) + return + else: + subdir = _archive_subdir_for_result(result) + if subdir is None: + return + + dest_dir = os.path.join(archive_directory, subdir) + try: + dest_path = _move_file_to_archive(file_path, dest_dir) + except Exception as e: + logger.error(f"Error moving {file_path} to the archive: {e}") + return + logger.debug(f"Archived {file_path} to {dest_path}") + + # All known INI config section names, used for env var resolution. _KNOWN_SECTIONS = frozenset( { @@ -483,6 +671,8 @@ def _parse_config(config: ConfigParser, opts): ) if "output" in general_config: opts.output = _expand_path(general_config["output"]) + if "archive_directory" in general_config: + opts.archive_directory = _expand_path(general_config["archive_directory"]) if "aggregate_json_filename" in general_config: opts.aggregate_json_filename = general_config["aggregate_json_filename"] if "failure_json_filename" in general_config: @@ -2131,6 +2321,7 @@ def _main(): maildir_create=False, log_file=args.log_file, n_procs=1, + archive_directory=None, ip_db_path=None, ipinfo_url=None, ipinfo_api_token=None, @@ -2307,6 +2498,8 @@ def _main(): signal.signal(signal.SIGINT, _handle_sigint) file_paths = _expand_file_path_args(args.file_path, recursive=args.recursive) + if opts.archive_directory: + file_paths = _exclude_archived_paths(file_paths, opts.archive_directory) mbox_paths = [] for file_path in file_paths: @@ -2354,6 +2547,8 @@ def _main(): failure_reports.append(result["report"]) elif result["report_type"] == "smtp_tls": smtp_tls_reports.append(result["report"]) + if opts.archive_directory: + _archive_processed_file(file_path, opts.archive_directory, result) if pbar is not None: pbar.close() diff --git a/tests/test_cli.py b/tests/test_cli.py index de067b76..6a126a5f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -6,6 +6,7 @@ import json import logging import os import signal +import stat import sys import tempfile import unittest @@ -26,7 +27,7 @@ import parsedmarc import parsedmarc.cli import parsedmarc.elastic import parsedmarc.opensearch as opensearch_module -from parsedmarc.types import AggregateReport +from parsedmarc.types import AggregateReport, ParsedReport SAMPLE_AGGREGATE_REPORT_PATH = ( "samples/aggregate/!example.com!1538204542!1538463818.xml" @@ -1086,6 +1087,321 @@ class TestDirectoryFilePaths(unittest.TestCase): self.assertEqual(report_ids, {top_level_id, nested_id}) +class TestArchiveDirectory(unittest.TestCase): + """End-to-end coverage of issue #570: ``[general] archive_directory`` + moves successfully processed local report files into a dated + ``///`` tree, and unparseable + files into ``/Invalid/``. Runs the real ``_main()`` + entry point via ``patch.object(sys, "argv", ...)`` against on-disk + sample reports, per AGENTS.md's mock-at-SDK-boundary rule (there is no + external SDK boundary in this code path to mock).""" + + AGGREGATE_SAMPLE_1 = "samples/aggregate/!example.com!1538204542!1538463818.xml" + AGGREGATE_SAMPLE_2 = ( + "samples/aggregate/!large-example.com!1711897200!1711983600.xml" + ) + # Not dmarc_ruf_report_linkedin.eml: that sample begins with a + # "From dmarc-noreply@linkedin.com ..." mbox envelope line, which is + # why Python's mailbox module reads it as a (single-message) mbox + # file and is_mbox() classifies it as an mbox — routing it through + # _main()'s mbox_paths branch instead of the direct-file archiving + # path this test exercises (mbox files are intentionally never + # archived — see the archive_directory docs in docs/source/usage.md). + # The sharepoint sample below starts with a MIME header instead, so + # it isn't mbox-classified. + FAILURE_SAMPLE = ( + "samples/failure/DMARC Failure Report for domain.de " + "(mail-from=sharepoint@domain.de, ip=10.10.10.10).eml" + ) + SMTP_TLS_SAMPLE = "samples/smtp_tls/rfc8460.json" + + def setUp(self): + # SEEN_AGGREGATE_REPORT_IDS is a module-level ExpiringDict that + # dedupes report IDs across parses within one process; clear it so + # a report "seen" by an earlier test isn't silently dropped here, + # and so tests within this class don't interfere with each other. + # Precedent: TestDirectoryFilePaths.setUp above. + parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear() + self.addCleanup(parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear) + self._env_patcher = patch.dict( + os.environ, {"GITHUB_ACTIONS": "true"}, clear=False + ) + self._env_patcher.start() + self.addCleanup(self._env_patcher.stop) + + def _copy_sample(self, src_path, dest_dir, dest_basename=None): + dest_basename = dest_basename or os.path.basename(src_path) + dest_path = os.path.join(dest_dir, dest_basename) + with open(src_path, "rb") as src, open(dest_path, "wb") as dst: + dst.write(src.read()) + return dest_path + + def _expected_subdir(self, sample_path): + """Parse *sample_path* the same way the CLI does and return the + ``//`` subdirectory ``_archive_subdir_for_result`` + computes for it. Computed dynamically rather than hardcoded because + aggregate ``begin_date`` is local-time, so month buckets are + timezone-dependent.""" + result = parsedmarc.parse_report_file(sample_path, offline=True) + subdir = parsedmarc.cli._archive_subdir_for_result(result) + assert subdir is not None + return subdir + + def _write_config( + self, tmp_dir, output_dirname, archive_dirname=None, n_procs=None + ): + cfg_path = os.path.join(tmp_dir, "parsedmarc.ini") + output_dir = os.path.join(tmp_dir, output_dirname) + config_text = ( + f"[general]\noffline = True\nsilent = True\noutput = {output_dir}\n" + ) + archive_dir = None + if archive_dirname is not None: + archive_dir = os.path.join(tmp_dir, archive_dirname) + config_text += f"archive_directory = {archive_dir}\n" + if n_procs is not None: + config_text += f"n_procs = {n_procs}\n" + with open(cfg_path, "w") as f: + f.write(config_text) + return cfg_path, output_dir, archive_dir + + def test_archive_moves_all_three_types(self): + """Aggregate, failure, and SMTP TLS report files given as direct + ``file_path`` arguments are moved into + ``////`` after a + successful parse, the source files are gone from the input + directory, and the aggregate JSON output is still produced.""" + with tempfile.TemporaryDirectory() as tmp_dir: + input_dir = os.path.join(tmp_dir, "input") + os.makedirs(input_dir) + agg1 = self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir) + agg2 = self._copy_sample(self.AGGREGATE_SAMPLE_2, input_dir) + failure = self._copy_sample(self.FAILURE_SAMPLE, input_dir) + smtp_tls = self._copy_sample(self.SMTP_TLS_SAMPLE, input_dir) + + cfg_path, output_dir, archive_dir = self._write_config( + tmp_dir, "output", archive_dirname="archive" + ) + assert archive_dir is not None + + with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]): + parsedmarc.cli._main() + + for src, sample_path in ( + (agg1, self.AGGREGATE_SAMPLE_1), + (agg2, self.AGGREGATE_SAMPLE_2), + (failure, self.FAILURE_SAMPLE), + (smtp_tls, self.SMTP_TLS_SAMPLE), + ): + subdir = self._expected_subdir(sample_path) + dest = os.path.join(archive_dir, subdir, os.path.basename(src)) + self.assertTrue(os.path.isfile(dest), f"missing {dest}") + self.assertFalse(os.path.isfile(src)) + + self.assertTrue(os.path.isfile(os.path.join(output_dir, "aggregate.json"))) + + def test_failed_parse_moved_to_invalid(self): + """A file that fails to parse (garbage content) is moved to + ``/Invalid/`` rather than left in place, and + the run completes without raising.""" + with tempfile.TemporaryDirectory() as tmp_dir: + input_dir = os.path.join(tmp_dir, "input") + os.makedirs(input_dir) + garbage_path = os.path.join(input_dir, "garbage.xml") + with open(garbage_path, "wb") as f: + f.write(b"not a report") + + cfg_path, output_dir, archive_dir = self._write_config( + tmp_dir, "output", archive_dirname="archive" + ) + assert archive_dir is not None + + with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]): + parsedmarc.cli._main() + + dest = os.path.join(archive_dir, "Invalid", "garbage.xml") + self.assertTrue(os.path.isfile(dest)) + self.assertFalse(os.path.isfile(garbage_path)) + + def test_collision_appends_numeric_suffix(self): + """A destination file that already exists at the computed archive + path is never overwritten: the newly archived file gets a + numeric suffix appended before its extension instead, and the + pre-existing file's content is untouched.""" + with tempfile.TemporaryDirectory() as tmp_dir: + input_dir = os.path.join(tmp_dir, "input") + os.makedirs(input_dir) + src = self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir) + + cfg_path, output_dir, archive_dir = self._write_config( + tmp_dir, "output", archive_dirname="archive" + ) + assert archive_dir is not None + + subdir = self._expected_subdir(self.AGGREGATE_SAMPLE_1) + dest_dir = os.path.join(archive_dir, subdir) + os.makedirs(dest_dir) + basename = os.path.basename(src) + preexisting_path = os.path.join(dest_dir, basename) + with open(preexisting_path, "wb") as f: + f.write(b"PREEXISTING DUMMY CONTENT") + + with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]): + parsedmarc.cli._main() + + with open(preexisting_path, "rb") as f: + self.assertEqual(f.read(), b"PREEXISTING DUMMY CONTENT") + + base, ext = os.path.splitext(basename) + suffixed_path = os.path.join(dest_dir, f"{base}-1{ext}") + self.assertTrue(os.path.isfile(suffixed_path)) + self.assertFalse(os.path.isfile(src)) + + def test_second_run_skips_archived_files(self): + """Files already inside ``archive_directory`` are excluded from + the next run's ``file_path`` expansion, so the archive may safely + live inside an input directory without its own contents being + re-parsed and re-archived on a later run. SEEN_AGGREGATE_REPORT_IDS + is cleared between runs to prove the exclusion doesn't rely on + aggregate-report dedup (the failure/SMTP-TLS samples aren't + deduped at all, so they alone would already prove this, but + clearing it removes any doubt for the aggregate sample too).""" + with tempfile.TemporaryDirectory() as tmp_dir: + input_dir = os.path.join(tmp_dir, "input") + os.makedirs(input_dir) + self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir) + self._copy_sample(self.FAILURE_SAMPLE, input_dir) + self._copy_sample(self.SMTP_TLS_SAMPLE, input_dir) + + archive_dir = os.path.join(input_dir, "archive") + output_dir = os.path.join(tmp_dir, "output") + cfg_path = os.path.join(tmp_dir, "parsedmarc.ini") + config_text = ( + "[general]\noffline = True\nsilent = True\n" + f"output = {output_dir}\narchive_directory = {archive_dir}\n" + ) + with open(cfg_path, "w") as f: + f.write(config_text) + + def _archive_tree(): + tree = {} + for root, _dirs, files in os.walk(archive_dir): + for name in files: + path = os.path.join(root, name) + rel = os.path.relpath(path, archive_dir) + with open(path, "rb") as f: + tree[rel] = f.read() + return tree + + with patch.object( + sys, "argv", ["parsedmarc", "-c", cfg_path, "-r", input_dir] + ): + parsedmarc.cli._main() + + first_run_tree = _archive_tree() + self.assertEqual(len(first_run_tree), 3) + + parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear() + + with patch.object( + sys, "argv", ["parsedmarc", "-c", cfg_path, "-r", input_dir] + ): + parsedmarc.cli._main() + + second_run_tree = _archive_tree() + self.assertEqual(first_run_tree, second_run_tree) + for rel in second_run_tree: + self.assertNotIn("-1", os.path.basename(rel)) + + def test_archive_with_n_procs_2(self): + """The same archiving behavior as + ``test_archive_moves_all_three_types``, but with ``n_procs = 2`` + so the direct-file parsing path runs through the multiprocessing + pool.""" + with tempfile.TemporaryDirectory() as tmp_dir: + input_dir = os.path.join(tmp_dir, "input") + os.makedirs(input_dir) + agg1 = self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir) + agg2 = self._copy_sample(self.AGGREGATE_SAMPLE_2, input_dir) + failure = self._copy_sample(self.FAILURE_SAMPLE, input_dir) + smtp_tls = self._copy_sample(self.SMTP_TLS_SAMPLE, input_dir) + + cfg_path, output_dir, archive_dir = self._write_config( + tmp_dir, "output", archive_dirname="archive", n_procs=2 + ) + assert archive_dir is not None + + with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]): + parsedmarc.cli._main() + + for src, sample_path in ( + (agg1, self.AGGREGATE_SAMPLE_1), + (agg2, self.AGGREGATE_SAMPLE_2), + (failure, self.FAILURE_SAMPLE), + (smtp_tls, self.SMTP_TLS_SAMPLE), + ): + subdir = self._expected_subdir(sample_path) + dest = os.path.join(archive_dir, subdir, os.path.basename(src)) + self.assertTrue(os.path.isfile(dest), f"missing {dest}") + self.assertFalse(os.path.isfile(src)) + + self.assertTrue(os.path.isfile(os.path.join(output_dir, "aggregate.json"))) + + def test_no_archive_when_option_unset(self): + """Without ``archive_directory`` configured, input files are left + in place after processing and no ``archive`` directory is + created.""" + with tempfile.TemporaryDirectory() as tmp_dir: + input_dir = os.path.join(tmp_dir, "input") + os.makedirs(input_dir) + agg1 = self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir) + + cfg_path, output_dir, archive_dir = self._write_config(tmp_dir, "output") + self.assertIsNone(archive_dir) + + with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]): + parsedmarc.cli._main() + + self.assertTrue(os.path.isfile(agg1)) + self.assertFalse(os.path.isdir(os.path.join(tmp_dir, "archive"))) + + def test_duplicate_aggregate_reports_both_archived(self): + """The same aggregate report content under two different + filenames is deduplicated down to one report in the JSON output, + but both source files are archived normally: archiving runs for + every file that parsed successfully, independent of the + in-process report-ID dedup.""" + with tempfile.TemporaryDirectory() as tmp_dir: + input_dir = os.path.join(tmp_dir, "input") + os.makedirs(input_dir) + copy1 = self._copy_sample( + self.AGGREGATE_SAMPLE_1, input_dir, dest_basename="copy-a.xml" + ) + copy2 = self._copy_sample( + self.AGGREGATE_SAMPLE_1, input_dir, dest_basename="copy-b.xml" + ) + + cfg_path, output_dir, archive_dir = self._write_config( + tmp_dir, "output", archive_dirname="archive" + ) + assert archive_dir is not None + + with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]): + parsedmarc.cli._main() + + with open(os.path.join(output_dir, "aggregate.json")) as f: + reports = json.load(f) + self.assertEqual(len(reports), 1) + + subdir = self._expected_subdir(self.AGGREGATE_SAMPLE_1) + dest1 = os.path.join(archive_dir, subdir, "copy-a.xml") + dest2 = os.path.join(archive_dir, subdir, "copy-b.xml") + self.assertTrue(os.path.isfile(dest1)) + self.assertTrue(os.path.isfile(dest2)) + self.assertFalse(os.path.isfile(copy1)) + self.assertFalse(os.path.isfile(copy2)) + + class TestGmailAuthModes(unittest.TestCase): @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox") @patch("parsedmarc.cli.GmailConnection") @@ -3734,6 +4050,398 @@ class TestExpandPath(unittest.TestCase): self.assertEqual(_expand_path("relative/path"), "relative/path") +class TestArchiveSubdirForResult(unittest.TestCase): + """Unit tests for _archive_subdir_for_result (issue #570): maps a + parsed report to its ``//`` + archive subdirectory, using minimal synthetic report dicts rather + than full sample parses.""" + + def test_aggregate_report_january_zero_pads_month(self): + from parsedmarc.cli import _archive_subdir_for_result + + result = cast( + ParsedReport, + { + "report_type": "aggregate", + "report": { + "report_metadata": {"begin_date": "2023-01-05 00:00:00"}, + }, + }, + ) + self.assertEqual( + _archive_subdir_for_result(result), + os.path.join("2023", "01", "Aggregate"), + ) + + def test_failure_report_january_zero_pads_month(self): + from parsedmarc.cli import _archive_subdir_for_result + + result = cast( + ParsedReport, + { + "report_type": "failure", + "report": {"arrival_date_utc": "2023-01-05 12:00:00"}, + }, + ) + self.assertEqual( + _archive_subdir_for_result(result), + os.path.join("2023", "01", "Failure"), + ) + + def test_smtp_tls_report_uses_hyphenated_folder_name(self): + from parsedmarc.cli import _archive_subdir_for_result + + result = cast( + ParsedReport, + { + "report_type": "smtp_tls", + "report": {"begin_date": "2016-04-01T00:00:00Z"}, + }, + ) + self.assertEqual( + _archive_subdir_for_result(result), + os.path.join("2016", "04", "SMTP-TLS"), + ) + + def test_missing_date_key_returns_none(self): + from parsedmarc.cli import _archive_subdir_for_result + + result = cast( + ParsedReport, + { + "report_type": "aggregate", + "report": {"report_metadata": {}}, + }, + ) + self.assertIsNone(_archive_subdir_for_result(result)) + + def test_unparseable_date_returns_none(self): + from parsedmarc.cli import _archive_subdir_for_result + + result = cast( + ParsedReport, + { + "report_type": "failure", + "report": {"arrival_date_utc": "not-a-real-date"}, + }, + ) + self.assertIsNone(_archive_subdir_for_result(result)) + + def test_unknown_report_type_returns_none(self): + from parsedmarc.cli import _archive_subdir_for_result + + result = cast(ParsedReport, {"report_type": "unknown", "report": {}}) + self.assertIsNone(_archive_subdir_for_result(result)) + + +class TestExcludeArchivedPaths(unittest.TestCase): + """Unit tests for _exclude_archived_paths (issue #570): filters out + paths already inside the archive directory so a re-run doesn't + re-parse and re-archive its own previous output.""" + + def test_files_inside_archive_dir_are_excluded(self): + from parsedmarc.cli import _exclude_archived_paths + + with tempfile.TemporaryDirectory() as tmp_dir: + archive_dir = os.path.join(tmp_dir, "archive") + nested_dir = os.path.join(archive_dir, "2024", "01", "Aggregate") + os.makedirs(nested_dir) + inside_path = os.path.join(nested_dir, "report.xml") + with open(inside_path, "w") as f: + f.write("x") + outside_path = os.path.join(tmp_dir, "report.xml") + with open(outside_path, "w") as f: + f.write("x") + + result = _exclude_archived_paths([inside_path, outside_path], archive_dir) + + self.assertEqual(result, [outside_path]) + + def test_relative_paths_are_resolved_before_comparison(self): + from parsedmarc.cli import _exclude_archived_paths + + with tempfile.TemporaryDirectory() as tmp_dir: + archive_dir = os.path.join(tmp_dir, "archive") + os.makedirs(archive_dir) + inside_path = os.path.join(archive_dir, "report.xml") + with open(inside_path, "w") as f: + f.write("x") + + cwd = os.getcwd() + try: + os.chdir(tmp_dir) + result = _exclude_archived_paths( + [os.path.join("archive", "report.xml")], "archive" + ) + finally: + os.chdir(cwd) + + self.assertEqual(result, []) + + def test_files_outside_archive_dir_are_kept(self): + from parsedmarc.cli import _exclude_archived_paths + + with tempfile.TemporaryDirectory() as tmp_dir: + archive_dir = os.path.join(tmp_dir, "archive") + os.makedirs(archive_dir) + other_path = os.path.join(tmp_dir, "elsewhere", "report.xml") + os.makedirs(os.path.dirname(other_path)) + with open(other_path, "w") as f: + f.write("x") + + result = _exclude_archived_paths([other_path], archive_dir) + + self.assertEqual(result, [other_path]) + + def test_symlinked_archive_dir_still_excludes_real_path(self): + """A symlink doesn't defeat exclusion: if archive_directory is + configured through one spelling (e.g. a ``/data`` symlink) while + an input path is discovered through the real mount-point spelling + (e.g. ``/mnt/...``), ``os.path.realpath`` resolves both to the + same canonical path so the file already inside the archive is + still recognized and excluded rather than re-archived forever.""" + from parsedmarc.cli import _exclude_archived_paths + + with tempfile.TemporaryDirectory() as tmp_dir: + real_archive_dir = os.path.join(tmp_dir, "real_archive") + os.makedirs(real_archive_dir) + symlinked_archive_dir = os.path.join(tmp_dir, "archive_link") + os.symlink(real_archive_dir, symlinked_archive_dir) + + real_file_path = os.path.join(real_archive_dir, "report.xml") + with open(real_file_path, "w") as f: + f.write("x") + + # archive_directory is configured via the symlinked spelling, + # while the file is discovered via the real spelling. + result = _exclude_archived_paths([real_file_path], symlinked_archive_dir) + + self.assertEqual(result, []) + + def test_non_comparable_paths_are_kept(self): + """Per the Python docs, ``os.path.commonpath`` raises + ``ValueError`` when the paths "are on the different drives" + (Windows) or mix absolute and relative pathnames + (https://docs.python.org/3/library/os.path.html#os.path.commonpath). + Both inputs here are realpath()-resolved so the mix can't occur + on POSIX, leaving the different-drives case unreachable on Linux + CI — hence the simulated raise. A path that can't be compared + with the archive root can't be inside it, so it must be kept + for parsing, and the ValueError must not propagate.""" + from parsedmarc.cli import _exclude_archived_paths + + with tempfile.TemporaryDirectory() as tmp_dir: + archive_dir = os.path.join(tmp_dir, "archive") + os.makedirs(archive_dir) + report_path = os.path.join(tmp_dir, "report.xml") + with open(report_path, "w") as f: + f.write("x") + + with patch( + "parsedmarc.cli.os.path.commonpath", + side_effect=ValueError("Paths don't have the same drive"), + ): + result = _exclude_archived_paths([report_path], archive_dir) + + self.assertEqual(result, [report_path]) + + +class TestMoveFileToArchive(unittest.TestCase): + """Unit tests for _move_file_to_archive (issue #570): the collision + loop that appends a numeric suffix rather than overwriting an + existing destination file.""" + + def test_no_collision_keeps_original_basename(self): + from parsedmarc.cli import _move_file_to_archive + + with tempfile.TemporaryDirectory() as tmp_dir: + src_dir = os.path.join(tmp_dir, "src") + os.makedirs(src_dir) + src_path = os.path.join(src_dir, "report.xml") + with open(src_path, "w") as f: + f.write("content") + dest_dir = os.path.join(tmp_dir, "dest") + + dest_path = _move_file_to_archive(src_path, dest_dir) + + self.assertEqual(dest_path, os.path.join(dest_dir, "report.xml")) + + def test_collision_appends_dash_one(self): + from parsedmarc.cli import _move_file_to_archive + + with tempfile.TemporaryDirectory() as tmp_dir: + src_dir = os.path.join(tmp_dir, "src") + os.makedirs(src_dir) + src_path = os.path.join(src_dir, "report.xml") + with open(src_path, "w") as f: + f.write("new content") + dest_dir = os.path.join(tmp_dir, "dest") + os.makedirs(dest_dir) + existing_path = os.path.join(dest_dir, "report.xml") + with open(existing_path, "w") as f: + f.write("original content") + + dest_path = _move_file_to_archive(src_path, dest_dir) + + self.assertEqual(dest_path, os.path.join(dest_dir, "report-1.xml")) + with open(existing_path) as f: + self.assertEqual(f.read(), "original content") + with open(dest_path) as f: + self.assertEqual(f.read(), "new content") + + def test_two_collisions_append_dash_two(self): + from parsedmarc.cli import _move_file_to_archive + + with tempfile.TemporaryDirectory() as tmp_dir: + src_dir = os.path.join(tmp_dir, "src") + os.makedirs(src_dir) + src_path = os.path.join(src_dir, "report.xml") + with open(src_path, "w") as f: + f.write("newest content") + dest_dir = os.path.join(tmp_dir, "dest") + os.makedirs(dest_dir) + for name in ("report.xml", "report-1.xml"): + with open(os.path.join(dest_dir, name), "w") as f: + f.write(f"original {name}") + + dest_path = _move_file_to_archive(src_path, dest_dir) + + self.assertEqual(dest_path, os.path.join(dest_dir, "report-2.xml")) + with open(os.path.join(dest_dir, "report.xml")) as f: + self.assertEqual(f.read(), "original report.xml") + with open(os.path.join(dest_dir, "report-1.xml")) as f: + self.assertEqual(f.read(), "original report-1.xml") + + def test_failed_move_reraises_even_when_placeholder_cleanup_fails(self): + """When the move fails and removing the placeholder also fails, + the move failure still propagates: the OSError from the + best-effort cleanup must be swallowed, not allowed to mask the + actionable error. The move error is deliberately a non-OSError + type so the assertion proves which of the two exceptions + escaped; the placeholder is left behind, as expected when its + cleanup fails.""" + from parsedmarc.cli import _move_file_to_archive + + with tempfile.TemporaryDirectory() as tmp_dir: + src_path = os.path.join(tmp_dir, "report.xml") + with open(src_path, "w") as f: + f.write("content") + dest_dir = os.path.join(tmp_dir, "dest") + + with ( + patch( + "parsedmarc.cli.shutil.move", + side_effect=RuntimeError("move failed"), + ), + patch( + "parsedmarc.cli.os.remove", + side_effect=OSError("remove failed"), + ), + ): + with self.assertRaises(RuntimeError): + _move_file_to_archive(src_path, dest_dir) + + # The source file is untouched and the zero-byte placeholder + # survives its failed cleanup. + self.assertTrue(os.path.isfile(src_path)) + placeholder = os.path.join(dest_dir, "report.xml") + self.assertTrue(os.path.isfile(placeholder)) + self.assertEqual(os.path.getsize(placeholder), 0) + # A leftover placeholder must never be executable or + # group/other-accessible: it's created with mode 0o600, not + # os.open's 0o777 default. The requested mode is masked by + # the process umask, which only clears bits, so asserting on + # the owner-exec and group/other bits is umask-independent. + placeholder_mode = stat.S_IMODE(os.stat(placeholder).st_mode) + self.assertEqual(placeholder_mode & 0o177, 0) + + +class TestArchiveProcessedFile(unittest.TestCase): + """Unit tests for _archive_processed_file (issue #570): the + orchestrator that routes a processed file to Invalid/ on a parse + failure or the dated subdirectory on success, and never lets a move + failure abort the run.""" + + def test_exception_result_routes_to_invalid(self): + from parsedmarc.cli import _archive_processed_file + + with tempfile.TemporaryDirectory() as tmp_dir: + src_path = os.path.join(tmp_dir, "garbage.xml") + with open(src_path, "w") as f: + f.write("not a report") + archive_dir = os.path.join(tmp_dir, "archive") + + _archive_processed_file( + src_path, archive_dir, parsedmarc.ParserError("boom") + ) + + self.assertTrue( + os.path.isfile(os.path.join(archive_dir, "Invalid", "garbage.xml")) + ) + + def test_move_failure_is_logged_and_does_not_raise(self): + """A filesystem error while moving a file into the archive is + logged, not raised: the file has already been definitively + classified (parsed or failed), so a move error must not abort + the run and lose that work.""" + from parsedmarc.cli import _archive_processed_file + + with tempfile.TemporaryDirectory() as tmp_dir: + src_path = os.path.join(tmp_dir, "garbage.xml") + with open(src_path, "w") as f: + f.write("not a report") + archive_dir = os.path.join(tmp_dir, "archive") + + with patch("shutil.move", side_effect=OSError("disk full")): + with self.assertLogs("parsedmarc.log", level="ERROR") as log_ctx: + _archive_processed_file( + src_path, archive_dir, parsedmarc.ParserError("boom") + ) + + self.assertTrue(any("Error moving" in msg for msg in log_ctx.output)) + # The move never completed, so the source file is untouched. + self.assertTrue(os.path.isfile(src_path)) + + def test_unresolvable_subdir_leaves_file_in_place(self): + """When _archive_subdir_for_result can't determine a destination + (unknown report type or unparseable date, already warned about by + that helper), the file is left where it is rather than guessed + at.""" + from parsedmarc.cli import _archive_processed_file + + with tempfile.TemporaryDirectory() as tmp_dir: + src_path = os.path.join(tmp_dir, "report.json") + with open(src_path, "w") as f: + f.write("{}") + archive_dir = os.path.join(tmp_dir, "archive") + + result = cast(ParsedReport, {"report_type": "unknown", "report": {}}) + _archive_processed_file(src_path, archive_dir, result) + + self.assertTrue(os.path.isfile(src_path)) + + def test_non_parser_error_leaves_file_in_place(self): + """Only a ParserError (every parse-failure exception, including + InvalidSMTPTLSReport, subclasses it) routes to Invalid/. A + transient OSError — e.g. from _parse_report_file_job's broad + catch, not a report-parsing failure — leaves the file in place + so a later run can retry it, and creates no Invalid/ directory + at all, since renaming it would permanently sideline a file that + was never actually shown to be an invalid report.""" + from parsedmarc.cli import _archive_processed_file + + with tempfile.TemporaryDirectory() as tmp_dir: + src_path = os.path.join(tmp_dir, "report.xml") + with open(src_path, "w") as f: + f.write("") + archive_dir = os.path.join(tmp_dir, "archive") + + _archive_processed_file(src_path, archive_dir, OSError("transient")) + + self.assertTrue(os.path.isfile(src_path)) + self.assertFalse(os.path.isdir(archive_dir)) + + # --------------------------------------------------------------------------- # _parse_config: per-section INI → opts mapping # @@ -3873,6 +4581,33 @@ class TestParseConfigGeneral(unittest.TestCase): _parse_config(cp, opts) self.assertEqual(opts.normalize_timespan_threshold_hours, 48.0) + def test_general_archive_directory_expands_env_var(self): + """archive_directory goes through _expand_path, so a $VAR + reference in the INI value is expanded (issue #570).""" + from parsedmarc.cli import _parse_config + + cp = _config_with( + "general", {"archive_directory": "$SOME_ARCHIVE_VAR/reports-archive"} + ) + opts = _opts() + with patch.dict(os.environ, {"SOME_ARCHIVE_VAR": "/opt/dmarc"}): + _parse_config(cp, opts) + self.assertEqual(opts.archive_directory, "/opt/dmarc/reports-archive") + + def test_general_archive_directory_unset_leaves_attribute_absent(self): + """When archive_directory is absent from the INI, _parse_config + never sets opts.archive_directory at all — asserted here as the + attribute staying absent from this test's bare Namespace. (In the + real CLI the attribute pre-exists with the Namespace default of + None, so _parse_config leaving it untouched is what keeps + archiving disabled.)""" + from parsedmarc.cli import _parse_config + + cp = _config_with("general", {"silent": "false"}) + opts = _opts() + _parse_config(cp, opts) + self.assertFalse(hasattr(opts, "archive_directory")) + class TestParseConfigElasticsearch(unittest.TestCase): def test_elasticsearch_basic(self):