amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-08-15 03:43:27 +00:00
Code Issues Projects Releases Wiki Activity
Files
5b2dfa7d1ab7a0068514391e35e15449094a2caa
parsedmarc/parsedmarc/cli.py
T
Sean WhalenGitHubClaude Opus 5
 b304cf8639 Cover index_prefix_domain_map and index_suffix in index migrations (#868) (#869) 
* Cover index_prefix_domain_map and index_suffix in index migrations (#868)

The startup Elasticsearch/OpenSearch index migrations built their target
index names from the [elasticsearch]/[opensearch] index_prefix and
index_suffix options alone, while the save path also honors
general.index_prefix_domain_map. A multi-tenant deployment therefore ran
the backfill guard `count` against dmarc_aggregate*/smtp_tls*, patterns
matching none of its real <tenant>_dmarc_aggregate-* indexes. The query
passes allow_no_indices=True, so a zero-match wildcard returns count 0 --
indistinguishable from "already backfilled" -- and the backfill was
skipped silently, with no log line at any level.

Resolve migration index names through a new _migration_index_names()
helper that widens both configurable axes: one name per tenant prefix in
the map plus the unprefixed name (unmapped domains are still saved
unprefixed), and, when an index_suffix is set, the unsuffixed name
alongside the suffixed one so history predating the suffix is covered. A
configured index_prefix still wins outright and suppresses the map
fan-out, matching save-time precedence. The key normalization is now
shared with get_index_prefix() via _normalize_index_prefix(), so the
names parsedmarc migrates cannot drift from the ones it writes. The
resolved lists are logged at DEBUG, and the SIGHUP reload path passes the
freshly parsed map, so a newly onboarded tenant is covered without a
restart.

Also repair the legacy published_policy.fo migration, which has been
unable to complete since mapping types were removed in Elasticsearch 7
(and never existed in OpenSearch): it read the field mapping in the
type-keyed response shape, so the check always fell through, and its
put_mapping() call passed a doc_type argument neither current client
accepts. It now reads either response shape, uses each client's current
signature, and takes its index names in a separate legacy_fo_indexes
argument -- exact names, since 5.0.0 introduced date-suffixed index names
in the same release that fixed the fo declaration, but prefixed and
suffixed where configured, since both options date back to 4.1.0. Tenant
prefixes are excluded: index_prefix_domain_map arrived in 8.19.0, and
this migration renames the index it rebuilds. The Elasticsearch copy,
removed as unreachable during the #806 client migration, is restored now
that the cause is understood.

Finally, reject an index_prefix_domain_map YAML file that is not a
mapping of string tenant names to lists of domains, instead of raising
mid-save on a non-string key or silently matching the wrong domains on a
scalar value -- `in` on a str is a substring test, so "example.co"
matches "example.com"
(https://docs.python.org/3/reference/expressions.html#membership-test-operations).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Fix OpenSearch capitalization and spacing in the usage docs

Copilot review of #869: the `index_prefix_domain_map` option line spelled
"OpenSearch" as "Opensearch" and had a doubled space before the type. Both
predate this branch but sit inside a hunk it rewrites. Also wrapped the line
to match the continuation-indent style of every other option in the list, and
corrected the same misspelling in the multi-tenant section a few lines above
the paragraph this branch added.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Require index_prefix_domain_map domain lists to hold strings

Copilot review of #869: the shape check verified that each value was a
list but not what the list contained, so `tenant_a: [42]` passed and then
never compared equal to any domain the save path looks up -- the same
silent-misbehavior class the check exists to reject, and a contradiction
of the "any other shape is rejected at startup" claim in the docs.

Check the list's items too, and reword the error message, comment,
CHANGELOG and docs to state the rule the check now enforces: a mapping of
tenant names to lists of domain names, all strings.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Make the legacy fo migration retry-safe, and nest its mapping body

Copilot review of #869, verified against live Elasticsearch 8.19 and
OpenSearch 3 containers rather than mocks.

Retry safety (a real defect): an attempt interrupted between creating the
-v2 index and deleting the original left debris that made create() fail
with "resource already exists" on every later startup, inside the same try
that swallows the error -- so the index was never migrated and the debris
document survived. Reproduced on a live cluster. Discard a leftover target
first; that is safe precisely because reaching this point means the
original still holds the data, since it is deleted only once the reindex
has succeeded.

Mapping body: the reviewer's concern that a dotted key under `properties`
risks a runtime failure does not hold -- both clusters accept it and
produce a byte-identical mapping, with the dot expanded into
published_policy -> properties -> fo. Switch to the nested object form
anyway, since dot expansion is conditional on the object's `subobjects`
setting and this shape never is, and derive the object/leaf names from the
dotted constant so the write cannot drift from the field the read looks up.

Both fo-migration suites now build per-name Index mocks. A single shared
mock cannot express "the original exists but its migration target does
not", which is the ordinary case and the one the retry fix turns on; the
happy path now also asserts the target index is never deleted.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 20:48:00 -04:00

3352 lines
138 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""A CLI for parsing DMARC reports"""
import atexit
import functools
import http.client
import json
import logging
import os
import shutil
import signal
import sys
import time
from argparse import ArgumentParser, Namespace
from configparser import ConfigParser
from glob import escape as glob_escape, glob
from ssl import CERT_NONE, create_default_context
import httpx
import yaml
from azure.core.exceptions import ClientAuthenticationError
from kiota_abstractions.api_error import APIError
from tqdm import tqdm
from parsedmarc import (
IP_ADDRESS_CACHE,
REVERSE_DNS_MAP,
SEEN_AGGREGATE_REPORT_IDS,
InvalidDMARCReport,
ParserConfig,
ParserError,
__version__,
elastic,
email_results,
email_results_via_msgraph,
gelf,
get_dmarc_reports_from_mailbox,
get_dmarc_reports_from_mbox,
kafkaclient,
loganalytics,
opensearch,
postgres,
s3,
save_output,
splunk,
syslog,
watch_inbox,
webhook,
)
from parsedmarc.constants import DEFAULT_DNS_MAX_RETRIES, DEFAULT_DNS_TIMEOUT
from parsedmarc.log import logger
from parsedmarc.mail import (
AuthMethod,
GmailConnection,
IMAPConnection,
MaildirConnection,
MSGraphConnection,
)
from parsedmarc.parallel import _parse_report_file_job, parallel_map
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,
load_reverse_dns_map,
)
# Increase the max header limit for very large emails. `_MAXHEADERS` is a
# private stdlib attribute and may not exist in type stubs.
setattr(http.client, "_MAXHEADERS", 200)
formatter = logging.Formatter(
fmt="%(levelname)8s:%(filename)s:%(lineno)d:%(message)s",
datefmt="%Y-%m-%d:%H:%M:%S",
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
class ConfigurationError(Exception):
"""Raised when a configuration file has missing or invalid settings."""
pass
def _normalize_graph_auth_method(value: str) -> str:
"""Return the canonical :class:`AuthMethod` member name for *value*.
Matching is case-insensitive so config values like ``certificate`` are
accepted alongside the canonical ``Certificate``.
Raises:
ConfigurationError: When *value* does not match a known auth method.
"""
value_lower = value.lower()
for method in AuthMethod:
if method.name.lower() == value_lower:
return method.name
raise ConfigurationError(
"Invalid msgraph auth_method: {!r}. Valid values are: {}".format(
value, ", ".join(m.name for m in AuthMethod)
)
)
def _str_to_list(s):
"""Converts a comma separated string to a list"""
_list = s.split(",")
return list(map(lambda i: i.lstrip(), _list))
def _msgraph_request_id_suffix(error: Exception) -> str:
"""Returns ``" (request-id=..., client-request-id=...)"`` with only
the ids that are actually present, or ``""`` if neither is
available. Never raises."""
try:
inner_error = getattr(getattr(error, "error", None), "inner_error", None)
request_id = getattr(inner_error, "request_id", None)
client_request_id = getattr(inner_error, "client_request_id", None)
if not request_id:
headers = getattr(error, "response_headers", None) or {}
request_id = headers.get("request-id")
parts = []
if request_id:
parts.append(f"request-id={request_id}")
if client_request_id:
parts.append(f"client-request-id={client_request_id}")
if not parts:
return ""
return " ({})".format(", ".join(parts))
except Exception:
return ""
def _log_msgraph_failure(
error: Exception,
*,
stage: str,
mailbox: str | None,
tenant_id: str | None,
auth_method: str | None,
) -> None:
"""Logs a single clear ERROR line for a Microsoft Graph connection,
fetch, send, or watch failure, identifying the mailbox/tenant/auth
method and the Graph request-id/client-request-id when available.
The full traceback is preserved at --debug via a follow-up DEBUG
record. Never calls exit() - the call site keeps its own exit(1)."""
if isinstance(error, APIError):
detail = getattr(error, "primary_message", None) or error.message or str(error)
detail = " ".join(str(detail).split())
summary = (
f"{type(error).__name__} status={error.response_status_code}: {detail}"
)
else:
summary = "{}: {}".format(type(error).__name__, " ".join(str(error).split()))
logger.error(
"Microsoft Graph %s failed (mailbox=%s, tenant_id=%s, auth_method=%s): %s%s",
stage,
mailbox,
tenant_id,
auth_method,
summary,
_msgraph_request_id_suffix(error),
)
logger.debug("Microsoft Graph %s failure details:", stage, exc_info=True)
def _expand_path(p: str) -> str:
"""Expand ``~`` and ``$VAR`` references in a file path."""
return os.path.expanduser(os.path.expandvars(p))
def _expand_file_path_args(paths: list[str], recursive: bool = False) -> list[str]:
"""Expand CLI file-path arguments into a flat list of file paths.
A path to an existing file is taken literally, a path to an existing
directory is expanded to the files inside it (see below), and only a
non-existent path is treated as a glob pattern. This preserves
shell-style wildcard expansion (e.g. a quoted ``samples/*.xml``) while
ensuring that literal filenames containing glob metacharacters
(``[``, ``]``, ``*``, ``?``) are not silently dropped. Emailed DMARC
failure reports are frequently named like
``[Provider DMARC Failure Report] Subject.eml``; ``glob()`` treats the
brackets as a character class, matches nothing, and drops the file
(see <https://docs.python.org/3/library/glob.html>).
A directory is expanded to the files directly inside it, using the
same shell-glob semantics as ``<dir>/*`` (or ``<dir>/**`` when
``recursive`` is ``True``): dotfile entries are excluded, and
non-file entries (subdirectories) are filtered out. With
``recursive=False`` a subdirectory found this way is skipped with a
debug log rather than descended into. The directory component is
passed through ``glob.escape`` before being combined with the
wildcard so that directory names containing glob metacharacters
(``[``, ``]``, ``*``, ``?``) still expand correctly instead of being
treated as a character class or wildcard themselves.
``recursive`` also enables ``**`` to match any number of directories
(including none) in glob patterns supplied directly as arguments, per
the same stdlib glob semantics.
"""
expanded: list[str] = []
for path in paths:
if os.path.isdir(path):
pattern = os.path.join(glob_escape(path), "**" if recursive else "*")
for match in sorted(glob(pattern, recursive=recursive)):
if os.path.isfile(match):
expanded.append(match)
elif not recursive and os.path.isdir(match):
logger.debug(
"Skipping subdirectory %s (pass --recursive to descend)",
match,
)
elif os.path.exists(path):
expanded.append(path)
else:
expanded += glob(path, recursive=recursive)
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.
``<input>/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 ``<year>/<month>/<type folder>`` 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
``<archive_directory>/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 ``<year>/<month>/<type>``
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(
{
"general",
"mailbox",
"imap",
"msgraph",
"elasticsearch",
"opensearch",
"splunk_hec",
"kafka",
"smtp",
"s3",
"postgresql",
"syslog",
"gmail_api",
"maildir",
"log_analytics",
"gelf",
"webhook",
}
)
# Short aliases that don't follow the PARSEDMARC_{SECTION}_{KEY} pattern.
_ENV_ALIASES: dict[str, tuple[str, str]] = {
"DEBUG": ("general", "debug"),
"PARSEDMARC_DEBUG": ("general", "debug"),
}
# Real config keys whose own names end in ``_file``. For these the
# ``PARSEDMARC_..._FILE`` env var is the direct value (a path string),
# not a Docker-secret file reference. Keep in sync with ``_parse_config``
# whenever a new ``*_file`` config key is added.
_DIRECT_FILE_KEYS = frozenset(
[
"GENERAL_LOG_FILE",
"MSGRAPH_TOKEN_FILE",
"GMAIL_API_CREDENTIALS_FILE",
"GMAIL_API_TOKEN_FILE",
]
)
def _resolve_section_key(suffix: str) -> tuple:
"""Resolve an env var suffix like ``IMAP_PASSWORD`` to ``('imap', 'password')``.
Uses longest-prefix matching against known section names so that
multi-word sections like ``splunk_hec`` are handled correctly.
Returns ``(None, None)`` when no known section matches.
"""
suffix_lower = suffix.lower()
best_section = None
best_key = None
for section in _KNOWN_SECTIONS:
section_prefix = section + "_"
if suffix_lower.startswith(section_prefix):
key = suffix_lower[len(section_prefix) :]
if key and (best_section is None or len(section) > len(best_section)):
best_section = section
best_key = key
return best_section, best_key
def _read_secret_file(env_key: str, raw_path: str) -> str:
"""Read a Docker-secret file referenced by a ``PARSEDMARC_..._FILE`` env var.
Strips any trailing CR/LF from the file contents. Raises
``ConfigurationError`` (not a silent fallback) when the file is missing,
unreadable, or not valid UTF-8.
"""
path = _expand_path(raw_path)
try:
with open(path, encoding="utf-8") as f:
return f.read().rstrip("\r\n")
except (OSError, UnicodeDecodeError) as exc:
raise ConfigurationError(
f"Cannot read secret file for {env_key}: {path} ({exc.__class__.__name__})"
) from exc
def _apply_env_overrides(config: ConfigParser) -> None:
"""Inject ``PARSEDMARC_*`` environment variables into *config*.
Environment variables matching ``PARSEDMARC_{SECTION}_{KEY}`` override
(or create) the corresponding config-file values. Sections are created
automatically when they do not yet exist.
A ``PARSEDMARC_{SECTION}_{KEY}_FILE`` variant reads the value from the
referenced file (Docker / Kubernetes secret convention). When both the
direct variable and its ``_FILE`` companion are set, the file wins. The
handful of real config keys whose own names end in ``_file`` (see
``_DIRECT_FILE_KEYS``) keep their pre-existing direct-value semantics
and are not eligible for the secret-file wrap.
"""
prefix = "PARSEDMARC_"
file_suffix = "_FILE"
direct: dict[tuple[str, str], str] = {}
secrets: dict[tuple[str, str], str] = {}
for env_key, value in os.environ.items():
if env_key == "PARSEDMARC_CONFIG_FILE":
continue
if env_key in _ENV_ALIASES:
direct[_ENV_ALIASES[env_key]] = value
continue
if not env_key.startswith(prefix):
continue
key_body = env_key[len(prefix) :]
is_secret = key_body.endswith(file_suffix) and key_body not in _DIRECT_FILE_KEYS
if is_secret:
section, key = _resolve_section_key(key_body[: -len(file_suffix)])
else:
section, key = _resolve_section_key(key_body)
if section is None:
logger.debug("Ignoring unrecognized env var: %s", env_key)
continue
if is_secret:
value = _read_secret_file(env_key, value)
secrets[(section, key)] = value
else:
direct[(section, key)] = value
# _FILE entries win over direct ones: dict-unpack lets later mappings overwrite.
for (section, key), value in {**direct, **secrets}.items():
if not config.has_section(section):
config.add_section(section)
config.set(section, key, value)
logger.debug("Config override from env: [%s] %s", section, key)
def _configure_logging(log_level, log_file=None):
"""
Configure logging for the current process.
This is needed for child processes to properly log messages.
Args:
log_level: The logging level (e.g., logging.DEBUG, logging.WARNING)
log_file: Optional path to log file
"""
from parsedmarc.log import configure_logging
configure_logging(log_level, log_file)
# Loggers of the libraries that implement the mailbox and Microsoft Graph
# layers. parsedmarc only configures its own logger, so without this list
# their records — including azure-identity's AADSTS token-endpoint errors,
# which are what distinguish a local config problem from an Exchange
# Online / Entra ID one — are silently dropped even with --debug.
# The Graph SDK's kiota middleware (kiota_http etc.) is deliberately
# absent: it does not use Python logging (its observability is
# OpenTelemetry tracing), so there are no records to enable.
_DEPENDENCY_LOGGERS = (
"mailsuite",
"azure",
"msgraph",
"httpx",
"httpcore",
)
def _configure_dependency_logging(level: int) -> None:
"""Apply parsedmarc's logging verbosity to dependency loggers.
Follows the parsedmarc log level when ``--verbose``/``--debug`` makes it
more verbose than WARNING, and stays at WARNING otherwise, so dependency
warnings keep surfacing without adding noise at the default level.
Handlers are synced to exactly the parsedmarc logger's own handlers
(console and optional file), so dependency records reach the same
destinations, and a SIGHUP reload that swaps the log file neither
duplicates output nor keeps writing to a closed handler. Propagation
to the root logger is disabled so that a stray ``logging.basicConfig()``
anywhere in the process cannot double-print every dependency record.
CLI-only: library consumers configure logging themselves.
"""
dep_level = min(level, logging.WARNING)
for name in _DEPENDENCY_LOGGERS:
dep_logger = logging.getLogger(name)
dep_logger.setLevel(dep_level)
dep_logger.propagate = False
for existing in list(dep_logger.handlers):
if existing not in logger.handlers:
dep_logger.removeHandler(existing)
for wanted in logger.handlers:
if wanted not in dep_logger.handlers:
dep_logger.addHandler(wanted)
def _load_config(config_file: str | None = None) -> ConfigParser:
"""Load configuration from an INI file and/or environment variables.
Args:
config_file: Optional path to an .ini config file.
Returns:
A ``ConfigParser`` populated from the file (if given) and from any
``PARSEDMARC_*`` environment variables.
Raises:
ConfigurationError: If *config_file* is given but does not exist.
"""
config = ConfigParser(interpolation=None)
if config_file is not None:
abs_path = os.path.abspath(config_file)
if not os.path.exists(abs_path):
raise ConfigurationError(f"A file does not exist at {abs_path}")
if not os.access(abs_path, os.R_OK):
raise ConfigurationError(
f"Unable to read {abs_path} — check file permissions"
)
config.read(config_file)
_apply_env_overrides(config)
return config
def _parse_config(config: ConfigParser, opts):
"""Apply a loaded ``ConfigParser`` to *opts* in place.
Args:
config: A ``ConfigParser`` (from ``_load_config``).
opts: Namespace object to update with parsed values.
Returns:
index_prefix_domain_map or None
Raises:
ConfigurationError: If required settings are missing or invalid.
"""
opts.silent = True
index_prefix_domain_map = None
if "general" in config.sections():
general_config = config["general"]
if "silent" in general_config:
opts.silent = bool(general_config.getboolean("silent"))
if "normalize_timespan_threshold_hours" in general_config:
opts.normalize_timespan_threshold_hours = general_config.getfloat(
"normalize_timespan_threshold_hours"
)
if "index_prefix_domain_map" in general_config:
with open(_expand_path(general_config["index_prefix_domain_map"])) as f:
index_prefix_domain_map = yaml.safe_load(f)
# An empty file loads as None, which means "unset". Anything else
# must be a mapping of tenant name to a list of domain names, all
# strings: the save path iterates the keys as index name prefixes
# and tests `get_base_domain(...).lower() in <value>`. Every other
# shape fails silently rather than loudly -- a scalar value makes
# that an `in` on a str, which is a substring test, so it matches
# the wrong domains ("example.co" in "example.com" is True), and a
# non-string list item simply never compares equal to any domain.
if index_prefix_domain_map is not None and (
not isinstance(index_prefix_domain_map, dict)
or not all(
isinstance(key, str)
and isinstance(value, list)
and all(isinstance(domain, str) for domain in value)
for key, value in index_prefix_domain_map.items()
)
):
raise ConfigurationError(
"index_prefix_domain_map must be a YAML mapping of tenant "
"name to a list of domain names, all strings"
)
if "offline" in general_config:
opts.offline = bool(general_config.getboolean("offline"))
if "strip_attachment_payloads" in general_config:
opts.strip_attachment_payloads = bool(
general_config.getboolean("strip_attachment_payloads")
)
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:
opts.failure_json_filename = general_config["failure_json_filename"]
elif "forensic_json_filename" in general_config:
opts.failure_json_filename = general_config["forensic_json_filename"]
if "smtp_tls_json_filename" in general_config:
opts.smtp_tls_json_filename = general_config["smtp_tls_json_filename"]
if "aggregate_csv_filename" in general_config:
opts.aggregate_csv_filename = general_config["aggregate_csv_filename"]
if "failure_csv_filename" in general_config:
opts.failure_csv_filename = general_config["failure_csv_filename"]
elif "forensic_csv_filename" in general_config:
opts.failure_csv_filename = general_config["forensic_csv_filename"]
if "smtp_tls_csv_filename" in general_config:
opts.smtp_tls_csv_filename = general_config["smtp_tls_csv_filename"]
if "dns_timeout" in general_config:
opts.dns_timeout = general_config.getfloat("dns_timeout")
if opts.dns_timeout is None:
opts.dns_timeout = 2
if "dns_retries" in general_config:
opts.dns_retries = general_config.getint("dns_retries")
if opts.dns_retries is None:
opts.dns_retries = 0
if "dns_test_address" in general_config:
opts.dns_test_address = general_config["dns_test_address"]
if "nameservers" in general_config:
opts.nameservers = _str_to_list(general_config["nameservers"])
# nameservers pre-flight check
dummy_hostname = None
try:
dummy_hostname = get_reverse_dns(
opts.dns_test_address,
nameservers=opts.nameservers,
timeout=opts.dns_timeout,
)
except Exception as ns_error:
raise ConfigurationError(
f"DNS pre-flight check failed: {ns_error}"
) from ns_error
if not dummy_hostname:
raise ConfigurationError(
f"DNS pre-flight check failed: no PTR record for {opts.dns_test_address} from {opts.nameservers}"
)
if "save_aggregate" in general_config:
opts.save_aggregate = bool(general_config.getboolean("save_aggregate"))
if "save_failure" in general_config:
opts.save_failure = bool(general_config.getboolean("save_failure"))
elif "save_forensic" in general_config:
opts.save_failure = bool(general_config.getboolean("save_forensic"))
if "save_smtp_tls" in general_config:
opts.save_smtp_tls = bool(general_config.getboolean("save_smtp_tls"))
if "debug" in general_config:
opts.debug = bool(general_config.getboolean("debug"))
if "verbose" in general_config:
opts.verbose = bool(general_config.getboolean("verbose"))
if "warnings" in general_config:
opts.warnings = bool(general_config.getboolean("warnings"))
if "fail_on_output_error" in general_config:
opts.fail_on_output_error = bool(
general_config.getboolean("fail_on_output_error")
)
if "log_file" in general_config:
opts.log_file = _expand_path(general_config["log_file"])
if "n_procs" in general_config:
opts.n_procs = general_config.getint("n_procs")
if "ip_db_path" in general_config:
opts.ip_db_path = _expand_path(general_config["ip_db_path"])
else:
opts.ip_db_path = None
if "ipinfo_url" in general_config:
opts.ipinfo_url = general_config["ipinfo_url"]
elif "ip_db_url" in general_config:
# ``ip_db_url`` is the pre-9.10 name for the same option. Accept
# it as a deprecated alias; prefer ``ipinfo_url`` going forward.
opts.ipinfo_url = general_config["ip_db_url"]
logger.warning("[general] ip_db_url is deprecated; rename it to ipinfo_url")
if "ipinfo_api_token" in general_config:
opts.ipinfo_api_token = general_config["ipinfo_api_token"]
if "always_use_local_files" in general_config:
opts.always_use_local_files = bool(
general_config.getboolean("always_use_local_files")
)
if "local_reverse_dns_map_path" in general_config:
opts.reverse_dns_map_path = _expand_path(
general_config["local_reverse_dns_map_path"]
)
if "reverse_dns_map_url" in general_config:
opts.reverse_dns_map_url = general_config["reverse_dns_map_url"]
if "local_psl_overrides_path" in general_config:
opts.psl_overrides_path = _expand_path(
general_config["local_psl_overrides_path"]
)
if "psl_overrides_url" in general_config:
opts.psl_overrides_url = general_config["psl_overrides_url"]
if "prettify_json" in general_config:
opts.prettify_json = bool(general_config.getboolean("prettify_json"))
if "mailbox" in config.sections():
mailbox_config = config["mailbox"]
if "msgraph" in config.sections():
opts.mailbox_reports_folder = "Inbox"
if "reports_folder" in mailbox_config:
opts.mailbox_reports_folder = mailbox_config["reports_folder"]
if "archive_folder" in mailbox_config:
opts.mailbox_archive_folder = mailbox_config["archive_folder"]
if "watch" in mailbox_config:
opts.mailbox_watch = bool(mailbox_config.getboolean("watch"))
if "delete" in mailbox_config:
opts.mailbox_delete = bool(mailbox_config.getboolean("delete"))
if "delete_aggregate" in mailbox_config:
opts.mailbox_delete_aggregate = bool(
mailbox_config.getboolean("delete_aggregate")
)
if "delete_failure" in mailbox_config:
opts.mailbox_delete_failure = bool(
mailbox_config.getboolean("delete_failure")
)
if "delete_smtp_tls" in mailbox_config:
opts.mailbox_delete_smtp_tls = bool(
mailbox_config.getboolean("delete_smtp_tls")
)
if "delete_invalid" in mailbox_config:
opts.mailbox_delete_invalid = bool(
mailbox_config.getboolean("delete_invalid")
)
if "test" in mailbox_config:
opts.mailbox_test = bool(mailbox_config.getboolean("test"))
if "batch_size" in mailbox_config:
opts.mailbox_batch_size = mailbox_config.getint("batch_size")
if "check_timeout" in mailbox_config:
opts.mailbox_check_timeout = mailbox_config.getint("check_timeout")
if "max_unsaved_retries" in mailbox_config:
opts.mailbox_max_unsaved_retries = mailbox_config.getint(
"max_unsaved_retries"
)
if "since" in mailbox_config:
opts.mailbox_since = mailbox_config["since"]
if "imap" in config.sections():
imap_config = config["imap"]
if "watch" in imap_config:
logger.warning(
"Starting in 8.0.0, the watch option has been "
"moved from the imap configuration section to "
"the mailbox configuration section."
)
if "host" in imap_config:
opts.imap_host = imap_config["host"]
else:
raise ConfigurationError(
"host setting missing from the imap config section"
)
if "port" in imap_config:
opts.imap_port = imap_config.getint("port")
if "timeout" in imap_config:
opts.imap_timeout = imap_config.getint("timeout")
if "max_retries" in imap_config:
opts.imap_max_retries = imap_config.getint("max_retries")
if "ssl" in imap_config:
opts.imap_ssl = bool(imap_config.getboolean("ssl"))
if "skip_certificate_verification" in imap_config:
opts.imap_skip_certificate_verification = bool(
imap_config.getboolean("skip_certificate_verification")
)
if "user" in imap_config:
opts.imap_user = imap_config["user"]
else:
raise ConfigurationError(
"user setting missing from the imap config section"
)
if "password" in imap_config:
opts.imap_password = imap_config["password"]
else:
raise ConfigurationError(
"password setting missing from the imap config section"
)
if "reports_folder" in imap_config:
opts.mailbox_reports_folder = imap_config["reports_folder"]
logger.warning(
"Use of the reports_folder option in the imap "
"configuration section has been deprecated. "
"Use this option in the mailbox configuration "
"section instead."
)
if "archive_folder" in imap_config:
opts.mailbox_archive_folder = imap_config["archive_folder"]
logger.warning(
"Use of the archive_folder option in the imap "
"configuration section has been deprecated. "
"Use this option in the mailbox configuration "
"section instead."
)
if "watch" in imap_config:
opts.mailbox_watch = bool(imap_config.getboolean("watch"))
logger.warning(
"Use of the watch option in the imap "
"configuration section has been deprecated. "
"Use this option in the mailbox configuration "
"section instead."
)
if "delete" in imap_config:
logger.warning(
"Use of the delete option in the imap "
"configuration section has been deprecated. "
"Use this option in the mailbox configuration "
"section instead."
)
if "test" in imap_config:
opts.mailbox_test = bool(imap_config.getboolean("test"))
logger.warning(
"Use of the test option in the imap "
"configuration section has been deprecated. "
"Use this option in the mailbox configuration "
"section instead."
)
if "batch_size" in imap_config:
opts.mailbox_batch_size = imap_config.getint("batch_size")
logger.warning(
"Use of the batch_size option in the imap "
"configuration section has been deprecated. "
"Use this option in the mailbox configuration "
"section instead."
)
if "msgraph" in config.sections():
graph_config = config["msgraph"]
opts.graph_token_file = _expand_path(graph_config.get("token_file", ".token"))
if "auth_method" not in graph_config:
logger.info(
"auth_method setting missing from the "
"msgraph config section "
"defaulting to UsernamePassword"
)
opts.graph_auth_method = AuthMethod.UsernamePassword.name
else:
opts.graph_auth_method = _normalize_graph_auth_method(
graph_config["auth_method"]
)
if opts.graph_auth_method == AuthMethod.UsernamePassword.name:
if "user" in graph_config:
opts.graph_user = graph_config["user"]
else:
raise ConfigurationError(
"user setting missing from the msgraph config section"
)
if "password" in graph_config:
opts.graph_password = graph_config["password"]
else:
raise ConfigurationError(
"password setting missing from the msgraph config section"
)
if "client_secret" in graph_config:
opts.graph_client_secret = graph_config["client_secret"]
else:
raise ConfigurationError(
"client_secret setting missing from the msgraph config section"
)
if opts.graph_auth_method == AuthMethod.DeviceCode.name:
if "user" in graph_config:
opts.graph_user = graph_config["user"]
if opts.graph_auth_method != AuthMethod.UsernamePassword.name:
if "tenant_id" in graph_config:
opts.graph_tenant_id = graph_config["tenant_id"]
else:
raise ConfigurationError(
"tenant_id setting missing from the msgraph config section"
)
if opts.graph_auth_method == AuthMethod.ClientSecret.name:
if "client_secret" in graph_config:
opts.graph_client_secret = graph_config["client_secret"]
else:
raise ConfigurationError(
"client_secret setting missing from the msgraph config section"
)
if opts.graph_auth_method == AuthMethod.Certificate.name:
if "certificate_path" in graph_config:
opts.graph_certificate_path = _expand_path(
graph_config["certificate_path"]
)
else:
raise ConfigurationError(
"certificate_path setting missing from the msgraph config section"
)
if "certificate_password" in graph_config:
opts.graph_certificate_password = graph_config["certificate_password"]
if opts.graph_auth_method == AuthMethod.ClientAssertion.name:
if "client_assertion" in graph_config:
opts.graph_client_assertion = graph_config["client_assertion"]
else:
raise ConfigurationError(
"client_assertion setting missing from the msgraph config section"
)
if "client_id" in graph_config:
opts.graph_client_id = graph_config["client_id"]
else:
raise ConfigurationError(
"client_id setting missing from the msgraph config section"
)
if "mailbox" in graph_config:
opts.graph_mailbox = graph_config["mailbox"]
elif opts.graph_auth_method != AuthMethod.UsernamePassword.name:
raise ConfigurationError(
"mailbox setting missing from the msgraph config section"
)
if "graph_url" in graph_config:
opts.graph_url = graph_config["graph_url"]
elif "url" in graph_config:
opts.graph_url = graph_config["url"]
if "allow_unencrypted_storage" in graph_config:
opts.graph_allow_unencrypted_storage = bool(
graph_config.getboolean("allow_unencrypted_storage")
)
if "elasticsearch" in config:
elasticsearch_config = config["elasticsearch"]
if "hosts" in elasticsearch_config:
opts.elasticsearch_hosts = _str_to_list(elasticsearch_config["hosts"])
else:
raise ConfigurationError(
"hosts setting missing from the elasticsearch config section"
)
if "timeout" in elasticsearch_config:
timeout = elasticsearch_config.getfloat("timeout")
opts.elasticsearch_timeout = timeout
if "number_of_shards" in elasticsearch_config:
number_of_shards = elasticsearch_config.getint("number_of_shards")
opts.elasticsearch_number_of_shards = number_of_shards
if "number_of_replicas" in elasticsearch_config:
number_of_replicas = elasticsearch_config.getint("number_of_replicas")
opts.elasticsearch_number_of_replicas = number_of_replicas
if "index_suffix" in elasticsearch_config:
opts.elasticsearch_index_suffix = elasticsearch_config["index_suffix"]
if "index_prefix" in elasticsearch_config:
opts.elasticsearch_index_prefix = elasticsearch_config["index_prefix"]
if "monthly_indexes" in elasticsearch_config:
monthly = bool(elasticsearch_config.getboolean("monthly_indexes"))
opts.elasticsearch_monthly_indexes = monthly
if "ssl" in elasticsearch_config:
opts.elasticsearch_ssl = bool(elasticsearch_config.getboolean("ssl"))
if "cert_path" in elasticsearch_config:
opts.elasticsearch_ssl_cert_path = _expand_path(
elasticsearch_config["cert_path"]
)
if "skip_certificate_verification" in elasticsearch_config:
opts.elasticsearch_skip_certificate_verification = bool(
elasticsearch_config.getboolean("skip_certificate_verification")
)
if "user" in elasticsearch_config:
opts.elasticsearch_username = elasticsearch_config["user"]
if "password" in elasticsearch_config:
opts.elasticsearch_password = elasticsearch_config["password"]
# Until 8.20
if "apiKey" in elasticsearch_config:
opts.elasticsearch_api_key = elasticsearch_config["apiKey"]
# Since 8.20
if "api_key" in elasticsearch_config:
opts.elasticsearch_api_key = elasticsearch_config["api_key"]
if "serverless" in elasticsearch_config:
opts.elasticsearch_serverless = elasticsearch_config.getboolean(
"serverless"
)
if "opensearch" in config:
opensearch_config = config["opensearch"]
if "hosts" in opensearch_config:
opts.opensearch_hosts = _str_to_list(opensearch_config["hosts"])
else:
raise ConfigurationError(
"hosts setting missing from the opensearch config section"
)
if "timeout" in opensearch_config:
timeout = opensearch_config.getfloat("timeout")
opts.opensearch_timeout = timeout
if "number_of_shards" in opensearch_config:
number_of_shards = opensearch_config.getint("number_of_shards")
opts.opensearch_number_of_shards = number_of_shards
if "number_of_replicas" in opensearch_config:
number_of_replicas = opensearch_config.getint("number_of_replicas")
opts.opensearch_number_of_replicas = number_of_replicas
if "index_suffix" in opensearch_config:
opts.opensearch_index_suffix = opensearch_config["index_suffix"]
if "index_prefix" in opensearch_config:
opts.opensearch_index_prefix = opensearch_config["index_prefix"]
if "monthly_indexes" in opensearch_config:
monthly = bool(opensearch_config.getboolean("monthly_indexes"))
opts.opensearch_monthly_indexes = monthly
if "ssl" in opensearch_config:
opts.opensearch_ssl = bool(opensearch_config.getboolean("ssl"))
if "cert_path" in opensearch_config:
opts.opensearch_ssl_cert_path = _expand_path(opensearch_config["cert_path"])
if "skip_certificate_verification" in opensearch_config:
opts.opensearch_skip_certificate_verification = bool(
opensearch_config.getboolean("skip_certificate_verification")
)
if "user" in opensearch_config:
opts.opensearch_username = opensearch_config["user"]
if "password" in opensearch_config:
opts.opensearch_password = opensearch_config["password"]
# Until 8.20
if "apiKey" in opensearch_config:
opts.opensearch_api_key = opensearch_config["apiKey"]
# Since 8.20
if "api_key" in opensearch_config:
opts.opensearch_api_key = opensearch_config["api_key"]
if "auth_type" in opensearch_config:
opts.opensearch_auth_type = opensearch_config["auth_type"].strip().lower()
elif "authentication_type" in opensearch_config:
opts.opensearch_auth_type = (
opensearch_config["authentication_type"].strip().lower()
)
if "aws_region" in opensearch_config:
opts.opensearch_aws_region = opensearch_config["aws_region"].strip()
if "aws_service" in opensearch_config:
opts.opensearch_aws_service = opensearch_config["aws_service"].strip()
if "splunk_hec" in config.sections():
hec_config = config["splunk_hec"]
if "url" in hec_config:
opts.hec = hec_config["url"]
else:
raise ConfigurationError(
"url setting missing from the splunk_hec config section"
)
if "token" in hec_config:
opts.hec_token = hec_config["token"]
else:
raise ConfigurationError(
"token setting missing from the splunk_hec config section"
)
if "index" in hec_config:
opts.hec_index = hec_config["index"]
else:
raise ConfigurationError(
"index setting missing from the splunk_hec config section"
)
if "skip_certificate_verification" in hec_config:
opts.hec_skip_certificate_verification = bool(
hec_config.getboolean("skip_certificate_verification", fallback=False)
)
if "kafka" in config.sections():
kafka_config = config["kafka"]
if "hosts" in kafka_config:
opts.kafka_hosts = _str_to_list(kafka_config["hosts"])
else:
raise ConfigurationError(
"hosts setting missing from the kafka config section"
)
if "user" in kafka_config:
opts.kafka_username = kafka_config["user"]
if "password" in kafka_config:
opts.kafka_password = kafka_config["password"]
if "ssl" in kafka_config:
opts.kafka_ssl = bool(kafka_config.getboolean("ssl"))
if "skip_certificate_verification" in kafka_config:
kafka_verify = bool(
kafka_config.getboolean("skip_certificate_verification")
)
opts.kafka_skip_certificate_verification = kafka_verify
if "aggregate_topic" in kafka_config:
opts.kafka_aggregate_topic = kafka_config["aggregate_topic"]
else:
raise ConfigurationError(
"aggregate_topic setting missing from the kafka config section"
)
if "failure_topic" in kafka_config:
opts.kafka_failure_topic = kafka_config["failure_topic"]
elif "forensic_topic" in kafka_config:
opts.kafka_failure_topic = kafka_config["forensic_topic"]
else:
raise ConfigurationError(
"failure_topic setting missing from the kafka config section"
)
if "smtp_tls_topic" in kafka_config:
opts.kafka_smtp_tls_topic = kafka_config["smtp_tls_topic"]
else:
raise ConfigurationError(
"smtp_tls_topic setting missing from the kafka config section"
)
if "smtp" in config.sections():
smtp_config = config["smtp"]
if "host" in smtp_config:
opts.smtp_host = smtp_config["host"]
if "user" in smtp_config:
opts.smtp_user = smtp_config["user"]
else:
raise ConfigurationError(
"user setting missing from the smtp config section"
)
if "password" in smtp_config:
opts.smtp_password = smtp_config["password"]
else:
raise ConfigurationError(
"password setting missing from the smtp config section"
)
if "from" in smtp_config:
opts.smtp_from = smtp_config["from"]
else:
logger.critical("from setting missing from the smtp config section")
elif getattr(opts, "graph_client_id", None):
# host is SMTP-only; when [msgraph] is configured, the
# summary email is sent via the same Graph mailbox connection
# instead, so host/user/password/from are not required here.
pass
else:
raise ConfigurationError(
"host setting missing from the smtp config section"
)
if "port" in smtp_config:
opts.smtp_port = smtp_config.getint("port")
if "ssl" in smtp_config:
opts.smtp_ssl = bool(smtp_config.getboolean("ssl"))
if "skip_certificate_verification" in smtp_config:
smtp_verify = bool(smtp_config.getboolean("skip_certificate_verification"))
opts.smtp_skip_certificate_verification = smtp_verify
if "to" in smtp_config:
opts.smtp_to = _str_to_list(smtp_config["to"])
else:
logger.critical("to setting missing from the smtp config section")
if "subject" in smtp_config:
opts.smtp_subject = smtp_config["subject"]
if "attachment" in smtp_config:
opts.smtp_attachment = _expand_path(smtp_config["attachment"])
if "message" in smtp_config:
opts.smtp_message = smtp_config["message"]
if "s3" in config.sections():
s3_config = config["s3"]
if "bucket" in s3_config:
opts.s3_bucket = s3_config["bucket"]
else:
raise ConfigurationError(
"bucket setting missing from the s3 config section"
)
if "path" in s3_config:
opts.s3_path = s3_config["path"]
if opts.s3_path.startswith("/"):
opts.s3_path = opts.s3_path[1:]
if opts.s3_path.endswith("/"):
opts.s3_path = opts.s3_path[:-1]
else:
opts.s3_path = ""
if "region_name" in s3_config:
opts.s3_region_name = s3_config["region_name"]
if "endpoint_url" in s3_config:
opts.s3_endpoint_url = s3_config["endpoint_url"]
if "access_key_id" in s3_config:
opts.s3_access_key_id = s3_config["access_key_id"]
if "secret_access_key" in s3_config:
opts.s3_secret_access_key = s3_config["secret_access_key"]
if "postgresql" in config.sections():
pg_config = config["postgresql"]
if "connection_string" in pg_config:
opts.postgresql_connection_string = pg_config["connection_string"]
elif "host" in pg_config:
opts.postgresql_host = pg_config["host"]
if "port" in pg_config:
opts.postgresql_port = pg_config.getint("port")
if "user" in pg_config:
opts.postgresql_user = pg_config["user"]
if "password" in pg_config:
opts.postgresql_password = pg_config["password"]
if "database" in pg_config:
opts.postgresql_database = pg_config["database"]
else:
raise ConfigurationError(
"host (or connection_string) setting missing from the "
"postgresql config section"
)
if "syslog" in config.sections():
syslog_config = config["syslog"]
if "server" in syslog_config:
opts.syslog_server = syslog_config["server"]
else:
raise ConfigurationError(
"server setting missing from the syslog config section"
)
if "port" in syslog_config:
opts.syslog_port = syslog_config["port"]
else:
opts.syslog_port = 514
if "protocol" in syslog_config:
opts.syslog_protocol = syslog_config["protocol"]
else:
opts.syslog_protocol = "udp"
if "cafile_path" in syslog_config:
opts.syslog_cafile_path = _expand_path(syslog_config["cafile_path"])
if "certfile_path" in syslog_config:
opts.syslog_certfile_path = _expand_path(syslog_config["certfile_path"])
if "keyfile_path" in syslog_config:
opts.syslog_keyfile_path = _expand_path(syslog_config["keyfile_path"])
if "timeout" in syslog_config:
opts.syslog_timeout = float(syslog_config["timeout"])
else:
opts.syslog_timeout = 5.0
if "retry_attempts" in syslog_config:
opts.syslog_retry_attempts = int(syslog_config["retry_attempts"])
else:
opts.syslog_retry_attempts = 3
if "retry_delay" in syslog_config:
opts.syslog_retry_delay = int(syslog_config["retry_delay"])
else:
opts.syslog_retry_delay = 5
if "gmail_api" in config.sections():
gmail_api_config = config["gmail_api"]
gmail_creds = gmail_api_config.get("credentials_file")
opts.gmail_api_credentials_file = (
_expand_path(gmail_creds) if gmail_creds else gmail_creds
)
opts.gmail_api_token_file = _expand_path(
gmail_api_config.get("token_file", ".token")
)
opts.gmail_api_include_spam_trash = bool(
gmail_api_config.getboolean("include_spam_trash", False)
)
opts.gmail_api_paginate_messages = bool(
gmail_api_config.getboolean("paginate_messages", True)
)
default_gmail_api_scope = "https://www.googleapis.com/auth/gmail.modify"
opts.gmail_api_scopes = gmail_api_config.get("scopes", default_gmail_api_scope)
opts.gmail_api_scopes = _str_to_list(opts.gmail_api_scopes)
if "oauth2_port" in gmail_api_config:
opts.gmail_api_oauth2_port = gmail_api_config.getint("oauth2_port", 8080)
if "auth_mode" in gmail_api_config:
opts.gmail_api_auth_mode = gmail_api_config["auth_mode"].strip()
if "service_account_user" in gmail_api_config:
opts.gmail_api_service_account_user = gmail_api_config[
"service_account_user"
].strip()
elif "delegated_user" in gmail_api_config:
opts.gmail_api_service_account_user = gmail_api_config[
"delegated_user"
].strip()
if "maildir" in config.sections():
maildir_api_config = config["maildir"]
maildir_p = maildir_api_config.get(
"maildir_path", maildir_api_config.get("path")
)
opts.maildir_path = _expand_path(maildir_p) if maildir_p else maildir_p
opts.maildir_create = bool(
maildir_api_config.getboolean(
"maildir_create",
fallback=maildir_api_config.getboolean("create", fallback=False),
)
)
if "log_analytics" in config.sections():
log_analytics_config = config["log_analytics"]
opts.la_client_id = log_analytics_config.get("client_id")
opts.la_client_secret = log_analytics_config.get("client_secret")
opts.la_tenant_id = log_analytics_config.get("tenant_id")
opts.la_dce = log_analytics_config.get("dce")
opts.la_dcr_immutable_id = log_analytics_config.get("dcr_immutable_id")
opts.la_dcr_aggregate_stream = log_analytics_config.get("dcr_aggregate_stream")
opts.la_dcr_failure_stream = log_analytics_config.get(
"dcr_failure_stream"
) or log_analytics_config.get("dcr_forensic_stream")
opts.la_dcr_smtp_tls_stream = log_analytics_config.get("dcr_smtp_tls_stream")
if "gelf" in config.sections():
gelf_config = config["gelf"]
if "host" in gelf_config:
opts.gelf_host = gelf_config["host"]
else:
raise ConfigurationError(
"host setting missing from the gelf config section"
)
if "port" in gelf_config:
opts.gelf_port = gelf_config["port"]
else:
raise ConfigurationError(
"port setting missing from the gelf config section"
)
if "mode" in gelf_config:
opts.gelf_mode = gelf_config["mode"]
else:
raise ConfigurationError(
"mode setting missing from the gelf config section"
)
if "webhook" in config.sections():
webhook_config = config["webhook"]
if "aggregate_url" in webhook_config:
opts.webhook_aggregate_url = webhook_config["aggregate_url"]
if "failure_url" in webhook_config:
opts.webhook_failure_url = webhook_config["failure_url"]
elif "forensic_url" in webhook_config:
opts.webhook_failure_url = webhook_config["forensic_url"]
if "smtp_tls_url" in webhook_config:
opts.webhook_smtp_tls_url = webhook_config["smtp_tls_url"]
if "timeout" in webhook_config:
opts.webhook_timeout = webhook_config.getint("timeout")
return index_prefix_domain_map
class _ElasticsearchHandle:
"""Sentinel so Elasticsearch participates in _close_output_clients."""
def close(self):
try:
conn = elastic.connections.get_connection()
if not isinstance(conn, str):
conn.close()
except Exception:
pass
try:
elastic.connections.remove_connection("default")
except Exception:
pass
class _OpenSearchHandle:
"""Sentinel so OpenSearch participates in _close_output_clients."""
def close(self):
try:
conn = opensearch.connections.get_connection()
if not isinstance(conn, str):
conn.close()
except Exception:
pass
try:
opensearch.connections.remove_connection("default")
except Exception:
pass
def _normalize_index_prefix(prefix):
"""Normalize an ``index_prefix_domain_map`` key into an index name prefix.
Lowercases, strips surrounding whitespace and then surrounding
underscores, replaces the remaining spaces and hyphens with
underscores, and appends a trailing ``_``.
Shared by the save path (``get_index_prefix()`` in :func:`_main`) and the
migration path (:func:`_migration_index_names`) so that the indexes
parsedmarc migrates cannot drift from the ones it writes to.
A key that normalizes to the empty string (``"_"``, ``" "``) yields the
literal ``"_"``. That is deliberate: it is exactly the prefix the save
path produces for such a key, so it is the prefix its documents live
under.
The ``[elasticsearch]``/``[opensearch]`` ``index_prefix`` option is never
passed through this function -- the save path uses that option verbatim,
so the migration path must too.
Args:
prefix (str): A key from ``index_prefix_domain_map``.
Returns:
str: The index name prefix, including its trailing underscore.
"""
prefix = prefix.lower().strip().strip("_").replace(" ", "_").replace("-", "_")
return f"{prefix}_"
def _migration_index_names(
base_name, index_suffix, configured_prefix, index_prefix_domain_map
):
"""Resolve every index name an index migration should target.
Mirrors the way save time builds index names, which is
``{prefix}{base_name}_{index_suffix}-{date}``, and widens each of the
two configurable axes so that a migration cannot silently skip indexes
the deployment holds data in (issue #868).
**Suffix axis.** When ``index_suffix`` is set, the suffixed name (what
this deployment writes today) and the bare ``base_name`` are both
returned, suffixed first. The bare name covers
documents indexed before the suffix was configured, or under a previous
one; without it, ``dmarc_aggregate_prod*`` matches none of the
operator's own ``dmarc_aggregate-*`` indexes. Because callers turn each
name into an ``f"{name}*"`` pattern, the bare name's pattern is a strict
superset of the suffixed one: on a shared cluster it also matches other
deployments' suffixes, and on the first run after an upgrade both
patterns can submit an overlapping ``update_by_query``. That is safe --
the backfill scripts only set a field that is missing, and submissions
use ``conflicts="proceed"`` -- but it is a deliberate trade of
narrowness for coverage of the operator's own history.
**Prefix axis.** A configured ``index_prefix`` wins outright and
suppresses the ``index_prefix_domain_map`` fan-out, because such a
deployment writes only under that prefix and must not touch index
patterns it does not own. Truthiness decides, matching the save path's
``opts.*_index_prefix or get_index_prefix(report)``. Otherwise the
unprefixed name comes first, followed by one name per map key --
normalized by :func:`_normalize_index_prefix`, in map order. The
unprefixed name has to stay: aggregate and failure reports for a domain
that is absent from the map are still saved without a prefix, and
indexes predating the map exist for every report type.
``configured_prefix`` is used verbatim, never normalized, for parity
with the save path.
Args:
base_name (str): The unprefixed, unsuffixed index name, e.g.
``"dmarc_aggregate"``.
index_suffix (str | None): The configured ``index_suffix``, or
``None``/``""`` when none is configured.
configured_prefix (str | None): The configured ``index_prefix``, or
``None``/``""`` when none is configured.
index_prefix_domain_map (dict | None): The parsed
``general.index_prefix_domain_map``, or ``None`` when
multi-tenant prefixing is not configured.
Returns:
list: Index names, deduplicated, in first-seen order. With nothing
configured this is just ``[base_name]``.
"""
bases = [base_name]
if index_suffix:
bases.insert(0, f"{base_name}_{index_suffix}")
if configured_prefix:
prefixes = [configured_prefix]
else:
prefixes = [""]
for key in index_prefix_domain_map or {}:
prefix = _normalize_index_prefix(key)
if prefix not in prefixes:
prefixes.append(prefix)
names = []
for prefix in prefixes:
for base in bases:
name = f"{prefix}{base}"
if name not in names:
names.append(name)
return names
def _init_output_clients(opts, index_prefix_domain_map=None):
"""Create output clients based on current opts.
Args:
opts: Namespace of parsed configuration values.
index_prefix_domain_map (dict | None): The parsed
``general.index_prefix_domain_map``. ``None`` -- the default --
means multi-tenant prefixing is not configured, so Elasticsearch
and OpenSearch index migrations target only the names derived
from ``index_prefix``/``index_suffix``.
Returns:
dict of client instances keyed by name.
Raises:
ConfigurationError: If a required output client cannot be created.
"""
clients = {}
try:
if opts.s3_bucket:
logger.debug("Initializing S3 client: bucket=%s", opts.s3_bucket)
clients["s3_client"] = s3.S3Client(
bucket_name=opts.s3_bucket,
bucket_path=opts.s3_path,
region_name=opts.s3_region_name,
endpoint_url=opts.s3_endpoint_url,
access_key_id=opts.s3_access_key_id,
secret_access_key=opts.s3_secret_access_key,
)
except Exception as e:
raise RuntimeError(f"S3: {e}") from e
try:
if opts.postgresql_host or opts.postgresql_connection_string:
logger.debug("Initializing PostgreSQL client")
pg_client = postgres.PostgreSQLClient(
connection_string=opts.postgresql_connection_string,
host=opts.postgresql_host,
port=int(opts.postgresql_port or 5432),
user=opts.postgresql_user,
password=opts.postgresql_password,
database=opts.postgresql_database,
)
pg_client.create_tables()
clients["postgresql_client"] = pg_client
except Exception as e:
raise RuntimeError(f"PostgreSQL: {e}") from e
try:
if opts.syslog_server:
logger.debug(
"Initializing syslog client: server=%s:%s",
opts.syslog_server,
opts.syslog_port,
)
clients["syslog_client"] = syslog.SyslogClient(
server_name=opts.syslog_server,
server_port=int(opts.syslog_port),
protocol=opts.syslog_protocol or "udp",
cafile_path=opts.syslog_cafile_path,
certfile_path=opts.syslog_certfile_path,
keyfile_path=opts.syslog_keyfile_path,
timeout=opts.syslog_timeout if opts.syslog_timeout is not None else 5.0,
retry_attempts=opts.syslog_retry_attempts
if opts.syslog_retry_attempts is not None
else 3,
retry_delay=opts.syslog_retry_delay
if opts.syslog_retry_delay is not None
else 5,
)
except Exception as e:
raise RuntimeError(f"Syslog: {e}") from e
if opts.hec:
if opts.hec_token is None or opts.hec_index is None:
raise ConfigurationError(
"HEC token and HEC index are required when using HEC URL"
)
try:
logger.debug("Initializing Splunk HEC client: url=%s", opts.hec)
verify = True
if opts.hec_skip_certificate_verification:
verify = False
clients["hec_client"] = splunk.HECClient(
opts.hec, opts.hec_token, opts.hec_index, verify=verify
)
except Exception as e:
raise RuntimeError(f"Splunk HEC: {e}") from e
try:
if opts.kafka_hosts:
logger.debug("Initializing Kafka client: hosts=%s", opts.kafka_hosts)
ssl_context = None
if opts.kafka_skip_certificate_verification:
logger.debug("Skipping Kafka certificate verification")
ssl_context = create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = CERT_NONE
clients["kafka_client"] = kafkaclient.KafkaClient(
opts.kafka_hosts,
username=opts.kafka_username,
password=opts.kafka_password,
ssl_context=ssl_context,
)
except Exception as e:
raise RuntimeError(f"Kafka: {e}") from e
try:
if opts.gelf_host:
logger.debug(
"Initializing GELF client: host=%s:%s",
opts.gelf_host,
opts.gelf_port,
)
clients["gelf_client"] = gelf.GelfClient(
host=opts.gelf_host,
port=int(opts.gelf_port),
mode=opts.gelf_mode,
)
except Exception as e:
raise RuntimeError(f"GELF: {e}") from e
try:
if (
opts.webhook_aggregate_url
or opts.webhook_failure_url
or opts.webhook_smtp_tls_url
):
logger.debug("Initializing webhook client")
clients["webhook_client"] = webhook.WebhookClient(
aggregate_url=opts.webhook_aggregate_url,
failure_url=opts.webhook_failure_url,
smtp_tls_url=opts.webhook_smtp_tls_url,
timeout=opts.webhook_timeout,
)
except Exception as e:
raise RuntimeError(f"Webhook: {e}") from e
# Elasticsearch and OpenSearch mutate module-level global state via
# connections.create_connection(), which cannot be rolled back if a later
# step fails. Initialise them last so that all other clients are created
# successfully first; this minimizes the window for partial-init problems
# during config reload.
if opts.save_aggregate or opts.save_failure or opts.save_smtp_tls:
try:
if opts.elasticsearch_hosts:
logger.debug(
"Initializing Elasticsearch client: hosts=%s, ssl=%s",
opts.elasticsearch_hosts,
opts.elasticsearch_ssl,
)
es_aggregate_indexes = _migration_index_names(
"dmarc_aggregate",
opts.elasticsearch_index_suffix,
opts.elasticsearch_index_prefix,
index_prefix_domain_map,
)
es_failure_indexes = _migration_index_names(
"dmarc_failure",
opts.elasticsearch_index_suffix,
opts.elasticsearch_index_prefix,
index_prefix_domain_map,
)
es_smtp_tls_indexes = _migration_index_names(
"smtp_tls",
opts.elasticsearch_index_suffix,
opts.elasticsearch_index_prefix,
index_prefix_domain_map,
)
# The legacy published_policy.fo migration gets the same
# names minus the tenant fan-out: index_prefix_domain_map
# arrived in 8.19.0, long after 5.0.0 fixed the mapping, so
# no index it names can carry the old one.
es_legacy_fo_indexes = _migration_index_names(
"dmarc_aggregate",
opts.elasticsearch_index_suffix,
opts.elasticsearch_index_prefix,
None,
)
elastic_timeout_value = (
float(opts.elasticsearch_timeout)
if opts.elasticsearch_timeout is not None
else 60.0
)
elastic.set_hosts(
opts.elasticsearch_hosts,
use_ssl=opts.elasticsearch_ssl,
ssl_cert_path=opts.elasticsearch_ssl_cert_path,
skip_certificate_verification=opts.elasticsearch_skip_certificate_verification,
username=opts.elasticsearch_username,
password=opts.elasticsearch_password,
api_key=opts.elasticsearch_api_key,
timeout=elastic_timeout_value,
serverless=opts.elasticsearch_serverless,
)
logger.debug(
"Elasticsearch index migration targets: aggregate=%s, "
"failure=%s, smtp_tls=%s, legacy_fo=%s",
es_aggregate_indexes,
es_failure_indexes,
es_smtp_tls_indexes,
es_legacy_fo_indexes,
)
elastic.migrate_indexes(
aggregate_indexes=es_aggregate_indexes,
failure_indexes=es_failure_indexes,
smtp_tls_indexes=es_smtp_tls_indexes,
legacy_fo_indexes=es_legacy_fo_indexes,
)
clients["elasticsearch"] = _ElasticsearchHandle()
except Exception as e:
raise RuntimeError(f"Elasticsearch: {e}") from e
try:
if opts.opensearch_hosts:
logger.debug(
"Initializing OpenSearch client: hosts=%s, ssl=%s",
opts.opensearch_hosts,
opts.opensearch_ssl,
)
os_aggregate_indexes = _migration_index_names(
"dmarc_aggregate",
opts.opensearch_index_suffix,
opts.opensearch_index_prefix,
index_prefix_domain_map,
)
os_failure_indexes = _migration_index_names(
"dmarc_failure",
opts.opensearch_index_suffix,
opts.opensearch_index_prefix,
index_prefix_domain_map,
)
os_smtp_tls_indexes = _migration_index_names(
"smtp_tls",
opts.opensearch_index_suffix,
opts.opensearch_index_prefix,
index_prefix_domain_map,
)
# The legacy published_policy.fo migration gets the same
# names minus the tenant fan-out: index_prefix_domain_map
# arrived in 8.19.0, long after 5.0.0 fixed the mapping, so
# no index it names can carry the old one.
os_legacy_fo_indexes = _migration_index_names(
"dmarc_aggregate",
opts.opensearch_index_suffix,
opts.opensearch_index_prefix,
None,
)
opensearch_timeout_value = (
float(opts.opensearch_timeout)
if opts.opensearch_timeout is not None
else 60.0
)
opensearch.set_hosts(
opts.opensearch_hosts,
use_ssl=opts.opensearch_ssl,
ssl_cert_path=opts.opensearch_ssl_cert_path,
skip_certificate_verification=opts.opensearch_skip_certificate_verification,
username=opts.opensearch_username,
password=opts.opensearch_password,
api_key=opts.opensearch_api_key,
timeout=opensearch_timeout_value,
auth_type=opts.opensearch_auth_type,
aws_region=opts.opensearch_aws_region,
aws_service=opts.opensearch_aws_service,
)
logger.debug(
"OpenSearch index migration targets: aggregate=%s, "
"failure=%s, smtp_tls=%s, legacy_fo=%s",
os_aggregate_indexes,
os_failure_indexes,
os_smtp_tls_indexes,
os_legacy_fo_indexes,
)
opensearch.migrate_indexes(
aggregate_indexes=os_aggregate_indexes,
failure_indexes=os_failure_indexes,
smtp_tls_indexes=os_smtp_tls_indexes,
legacy_fo_indexes=os_legacy_fo_indexes,
)
clients["opensearch"] = _OpenSearchHandle()
except Exception as e:
raise RuntimeError(f"OpenSearch: {e}") from e
return clients
def _close_output_clients(clients):
"""Close output clients that hold persistent connections.
Clients that do not expose a ``close`` method are silently skipped.
Errors during closing are logged as warnings and do not propagate.
Idempotent: each client is popped as it is closed, so a second call
(e.g. the trailing close plus the atexit safety net) is a no-op.
Args:
clients: dict of client instances returned by :func:`_init_output_clients`.
"""
while clients:
name, client = clients.popitem()
if hasattr(client, "close"):
try:
client.close()
except Exception:
logger.warning("Error closing %s", name, exc_info=True)
def _build_parser_config(opts: Namespace) -> ParserConfig:
"""Builds the single ParserConfig for this run from parsed opts, bound to
the process-wide default caches (the parsedmarc module globals).
"""
return ParserConfig(
offline=opts.offline,
ip_db_path=opts.ip_db_path,
always_use_local_files=opts.always_use_local_files,
reverse_dns_map_path=opts.reverse_dns_map_path,
reverse_dns_map_url=opts.reverse_dns_map_url,
psl_overrides_path=opts.psl_overrides_path,
psl_overrides_url=opts.psl_overrides_url,
nameservers=opts.nameservers,
dns_timeout=(
float(opts.dns_timeout)
if opts.dns_timeout is not None
else DEFAULT_DNS_TIMEOUT
),
dns_retries=(
int(opts.dns_retries)
if opts.dns_retries is not None
else DEFAULT_DNS_MAX_RETRIES
),
strip_attachment_payloads=opts.strip_attachment_payloads,
normalize_timespan_threshold_hours=(
float(opts.normalize_timespan_threshold_hours)
if opts.normalize_timespan_threshold_hours is not None
else 24.0
),
ip_address_cache=IP_ADDRESS_CACHE,
seen_aggregate_report_ids=SEEN_AGGREGATE_REPORT_IDS,
reverse_dns_map=REVERSE_DNS_MAP,
)
def _main():
"""Called when the module is executed"""
def get_index_prefix(report):
domain = None
if index_prefix_domain_map is None:
return None
if "policy_published" in report:
domain = report["policy_published"]["domain"]
elif "reported_domain" in report:
domain = report["reported_domain"]
elif report.get("policies"):
# Guarded with .get() truthiness: parse_smtp_tls_report_json()
# accepts a report whose policies list is empty, which would
# make [0] raise IndexError here. Such a report has no domain
# to map, so it falls through to return None like any other
# unmappable report.
domain = report["policies"][0]["policy_domain"]
if domain:
domain = get_base_domain(domain)
if domain:
domain = domain.lower()
for key in index_prefix_domain_map:
if domain in index_prefix_domain_map[key]:
return _normalize_index_prefix(key)
return None
def filter_smtp_tls_reports_for_index_prefix(tls_reports):
"""Drop SMTP TLS reports whose domain isn't covered by
``index_prefix_domain_map``.
Shared by ``process_reports()`` (which filters each batch it saves)
and by the combined ``parsing_results`` that feeds
``email_results()``. Mailbox batches are saved inside
``get_dmarc_reports_from_mailbox()``, so the dicts
``process_reports()`` filters in place are no longer the same
objects as the combined results assembled afterward -- without this,
the emailed summary would list SMTP TLS reports that were
deliberately excluded from every output destination.
"""
if index_prefix_domain_map is None:
return tls_reports
filtered_tls = []
for report in tls_reports:
if get_index_prefix(report) is not None:
filtered_tls.append(report)
else:
domain = "unknown"
if "policies" in report and report["policies"]:
domain = report["policies"][0].get("policy_domain", "unknown")
logger.debug(
"Ignoring SMTP TLS report for domain not in "
"index_prefix_domain_map: %s",
domain,
)
return filtered_tls
def process_reports(reports_):
"""Write ``reports_`` to every configured output destination.
Returns the list of human-readable output-error messages recorded
along the way -- empty when every destination accepted the reports.
Callers use that as the "was this batch saved?" signal; see
``mailbox_save_callback()``.
"""
output_errors = []
def log_output_error(destination, error):
message = f"{destination} Error: {error}"
logger.error(message)
output_errors.append(message)
if index_prefix_domain_map is not None:
reports_["smtp_tls_reports"] = filter_smtp_tls_reports_for_index_prefix(
reports_.get("smtp_tls_reports", [])
)
indent_value = 2 if opts.prettify_json else None
output_str = (
f"{json.dumps(reports_, ensure_ascii=False, indent=indent_value)}\n"
)
if not opts.silent:
print(output_str)
if opts.output:
try:
save_output(
reports_,
output_directory=opts.output,
aggregate_json_filename=opts.aggregate_json_filename,
failure_json_filename=opts.failure_json_filename,
smtp_tls_json_filename=opts.smtp_tls_json_filename,
aggregate_csv_filename=opts.aggregate_csv_filename,
failure_csv_filename=opts.failure_csv_filename,
smtp_tls_csv_filename=opts.smtp_tls_csv_filename,
)
except (OSError, ValueError) as error_:
# The only output destination that was not already caught:
# a full disk or an unwritable directory used to crash the
# run outright, and now that a failed save holds mailbox
# messages back it also has to be recorded like any other
# destination's failure rather than escaping.
log_output_error("File output", str(error_))
kafka_client = clients.get("kafka_client")
s3_client = clients.get("s3_client")
syslog_client = clients.get("syslog_client")
hec_client = clients.get("hec_client")
gelf_client = clients.get("gelf_client")
webhook_client = clients.get("webhook_client")
pg_client = clients.get("postgresql_client")
kafka_aggregate_topic = opts.kafka_aggregate_topic
kafka_failure_topic = opts.kafka_failure_topic
kafka_smtp_tls_topic = opts.kafka_smtp_tls_topic
if opts.save_aggregate:
for report in reports_["aggregate_reports"]:
try:
if opts.elasticsearch_hosts:
shards = opts.elasticsearch_number_of_shards
replicas = opts.elasticsearch_number_of_replicas
elastic.save_aggregate_report_to_elasticsearch(
report,
index_suffix=opts.elasticsearch_index_suffix,
index_prefix=opts.elasticsearch_index_prefix
or get_index_prefix(report),
monthly_indexes=opts.elasticsearch_monthly_indexes,
number_of_shards=shards,
number_of_replicas=replicas,
)
except elastic.AlreadySaved as warning:
logger.warning(warning.__str__())
except elastic.ElasticsearchError as error_:
log_output_error("Elasticsearch", error_.__str__())
except Exception as error_:
log_output_error("Elasticsearch exception", error_.__str__())
try:
if opts.opensearch_hosts:
shards = opts.opensearch_number_of_shards
replicas = opts.opensearch_number_of_replicas
opensearch.save_aggregate_report_to_opensearch(
report,
index_suffix=opts.opensearch_index_suffix,
index_prefix=opts.opensearch_index_prefix
or get_index_prefix(report),
monthly_indexes=opts.opensearch_monthly_indexes,
number_of_shards=shards,
number_of_replicas=replicas,
)
except opensearch.AlreadySaved as warning:
logger.warning(warning.__str__())
except opensearch.OpenSearchError as error_:
log_output_error("OpenSearch", error_.__str__())
except Exception as error_:
log_output_error("OpenSearch exception", error_.__str__())
try:
if kafka_client:
kafka_client.save_aggregate_reports_to_kafka(
report, kafka_aggregate_topic
)
except Exception as error_:
log_output_error("Kafka", error_.__str__())
try:
if s3_client:
s3_client.save_aggregate_report_to_s3(report)
except Exception as error_:
log_output_error("S3", error_.__str__())
try:
if pg_client:
pg_client.save_aggregate_report_to_postgresql(report)
except postgres.AlreadySaved as warning:
logger.warning(warning.__str__())
except postgres.PostgreSQLError as error_:
log_output_error("PostgreSQL", error_.__str__())
try:
if syslog_client:
syslog_client.save_aggregate_report_to_syslog(report)
except Exception as error_:
log_output_error("Syslog", error_.__str__())
try:
if gelf_client:
gelf_client.save_aggregate_report_to_gelf(report)
except Exception as error_:
log_output_error("GELF", error_.__str__())
try:
if opts.webhook_aggregate_url and webhook_client:
indent_value = 2 if opts.prettify_json else None
webhook_client.save_aggregate_report_to_webhook(
json.dumps(report, ensure_ascii=False, indent=indent_value)
)
except Exception as error_:
log_output_error("Webhook", error_.__str__())
if hec_client:
try:
aggregate_reports_ = reports_["aggregate_reports"]
if len(aggregate_reports_) > 0:
hec_client.save_aggregate_reports_to_splunk(aggregate_reports_)
except splunk.SplunkError as e:
log_output_error("Splunk HEC", e.__str__())
if opts.save_failure:
for report in reports_["failure_reports"]:
try:
shards = opts.elasticsearch_number_of_shards
replicas = opts.elasticsearch_number_of_replicas
if opts.elasticsearch_hosts:
elastic.save_failure_report_to_elasticsearch(
report,
index_suffix=opts.elasticsearch_index_suffix,
index_prefix=opts.elasticsearch_index_prefix
or get_index_prefix(report),
monthly_indexes=opts.elasticsearch_monthly_indexes,
number_of_shards=shards,
number_of_replicas=replicas,
)
except elastic.AlreadySaved as warning:
logger.warning(warning.__str__())
except elastic.ElasticsearchError as error_:
log_output_error("Elasticsearch", error_.__str__())
except InvalidDMARCReport as error_:
log_output_error("Invalid DMARC report", error_.__str__())
try:
shards = opts.opensearch_number_of_shards
replicas = opts.opensearch_number_of_replicas
if opts.opensearch_hosts:
opensearch.save_failure_report_to_opensearch(
report,
index_suffix=opts.opensearch_index_suffix,
index_prefix=opts.opensearch_index_prefix
or get_index_prefix(report),
monthly_indexes=opts.opensearch_monthly_indexes,
number_of_shards=shards,
number_of_replicas=replicas,
)
except opensearch.AlreadySaved as warning:
logger.warning(warning.__str__())
except opensearch.OpenSearchError as error_:
log_output_error("OpenSearch", error_.__str__())
except InvalidDMARCReport as error_:
log_output_error("Invalid DMARC report", error_.__str__())
try:
if kafka_client:
kafka_client.save_failure_reports_to_kafka(
report, kafka_failure_topic
)
except Exception as error_:
log_output_error("Kafka", error_.__str__())
try:
if s3_client:
s3_client.save_failure_report_to_s3(report)
except Exception as error_:
log_output_error("S3", error_.__str__())
try:
if pg_client:
pg_client.save_failure_report_to_postgresql(report)
except postgres.AlreadySaved as warning:
logger.warning(warning.__str__())
except postgres.PostgreSQLError as error_:
log_output_error("PostgreSQL", error_.__str__())
try:
if syslog_client:
syslog_client.save_failure_report_to_syslog(report)
except Exception as error_:
log_output_error("Syslog", error_.__str__())
try:
if gelf_client:
gelf_client.save_failure_report_to_gelf(report)
except Exception as error_:
log_output_error("GELF", error_.__str__())
try:
if opts.webhook_failure_url and webhook_client:
indent_value = 2 if opts.prettify_json else None
webhook_client.save_failure_report_to_webhook(
json.dumps(report, ensure_ascii=False, indent=indent_value)
)
except Exception as error_:
log_output_error("Webhook", error_.__str__())
if hec_client:
try:
failure_reports_ = reports_["failure_reports"]
if len(failure_reports_) > 0:
hec_client.save_failure_reports_to_splunk(failure_reports_)
except splunk.SplunkError as e:
log_output_error("Splunk HEC", e.__str__())
if opts.save_smtp_tls:
for report in reports_["smtp_tls_reports"]:
try:
shards = opts.elasticsearch_number_of_shards
replicas = opts.elasticsearch_number_of_replicas
if opts.elasticsearch_hosts:
elastic.save_smtp_tls_report_to_elasticsearch(
report,
index_suffix=opts.elasticsearch_index_suffix,
index_prefix=opts.elasticsearch_index_prefix
or get_index_prefix(report),
monthly_indexes=opts.elasticsearch_monthly_indexes,
number_of_shards=shards,
number_of_replicas=replicas,
)
except elastic.AlreadySaved as warning:
logger.warning(warning.__str__())
except elastic.ElasticsearchError as error_:
log_output_error("Elasticsearch", error_.__str__())
except InvalidDMARCReport as error_:
log_output_error("Invalid DMARC report", error_.__str__())
try:
shards = opts.opensearch_number_of_shards
replicas = opts.opensearch_number_of_replicas
if opts.opensearch_hosts:
opensearch.save_smtp_tls_report_to_opensearch(
report,
index_suffix=opts.opensearch_index_suffix,
index_prefix=opts.opensearch_index_prefix
or get_index_prefix(report),
monthly_indexes=opts.opensearch_monthly_indexes,
number_of_shards=shards,
number_of_replicas=replicas,
)
except opensearch.AlreadySaved as warning:
logger.warning(warning.__str__())
except opensearch.OpenSearchError as error_:
log_output_error("OpenSearch", error_.__str__())
except InvalidDMARCReport as error_:
log_output_error("Invalid DMARC report", error_.__str__())
try:
if kafka_client:
kafka_client.save_smtp_tls_reports_to_kafka(
[report], kafka_smtp_tls_topic
)
except Exception as error_:
log_output_error("Kafka", error_.__str__())
try:
if s3_client:
s3_client.save_smtp_tls_report_to_s3(report)
except Exception as error_:
log_output_error("S3", error_.__str__())
try:
if pg_client:
pg_client.save_smtp_tls_report_to_postgresql(report)
except postgres.AlreadySaved as warning:
logger.warning(warning.__str__())
except postgres.PostgreSQLError as error_:
log_output_error("PostgreSQL", error_.__str__())
try:
if syslog_client:
syslog_client.save_smtp_tls_report_to_syslog(report)
except Exception as error_:
log_output_error("Syslog", error_.__str__())
try:
if gelf_client:
gelf_client.save_smtp_tls_report_to_gelf(report)
except Exception as error_:
log_output_error("GELF", error_.__str__())
try:
if opts.webhook_smtp_tls_url and webhook_client:
indent_value = 2 if opts.prettify_json else None
webhook_client.save_smtp_tls_report_to_webhook(
json.dumps(report, ensure_ascii=False, indent=indent_value)
)
except Exception as error_:
log_output_error("Webhook", error_.__str__())
if hec_client:
try:
smtp_tls_reports_ = reports_["smtp_tls_reports"]
if len(smtp_tls_reports_) > 0:
hec_client.save_smtp_tls_reports_to_splunk(smtp_tls_reports_)
except splunk.SplunkError as e:
log_output_error("Splunk HEC", e.__str__())
if opts.la_dce:
try:
la_client = loganalytics.LogAnalyticsClient(
client_id=opts.la_client_id,
client_secret=opts.la_client_secret,
tenant_id=opts.la_tenant_id,
dce=opts.la_dce,
dcr_immutable_id=opts.la_dcr_immutable_id,
dcr_aggregate_stream=opts.la_dcr_aggregate_stream,
dcr_failure_stream=opts.la_dcr_failure_stream,
dcr_smtp_tls_stream=opts.la_dcr_smtp_tls_stream,
)
la_client.publish_results(
reports_,
opts.save_aggregate,
opts.save_failure,
opts.save_smtp_tls,
)
except loganalytics.LogAnalyticsException as e:
log_output_error("Log Analytics", e.__str__())
except Exception as e:
log_output_error("Log Analytics", f"Unknown publishing error: {e}")
if opts.fail_on_output_error and output_errors:
raise ParserError(
"Output destination failures detected: {}".format(
" | ".join(output_errors)
)
)
return output_errors
arg_parser = ArgumentParser(description="Parses DMARC reports")
arg_parser.add_argument(
"-c",
"--config-file",
help="a path to a configuration file (--silent implied)",
)
arg_parser.add_argument(
"file_path",
nargs="*",
help="one or more paths to aggregate or failure report files, "
"emails, mbox files, or directories containing them",
)
arg_parser.add_argument(
"-r",
"--recursive",
action="store_true",
help="search directories given as file_path recursively, and "
"enable '**' recursion in glob patterns",
)
strip_attachment_help = "remove attachment payloads from failure report output"
arg_parser.add_argument(
"--strip-attachment-payloads", help=strip_attachment_help, action="store_true"
)
arg_parser.add_argument(
"-o", "--output", help="write output files to the given directory"
)
arg_parser.add_argument(
"--aggregate-json-filename",
help="filename for the aggregate JSON output file",
default="aggregate.json",
)
arg_parser.add_argument(
"--failure-json-filename",
help="filename for the failure JSON output file",
default="failure.json",
)
arg_parser.add_argument(
"--smtp-tls-json-filename",
help="filename for the SMTP TLS JSON output file",
default="smtp_tls.json",
)
arg_parser.add_argument(
"--aggregate-csv-filename",
help="filename for the aggregate CSV output file",
default="aggregate.csv",
)
arg_parser.add_argument(
"--failure-csv-filename",
help="filename for the failure CSV output file",
default="failure.csv",
)
arg_parser.add_argument(
"--smtp-tls-csv-filename",
help="filename for the SMTP TLS CSV output file",
default="smtp_tls.csv",
)
arg_parser.add_argument(
"-n", "--nameservers", nargs="+", help="nameservers to query"
)
arg_parser.add_argument(
"-t",
"--dns_timeout",
help="number of seconds to wait for an answer from DNS (default: 2.0)",
type=float,
default=2.0,
)
arg_parser.add_argument(
"--dns-retries",
dest="dns_retries",
help="number of times to retry DNS queries on timeout or other "
"transient errors (default: 0)",
type=int,
default=0,
)
arg_parser.add_argument(
"--offline",
action="store_true",
help="do not make online queries for geolocation or DNS",
)
arg_parser.add_argument(
"-s", "--silent", action="store_true", help="only print errors"
)
arg_parser.add_argument(
"-w",
"--warnings",
action="store_true",
help="print warnings in addition to errors",
)
arg_parser.add_argument(
"--verbose", action="store_true", help="more verbose output"
)
arg_parser.add_argument(
"--debug", action="store_true", help="print debugging information"
)
arg_parser.add_argument("--log-file", default=None, help="output logging to a file")
arg_parser.add_argument(
"--no-prettify-json",
action="store_false",
dest="prettify_json",
help="output JSON in a single line without indentation",
)
arg_parser.add_argument("-v", "--version", action="version", version=__version__)
aggregate_reports = []
failure_reports = []
smtp_tls_reports = []
args = arg_parser.parse_args()
opts = Namespace(
file_path=args.file_path,
config_file=args.config_file,
offline=args.offline,
strip_attachment_payloads=args.strip_attachment_payloads,
output=args.output,
aggregate_csv_filename=args.aggregate_csv_filename,
aggregate_json_filename=args.aggregate_json_filename,
failure_csv_filename=args.failure_csv_filename,
failure_json_filename=args.failure_json_filename,
smtp_tls_json_filename=args.smtp_tls_json_filename,
smtp_tls_csv_filename=args.smtp_tls_csv_filename,
nameservers=args.nameservers,
dns_test_address="1.1.1.1",
silent=args.silent,
warnings=args.warnings,
dns_timeout=args.dns_timeout,
dns_retries=args.dns_retries,
debug=args.debug,
verbose=args.verbose,
prettify_json=args.prettify_json,
save_aggregate=False,
save_failure=False,
save_smtp_tls=False,
mailbox_reports_folder="INBOX",
mailbox_archive_folder="Archive",
mailbox_watch=False,
mailbox_delete=False,
# None means "unset": each per-report-type flag inherits mailbox_delete
# in get_dmarc_reports_from_mailbox, so an explicit False (opting one
# type out of a global delete = true) stays distinct from being unset.
mailbox_delete_aggregate=None,
mailbox_delete_failure=None,
mailbox_delete_smtp_tls=None,
mailbox_delete_invalid=None,
mailbox_test=False,
mailbox_batch_size=10,
mailbox_check_timeout=30,
mailbox_max_unsaved_retries=2,
mailbox_since=None,
imap_host=None,
imap_skip_certificate_verification=False,
imap_ssl=True,
imap_port=993,
imap_timeout=30,
imap_max_retries=4,
imap_user=None,
imap_password=None,
graph_auth_method=None,
graph_user=None,
graph_password=None,
graph_client_id=None,
graph_client_secret=None,
graph_certificate_path=None,
graph_certificate_password=None,
graph_client_assertion=None,
graph_tenant_id=None,
graph_mailbox=None,
graph_allow_unencrypted_storage=False,
graph_url="https://graph.microsoft.com",
hec=None,
hec_token=None,
hec_index=None,
hec_skip_certificate_verification=False,
elasticsearch_hosts=None,
elasticsearch_timeout=60,
elasticsearch_number_of_shards=1,
elasticsearch_number_of_replicas=0,
elasticsearch_index_suffix=None,
elasticsearch_index_prefix=None,
elasticsearch_ssl=True,
elasticsearch_ssl_cert_path=None,
elasticsearch_skip_certificate_verification=False,
elasticsearch_monthly_indexes=False,
elasticsearch_username=None,
elasticsearch_password=None,
elasticsearch_api_key=None,
elasticsearch_serverless=False,
opensearch_hosts=None,
opensearch_timeout=60,
opensearch_number_of_shards=1,
opensearch_number_of_replicas=0,
opensearch_index_suffix=None,
opensearch_index_prefix=None,
opensearch_ssl=True,
opensearch_ssl_cert_path=None,
opensearch_skip_certificate_verification=False,
opensearch_monthly_indexes=False,
opensearch_username=None,
opensearch_password=None,
opensearch_api_key=None,
opensearch_auth_type="basic",
opensearch_aws_region=None,
opensearch_aws_service="es",
kafka_hosts=None,
kafka_username=None,
kafka_password=None,
kafka_aggregate_topic=None,
kafka_failure_topic=None,
kafka_smtp_tls_topic=None,
kafka_ssl=False,
kafka_skip_certificate_verification=False,
smtp_host=None,
smtp_port=25,
smtp_ssl=False,
smtp_skip_certificate_verification=False,
smtp_user=None,
smtp_password=None,
smtp_from=None,
smtp_to=[],
smtp_subject="parsedmarc report",
smtp_attachment=None,
smtp_message="Please see the attached DMARC results.",
s3_bucket=None,
s3_path=None,
s3_region_name=None,
s3_endpoint_url=None,
s3_access_key_id=None,
s3_secret_access_key=None,
syslog_server=None,
syslog_port=None,
syslog_protocol=None,
syslog_cafile_path=None,
syslog_certfile_path=None,
syslog_keyfile_path=None,
syslog_timeout=None,
syslog_retry_attempts=None,
syslog_retry_delay=None,
gmail_api_credentials_file=None,
gmail_api_token_file=None,
gmail_api_include_spam_trash=False,
gmail_api_paginate_messages=True,
gmail_api_scopes=[],
gmail_api_oauth2_port=8080,
gmail_api_auth_mode="installed_app",
gmail_api_service_account_user=None,
maildir_path=None,
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,
always_use_local_files=False,
reverse_dns_map_path=None,
reverse_dns_map_url=None,
psl_overrides_path=None,
psl_overrides_url=None,
la_client_id=None,
la_client_secret=None,
la_tenant_id=None,
la_dce=None,
la_dcr_immutable_id=None,
la_dcr_aggregate_stream=None,
la_dcr_failure_stream=None,
la_dcr_smtp_tls_stream=None,
gelf_host=None,
gelf_port=None,
gelf_mode=None,
webhook_aggregate_url=None,
webhook_failure_url=None,
webhook_smtp_tls_url=None,
webhook_timeout=60,
normalize_timespan_threshold_hours=24.0,
postgresql_host=None,
postgresql_port=5432,
postgresql_user=None,
postgresql_password=None,
postgresql_database=None,
postgresql_connection_string=None,
fail_on_output_error=False,
)
# Snapshot opts as set from CLI args / hardcoded defaults, before any config
# file is applied. On each SIGHUP reload we restore this baseline first so
# that sections removed from the config file actually take effect.
opts_from_cli = Namespace(**vars(opts))
index_prefix_domain_map = None
config_file = args.config_file or os.environ.get("PARSEDMARC_CONFIG_FILE")
has_env_config = any(
k.startswith("PARSEDMARC_") and k != "PARSEDMARC_CONFIG_FILE"
for k in os.environ
)
if config_file or has_env_config:
try:
config = _load_config(config_file)
index_prefix_domain_map = _parse_config(config, opts)
except ConfigurationError as e:
logger.critical(str(e))
exit(-1)
logger.setLevel(logging.ERROR)
if opts.warnings:
logger.setLevel(logging.WARNING)
if opts.verbose:
logger.setLevel(logging.INFO)
if opts.debug:
logger.setLevel(logging.DEBUG)
if opts.log_file:
try:
fh = logging.FileHandler(opts.log_file, "a")
formatter = logging.Formatter(
"%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s"
)
fh.setFormatter(formatter)
logger.addHandler(fh)
except Exception as error:
logger.warning(f"Unable to write to log file: {error}")
opts.active_log_file = opts.log_file
_configure_dependency_logging(logger.level)
if (
opts.imap_host is None
and opts.graph_client_id is None
and opts.gmail_api_credentials_file is None
and opts.maildir_path is None
and len(opts.file_path) == 0
):
logger.error("You must supply input files or a mailbox connection")
exit(1)
logger.info("Starting parsedmarc")
load_ip_db(
always_use_local_file=opts.always_use_local_files,
local_file_path=opts.ip_db_path,
url=opts.ipinfo_url,
offline=opts.offline,
)
if opts.ipinfo_api_token and not opts.offline:
try:
configure_ipinfo_api(opts.ipinfo_api_token)
except InvalidIPinfoAPIKey as e:
logger.critical(str(e))
exit(1)
load_psl_overrides(
always_use_local_file=opts.always_use_local_files,
local_file_path=opts.psl_overrides_path,
url=opts.psl_overrides_url,
offline=opts.offline,
)
# Initialize output clients (with retry for transient connection errors)
clients = {}
max_retries = 4
retry_delay = 5
for attempt in range(max_retries + 1):
try:
clients = _init_output_clients(
opts, index_prefix_domain_map=index_prefix_domain_map
)
break
except ConfigurationError as e:
logger.critical(str(e))
exit(1)
except Exception as error_:
if attempt < max_retries:
logger.warning(
"Output client error (attempt %d/%d, retrying in %ds): %s",
attempt + 1,
max_retries + 1,
retry_delay,
error_,
)
time.sleep(retry_delay)
retry_delay *= 2
else:
logger.error(f"Output client error: {error_}")
exit(1)
# Always close output clients on the way out (normal return,
# exit(N), uncaught exception, or SystemExit from a signal-driven
# shutdown). atexit does NOT fire on os._exit(130) — that's
# intentional for the SIGINT double-tap. The lambda closes whatever
# `clients` currently points at, so a SIGHUP reload that swaps the
# dict in-place is still covered.
atexit.register(lambda: _close_output_clients(clients))
# Signal handlers set a cooperative flag polled at safe checkpoints:
# the one-shot loops check it between batches; the watch loop relies
# on the mailbox backend polling `config_reloading` (which includes
# this flag) between checks, including inside the IMAP IDLE loop, so
# the current batch finishes before the watcher exits. SIGINT is a
# "double tap": the first press is graceful, the second short-circuits
# to os._exit(130). os._exit is async-signal-safe; sys.exit and
# logging are not, so the handlers only set flags / call os._exit.
_reload_requested = False
_shutdown_requested = False
_sigint_count = 0
def _handle_sighup(signum, frame):
nonlocal _reload_requested
_reload_requested = True
def _handle_sigterm(signum, frame):
nonlocal _shutdown_requested
_shutdown_requested = True
def _handle_sigint(signum, frame):
nonlocal _shutdown_requested, _sigint_count
_sigint_count += 1
if _sigint_count >= 2:
os._exit(130)
_shutdown_requested = True
if hasattr(signal, "SIGHUP"):
signal.signal(signal.SIGHUP, _handle_sighup)
signal.signal(signal.SIGTERM, _handle_sigterm)
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:
if is_mbox(file_path):
mbox_paths.append(file_path)
file_paths = list(set(file_paths))
mbox_paths = list(set(mbox_paths))
for mbox_path in mbox_paths:
file_paths.remove(mbox_path)
pbar = None
if sys.stderr.isatty() and len(file_paths) > 0:
pbar = tqdm(total=len(file_paths))
n_procs = int(opts.n_procs or 1)
if n_procs < 1:
n_procs = 1
parser_config = _build_parser_config(opts)
func = functools.partial(_parse_report_file_job, config=parser_config)
for file_path, result in parallel_map(
func, file_paths, n_procs, should_stop=lambda: _shutdown_requested
):
if pbar is not None:
pbar.update(1)
if isinstance(result, Exception):
logger.error(f"Failed to parse {file_path} - {result}")
else:
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
aggregate_reports.append(result["report"])
else:
logger.debug(
"Skipping duplicate aggregate report "
f"from {report_org} with ID: {report_id}"
)
elif result["report_type"] == "failure":
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()
if _shutdown_requested:
# Anything already parsed is still in aggregate_reports /
# failure_reports / smtp_tls_reports and will go through
# process_reports() in the cleanup path so we don't lose work
# the operator already paid for.
logger.info("Shutdown requested, stopping file processing early")
for mbox_path in mbox_paths:
if _shutdown_requested:
logger.info("Shutdown requested, skipping remaining mbox files")
break
reports = get_dmarc_reports_from_mbox(
mbox_path,
config=parser_config,
n_procs=n_procs,
)
aggregate_reports += reports["aggregate_reports"]
failure_reports += reports["failure_reports"]
smtp_tls_reports += reports["smtp_tls_reports"]
# Snapshot of the file/mbox-derived reports, taken before the mailbox
# block below appends anything fetched from a live mailbox connection.
# Mailbox batches are handed to process_reports() by
# mailbox_save_callback() before get_dmarc_reports_from_mailbox() even
# returns -- that is what lets it decide whether archiving is safe -- so
# the final process_reports() call runs on this snapshot only, or the
# mailbox-derived reports would be saved twice.
file_parsing_results: ParsingResults = {
"aggregate_reports": list(aggregate_reports),
"failure_reports": list(failure_reports),
"smtp_tls_reports": list(smtp_tls_reports),
}
mailbox_connection = None
msgraph_connection: MSGraphConnection | None = None
mailbox_batch_size_value = 10
mailbox_check_timeout_value = 30
mailbox_max_unsaved_retries_value = 2
if opts.imap_host:
try:
if opts.imap_user is None or opts.imap_password is None:
logger.error(
"IMAP user and password must be specified if host is specified"
)
exit(1)
ssl = True
verify = True
if opts.imap_skip_certificate_verification:
logger.debug("Skipping IMAP certificate verification")
verify = False
if not opts.imap_ssl:
ssl = False
imap_timeout = (
int(opts.imap_timeout) if opts.imap_timeout is not None else 30
)
imap_max_retries = (
int(opts.imap_max_retries) if opts.imap_max_retries is not None else 4
)
imap_port_value = int(opts.imap_port) if opts.imap_port is not None else 993
mailbox_connection = IMAPConnection(
host=opts.imap_host,
port=imap_port_value,
ssl=ssl,
verify=verify,
timeout=imap_timeout,
max_retries=imap_max_retries,
user=opts.imap_user,
password=opts.imap_password,
)
except Exception:
logger.exception("IMAP Error")
exit(1)
if opts.graph_client_id:
try:
mailbox = opts.graph_mailbox or opts.graph_user
# Redacted connection summary: enough to spot a wrong
# tenant/client/mailbox at a glance, before any network I/O,
# so a hang during credential construction leaves a trace.
# Secret values are never logged.
logger.info(
"Connecting to Microsoft Graph (auth_method=%s, tenant_id=%s, "
"client_id=%s, mailbox=%s, graph_url=%s)",
opts.graph_auth_method,
opts.graph_tenant_id,
opts.graph_client_id,
mailbox,
opts.graph_url,
)
logger.debug(
"Microsoft Graph auth details: username=%s, "
"certificate_path=%s, certificate_password %s, "
"client_secret %s, password %s, client_assertion %s, "
"token_file=%s, allow_unencrypted_storage=%s",
opts.graph_user,
opts.graph_certificate_path,
"set" if opts.graph_certificate_password else "not set",
"set" if opts.graph_client_secret else "not set",
"set" if opts.graph_password else "not set",
"set" if opts.graph_client_assertion else "not set",
opts.graph_token_file,
bool(opts.graph_allow_unencrypted_storage),
)
connect_start = time.monotonic()
mailbox_connection = MSGraphConnection(
auth_method=opts.graph_auth_method,
mailbox=mailbox,
tenant_id=opts.graph_tenant_id,
client_id=opts.graph_client_id,
client_secret=opts.graph_client_secret,
certificate_path=opts.graph_certificate_path,
certificate_password=opts.graph_certificate_password,
client_assertion=opts.graph_client_assertion,
username=opts.graph_user,
password=opts.graph_password,
token_file=opts.graph_token_file,
allow_unencrypted_storage=bool(opts.graph_allow_unencrypted_storage),
graph_url=opts.graph_url,
token_cache_name="parsedmarc",
)
# App-only methods (ClientSecret/Certificate) construct their
# credential lazily; the first token request happens on the
# first mailbox call, so failures can still surface later.
logger.info(
"Microsoft Graph connection initialized in %.2f seconds",
time.monotonic() - connect_start,
)
msgraph_connection = mailbox_connection
except (ClientAuthenticationError, APIError, httpx.HTTPError) as error:
_log_msgraph_failure(
error,
stage="connection",
mailbox=opts.graph_mailbox or opts.graph_user,
tenant_id=opts.graph_tenant_id,
auth_method=opts.graph_auth_method,
)
exit(1)
except Exception:
logger.exception("MS Graph Error")
exit(1)
if opts.gmail_api_credentials_file:
# Any effective delete flag needs the deletion scope: the per-report-type
# flags inherit mailbox_delete when unset (None), so this reduces to
# mailbox_delete alone when none of them is set. The scope is a
# mailbox-wide capability grant rather than a per-type one, so when it
# is missing every flag is turned off explicitly.
per_type_delete_opts = (
"mailbox_delete_aggregate",
"mailbox_delete_failure",
"mailbox_delete_smtp_tls",
"mailbox_delete_invalid",
)
if any(
opts.mailbox_delete if getattr(opts, name) is None else getattr(opts, name)
for name in per_type_delete_opts
):
if "https://mail.google.com/" not in opts.gmail_api_scopes:
logger.error(
"Message deletion requires scope"
" 'https://mail.google.com/'. "
"Add the scope and remove token file "
"to acquire proper access."
)
opts.mailbox_delete = False
for name in per_type_delete_opts:
setattr(opts, name, False)
try:
mailbox_connection = GmailConnection(
credentials_file=opts.gmail_api_credentials_file,
token_file=opts.gmail_api_token_file,
scopes=opts.gmail_api_scopes,
include_spam_trash=opts.gmail_api_include_spam_trash,
paginate_messages=opts.gmail_api_paginate_messages,
reports_folder=opts.mailbox_reports_folder,
oauth2_port=opts.gmail_api_oauth2_port,
auth_mode=opts.gmail_api_auth_mode,
service_account_user=opts.gmail_api_service_account_user,
)
except Exception:
logger.exception("Gmail API Error")
exit(1)
if opts.maildir_path:
try:
mailbox_connection = MaildirConnection(
maildir_path=opts.maildir_path,
maildir_create=opts.maildir_create,
)
except Exception:
logger.exception("Maildir Error")
exit(1)
if mailbox_connection:
mailbox_batch_size_value = (
int(opts.mailbox_batch_size) if opts.mailbox_batch_size is not None else 10
)
mailbox_check_timeout_value = (
int(opts.mailbox_check_timeout)
if opts.mailbox_check_timeout is not None
else 30
)
mailbox_max_unsaved_retries_value = (
int(opts.mailbox_max_unsaved_retries)
if opts.mailbox_max_unsaved_retries is not None
else 2
)
def mailbox_save_callback(batch: ParsingResults) -> bool:
"""Save one mailbox batch and report whether it actually landed.
Passed to ``get_dmarc_reports_from_mailbox()`` as ``save_callback``
and to ``watch_inbox()`` as its ``callback``. Returning ``False``
when any output destination failed keeps that batch's messages in
the mailbox to be retried, instead of archiving or deleting reports
that were never persisted anywhere (#242). With
``fail_on_output_error`` enabled it never returns ``False``:
``process_reports()`` raises ``ParserError`` instead, which the
library treats as "unsaved" too (same retention and retry-cap
bookkeeping) before the exception propagates back out here.
"""
return not process_reports(batch)
if mailbox_connection and not _shutdown_requested:
try:
reports = get_dmarc_reports_from_mailbox(
connection=mailbox_connection,
delete=opts.mailbox_delete,
delete_aggregate=opts.mailbox_delete_aggregate,
delete_failure=opts.mailbox_delete_failure,
delete_smtp_tls=opts.mailbox_delete_smtp_tls,
delete_invalid=opts.mailbox_delete_invalid,
batch_size=mailbox_batch_size_value,
reports_folder=opts.mailbox_reports_folder,
archive_folder=opts.mailbox_archive_folder,
test=opts.mailbox_test,
since=opts.mailbox_since,
config=parser_config,
n_procs=n_procs,
save_callback=mailbox_save_callback,
max_unsaved_retries=mailbox_max_unsaved_retries_value,
)
aggregate_reports += reports["aggregate_reports"]
failure_reports += reports["failure_reports"]
smtp_tls_reports += reports["smtp_tls_reports"]
except ParserError as error:
# fail_on_output_error turns a failed batch save into a
# ParserError inside mailbox_save_callback; it reaches here
# through get_dmarc_reports_from_mailbox, which leaves the
# batch's messages in the mailbox on its way out.
logger.error(error.__str__())
sys.exit(1)
except (ClientAuthenticationError, APIError, httpx.HTTPError) as error:
if msgraph_connection is None:
logger.exception("Mailbox Error")
else:
_log_msgraph_failure(
error,
stage="mailbox fetch",
mailbox=opts.graph_mailbox or opts.graph_user,
tenant_id=opts.graph_tenant_id,
auth_method=opts.graph_auth_method,
)
exit(1)
except Exception:
logger.exception("Mailbox Error")
exit(1)
# Filtered here rather than relying on process_reports()'s in-place
# filtering: the dicts it filters are the file snapshot and the mailbox
# batches, not this combined dict, which exists only to feed
# email_results() / email_results_via_msgraph() below.
parsing_results: ParsingResults = {
"aggregate_reports": aggregate_reports,
"failure_reports": failure_reports,
"smtp_tls_reports": filter_smtp_tls_reports_for_index_prefix(smtp_tls_reports),
}
file_results_nonempty = bool(
file_parsing_results["aggregate_reports"]
or file_parsing_results["failure_reports"]
or file_parsing_results["smtp_tls_reports"]
)
# Mailbox-derived reports were already saved by mailbox_save_callback;
# only file/mbox-derived reports are left to save here. With a mailbox
# connection and nothing from files, skip the call entirely so the run
# doesn't print a second, empty JSON blob.
if file_results_nonempty or not mailbox_connection:
try:
process_reports(file_parsing_results)
except ParserError as error:
logger.error(error.__str__())
sys.exit(1)
smtp_to_value = (
list(opts.smtp_to)
if isinstance(opts.smtp_to, list)
else _str_to_list(str(opts.smtp_to))
)
has_reports = bool(
parsing_results["aggregate_reports"]
or parsing_results["failure_reports"]
or parsing_results["smtp_tls_reports"]
)
if not has_reports and (
opts.smtp_host or (msgraph_connection is not None and smtp_to_value)
):
logger.info("No reports were parsed; skipping the results email")
elif opts.smtp_host:
try:
verify = True
if opts.smtp_skip_certificate_verification:
verify = False
smtp_port_value = int(opts.smtp_port) if opts.smtp_port is not None else 25
email_results(
parsing_results,
opts.smtp_host,
opts.smtp_from,
smtp_to_value,
port=smtp_port_value,
verify=verify,
username=opts.smtp_user,
password=opts.smtp_password,
subject=opts.smtp_subject,
require_encryption=opts.smtp_ssl,
attachment_filename=opts.smtp_attachment,
message=opts.smtp_message,
)
except Exception:
logger.exception("Failed to email results")
exit(1)
elif msgraph_connection is not None and smtp_to_value:
try:
email_results_via_msgraph(
parsing_results,
msgraph_connection,
smtp_to_value,
subject=opts.smtp_subject,
attachment_filename=opts.smtp_attachment,
message=opts.smtp_message,
)
except (ClientAuthenticationError, APIError, httpx.HTTPError) as error:
_log_msgraph_failure(
error,
stage="message send",
mailbox=opts.graph_mailbox or opts.graph_user,
tenant_id=opts.graph_tenant_id,
auth_method=opts.graph_auth_method,
)
exit(1)
except Exception:
logger.exception("Failed to email results via Microsoft Graph")
exit(1)
if mailbox_connection and opts.mailbox_watch:
logger.info("Watching for email - Ctrl-C once to quit, twice to force")
while True:
# Re-check mailbox_watch in case a config reload disabled watch mode
if not opts.mailbox_watch:
logger.info(
"Mailbox watch disabled in reloaded configuration, stopping watcher"
)
break
try:
# `config_reloading` returns True on SIGHUP (reload) or
# SIGTERM/SIGINT (shutdown); the backend polls it between
# checks — including inside the IMAP IDLE loop — and returns
# at a safe boundary once the current batch is processed.
watch_inbox(
mailbox_connection=mailbox_connection,
callback=mailbox_save_callback,
reports_folder=opts.mailbox_reports_folder,
archive_folder=opts.mailbox_archive_folder,
delete=opts.mailbox_delete,
delete_aggregate=opts.mailbox_delete_aggregate,
delete_failure=opts.mailbox_delete_failure,
delete_smtp_tls=opts.mailbox_delete_smtp_tls,
delete_invalid=opts.mailbox_delete_invalid,
test=opts.mailbox_test,
check_timeout=mailbox_check_timeout_value,
batch_size=mailbox_batch_size_value,
since=opts.mailbox_since,
config=parser_config,
config_reloading=lambda: _reload_requested or _shutdown_requested,
n_procs=n_procs,
max_unsaved_retries=mailbox_max_unsaved_retries_value,
)
except FileExistsError as error:
logger.error(f"{error.__str__()}")
exit(1)
except ParserError as error:
logger.error(error.__str__())
exit(1)
except (ClientAuthenticationError, APIError, httpx.HTTPError) as error:
if msgraph_connection is None:
logger.exception("Mailbox Error")
else:
_log_msgraph_failure(
error,
stage="mailbox watch",
mailbox=opts.graph_mailbox or opts.graph_user,
tenant_id=opts.graph_tenant_id,
auth_method=opts.graph_auth_method,
)
exit(1)
# Prioritize shutdown over reload if both flags are set (e.g.
# SIGHUP followed by SIGTERM). atexit closes output clients.
if _shutdown_requested:
logger.info("Shutdown requested, exiting watch loop")
break
if not _reload_requested:
break
# Reload configuration — emit the log message here (not in the
# signal handler, which is not async-signal-safe), then clear the
# flag so that any new SIGHUP arriving while we reload will be
# captured for the next iteration rather than being silently dropped.
logger.info("SIGHUP received, config will reload after current batch")
_reload_requested = False
logger.info("Reloading configuration...")
try:
# Build a fresh opts starting from CLI-only defaults so that
# sections removed from the config file actually take effect.
new_opts = Namespace(**vars(opts_from_cli))
new_config = _load_config(config_file)
new_index_prefix_domain_map = _parse_config(new_config, new_opts)
new_clients = _init_output_clients(
new_opts, index_prefix_domain_map=new_index_prefix_domain_map
)
# All steps succeeded — commit the changes atomically.
_close_output_clients(clients)
clients = new_clients
index_prefix_domain_map = new_index_prefix_domain_map
# Reload the reverse DNS map so changes to the
# map path/URL in the config take effect. PSL overrides
# are reloaded alongside it so map entries that depend on
# a folded base domain keep working.
load_reverse_dns_map(
REVERSE_DNS_MAP,
always_use_local_file=new_opts.always_use_local_files,
local_file_path=new_opts.reverse_dns_map_path,
url=new_opts.reverse_dns_map_url,
offline=new_opts.offline,
psl_overrides_path=new_opts.psl_overrides_path,
psl_overrides_url=new_opts.psl_overrides_url,
)
# Reload the IP database so changes to the
# db path/URL in the config take effect.
load_ip_db(
always_use_local_file=new_opts.always_use_local_files,
local_file_path=new_opts.ip_db_path,
url=new_opts.ipinfo_url,
offline=new_opts.offline,
)
# Re-apply IPinfo API settings. Passing a falsy token disables
# the API; a rotated token picks up here too. An invalid token
# is fatal even on reload — the operator asked for it.
try:
configure_ipinfo_api(
new_opts.ipinfo_api_token if not new_opts.offline else None,
)
except InvalidIPinfoAPIKey as e:
logger.critical(str(e))
exit(1)
for k, v in vars(new_opts).items():
setattr(opts, k, v)
parser_config = _build_parser_config(opts)
# Update watch parameters from reloaded config
mailbox_batch_size_value = (
int(opts.mailbox_batch_size)
if opts.mailbox_batch_size is not None
else 10
)
mailbox_check_timeout_value = (
int(opts.mailbox_check_timeout)
if opts.mailbox_check_timeout is not None
else 30
)
mailbox_max_unsaved_retries_value = (
int(opts.mailbox_max_unsaved_retries)
if opts.mailbox_max_unsaved_retries is not None
else 2
)
# Update log level
logger.setLevel(logging.ERROR)
if opts.warnings:
logger.setLevel(logging.WARNING)
if opts.verbose:
logger.setLevel(logging.INFO)
if opts.debug:
logger.setLevel(logging.DEBUG)
# Refresh FileHandler if log_file changed
old_log_file = getattr(opts, "active_log_file", None)
new_log_file = opts.log_file
if old_log_file != new_log_file:
# Remove old FileHandlers
for h in list(logger.handlers):
if isinstance(h, logging.FileHandler):
h.close()
logger.removeHandler(h)
# Add new FileHandler if configured
if new_log_file:
try:
fh = logging.FileHandler(new_log_file, "a")
file_formatter = logging.Formatter(
"%(asctime)s - %(levelname)s"
" - [%(filename)s:%(lineno)d] - %(message)s"
)
fh.setFormatter(file_formatter)
logger.addHandler(fh)
except Exception as log_error:
logger.warning(f"Unable to write to log file: {log_error}")
opts.active_log_file = new_log_file
_configure_dependency_logging(logger.level)
logger.info("Configuration reloaded successfully")
except Exception:
logger.exception(
"Config reload failed, continuing with previous config"
)
# Close output clients on the success path (one-shot or graceful
# watch-loop exit). atexit-registered above is the safety net for
# exit(1) / uncaught-exception paths.
_close_output_clients(clients)
if __name__ == "__main__":
_main()
Reference in New Issue View Git Blame Copy Permalink