amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-09-13 17:38:00 +00:00
Code Issues Projects Releases Wiki Activity
Files
aedc7ee48b009f7cd880c19f83959ed8785fdad4
parsedmarc/parsedmarc/cli.py
T
Sean WhalenandClaude Fable 5.1 550244c6d7 fix: roll back the search backends' default alias when output init fails (#906) 
_init_output_clients() is now all-or-nothing against the Elasticsearch
and OpenSearch connection registries: either it returns the fully built
client dict, or it closes everything it built and leaves both "default"
aliases naming exactly what they named on entry.

elastic.set_hosts()/opensearch.set_hosts() end in
connections.create_connection(), which registers the new client under
the process-wide "default" alias the moment it is constructed
(elasticsearch/dsl/connections.py:81-88 and
opensearchpy/connection/connections.py:87-95, as installed: 8.19.3 /
3.2.0). Everything that runs after that -- the index migration, and
every output configured later -- can still fail. On the SIGHUP reload
path _main() builds the replacement clients before closing the old ones
and, when the build raises, logs "Config reload failed, continuing with
previous config" and keeps the old opts. But the alias had already been
handed to the new client, so every save -- elastic.py's Search and
Document.save() calls, which resolve the "default" alias through
elasticsearch/dsl/_sync/document.py:99-100 -- reached the new cluster
while the index prefixes/suffixes and index_prefix_domain_map still came
from the old configuration. The half-built client was never
closed either, nor was any client built earlier in the same failed call
(that half also leaked on every attempt of the startup retry loop).

The handler tears down first and restores second -- with the teardown in
a try/finally, since a second Ctrl-C landing in it propagates straight
through _close_output_clients, which swallows only Exception -- and the
two steps are not interchangeable. With an _ElasticsearchHandle already in `clients`,
teardown closes its client and releases the alias -- which still names
that client -- and the restore then re-registers the previous client
into an unset alias. Restoring first would put the previous client back
and only then close the handle, which rests the whole rollback on the
handle declining to touch an alias that no longer names its own client:
true only since #902, and a property of the handle rather than of this
function. Tearing down first keeps the guarantee local. Both failure
points are traced in the comment on the handler.

Closing the discarded client is best-effort, and closing it twice is
safe: Elasticsearch.close() -> Transport.close() closes each node's
urllib3 pool (elastic_transport/_transport.py:499-504,
_node/_http_urllib3.py:224-228, and urllib3 pool close() is a no-op once
cleared), and OpenSearch's connection close() guards on `if self.pool`
(opensearchpy/connection/http_urllib3.py:323-328). Taking the snapshot
cannot itself open a connection: get_connection() lazily builds a client
from kwargs left behind by configure() (dsl/connections.py:90-115), and
parsedmarc never calls configure().

The guard catches BaseException rather than Exception because
migrate_indexes() wraps every cluster call in `except Exception` and
logs a warning (elastic.py:824-829, and again in each of the per-index
loops that follow), so what escapes the Elasticsearch block after
set_hosts() is, in practice, a KeyboardInterrupt landing in one of those
calls --
which is what the regression tests inject at the SDK transport
boundary. For the same reason, closing the discarded client swallows
BaseException: an interrupt there must not cost the alias its hand-back,
and the original failure is re-raised afterwards either way.

The alias is not the only module-level state set_hosts() writes:
elastic.set_hosts() also assigns elastic._SERVERLESS (elastic.py:620-622)
-- before it constructs the client, so it can be stale even on a failure
that never reached the registry -- and create_indexes() consults it
(elastic.py:652-659) to decide whether to strip the shard settings
Serverless rejects. It is snapshotted and restored alongside the alias.
opensearch.py has no equivalent (no `global` statement in the file).

Shape: the existing body keeps its indentation as _build_output_clients(),
which fills a `clients` dict the caller passes in and is explicitly not
transactional; the rollback lives in a short _init_output_clients()
wrapper that owns that dict on both paths. Passing the dict in is what
lets the wrapper close what was already built after the build raises. The
public name is unchanged, so the tests and _main() call sites are
untouched.

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-11 08:40:36 -04:00

3756 lines
158 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 _search_alias_snapshot() -> list[tuple[Any, Any, Any]]:
"""Record the module-level state each search backend's set_hosts() writes.
``elastic.set_hosts()`` and ``opensearch.set_hosts()`` register the client
they build under their SDK's process-wide ``default`` connection alias,
and the save path resolves that alias on every write.
:func:`_init_output_clients` takes this snapshot before it touches either
registry, so that a failure part-way through can put back exactly what it
found.
Returns:
list: One ``(module, client, serverless)`` triple per backend.
``client`` is what that backend's ``default`` alias names, or ``None``
when the alias is unset. ``serverless`` is ``elastic._SERVERLESS`` for
the Elasticsearch backend and ``None`` for OpenSearch, which has no
equivalent. Carrying the module itself, rather than a name to look it
up by, is what lets :func:`_restore_search_aliases` put each client
back into the registry it came from without a second lookup to
disagree with. A backend whose module is ``None`` -- its optional
extra is not installed, see the guarded imports at the top of this
module -- has no state to snapshot and is left out of the list
entirely.
"""
snapshot: list[tuple[Any, Any, Any]] = []
for module in (elastic, opensearch):
if module is None:
continue
try:
# get_connection() does not only look the alias up: it also
# *constructs* a client from kwargs stashed by an earlier
# connections.configure() call. parsedmarc never calls
# configure() -- both set_hosts() implementations register their
# client with create_connection() -- so each registry's
# ``_kwargs`` stays empty and this can only return an
# already-registered client or raise KeyError. Taking the
# snapshot can never itself open a connection.
connection = module.connections.get_connection("default")
except KeyError:
connection = None
# The alias is not the only module-level state set_hosts() writes:
# elastic.set_hosts() also assigns elastic._SERVERLESS, which
# elastic.create_indexes() consults to decide whether to strip the
# shard settings Serverless rejects. Reaching into a sibling module's
# private is the same liberty this function already takes with
# ``connections``; without it, a failed reload that flipped
# ``[elasticsearch] serverless`` would pair the restored old client
# with the new config's flag. opensearch.py declares no module
# globals at all (no ``global`` statement in the file), so there is
# nothing to pair with it.
serverless = elastic._SERVERLESS if module is elastic else None
snapshot.append((module, connection, serverless))
return snapshot
def _restore_search_aliases(snapshot: list[tuple[Any, Any, Any]]) -> None:
"""Put the state in *snapshot* back the way it was when it was taken.
``elastic._SERVERLESS`` is put back first and unconditionally; the alias
then has three cases per backend. The alias still names the client it
named before, so there is nothing to do. A different client has taken it
over -- that client is
closed, and the previous client is registered again, or the alias is
removed outright when there was no previous client. Or the alias is unset
because a fully built handle released it during the teardown that runs
first -- nothing left to close, and the previous client is simply
registered again; this is the ordinary path when the failure came after a
search backend was fully built.
Closing is best-effort and silent: the client being closed here is the
half-built one for the configuration that just failed, it is being
discarded either way, and a teardown error is not actionable -- while
handing the alias back is what keeps the still-running configuration
writing where it thinks it is, so it must happen either way. Closing the
same client twice is safe -- ``_close_output_clients`` may already have
closed it through its handle -- because both SDKs' ``close()`` are
idempotent: ``Elasticsearch.close()`` closes each node's urllib3 pool,
whose ``close()`` is a no-op once cleared, and ``OpenSearch.close()``
guards on ``if self.pool``.
Args:
snapshot (list): The return value of :func:`_search_alias_snapshot`.
"""
for module, previous, previous_serverless in snapshot:
if elastic is not None and module is elastic:
# Unconditionally, and before the alias: set_hosts() assigns
# _SERVERLESS before it constructs the client, so it can be stale
# even on a failure that never reached the registry.
elastic._SERVERLESS = previous_serverless
try:
current = module.connections.get_connection("default")
except KeyError:
current = None
if current is previous:
# Nothing took the alias over, including the common case of
# both being None. Leave it alone.
continue
if current is not None:
try:
current.close()
except BaseException:
# Best-effort; see the docstring. BaseException, not
# Exception, for the same reason the caller's teardown is
# wrapped in try/finally: a Ctrl-C landing in close() must
# not cost the alias its hand-back, and the exception the
# caller re-raises afterwards still reports the failure.
pass
if previous is None:
# Cannot raise KeyError: the alias was just resolved out of this
# registry's own ``_conns``, and closing a client does not touch
# the registry, so it is still there.
module.connections.remove_connection("default")
else:
module.connections.add_connection("default", previous)
def _build_output_clients(opts, clients, index_prefix_domain_map=None):
"""Create output clients based on current opts, into *clients*.
Deliberately not transactional: it fills *clients* as it goes and, when a
step fails, leaves behind both the clients it had already built and any
change ``elastic.set_hosts()``/``opensearch.set_hosts()`` made to their
SDKs' module-level state. Undoing that is :func:`_init_output_clients`'s
job, which is why *clients* is a parameter -- the caller owns the dict on
the failure path too, and can close what is in it. Call
:func:`_init_output_clients`, not this.
Args:
opts: Namespace of parsed configuration values.
clients (dict): The dict to fill, keyed by client name. Filled in
place, and also returned.
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: *clients*, filled.
Raises:
ConfigurationError: If a required output client cannot be created.
RuntimeError: If constructing an output client fails, chained to the
error the SDK raised.
"""
# 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, in two
# places rather than one: connections.create_connection() registers the
# new client under the ``default`` alias, and elastic.set_hosts() also
# assigns elastic._SERVERLESS. _init_output_clients() rolls both back if
# a later step fails. They are still initialized last, so that a failure
# in any other output happens before either registry has been touched.
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 _init_output_clients(opts, index_prefix_domain_map=None):
"""Create output clients based on current opts, all-or-nothing.
Either every configured client is built and returned, or the clients built
so far are closed and the module-level state that
``elastic.set_hosts()``/``opensearch.set_hosts()`` mutate -- each
backend's ``default`` connection alias, and ``elastic._SERVERLESS`` -- is
left exactly as it was on entry.
That guarantee is what the SIGHUP reload in :func:`_main` needs. It builds
the replacement clients before closing the old ones and keeps running with
the old ``opts`` if the build fails. But ``set_hosts()`` registers its
client under the ``default`` alias as soon as it is constructed, well
before the rest of the initialization can fail -- on an output configured
later failing to build, or on a Ctrl-C landing in the index migration.
Without the rollback, such a reload left every subsequent save resolving
that alias to the *new* cluster while ``opts`` stayed old, and leaked the
half-built client.
Args:
opts: Namespace of parsed configuration values.
index_prefix_domain_map (dict | None): The parsed
``general.index_prefix_domain_map``; see
:func:`_build_output_clients`.
Returns:
dict of client instances keyed by name.
Raises:
ConfigurationError: If a required output client cannot be created.
RuntimeError: If constructing an output client fails, chained to the
error the SDK raised.
"""
previous_search_state = _search_alias_snapshot()
clients: dict[str, Any] = {}
try:
return _build_output_clients(
opts, clients, index_prefix_domain_map=index_prefix_domain_map
)
except BaseException:
# Teardown first, then restore. The order is deliberate, and the two
# steps are not interchangeable. Tracing the two points at which a
# failure can leave a ``default`` alias pointing at a new client:
#
# (1) Inside the Elasticsearch block after set_hosts(), before the
# handle exists. Teardown closes the outputs built earlier and
# leaves the alias alone -- nothing in ``clients`` owns it -- and
# the restore then closes the new client and re-registers the old
# one. Either order reaches that state.
#
# (2) In a later step, with _ElasticsearchHandle already in
# ``clients`` owning the new client that holds the alias.
# Teardown first: the handle closes its client and, seeing the
# alias still name that client, releases the alias; the restore
# then finds the alias unset and re-registers the old client,
# which was never closed. Restoring first would instead put the
# old client back and only then close the handle -- leaving the
# whole rollback resting on the handle declining to touch an
# alias that no longer names its own client. It does decline
# today, but that is a property of _ElasticsearchHandle.close(),
# not of this function: before #902 the handle re-resolved the
# alias at close time and would have deleted the registration
# restored a moment earlier, leaving every later save raising
# KeyError. Tearing down first keeps this guarantee local.
#
# BaseException, not Exception: elastic.migrate_indexes() catches
# Exception around every cluster call and logs a warning, so the
# failure that escapes the Elasticsearch block after set_hosts() is,
# in practice, a KeyboardInterrupt landing in one of them. And try/finally,
# because a second Ctrl-C arriving during the teardown propagates
# straight through _close_output_clients, which swallows only
# Exception; a plain statement sequence would then skip the restore
# and leave behind exactly the state this function exists to prevent.
try:
_close_output_clients(clients)
finally:
_restore_search_aliases(previous_search_state)
raise
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