amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-09-11 00:18:02 +00:00
Code Issues Projects Releases Wiki Activity
Files
0f3b973be57fd44e8c1ae352eb42af37cf4e2d84
parsedmarc/parsedmarc/cli.py
T
Sean WhalenandClaude Opus 5 0f3b973be5 fix: close the search client a handle owns, not the default alias (#902) 
_ElasticsearchHandle.close() and _OpenSearchHandle.close() re-resolved
the client library's "default" connection alias at close time. That is
wrong on the SIGHUP reload path, which is not a shutdown path:
_main's reload block calls _init_output_clients() -- and therefore
set_hosts() -> connections.create_connection() -- before
_close_output_clients() on the old clients, deliberately, so a broken
new config leaves the old clients running. By then the alias names the
*new* client, so the old handle closed the new client and then dropped
the alias entirely. Every later save raised
KeyError("There is no connection with alias 'default'.") until
parsedmarc was restarted, and the original client was leaked.

Verified against the installed SDK sources rather than the docs
(AGENTS.md, "Verify bug claims against authoritative sources", point 3):

  elasticsearch/dsl/connections.py (elasticsearch 9.x) and
  opensearchpy/connection/connections.py (opensearch-py 3.x) both define

      def create_connection(self, alias="default", **kwargs):
          conn = self._conns[alias] = <Client>(**kwargs)

  -- so create_connection() overwrites whatever the alias held, and
  returns the very object it stored (elasticsearch.dsl passes it through
  _with_user_agent(), which mutates headers and returns the same object,
  so identity is preserved) --

      def remove_connection(self, alias):
          errors = 0
          for d in (self._conns, self._kwargs):
              try:
                  del d[alias]
              except KeyError:
                  errors += 1
          if errors == 2:
              raise KeyError(f"There is no connection with alias {alias!r}.")

  -- i.e. it deletes the alias outright, with no regard for which client
  currently holds it, and

      def get_connection(self, alias="default"):
          if not isinstance(alias, str):
              return alias            # (opensearch-py; elasticsearch.dsl
                                      #  returns _with_user_agent(alias))
          try:
              return self._conns[alias]
          ...
          raise KeyError(f"There is no connection with alias {alias!r}.")

  -- a str only comes *back* out of get_connection() when a str was
  passed *in* as the alias, which these call sites never do; the client
  stored under "default" is always the object create_connection() built.
  The old `isinstance(conn, str)` guard was therefore dead code and is
  removed rather than carried forward (AGENTS.md: delete unreachable
  branches, don't hide them). add_connection(alias, conn) stores an
  arbitrary object as-is, which is what the new tests use.

set_hosts() now returns the client it registered, _init_output_clients()
passes that client to the handle, and close() closes that object and
gives up the alias only when the registry's current "default" *is* that
same object (identity check). The init/close order in the reload block
is unchanged.

tests/test_cli.py gains TestSearchBackendHandles, which drives the real
connection registries (the SDK boundary) with fake client objects, so no
network is touched: per backend, a reload-shaped test asserts the old
client is closed, the new one is not, and the alias still resolves to the
new client; a shutdown-shaped test asserts the alias is released when it
is still ours; and a best-effort test asserts a raising close() neither
propagates nor blocks the alias release. Each test restores whatever the
process-wide registry held beforehand. All six fail against the unfixed
handles.

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

3549 lines
147 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""A CLI for parsing DMARC and SMTP TLS 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
from typing import TYPE_CHECKING, Any
import httpx
import yaml
from tqdm import tqdm
from parsedmarc import (
IP_ADDRESS_CACHE,
REVERSE_DNS_MAP,
SEEN_AGGREGATE_REPORT_IDS,
InvalidDMARCReport,
ParserConfig,
ParserError,
__version__,
email_results,
email_results_via_msgraph,
get_dmarc_reports_from_mailbox,
get_dmarc_reports_from_mbox,
postgres,
save_output,
splunk,
syslog,
watch_inbox,
webhook,
)
# Output integrations that need an optional extra (issue #883). Each one
# imports a third-party SDK at module level -- elastic.py and
# opensearch.py define DSL Document classes there, so they cannot guard
# the import internally the way postgres.py does -- and so the guard
# lives here, at the import site. A missing module becomes ``None``; the
# config sections that would use it fail fast in _init_output_clients()
# with a pip-install hint. splunk, syslog, webhook, and postgres are
# imported eagerly above: they need only httpx, the standard library, or
# their own internal guard.
if TYPE_CHECKING:
from parsedmarc import elastic, gelf, kafkaclient, loganalytics, opensearch, s3
else:
try:
from parsedmarc import elastic
except ModuleNotFoundError:
elastic = None
try:
from parsedmarc import gelf
except ModuleNotFoundError:
gelf = None
try:
from parsedmarc import kafkaclient
except ModuleNotFoundError:
kafkaclient = None
try:
from parsedmarc import loganalytics
except ModuleNotFoundError:
loganalytics = None
try:
from parsedmarc import opensearch
except ModuleNotFoundError:
opensearch = None
try:
from parsedmarc import s3
except ModuleNotFoundError:
s3 = None
# Microsoft Graph error types, used only in ``except`` clauses and one
# ``isinstance`` check around Graph mailbox calls. Without the msgraph
# extra a Graph connection cannot be constructed at all (its
# parsedmarc.mail placeholder raises the extra's ImportError), so those
# handlers are unreachable and the placeholders below are never matched.
if TYPE_CHECKING:
from azure.core.exceptions import ClientAuthenticationError
from kiota_abstractions.api_error import APIError
else:
try:
from azure.core.exceptions import ClientAuthenticationError
except ModuleNotFoundError:
class ClientAuthenticationError(Exception):
"""Never-raised placeholder for the absent msgraph extra."""
try:
from kiota_abstractions.api_error import APIError
except ModuleNotFoundError:
class APIError(Exception):
"""Never-raised placeholder for the absent msgraph extra."""
from parsedmarc.constants import DEFAULT_DNS_MAX_RETRIES, DEFAULT_DNS_TIMEOUT
from parsedmarc.log import logger
import parsedmarc.mail
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 _missing_extra_hint(section: str, extra: str) -> str:
"""Return the error message for a config section whose extra is missing.
Args:
section (str): The INI section name, without brackets.
extra (str): The name of the extra that provides the section's
integration.
Returns:
str: A message naming the section, the extra, and the exact pip
command that installs it.
"""
return (
f"The [{section}] configuration section requires the {extra} extra: "
f"pip install parsedmarc[{extra}]"
)
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 sys.exit() - the call site keeps its own sys.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.
A ``path`` argument that matches nothing (a plain path that does
not exist and is not a glob match, or a directory/glob pattern
with no matches) is logged as a WARNING naming that argument,
so a typo'd path in a cron job doesn't silently "succeed" while
processing nothing forever. An empty ``paths`` list (e.g. a
mailbox-only run with no file arguments) logs nothing.
"""
expanded: list[str] = []
for path in paths:
before = len(expanded)
if os.path.isdir(path):
pattern = os.path.join(glob_escape(path), "**" if recursive else "*")
for match in sorted(glob(pattern, recursive=recursive)):
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)
if len(expanded) == before:
logger.warning("No files matched %s", path)
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 or
is not readable, or if a ``PARSEDMARC_..._FILE`` secret file
cannot be read.
"""
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():
# Without the msgraph extra, parsedmarc.mail binds a placeholder
# class (outside the MailboxConnection hierarchy) whose
# construction raises mailsuite's ImportError — fail fast here
# with parsedmarc's own install hint instead. Checked on the
# parsedmarc.mail module, the authoritative source of the
# placeholder state, not this module's rebound name (which tests
# replace with mocks).
if not issubclass(
parsedmarc.mail.MSGraphConnection, parsedmarc.mail.MailboxConnection
):
raise ConfigurationError(_missing_extra_hint("msgraph", "msgraph"))
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():
# Same placeholder detection as the msgraph section above.
if not issubclass(
parsedmarc.mail.GmailConnection, parsedmarc.mail.MailboxConnection
):
raise ConfigurationError(_missing_extra_hint("gmail_api", "gmail"))
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:
"""Owns the Elasticsearch client so it participates in _close_output_clients.
Holds the client that ``elastic.set_hosts()`` registered under the
``default`` connection alias, rather than re-resolving that alias when
it is closed. _close_output_clients is not only a shutdown path: the
SIGHUP reload in _main deliberately builds the replacement clients
*before* closing the old ones, and building them re-registers the
``default`` alias, so by close time the alias names the new client.
Args:
connection: The client returned by ``elastic.set_hosts()``.
"""
def __init__(self, connection: Any):
self._connection = connection
def close(self):
try:
self._connection.close()
except Exception:
# Best-effort, and deliberately silent: this is the first of
# two independent teardown steps, and swallowing here is what
# lets the second one still run. Nothing reports this error --
# _close_output_clients logs a warning only if close() itself
# raises, which it cannot while this handler swallows.
pass
try:
# Give up the alias only while it still names our own client.
# elasticsearch.dsl's Connections.remove_connection() deletes
# the alias outright, so removing it after a reload had pointed
# it at a new client would leave that client unreachable, with
# every later save raising KeyError.
if elastic.connections.get_connection("default") is self._connection:
elastic.connections.remove_connection("default")
except Exception:
# Best-effort and silent for the same reason as above: a
# failure to give up the alias is not actionable during
# teardown, and it is not reported anywhere either.
pass
class _OpenSearchHandle:
"""Owns the OpenSearch client so it participates in _close_output_clients.
Holds the client that ``opensearch.set_hosts()`` registered under the
``default`` connection alias; see _ElasticsearchHandle for why the
alias is not re-resolved at close time.
Args:
connection: The client returned by ``opensearch.set_hosts()``.
"""
def __init__(self, connection: Any):
self._connection = connection
def close(self):
try:
self._connection.close()
except Exception:
# Best-effort, and deliberately silent: this is the first of
# two independent teardown steps, and swallowing here is what
# lets the second one still run. Nothing reports this error --
# _close_output_clients logs a warning only if close() itself
# raises, which it cannot while this handler swallows.
pass
try:
# Only while the alias still names our own client; see
# _ElasticsearchHandle.close().
if opensearch.connections.get_connection("default") is self._connection:
opensearch.connections.remove_connection("default")
except Exception:
# Best-effort and silent for the same reason as above: a
# failure to give up the alias is not actionable during
# teardown, and it is not reported anywhere either.
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 = {}
# Each check below is deliberately outside the try/except that wraps
# its constructor: those handlers re-raise everything as RuntimeError,
# which would bury the install hint.
if opts.s3_bucket and s3 is None:
raise ConfigurationError(_missing_extra_hint("s3", "s3"))
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
# postgres.py guards its own psycopg import, so the module is always
# importable; check the SDK here so a missing extra is a fail-fast
# ConfigurationError rather than a PostgreSQLError that the startup
# retry loop would retry for over a minute before exiting.
if (
opts.postgresql_host or opts.postgresql_connection_string
) and postgres.psycopg is None:
raise ConfigurationError(_missing_extra_hint("postgresql", "postgresql"))
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
if opts.kafka_hosts and kafkaclient is None:
raise ConfigurationError(_missing_extra_hint("kafka", "kafka"))
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
if opts.gelf_host and gelf is None:
raise ConfigurationError(_missing_extra_hint("gelf", "gelf"))
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
# The Log Analytics client is built per batch in process_reports(),
# under the same opts.la_dce guard. Checking it here means a missing
# extra is reported at startup -- and again on a SIGHUP reload --
# rather than once reports are already in hand.
if opts.la_dce and loganalytics is None:
raise ConfigurationError(_missing_extra_hint("log_analytics", "loganalytics"))
# 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:
# Scoped to the same condition as the constructors below, which is
# also the condition under which process_reports() dereferences
# these modules to save reports.
if opts.elasticsearch_hosts and elastic is None:
raise ConfigurationError(_missing_extra_hint("elasticsearch", "elastic"))
if opts.opensearch_hosts and opensearch is None:
raise ConfigurationError(_missing_extra_hint("opensearch", "opensearch"))
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
)
elasticsearch_connection = 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(
elasticsearch_connection
)
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_connection = 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(opensearch_connection)
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()``. With ``fail_on_output_error`` enabled,
a non-empty list is raised as ``ParserError`` instead of returned.
"""
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 and SMTP TLS 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, failure, or SMTP TLS 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: IP addresses, https:// URLs (DNS over "
"HTTPS), and/or tls://ip[:port][#hostname] (DNS over TLS)",
)
arg_parser.add_argument(
"-t",
"--dns_timeout",
"--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))
sys.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")
sys.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))
sys.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))
sys.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_}")
sys.exit(1)
# Always close output clients on the way out (normal return,
# sys.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"
)
sys.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")
sys.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,
)
sys.exit(1)
except Exception:
logger.exception("MS Graph Error")
sys.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")
sys.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")
sys.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,
)
sys.exit(1)
except Exception:
logger.exception("Mailbox Error")
sys.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")
sys.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,
)
sys.exit(1)
except Exception:
logger.exception("Failed to email results via Microsoft Graph")
sys.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__()}")
sys.exit(1)
except ParserError as error:
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 watch",
mailbox=opts.graph_mailbox or opts.graph_user,
tenant_id=opts.graph_tenant_id,
auth_method=opts.graph_auth_method,
)
sys.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))
sys.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
# sys.exit(1) / uncaught-exception paths.
_close_output_clients(clients)
if __name__ == "__main__":
_main()
Reference in New Issue View Git Blame Copy Permalink