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

Merge branch 'master' of https://github.com/domainaware/parsedmarc

Browse Source
This commit is contained in:
Sean Whalen
2026-09-12 13:59:05 -04:00
parent aedc7ee48b b41230dc07
commit 281a3cf3b1
6 changed files with 1426 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+3 -1
View File
@@ -7,14 +7,16 @@
- Upgrade the `mailsuite` requirement to `>=2.3.2`, which sets the `IMAPClient` version to `>=4.0.0`, fixing support for Python 3.14 (Closes #838)
- **The prebuilt Docker image (`ghcr.io/domainaware/parsedmarc`) is roughly 40% smaller to pull** ([#893](https://github.com/domainaware/parsedmarc/pull/893)). The runtime stage copied the built wheel out of the build stage and deleted it again at the end of the next `RUN`, but a `RUN` can only write a whiteout over a layer an earlier instruction already committed: the wheel shipped in every published image and every `docker pull` downloaded it (10,713,473 bytes of the 11.0.0 image, on both architectures). The wheel is now bind-mounted from the build stage instead, and a bind mount is never committed to a layer. `pip install` also runs with `--no-cache-dir`, which drops a further ~99 MB of pip's download cache that the image had been carrying in the same layer as `site-packages`. Measured on linux/amd64: 272,138,724 compressed bytes across six layers before, 163,757,439 across five after.
- **`parse_report_file()`, `extract_report()`, and `parse_aggregate_report_file()` no longer close a file object supplied by the caller, on success or failure.** A path input is still opened and closed by the function, and a `bytes`/`bytearray`/`memoryview` input is still wrapped in an internally-created `BytesIO` that the function closes; but a file-like object the caller passed in may be seeked, `tell()`'d, or reused afterward, and closing it out from under the caller (as these functions previously did on the success path) makes that impossible. `parse_aggregate_report_file()` forwards its input straight to `extract_report()`, so it inherits the fix without changes of its own. `extract_report()`'s handling of a caller-supplied *non-seekable* stream is unaffected: its contents are still copied into an internally-owned buffer that the function closes, since that buffer was never the caller's own handle.
- **Bare `exit(...)` calls are now `sys.exit(...)` throughout the CLI and the maintainer map-tooling scripts.** `exit` is installed into `builtins` by the `site` module, not just for interactive sessions, so it isn't guaranteed to exist under `python -S` or an embedded interpreter that skips `site` — the failure paths that called it would raise `NameError` instead of actually exiting. `sys.exit` is always available and was already used elsewhere in `cli.py` and `find_unknown_base_reverse_dns.py`. Also removed the dead, immediately-recomputed `index_date` stores in the Elasticsearch and OpenSearch aggregate-report savers; behavior is unchanged.
### Bug fixes
- `find_unknown_base_reverse_dns.py`'s missing-file checks for `base_reverse_dns_map.csv` and the `known_unknown`/PSL-override lists printed a clean error message but fell through into an unhandled `FileNotFoundError` traceback instead of exiting.
- **`parse_report_file()` now closes the file handle it opens itself for a path input if reading it raises.** When `input_` is a path, the function opened the file, read it, and closed it with no exception handling in between; an exception raised by `read()` (e.g. an `OSError` from the underlying storage) skipped the close, so the descriptor was left to be released only when Python's garbage collector eventually finalized the object — CPython's `io.IOBase.__del__` closes an unclosed file on finalization (<https://docs.python.org/3/library/io.html>) — rather than being closed deterministically. This is the pattern CodeQL's `py/file-not-closed` query flags, found in a local code-quality scan. The path branch now opens the file with a `with` block, so the handle is closed on both the success and exception paths. A file-like object or bytes buffer supplied by the caller is unaffected: as before, it is closed only after a successful read, and left open if `read()` raises.
- **`parse_report_file()` now closes the file handle it opens itself for a path input if reading it raises.** When `input_` is a path, the function opened the file, read it, and closed it with no exception handling in between; an exception raised by `read()` (e.g. an `OSError` from the underlying storage) skipped the close, so the descriptor was left to be released only when Python's garbage collector eventually finalized the object — CPython's `io.IOBase.__del__` closes an unclosed file on finalization (<https://docs.python.org/3/library/io.html>) — rather than being closed deterministically. This is the pattern CodeQL's `py/file-not-closed` query flags, found in a local code-quality scan. The path branch now opens the file with a `with` block, so the handle is closed on both the success and exception paths. (A caller-supplied file-like object's closing behavior changed again later in this same Unreleased version — see the *Changes* section above.)
- **A SIGHUP configuration reload no longer breaks every subsequent save to Elasticsearch or OpenSearch.** With `[elasticsearch]` or `[opensearch]` configured in watch mode, reloading the configuration re-registered the search client under the client library's `default` connection alias and then closed the previous run's clients. The close step re-resolved that alias instead of remembering the client it was created for, so it closed the *newly built* client and deleted the `default` alias outright — after which every report save failed with `KeyError: "There is no connection with alias 'default'."` until parsedmarc was restarted, and the original client was left open. Each backend's handle now closes the exact client it was created for and gives up the alias only while the alias still points at that client. Both the Elasticsearch and OpenSearch backends were affected.
- **A configuration reload that fails part-way no longer leaves reports being written to the new Elasticsearch or OpenSearch hosts under the old configuration.** The search client is registered under the client library's process-wide `default` connection alias as soon as it is constructed — before the index migration runs, and before the outputs configured after it are created. When a later step then failed on a SIGHUP reload — for example an `[opensearch]` section that cannot build its client (an unsupported `auth_type`, `awssigv4` without an `aws_region`, AWS credentials that will not load) — parsedmarc logged `Config reload failed, continuing with previous config` and kept the old configuration. But the alias had already been handed to the new client, so every subsequent save resolved it to the *new* hosts while the index prefixes, suffixes, and `index_prefix_domain_map` still came from the old configuration. Reports were silently written to a destination that was never successfully configured, with the log saying nothing had changed. The half-built client was never closed either, nor were the other clients (S3, Kafka, PostgreSQL, and so on) built earlier in the same failed reload — the same leak occurred on every attempt of the startup retry loop, which calls the same function. Building the output clients is now all-or-nothing: if any step fails, everything built so far is closed and the module-level state the search backends keep is put back — each configured backend's `default` alias restored to exactly the client it named beforehand, and the Elasticsearch `serverless` flag (which decides whether `number_of_shards`/`number_of_replicas` are sent when an index is created) to its old value — so a failed reload no longer changes where reports are written, or how indexes are created.
- **A SIGHUP configuration reload that fails after the replacement output clients are built now really does keep the previous configuration, instead of leaving the new clients live and the old ones closed.** The reload built the replacement clients, then immediately closed the old ones and swapped them in — and only afterwards reloaded the reverse DNS map and the PSL overrides, reloaded the IP database, re-applied the IPinfo API token, copied the new values onto `opts`, rebuilt the `ParserConfig`, and converted the watch parameters. A failure in any of those (a typo in `local_reverse_dns_map_path` or `local_psl_overrides_path`, an unreadable map or overrides file, an IP database that cannot be resolved) was caught and logged as `Config reload failed, continuing with previous config`, which was not true: the new output clients were already receiving every report, the old clients were closed and could not be brought back, `opts` could be half-applied, and `load_reverse_dns_map()`/`load_psl_overrides()` — both of which empty their target before reading anything — had left one of the two empty. A bad map path emptied the reverse DNS map, so reverse DNS lookups stopped resolving to known services until some later uncached lookup happened to refill the map from the *previous* configuration's paths (`get_service_from_reverse_dns_base_domain()` and `get_ip_address_info()` both reload a map they find empty). A bad PSL overrides path emptied the override list without ever touching the map, so nothing triggered that lazy reload and base domains were folded without the overrides until parsedmarc was restarted. The whole reload is now staged before anything live is touched: the map is loaded into a fresh dict, the `parsedmarc.utils` globals the loaders assign are snapshotted, the replacement `log_file` is opened during staging, and the commit is a run of plain assignments followed by the logging refresh, whose one remaining failure point — closing the replaced log file — is caught and logged rather than allowed to abort a reload that is already live. A `log_file` that cannot be opened now aborts the reload with the previous log file still attached and still receiving logs; previously the old handler had already been removed, the failed open was only a warning, file logging stayed dead, and a later reload of the same path was a no-op that never retried it. Relatedly, a `log_file` that could not be opened at startup — a warning there, since there is no previous configuration to keep — is now retried by the next reload even when the setting is unchanged; startup used to record the unopened file as the active one, so the reload saw nothing to do. On failure the replacement clients are closed, the log file opened for the reload is closed (best-effort, so a close error cannot skip the restores that follow it), the search backends' `default` connection alias is handed back to the old client, and the `utils` globals are restored, so the log message is accurate. The clients the reload replaces are now closed last, after the new configuration is live, and a failure to close one is logged without disturbing the reload. [#906](https://github.com/domainaware/parsedmarc/pull/906) fixed the other half of this — building the clients themselves is all-or-nothing.
## 11.0.1
docs/source/usage.md
+30 -9
View File
@@ -1383,7 +1383,7 @@ for that batch have completed. The following settings are reloaded:
- Multi-tenant index prefix domain map (`index_prefix_domain_map` —
the referenced YAML file is re-read on reload)
- DNS and GeoIP settings (`nameservers`, `dns_timeout`, `ip_db_path`,
`ip_db_url`, `offline`, etc.)
`ipinfo_url`, `offline`, etc.)
- Processing flags (`strip_attachment_payloads`, `batch_size`,
`check_timeout`, etc.)
- Log level (`debug`, `verbose`, `warnings`, `silent`)
@@ -1392,16 +1392,37 @@ Mailbox connection settings (IMAP host/credentials, Microsoft Graph,
Gmail API, Maildir path) are **not** reloaded — changing those still
requires a full restart.
On a **successful** reload, existing output client connections are
closed and new ones are created from the updated configuration. The
service then resumes watching with the new settings.
On a **successful** reload, the output clients for the updated
configuration are created first, and the connections they replace are
closed once the new configuration is live. The service then resumes
watching with the new settings.
If the new configuration file contains errors (missing required
settings, unreachable output destinations, etc.), the **entire reload
is aborted** — no output clients are replaced and the previous
configuration remains fully active. This means a typo in one section
will not take down an otherwise working setup. Check the logs for
details:
settings, an output client that cannot be built — a PostgreSQL server,
Kafka broker, or TCP/TLS syslog server that cannot be reached, or an
`[opensearch]` `auth_type` that is not supported — an unreadable reverse
DNS map or PSL overrides file, a `log_file` that cannot be opened for
writing, a non-numeric `batch_size`, etc.), the **entire reload is
aborted** — any output clients built for the new configuration are
closed again, and the previous configuration remains fully active: the
old output clients stay open and connected, and the reverse DNS map, the
PSL overrides, the IP database selection, and every other setting keep
the values they had before the `SIGHUP`. This means a typo in one
section will not take down an otherwise working setup. Unlike startup,
where an unwritable `log_file` is only a warning, on reload it is one of
the errors that abort the reload, so the previous log file keeps
receiving logs — including the one explaining why. A `log_file` that
could not be opened when parsedmarc started (a warning at startup) is
retried by the next reload even if the setting is unchanged. (One thing
an aborted reload does not put back: if it got as far as downloading a
new IP database from a changed `ipinfo_url`, that file stays in the
shared cache directory. Which database parsedmarc *uses* is unchanged,
and the cached file is the same one the next restart would download.)
Elasticsearch and OpenSearch are the exception to "cannot be reached":
their clients connect lazily and the index migration only warns when the
cluster is unreachable, so a reload that points at an unreachable
cluster still commits, and the failure shows up on the first save
instead. Check the logs for details:
```bash
journalctl -u parsedmarc.service -r
parsedmarc/__init__.py
+39 -6
View File
@@ -1223,6 +1223,16 @@ def extract_report(content: bytes | str | BinaryIO) -> str:
Extracts report text from zip- or gzip-compressed content, and returns
plain XML or JSON content decoded as-is.
A caller-supplied file-like object is read from but never closed by
this function, on either the success or the exception path; it is
left open and positioned wherever the function's own reads left it,
so the caller is free to seek(0) and reuse it. A seekable stream is
seeked to position 0 unconditionally (not back to wherever the caller
had left it) after its 6-byte header is sniffed, and then read from
there, so a successful call typically leaves it at EOF. A non-seekable
stream is drained into a buffer this function creates and closes
itself; the caller's stream is left open but exhausted.
Args:
content: The report as a base64-encoded string, file-like object,
or bytes. A string that is not valid base64 is returned
@@ -1232,6 +1242,12 @@ def extract_report(content: bytes | str | BinaryIO) -> str:
str: The extracted text
"""
file_object: BinaryIO | None = None
# True while file_object is a BytesIO this function created itself
# (from a str/bytes input, or as a buffer copied from a non-seekable
# caller stream); it is closed in the ``finally`` below. It is set to
# False when file_object instead aliases a caller-supplied seekable
# stream, which this function must never close.
owns_file_object = True
header: bytes
try:
if isinstance(content, str):
@@ -1265,7 +1281,10 @@ def extract_report(content: bytes | str | BinaryIO) -> str:
raise ParserError("File objects must be opened in binary (rb) mode")
header = bytes(header_raw)
stream.seek(0)
# file_object aliases the caller's own stream here, so it
# must not be closed by this function.
file_object = stream
owns_file_object = False
else:
header_raw = stream.read(6)
if isinstance(header_raw, str):
@@ -1299,7 +1318,7 @@ def extract_report(content: bytes | str | BinaryIO) -> str:
f"Invalid archive file: {error.__str__()}{_exc_origin(error)}"
) from error
finally:
if file_object:
if file_object and owns_file_object:
try:
file_object.close()
except Exception:
@@ -1340,6 +1359,11 @@ def parse_aggregate_report_file(
"""Parses a file at the given path, a file-like object, or bytes as an
aggregate DMARC report
``_input`` is forwarded to ``extract_report()`` unchanged, so a
caller-supplied file-like object is read from but never closed, on
either the success or the exception path; see ``extract_report()``
for exactly how such an object is left positioned afterward.
Args:
_input (str | bytes | IO): A path to a file, a file-like object, or bytes
offline (bool): Do not query online for geolocation or DNS
@@ -2283,6 +2307,11 @@ def parse_report_file(
"""Parses a DMARC aggregate report, DMARC failure report, or SMTP TLS
report from a file at the given path, a file-like object, or bytes
A path is opened and closed by this function. A caller-supplied file
object is read from but never closed, on either the success or the
exception path; it is left open and positioned wherever its own
``read()`` left it, so the caller is free to ``seek(0)`` and reuse it.
Args:
input_ (str | os.PathLike | bytes | BinaryIO): A path to a file,
a file-like object, or bytes
@@ -2332,14 +2361,18 @@ def parse_report_file(
content = file_object.read()
else:
if isinstance(input_, (bytes, bytearray, memoryview)):
# The BytesIO wrapper is created here, so this function owns
# it and closes it once the bytes have been read out.
file_object = BytesIO(bytes(input_))
content = file_object.read()
file_object.close()
else:
# A caller-supplied file-like object is only closed on success,
# matching long-standing behavior; it is left open if read()
# raises.
# A caller-supplied file-like object is never closed by this
# function, on success or failure: the caller may want to
# seek(0) and retry, inspect tell(), or otherwise reuse the
# handle afterward.
file_object = input_
content = file_object.read()
file_object.close()
content = file_object.read()
if content.startswith(MAGIC_ZIP) or content.startswith(MAGIC_GZIP):
content = extract_report(content)
parsedmarc/cli.py
+326 -47
View File
@@ -121,8 +121,10 @@ from parsedmarc.mail import (
)
from parsedmarc.parallel import _parse_report_file_job, parallel_map
from parsedmarc.types import ParsedReport, ParsingResults
import parsedmarc.utils
from parsedmarc.utils import (
InvalidIPinfoAPIKey,
ReverseDNSMap,
configure_ipinfo_api,
get_base_domain,
get_reverse_dns,
@@ -1813,6 +1815,60 @@ def _restore_search_aliases(snapshot: list[tuple[Any, Any, Any]]) -> None:
module.connections.add_connection("default", previous)
def _utils_globals_snapshot() -> dict[str, Any]:
"""Record the ``parsedmarc.utils`` module-level state the config loaders
write, so a failed SIGHUP reload can put it back.
Three globals, enumerated from the loaders' bodies rather than from
their docstrings:
* ``psl_overrides`` -- ``load_psl_overrides()`` clears the list and
repopulates it in place. It declares no ``global``: the list object
itself is the shared state, which is why this is snapshotted by value.
``load_reverse_dns_map()`` calls that loader before it reads anything
of its own, so staging the map into a fresh dict does *not* keep the
overrides list out of the reload's blast radius.
``get_base_domain()`` reads it on every lookup.
* ``_IP_DB_PATH`` -- the only global ``load_ip_db()`` assigns.
``_get_ip_database_path()`` falls back to it whenever the caller's
own ``ip_db_path`` is unset or names a file that is not there, which
is every MMDB lookup in the common case of no ``ip_db_path``.
* ``_IPINFO_API_TOKEN`` -- the only global ``configure_ipinfo_api()``
assigns, and it is assigned *before* the token probe that can fail,
so a rejected probe leaves the new token behind.
``_LAST_LOGGED_IP_DB_PATH`` is deliberately not covered: it is assigned
by ``_get_ip_database_path()``, which none of the loaders the reload
calls reach, and it only decides whether the selected database path is
logged again.
Returns:
dict: The values, to be handed to :func:`_restore_utils_globals`.
"""
return {
"psl_overrides": list(parsedmarc.utils.psl_overrides),
"_IP_DB_PATH": parsedmarc.utils._IP_DB_PATH,
"_IPINFO_API_TOKEN": parsedmarc.utils._IPINFO_API_TOKEN,
}
def _restore_utils_globals(snapshot: dict[str, Any]) -> None:
"""Put the state in *snapshot* back the way it was when it was taken.
``psl_overrides`` is restored in place, not rebound, for the same reason
it is snapshotted by value: ``load_psl_overrides()`` mutates that one
list object, and rebinding ``parsedmarc.utils.psl_overrides`` would
leave any already-bound reference to the old object holding the failed
reload's contents.
Args:
snapshot (dict): The return value of :func:`_utils_globals_snapshot`.
"""
parsedmarc.utils.psl_overrides[:] = snapshot["psl_overrides"]
parsedmarc.utils._IP_DB_PATH = snapshot["_IP_DB_PATH"]
parsedmarc.utils._IPINFO_API_TOKEN = snapshot["_IPINFO_API_TOKEN"]
def _build_output_clients(opts, clients, index_prefix_domain_map=None):
"""Create output clients based on current opts, into *clients*.
@@ -3031,6 +3087,12 @@ def _main():
logger.setLevel(logging.INFO)
if opts.debug:
logger.setLevel(logging.DEBUG)
# The log file currently being written -- what a SIGHUP reload compares
# the new config's log_file against. None when no file is attached,
# including when the configured one could not be opened, so that a
# reload naming the same path tries again once the operator has fixed
# it.
opts.active_log_file = None
if opts.log_file:
try:
fh = logging.FileHandler(opts.log_file, "a")
@@ -3039,10 +3101,9 @@ def _main():
)
fh.setFormatter(formatter)
logger.addHandler(fh)
opts.active_log_file = opts.log_file
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 (
@@ -3634,7 +3695,26 @@ def _main():
logger.info("SIGHUP received, config will reload after current batch")
_reload_requested = False
logger.info("Reloading configuration...")
# Both snapshots are taken before the try, because the rollback
# in the handler needs them and neither can fail: taking the
# search snapshot only resolves an alias (KeyError caught) and
# reads elastic._SERVERLESS, and taking the utils snapshot only
# copies three module attributes.
previous_search_state = _search_alias_snapshot()
previous_utils_state = _utils_globals_snapshot()
# Also bound before the try: the rollback closes whatever was
# built, and an empty dict makes that a no-op when the failure
# came before (or from inside) _init_output_clients().
new_clients: dict[str, Any] = {}
# Bound before the try for the same reason: the rollback closes
# the replacement log file handler if phase 1 opened one, and
# ``None`` means there is nothing to close.
staged_log_handler: logging.FileHandler | None = None
try:
# Phase 1 -- stage every fallible step off to the side.
# Nothing the running configuration reads is written here,
# so any failure below is recoverable by the handler.
#
# 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))
@@ -3644,17 +3724,78 @@ def _main():
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
# Derived values first, because they are the cheapest steps
# to fail and the only ones that write nothing at all: they
# read new_opts and nothing else. int() raises on a
# non-numeric value, and running that before the loaders
# keeps such a failure from reaching load_ip_db(), whose
# download is the one part of phase 1 with a side effect
# outside this process (see the rollback comment).
new_parser_config = _build_parser_config(new_opts)
new_batch_size = (
int(new_opts.mailbox_batch_size)
if new_opts.mailbox_batch_size is not None
else 10
)
new_check_timeout = (
int(new_opts.mailbox_check_timeout)
if new_opts.mailbox_check_timeout is not None
else 30
)
new_max_unsaved_retries = (
int(new_opts.mailbox_max_unsaved_retries)
if new_opts.mailbox_max_unsaved_retries is not None
else 2
)
# Open the replacement log file, if the config names a
# different one. Opening it is the one thing in the logging
# refresh that can fail on a bad path, and constructing the
# handler opens the file immediately (FileHandler's default
# ``delay=False``), so opening it here is what leaves phase
# 2 with nothing fallible to do but the close. Nothing in
# the running configuration is touched: the handler is not
# attached to the logger until the commit. It sits before
# the loaders because its only side effect outside this
# process is opening the new log file for append, which is
# smaller than load_ip_db()'s download.
#
# ``opts`` is still the pre-reload namespace here -- phase 2
# is what copies new_opts onto it -- which is deliberate:
# the comparison is between the file being written now and
# the one the new config names.
#
# Note the asymmetry with startup, where an unwritable log
# file is only a warning: there is no previous configuration
# to keep there. Here there is, so it is a reload failure --
# and the traceback lands in the old log file, which is
# still attached.
old_log_file = getattr(opts, "active_log_file", None)
new_log_file = new_opts.log_file
if old_log_file != new_log_file and new_log_file:
staged_log_handler = logging.FileHandler(new_log_file, "a")
staged_log_handler.setFormatter(
logging.Formatter(
"%(asctime)s - %(levelname)s"
" - [%(filename)s:%(lineno)d] - %(message)s"
)
)
# 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.
#
# Into a fresh dict, not REVERSE_DNS_MAP: load_reverse_dns_map()
# clears the dict it is handed before it has read anything, so
# loading straight into the live map would empty it and leave
# it empty if the read then failed. The overrides list has no
# such seam -- load_psl_overrides() clears and repopulates one
# module-level list in place -- which is what the utils
# snapshot above is for.
staged_reverse_dns_map: ReverseDNSMap = {}
load_reverse_dns_map(
REVERSE_DNS_MAP,
staged_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,
@@ -3664,7 +3805,11 @@ def _main():
)
# Reload the IP database so changes to the
# db path/URL in the config take effect.
# db path/URL in the config take effect. Assigns
# parsedmarc.utils._IP_DB_PATH, which the utils snapshot
# covers, and -- when the download succeeds -- rewrites the
# shared cache file, which nothing can cover. Last but one
# for that reason.
load_ip_db(
always_use_local_file=new_opts.always_use_local_files,
local_file_path=new_opts.ip_db_path,
@@ -3672,9 +3817,13 @@ def _main():
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.
# Re-apply IPinfo API settings, last, because an invalid
# token is fatal even on reload -- the operator asked for it
# -- and nothing should have to be unwound behind it.
# Passing a falsy token disables the API; a rotated token
# picks up here too. The exit is a SystemExit, not an
# Exception: it deliberately skips the rollback below,
# because the process is leaving.
try:
configure_ipinfo_api(
new_opts.ipinfo_api_token if not new_opts.offline else None,
@@ -3682,28 +3831,146 @@ def _main():
except InvalidIPinfoAPIKey as e:
logger.critical(str(e))
sys.exit(1)
except Exception:
# Rollback. Everything phase 1 wrote in this process is
# covered by the two snapshots -- the search backends'
# ``default`` alias and ``elastic._SERVERLESS`` (both
# through _init_output_clients), and the three
# parsedmarc.utils globals the loaders assign -- and all of
# it is put back here, in that order.
#
# One side effect is outside them and cannot be undone: a
# load_ip_db() whose download succeeds rewrites the shared
# cache file at <tmp>/parsedmarc/ipinfo_lite.mmdb before it
# assigns _IP_DB_PATH. Phase 1 runs that step last but one
# so that as little as possible can fail behind it, and the
# file it leaves is the same refresh startup performs on
# every launch -- the configured URL's current database --
# so the cost of not undoing it is bounded to that.
#
# ``Exception``, where _init_output_clients catches
# ``BaseException`` (#906): that function also runs at
# startup, before the signal handlers exist, so a Ctrl-C can
# land inside it. Here it cannot -- watch mode has replaced
# SIGINT with a handler that sets a flag and, on a second
# press, calls os._exit(130) -- and the one BaseException
# that does reach this point, the SystemExit from an invalid
# IPinfo token, must not be rolled back.
#
# The staged log file handler, if one was opened, is closed
# here too, best-effort: it was never attached to the
# logger, so no record ever reached it and there is
# nothing buffered to lose, but the close itself can still
# raise (a full disk, a stale network mount), and that
# error is logged and swallowed so it cannot skip the two
# restores below it.
#
# Close the new clients *before* restoring the alias, never
# after: a fully built _ElasticsearchHandle owns the new
# client and releases the alias as it closes, so closing
# first leaves the alias unset for the restore to re-register
# the old client into. Restoring first would hand the alias
# back and then close the handle, which would be safe only
# because of how _ElasticsearchHandle.close() happens to
# behave today (see _init_output_clients, and #902).
#
# Per-step end state, reading down phase 1 -- in every case
# `clients` is still the old dict with every old client open,
# REVERSE_DNS_MAP still holds the pre-reload entries (it was
# never handed to the loader), and opts, parser_config and
# the three watch parameters are untouched, because all of
# those are written in phase 2:
#
# _load_config / _parse_config: no state written yet.
# new_clients is still {}, so the close is a no-op; the
# alias and the utils globals still hold what the snapshots
# recorded, so both restores are no-ops too.
# _init_output_clients: rolled itself back before re-raising
# (#906), and never assigned new_clients, so this is the
# same no-op pair -- restoring an unchanged snapshot twice
# is harmless.
# _build_parser_config / the int() conversions: the new
# clients hold the alias, so they are closed and the old
# client is registered again. The utils globals were not
# reached, so their restore is still a no-op.
# The log file open: as above, and the handler was never
# attached to the logger, so it has nothing buffered to
# lose -- the logger still holds the old FileHandler,
# which phase 2 never got to remove. The close itself is
# best-effort (logged and swallowed), so a close that
# fails cannot skip the two restores below it.
# load_reverse_dns_map: as above, plus psl_overrides, which
# load_psl_overrides() cleared and may have refilled from
# the new config; restoring it by value is load-bearing
# here, and the staged map dict is simply dropped.
# load_ip_db: as above. Each of its four assignments to
# _IP_DB_PATH is the last thing on its path -- three are
# followed by a return and the fourth, the bundled
# fallback, by a log line and the end of the function --
# so a raise from this loader leaves it unchanged.
# configure_ipinfo_api: the last step, so a _IP_DB_PATH or
# _IPINFO_API_TOKEN that differs from the snapshot can
# only be rolled back from here. In practice this may be
# unreachable: the documented failure is
# InvalidIPinfoAPIKey, which exits above without rolling
# back, and _ipinfo_api_lookup() swallows every network
# and decoding error. Those two restores are therefore
# defensive -- a rollback's job is to leave state as it
# found it, and whatever step is added after this one
# makes them load-bearing again. _utils_globals_snapshot's
# own tests are what guard them.
_close_output_clients(new_clients)
if staged_log_handler is not None:
try:
staged_log_handler.close()
except Exception as close_error:
logger.warning(
"Unable to close the log file opened for the "
f"reload: {close_error}"
)
_restore_search_aliases(previous_search_state)
_restore_utils_globals(previous_utils_state)
logger.exception(
"Config reload failed, continuing with previous config"
)
else:
# Phase 2 -- all steps succeeded; commit the changes
# atomically. Nothing from here to the end of this block may
# raise, since there is no going back once the first
# statement lands: the commit itself is assignments only, and
# the logging refresh that follows it guards the one call
# that can fail, closing the replaced log file -- the
# replacement was opened back in phase 1. That is also why
# this is an ``else`` and not the tail of the ``try`` -- a
# rollback running over committed state would close the
# clients that are now live.
# In place, never rebound. Rebinding would only move this
# module's own name -- cli.py imports REVERSE_DNS_MAP from
# parsedmarc, so `REVERSE_DNS_MAP = staged` here would
# desync it from parsedmarc.REVERSE_DNS_MAP and from
# config.REVERSE_DNS_MAP, which is the object
# ParserConfig.__setstate__ rebinds to when a worker
# unpickles a config. The configs that name this dict are
# built elsewhere and keep whichever object they were given:
# _build_parser_config() and parsedmarc._resolve_config()
# pass it explicitly (the dataclass field's own default is a
# fresh dict, via default_factory), and new_parser_config
# was built back in phase 1 holding the pre-reload object.
REVERSE_DNS_MAP.clear()
REVERSE_DNS_MAP.update(staged_reverse_dns_map)
for k, v in vars(new_opts).items():
setattr(opts, k, v)
parser_config = _build_parser_config(opts)
old_clients = clients
clients = new_clients
index_prefix_domain_map = new_index_prefix_domain_map
parser_config = new_parser_config
# 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
)
mailbox_batch_size_value = new_batch_size
mailbox_check_timeout_value = new_check_timeout
mailbox_max_unsaved_retries_value = new_max_unsaved_retries
# Update log level
logger.setLevel(logging.ERROR)
@@ -3714,36 +3981,48 @@ def _main():
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
# Refresh FileHandler if log_file changed. Both names were
# computed in phase 1, against the configuration that was
# running then.
if old_log_file != new_log_file:
# Remove old FileHandlers
for h in list(logger.handlers):
if isinstance(h, logging.FileHandler):
h.close()
close_error = None
try:
h.close()
except Exception as error_:
# Flushing the replaced log file can fail --
# a full disk, a stale network mount. The
# reloaded configuration is already live by
# this point, so raising would kill the
# watcher over a log file, and rolling back
# to report a failed reload would be a lie.
close_error = error_
# Detach first, then report: the handler that
# just failed to close is no place to send the
# record describing that failure.
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}")
if close_error is not None:
logger.warning(
f"Unable to close the log file: {close_error}"
)
# Attach the handler phase 1 opened, if the new
# configuration names a log file at all.
if staged_log_handler is not None:
logger.addHandler(staged_log_handler)
opts.active_log_file = new_log_file
_configure_dependency_logging(logger.level)
# Phase 3 -- the new configuration is live and correct, so
# the clients it replaced can go. Last, and best effort:
# _close_output_clients() logs close errors and swallows
# them, so a dying connection on the way out cannot undo
# the commit above.
_close_output_clients(old_clients)
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
tests/test_cli.py
+926
View File
@@ -8,6 +8,7 @@ import io
import json
import logging
import os
import shutil
import signal
import stat
import sys
@@ -37,6 +38,8 @@ import parsedmarc.elastic
import parsedmarc.log
import parsedmarc.mail
import parsedmarc.opensearch as opensearch_module
import parsedmarc.resources.ipinfo
import parsedmarc.utils
from parsedmarc.types import AggregateReport, ParsedReport
SAMPLE_AGGREGATE_REPORT_PATH = (
@@ -3898,6 +3901,929 @@ watch = true
self.assertEqual(second_config.dns_timeout, 42.0)
class _RaisingCloseClient:
"""An output client whose close() fails, e.g. a Kafka producer whose
broker connection is already gone. Unlike the search handles, a plain
client like this lets the error reach _close_output_clients(), which is
where the SIGHUP reload closes the configuration it just replaced."""
def __init__(self):
self.close_count = 0
def close(self):
self.close_count += 1
raise RuntimeError("broker connection already gone")
class _RaisingCloseFileHandler(logging.FileHandler):
"""A log FileHandler whose close() fails, the way flushing to a full
disk or a stale network mount does. Counts the attempts so a test can
tell "close() was tried and failed" from "close() was never called"."""
def __init__(self, filename):
super().__init__(filename, "a")
self.close_attempts = 0
def close(self):
self.close_attempts += 1
raise OSError("No space left on device")
@unittest.skipUnless(
hasattr(signal, "SIGHUP"),
"SIGHUP not available on this platform",
)
class TestSighupReloadAtomicity(unittest.TestCase):
"""A SIGHUP reload is all-or-nothing.
The reload builds the replacement output clients first and only then
re-reads the reverse DNS map, the PSL overrides and the IP database,
re-applies opts, and rebuilds the ParserConfig. Committing the clients
before those steps ran meant a failure in any of them logged "Config
reload failed, continuing with previous config" while the *new* clients
were already live, the old ones were closed and unrecoverable, and the
reverse DNS map had been emptied by the loader that then failed to
refill it. #906 made _init_output_clients() itself all-or-nothing; these
tests cover the same guarantee for the whole reload.
Each test drives the real _main() through one SIGHUP and reads the state
the *next* watch_inbox() call sees, which is the first point at which the
running configuration is observable from outside.
"""
def setUp(self):
self._stdout_patch = patch("sys.stdout", new_callable=io.StringIO)
self._stderr_patch = patch("sys.stderr", new_callable=io.StringIO)
self._stdout_patch.start()
self._stderr_patch.start()
self.addCleanup(self._stderr_patch.stop)
self.addCleanup(self._stdout_patch.stop)
# _main() sets the parsedmarc logger's level and swaps its
# FileHandlers, and _configure_dependency_logging() copies its
# handler list onto every dependency logger -- including the
# temporary handler assertLogs installs. Snapshot all of them here,
# before anything runs, so no test leaves a dead handler behind for
# the next one.
loggers = [parsedmarc.log.logger] + [
logging.getLogger(name) for name in parsedmarc.cli._DEPENDENCY_LOGGERS
]
saved_logging = [
(lg, lg.level, list(lg.handlers), lg.propagate, lg.disabled)
for lg in loggers
]
def _restore_logging():
for lg, level, handlers, propagate, disabled in saved_logging:
for stray in lg.handlers:
# A reload that changes log_file leaves an open
# FileHandler behind; close it rather than waiting for
# the garbage collector to release the descriptor.
if isinstance(stray, logging.FileHandler) and (
stray not in handlers
):
stray.close()
lg.setLevel(level)
lg.handlers[:] = handlers
lg.propagate = propagate
lg.disabled = disabled
self.addCleanup(_restore_logging)
parsedmarc.log.logger.disabled = False
# _main() installs SIGHUP/SIGTERM/SIGINT handlers that would
# otherwise outlive the test and answer signals raised by later
# ones. Registered before the first _main() call installs them.
for sig in (signal.SIGHUP, signal.SIGTERM, signal.SIGINT):
self.addCleanup(signal.signal, sig, signal.getsignal(sig))
# Process-wide state a reload writes. Restored explicitly rather
# than through _utils_globals_snapshot()/_restore_utils_globals(),
# which are themselves under test here.
original_map = dict(parsedmarc.REVERSE_DNS_MAP)
def _restore_map():
parsedmarc.REVERSE_DNS_MAP.clear()
parsedmarc.REVERSE_DNS_MAP.update(original_map)
self.addCleanup(_restore_map)
saved_overrides = list(parsedmarc.utils.psl_overrides)
saved_ip_db_path = parsedmarc.utils._IP_DB_PATH
saved_ipinfo_token = parsedmarc.utils._IPINFO_API_TOKEN
def _restore_utils():
parsedmarc.utils.psl_overrides[:] = saved_overrides
parsedmarc.utils._IP_DB_PATH = saved_ip_db_path
parsedmarc.utils._IPINFO_API_TOKEN = saved_ipinfo_token
self.addCleanup(_restore_utils)
# A distinctive map entry, so "the map still holds what it held
# before the reload" is an assertion about this dict and not about
# whatever the bundled CSV happens to contain.
parsedmarc.REVERSE_DNS_MAP.clear()
parsedmarc.REVERSE_DNS_MAP["before.example.com"] = {
"name": "Before Reload",
"type": "test",
}
def _write_file(self, contents, suffix):
with tempfile.NamedTemporaryFile("w", suffix=suffix, delete=False) as f:
f.write(contents)
path = f.name
self.addCleanup(lambda: os.path.exists(path) and os.remove(path))
return path
def _fake_search_clients(self, extra_old_clients=None):
"""Stand in for _init_output_clients() against the *real* search
connection registry.
The real thing calls ``elastic.set_hosts()``, which registers the
client it builds under the process-wide ``default`` alias the moment
it is constructed, and stores it in the returned dict wrapped in an
_ElasticsearchHandle. That registry and that handle are the seam the
reload's rollback is written against, so both are real here and only
the cluster connection is a stand-in.
Returns:
tuple: the side effect to give the _init_output_clients patch,
the old connection, and the new one.
"""
connections = parsedmarc.elastic.connections
_isolate_default_alias(self, connections)
old_conn = _FakeSearchConnection("old")
new_conn = _FakeSearchConnection("new")
self._search_conns = (old_conn, new_conn)
calls = []
def init_clients(opts, index_prefix_domain_map=None):
conn = old_conn if not calls else new_conn
calls.append(conn)
connections.add_connection("default", cast(Elasticsearch, conn))
clients = {
"elasticsearch": parsedmarc.cli._ElasticsearchHandle(conn),
}
if conn is old_conn and extra_old_clients:
clients.update(extra_old_clients)
return clients
return init_clients, old_conn, new_conn
def _current_alias(self):
try:
return parsedmarc.elastic.connections.get_connection("default")
except KeyError:
return None
def _capture_state(self):
"""Everything the reload can write, as the running configuration
sees it. Read at the moment the reload is over rather than after
_main() returns, because the trailing _close_output_clients() would
otherwise blur which clients the reload itself closed."""
old_conn, new_conn = self._search_conns
return SimpleNamespace(
reverse_dns_map=dict(parsedmarc.REVERSE_DNS_MAP),
psl_overrides=list(parsedmarc.utils.psl_overrides),
ip_db_path=parsedmarc.utils._IP_DB_PATH,
ipinfo_api_token=parsedmarc.utils._IPINFO_API_TOKEN,
alias=self._current_alias(),
old_closed=old_conn.close_count,
new_closed=new_conn.close_count,
# Read here rather than after _main() returns: assertLogs()
# restores the logger's handler list on the way out, so the
# handlers a reload attached or detached are only visible from
# inside the run.
log_handlers=list(parsedmarc.log.logger.handlers),
)
def _drive_reload(
self, initial_config, reload_config, init_clients, before_signal=None
):
"""Run _main() through exactly one SIGHUP reload.
The first watch_inbox() call rewrites the config file the way an
operator would, runs *before_signal* if one was given -- the seam
for anything that has to be set up inside the run, after
assertLogs() has replaced the logger's handlers -- and signals
SIGHUP. The second records the state the reload left behind and
signals SIGTERM, so the watch loop breaks and _main() returns
normally -- which runs its trailing
_close_output_clients(clients). That trailing close is what makes
"``clients`` still names the old dict" observable: the clients it
closes are, by definition, the ones the run ended with.
Returns:
tuple: the captured state, the watch_inbox mock, and the
captured log output.
"""
cfg_path = self._write_file(initial_config, ".ini")
captured = {}
watch_calls = []
def watch_side_effect(*args, **kwargs):
watch_calls.append(kwargs)
if len(watch_calls) == 1:
with open(cfg_path, "w") as f:
f.write(reload_config)
if before_signal is not None:
before_signal()
os.kill(os.getpid(), signal.SIGHUP)
return
captured["state"] = self._capture_state()
os.kill(os.getpid(), signal.SIGTERM)
with (
patch("parsedmarc.cli.IMAPConnection", return_value=object()),
patch(
"parsedmarc.cli.get_dmarc_reports_from_mailbox",
return_value={
"aggregate_reports": [],
"failure_reports": [],
"smtp_tls_reports": [],
},
),
patch(
"parsedmarc.cli.watch_inbox", side_effect=watch_side_effect
) as mock_watch,
patch("parsedmarc.cli._init_output_clients", side_effect=init_clients),
patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]),
self.assertLogs(parsedmarc.log.logger, level="INFO") as logs,
):
parsedmarc.cli._main()
self.assertEqual(mock_watch.call_count, 2)
self.assertIn("state", captured)
return captured["state"], mock_watch, logs.output
_CONFIG_TEMPLATE = """[general]
debug = true
offline = true
dns_timeout = {dns_timeout}
{general_extra}
[imap]
host = imap.example.com
user = user
password = pass
[mailbox]
watch = true
batch_size = {batch_size}
"""
def _initial_config(self, general_extra=""):
"""The configuration the run starts with: batch_size 3 and a 5
second DNS timeout."""
return self._CONFIG_TEMPLATE.format(
dns_timeout="5.0", general_extra=general_extra, batch_size=3
)
def _reload_config(self, general_extra=""):
"""What the operator edits the file to before signalling: every
value here differs from the initial config, so committing half of
them would show."""
return self._CONFIG_TEMPLATE.format(
dns_timeout="42.0", general_extra=general_extra, batch_size=9
)
def _assert_previous_config_still_running(self, state, mock_watch, logs):
"""The claim "continuing with previous config" makes, item by
item."""
self.assertTrue(
any("continuing with previous config" in line for line in logs),
logs,
)
self.assertEqual(
state.reverse_dns_map,
{"before.example.com": {"name": "Before Reload", "type": "test"}},
)
second_call = mock_watch.call_args_list[1].kwargs
self.assertEqual(second_call["batch_size"], 3)
self.assertEqual(second_call["config"].dns_timeout, 5.0)
def testFailedReloadKeepsOldClientsWhenMapFileIsMissing(self):
"""A reverse DNS map path that does not exist fails the reload after
the replacement clients were built: the new clients must be closed
and the old ones left running and still registered under the
``default`` alias.
open() on the missing file is the injected failure -- the loader
catches httpx and CSV errors but not a missing local file, which is
exactly what a typo in ``reverse_dns_map_path`` produces.
"""
init_clients, old_conn, new_conn = self._fake_search_clients()
state, mock_watch, logs = self._drive_reload(
self._initial_config(),
self._reload_config(
general_extra="local_reverse_dns_map_path = /nonexistent/map.csv\n"
),
init_clients,
)
# The new clients were built, then closed and stripped of the alias.
self.assertEqual(state.new_closed, 1)
# The old ones are untouched and still hold the alias, so every save
# between now and the next reload reaches the cluster opts describes.
self.assertEqual(state.old_closed, 0)
self.assertIs(state.alias, old_conn)
self._assert_previous_config_still_running(state, mock_watch, logs)
# _main()'s trailing close ran against the old clients, which is
# only possible if `clients` still named them -- and it did not
# close the discarded clients a second time.
self.assertEqual(old_conn.close_count, 1)
self.assertEqual(new_conn.close_count, 1)
def testFailedReloadRestoresPslOverrides(self):
"""load_psl_overrides() empties the module-level list before it
reads anything, so a missing overrides file leaves it empty -- and
load_reverse_dns_map() calls it first thing, which is why staging
the map into a fresh dict is not enough on its own.
The rollback must put the list back by value, or every later
get_base_domain() call would fold domains without the overrides.
"""
init_clients, old_conn, new_conn = self._fake_search_clients()
overrides_before = list(parsedmarc.utils.psl_overrides)
self.assertGreater(len(overrides_before), 0)
state, mock_watch, logs = self._drive_reload(
self._initial_config(),
self._reload_config(
general_extra="local_psl_overrides_path = /nonexistent/psl_overrides.txt\n"
),
init_clients,
)
self.assertEqual(state.psl_overrides, overrides_before)
self.assertEqual(state.new_closed, 1)
self.assertEqual(state.old_closed, 0)
self.assertIs(state.alias, old_conn)
self._assert_previous_config_still_running(state, mock_watch, logs)
def testFailedReloadKeepsOldClientsWhenIpDatabaseIsUnavailable(self):
"""load_ip_db() falls back to the bundled MMDB when the configured
file, the download and the cache are all unavailable; an install
without that bundled resource makes the fallback itself raise.
This is the deepest point a reload can fail and still be rolled
back: only configure_ipinfo_api() runs after it, and its documented
failure (an invalid token) exits the process rather than returning
to the watch loop. By the time it fails, the replacement clients
hold the ``default`` alias, the reverse DNS map has been staged from
a different file, and load_psl_overrides() has already replaced the
live overrides list -- so this is also where the most restores are
load-bearing at once.
"""
init_clients, old_conn, new_conn = self._fake_search_clients()
overrides_before = list(parsedmarc.utils.psl_overrides)
self.assertNotIn("reload-only.example", overrides_before)
# Files the reload names and the run must end up not using.
reload_map_path = self._write_file(
"base_reverse_dns,name,type\nafter.example.com,After Reload,test\n",
".csv",
)
reload_overrides_path = self._write_file("reload-only.example\n", ".txt")
# The IP database the run starts with, and keeps.
ip_db_path = self._write_file("not really an mmdb", ".mmdb")
# An empty cache directory, so load_ip_db() cannot answer from a
# copy some earlier run downloaded and must reach the bundled file.
cache_dir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, cache_dir, True)
real_files = parsedmarc.utils.files
def files_without_bundled_mmdb(package):
# The importlib.resources boundary: the maps package (the
# reverse DNS map and the PSL overrides, both loaded earlier in
# the same reload) still resolves normally.
if package is parsedmarc.resources.ipinfo:
raise ModuleNotFoundError("parsedmarc.resources.ipinfo")
return real_files(package)
with (
patch("parsedmarc.utils.files", side_effect=files_without_bundled_mmdb),
patch("parsedmarc.utils.tempfile.gettempdir", return_value=cache_dir),
):
state, mock_watch, logs = self._drive_reload(
# Startup resolves the local file directly and never reaches
# the bundled fallback, so only the reload -- which drops
# ip_db_path -- fails.
self._initial_config(f"ip_db_path = {ip_db_path}\n"),
self._reload_config(
general_extra=(
f"local_reverse_dns_map_path = {reload_map_path}\n"
f"local_psl_overrides_path = {reload_overrides_path}\n"
)
),
init_clients,
)
# The two restores that carry weight at this depth: the overrides
# list load_psl_overrides() had already replaced, and the map, which
# was staged into a dict that is simply dropped -- the live one
# still holds the pre-reload entry rather than the reload's file.
self.assertEqual(state.psl_overrides, overrides_before)
self.assertNotIn("after.example.com", state.reverse_dns_map)
# load_ip_db() assigns _IP_DB_PATH only as the last thing on each
# of its paths, so a raise from it leaves the previous selection in
# place rather than a half-applied one; that the snapshot can put a
# *changed* value back is covered by TestUtilsGlobalsSnapshot.
self.assertEqual(state.ip_db_path, ip_db_path)
self.assertEqual(state.new_closed, 1)
self.assertEqual(state.old_closed, 0)
self.assertIs(state.alias, old_conn)
self._assert_previous_config_still_running(state, mock_watch, logs)
def testSuccessfulReloadCommitsEveryStagedValue(self):
"""The other half: when every step succeeds, the staged state is
what the run continues with -- new clients live and holding the
alias, old clients closed, the reverse DNS map replaced by the file
the new config names, and the new watch parameters and ParserConfig
in use.
(Passes against the pre-fix code too, necessarily: a reload with
nothing to roll back has to end in the same place either way. It is
here so that the failure tests above cannot be satisfied by a
reload that quietly stopped applying anything.)"""
init_clients, old_conn, new_conn = self._fake_search_clients()
map_path = self._write_file(
"base_reverse_dns,name,type\nafter.example.com,After Reload,test\n",
".csv",
)
state, mock_watch, logs = self._drive_reload(
self._initial_config(),
self._reload_config(
general_extra=f"local_reverse_dns_map_path = {map_path}\n"
),
init_clients,
)
self.assertTrue(
any("Configuration reloaded successfully" in line for line in logs),
logs,
)
self.assertEqual(
state.reverse_dns_map,
{"after.example.com": {"name": "After Reload", "type": "test"}},
)
self.assertIs(state.alias, new_conn)
self.assertEqual(state.old_closed, 1)
self.assertEqual(state.new_closed, 0)
second_call = mock_watch.call_args_list[1].kwargs
self.assertEqual(second_call["batch_size"], 9)
self.assertEqual(second_call["config"].dns_timeout, 42.0)
# The ParserConfig keeps the live map object, so the in-place
# refresh above has to reach the parser through it.
self.assertIs(second_call["config"].reverse_dns_map, parsedmarc.REVERSE_DNS_MAP)
# _main()'s trailing close ran against the new clients.
self.assertEqual(new_conn.close_count, 1)
def testOldClientCloseFailureDoesNotUndoTheReload(self):
"""Closing the replaced clients is the last step and is best effort:
a client whose close() raises is logged and the reloaded
configuration stays live.
(This one passes against the pre-fix code too -- there the same
close ran before the swap, and _close_output_clients() swallowed the
error either way. It guards the move: closing the old clients last
must not put the commit at risk.)
"""
raising_client = _RaisingCloseClient()
init_clients, old_conn, new_conn = self._fake_search_clients(
extra_old_clients={"kafka": raising_client}
)
map_path = self._write_file(
"base_reverse_dns,name,type\nafter.example.com,After Reload,test\n",
".csv",
)
state, mock_watch, logs = self._drive_reload(
self._initial_config(),
self._reload_config(
general_extra=f"local_reverse_dns_map_path = {map_path}\n"
),
init_clients,
)
self.assertEqual(raising_client.close_count, 1)
self.assertTrue(
any("Error closing kafka" in line for line in logs),
logs,
)
self.assertTrue(
any("Configuration reloaded successfully" in line for line in logs),
logs,
)
self.assertIs(state.alias, new_conn)
self.assertEqual(state.new_closed, 0)
self.assertEqual(state.old_closed, 1)
self.assertEqual(
state.reverse_dns_map,
{"after.example.com": {"name": "After Reload", "type": "test"}},
)
self.assertEqual(mock_watch.call_args_list[1].kwargs["batch_size"], 9)
def testLogFileCloseFailureDoesNotUndoTheReload(self):
"""Swapping the log file is the last thing the commit does, and
closing the replaced FileHandler can fail on a full disk or a stale
mount. Opening the replacement happens back in phase 1, so the close
is the only statement left in the commit that can raise, and it must
not take the reload -- which is already live by then -- with it."""
init_clients, old_conn, new_conn = self._fake_search_clients()
map_path = self._write_file(
"base_reverse_dns,name,type\nafter.example.com,After Reload,test\n",
".csv",
)
log_path = self._write_file("", ".log")
stale_handler = _RaisingCloseFileHandler(self._write_file("", ".log"))
# The subclass's close() raises; the unbound FileHandler.close()
# is what actually releases the descriptor.
self.addCleanup(logging.FileHandler.close, stale_handler)
state, mock_watch, logs = self._drive_reload(
self._initial_config(),
self._reload_config(
general_extra=(
f"local_reverse_dns_map_path = {map_path}\nlog_file = {log_path}\n"
)
),
init_clients,
# Inside the run: assertLogs() replaces the logger's handler
# list on the way in, so a handler attached before _main()
# started would not be there for the reload to close.
before_signal=lambda: parsedmarc.log.logger.addHandler(stale_handler),
)
self.assertEqual(stale_handler.close_attempts, 1)
self.assertTrue(
any("Unable to close the log file" in line for line in logs), logs
)
self.assertNotIn(stale_handler, state.log_handlers)
# The commit stands: new clients live and holding the alias, old
# ones closed, map replaced, new watch parameters in use.
self.assertTrue(
any("Configuration reloaded successfully" in line for line in logs), logs
)
self.assertIs(state.alias, new_conn)
self.assertEqual(state.new_closed, 0)
self.assertEqual(state.old_closed, 1)
self.assertEqual(
state.reverse_dns_map,
{"after.example.com": {"name": "After Reload", "type": "test"}},
)
self.assertEqual(mock_watch.call_args_list[1].kwargs["batch_size"], 9)
def testUnwritableLogFileAbortsTheReload(self):
"""The replacement log file is opened in phase 1, with the previous
FileHandler still attached, so a ``log_file`` that cannot be opened
is a reload failure rather than a warning logged after the old
handler is already gone.
Everything else in this reload would have succeeded -- it names a
readable map file and a new batch size -- so the previous
configuration still running afterwards is the log file open's doing.
The old handler is still attached and still open, which is what
makes the "Config reload failed" record reach the log file the
operator is tailing, and no second FileHandler was left behind by
the open that failed.
"""
init_clients, old_conn, new_conn = self._fake_search_clients()
map_path = self._write_file(
"base_reverse_dns,name,type\nafter.example.com,After Reload,test\n",
".csv",
)
# /proc has no subdirectories that can be created, so opening a file
# under a nonexistent one fails the way a bad ``log_file`` path
# does. Same trick as
# test_unwritable_log_file_logs_warning_does_not_raise.
unwritable_log = "/proc/nonexistent/parsedmarc-reload.log"
live_handler = logging.FileHandler(self._write_file("", ".log"), "a")
self.addCleanup(live_handler.close)
self.addCleanup(parsedmarc.log.logger.removeHandler, live_handler)
state, mock_watch, logs = self._drive_reload(
self._initial_config(),
self._reload_config(
general_extra=(
f"local_reverse_dns_map_path = {map_path}\n"
f"log_file = {unwritable_log}\n"
)
),
init_clients,
# Inside the run, for the same reason as the close-failure test:
# assertLogs() replaces the logger's handler list on the way in.
before_signal=lambda: parsedmarc.log.logger.addHandler(live_handler),
)
self.assertTrue(
any(
"Config reload failed, continuing with previous config" in line
for line in logs
),
logs,
)
self.assertFalse(
any("Configuration reloaded successfully" in line for line in logs),
logs,
)
# The log file the operator is tailing is still attached and still
# open, so the failure was reported into it -- and it is the only
# FileHandler on the logger, so nothing leaked from the failed open.
self.assertEqual(
[h for h in state.log_handlers if isinstance(h, logging.FileHandler)],
[live_handler],
)
self.assertTrue(
live_handler.stream is not None and not live_handler.stream.closed,
"the previous log file handler was closed",
)
# The rest of the reload was rolled back: the replacement clients
# are closed, the old ones still hold the alias, the map still holds
# its pre-reload entry, and the watch parameters are the originals.
self.assertEqual(state.new_closed, 1)
self.assertEqual(state.old_closed, 0)
self.assertIs(state.alias, old_conn)
self.assertNotIn("after.example.com", state.reverse_dns_map)
self._assert_previous_config_still_running(state, mock_watch, logs)
def testFailedReloadClosesTheStagedLogFileHandler(self):
"""A reload that opens the replacement log file and then fails on a
later step has to close the file it opened: the handler was never
attached to the logger, so closing it is the whole undo, and leaving
it open would leak a descriptor on every such reload.
The missing map path is the injected failure, as in
testFailedReloadKeepsOldClientsWhenMapFileIsMissing; this reload also
names a writable log file, so the staged handler exists by the time
that failure lands. Capturing it needs a real FileHandler subclass
rather than a mock, because the code the reload runs afterwards
checks handlers with isinstance().
"""
init_clients, old_conn, new_conn = self._fake_search_clients()
log_path = self._write_file("", ".log")
opened: list[logging.FileHandler] = []
class _CapturingFileHandler(logging.FileHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
opened.append(self)
with patch("logging.FileHandler", _CapturingFileHandler):
state, mock_watch, logs = self._drive_reload(
self._initial_config(),
self._reload_config(
general_extra=(
"local_reverse_dns_map_path = /nonexistent/map.csv\n"
f"log_file = {log_path}\n"
)
),
init_clients,
)
# The initial config names no log file, so the staging step is the
# only thing in the run that opens one.
self.assertEqual(len(opened), 1)
staged = opened[0]
self.addCleanup(staged.close)
self.assertEqual(staged.baseFilename, os.path.abspath(log_path))
# FileHandler.close() drops the stream it owns, so a None stream is
# the rollback having closed this handler.
self.assertIsNone(staged.stream)
# It was never attached, so there was nothing else to undo -- and
# the logger is still writing wherever it was before the SIGHUP.
self.assertNotIn(staged, state.log_handlers)
self.assertEqual(state.new_closed, 1)
self.assertEqual(state.old_closed, 0)
self.assertIs(state.alias, old_conn)
self._assert_previous_config_still_running(state, mock_watch, logs)
def testFailedReloadSurvivesAStagedLogFileHandlerThatCannotClose(self):
"""A rollback whose staged-handler close fails must still restore
the search alias and finish as "continuing with previous config",
instead of the close error escaping the handler and taking the
watcher down with the alias left unrestored.
Same missing-map injected failure as
testFailedReloadClosesTheStagedLogFileHandler, but the staged
handler here is a _RaisingCloseFileHandler, so the rollback's own
``staged_log_handler.close()`` call raises. Before the fix, that
raise propagated out of the ``except Exception:`` block, so
_restore_search_aliases() and _restore_utils_globals() below it
never ran and the exception itself replaced the "Config reload
failed" log message.
"""
init_clients, old_conn, new_conn = self._fake_search_clients()
log_path = self._write_file("", ".log")
opened: list[_RaisingCloseFileHandler] = []
class _CapturingRaisingCloseFileHandler(_RaisingCloseFileHandler):
# cli.py constructs the handler as FileHandler(path, "a"), and
# _RaisingCloseFileHandler always opens in append mode, so the
# mode argument only needs accepting.
def __init__(self, filename, mode="a"):
super().__init__(filename)
opened.append(self)
with patch("logging.FileHandler", _CapturingRaisingCloseFileHandler):
state, mock_watch, logs = self._drive_reload(
self._initial_config(),
self._reload_config(
general_extra=(
"local_reverse_dns_map_path = /nonexistent/map.csv\n"
f"log_file = {log_path}\n"
)
),
init_clients,
)
# The initial config names no log file, so the staging step is the
# only thing in the run that opens one.
self.assertEqual(len(opened), 1)
self.addCleanup(logging.FileHandler.close, opened[0])
self.assertEqual(opened[0].close_attempts, 1)
self.assertTrue(
any(
"Unable to close the log file opened for the reload" in line
for line in logs
),
logs,
)
# The close error was logged and swallowed rather than escaping, so
# the rollback still finished and reported itself normally.
self._assert_previous_config_still_running(state, mock_watch, logs)
# ... and the two restores that come after the close in source order
# still ran: the new clients are closed, the old ones still hold the
# alias, and the staged handler was never attached to the logger.
self.assertEqual(state.new_closed, 1)
self.assertEqual(state.old_closed, 0)
self.assertIs(state.alias, old_conn)
self.assertNotIn(opened[0], state.log_handlers)
def testLogFileUnopenableAtStartupIsRetriedOnReload(self):
"""A ``log_file`` that could not be opened when parsedmarc started is
only a warning there -- startup has no previous configuration to
keep, so it logs "Unable to write to log file" and carries on. The
next reload has to retry that open even though the operator never
changed the setting, once the directory that was missing exists.
Before this fix, startup recorded the unopened path onto
``opts.active_log_file`` anyway, so the reload's
``old_log_file != new_log_file`` comparison saw the same path on
both sides and treated the log file as nothing to refresh -- file
logging stayed dead until the process was restarted, even after the
directory was created. Recording ``None`` when the open fails makes
the comparison true, so the reload opens it for real.
"""
tmpdir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, tmpdir, True)
log_path = os.path.join(tmpdir, "later", "parsedmarc.log")
init_clients, old_conn, new_conn = self._fake_search_clients()
state, mock_watch, logs = self._drive_reload(
self._initial_config(general_extra=f"log_file = {log_path}\n"),
self._reload_config(general_extra=f"log_file = {log_path}\n"),
init_clients,
# The operator fixes the missing directory before the SIGHUP
# that _drive_reload sends right after this runs.
before_signal=lambda: os.makedirs(os.path.dirname(log_path)),
)
self.assertTrue(
any("Unable to write to log file" in line for line in logs), logs
)
self.assertTrue(
any("Configuration reloaded successfully" in line for line in logs),
logs,
)
file_handlers = [
h for h in state.log_handlers if isinstance(h, logging.FileHandler)
]
self.assertEqual(len(file_handlers), 1)
handler = file_handlers[0]
# assertLogs() already restored the logger's own handler list by
# the time this runs, so this handler is not attached to anything
# -- closing it is enough, with no removeHandler needed alongside
# it (contrast testUnwritableLogFileAbortsTheReload's live_handler,
# which stays attached and needs both).
self.addCleanup(handler.close)
self.assertEqual(handler.baseFilename, os.path.abspath(log_path))
self.assertEqual(state.old_closed, 1)
self.assertEqual(state.new_closed, 0)
self.assertIs(state.alias, new_conn)
self.assertTrue(os.path.exists(log_path))
def testLogFileOpenedAtStartupIsNotReopenedByAnUnchangedReload(self):
"""The other half of testLogFileUnopenableAtStartupIsRetriedOnReload:
a ``log_file`` startup *can* open gets recorded as the active one, so
a reload that leaves the setting unchanged does not reopen it.
This is what guards the ``opts.active_log_file = opts.log_file``
assignment moving inside the ``try`` -- if startup stopped recording
a successful open, ``old_log_file`` would stay ``None`` and every
reload would treat an unchanged, already-open ``log_file`` as a
replacement to stage, opening it a second time for no reason.
"""
log_path = self._write_file("", ".log")
opened: list[logging.FileHandler] = []
class _CapturingFileHandler(logging.FileHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
opened.append(self)
init_clients, old_conn, new_conn = self._fake_search_clients()
with patch("logging.FileHandler", _CapturingFileHandler):
state, mock_watch, logs = self._drive_reload(
self._initial_config(general_extra=f"log_file = {log_path}\n"),
self._reload_config(general_extra=f"log_file = {log_path}\n"),
init_clients,
)
self.assertFalse(
any("Unable to write to log file" in line for line in logs), logs
)
self.assertTrue(
any("Configuration reloaded successfully" in line for line in logs),
logs,
)
# Startup opened the file once; the reload, seeing the same path as
# its own active_log_file, did not stage a second FileHandler.
self.assertEqual(len(opened), 1)
file_handlers = [
h for h in state.log_handlers if isinstance(h, logging.FileHandler)
]
self.assertEqual(len(file_handlers), 1)
self.assertIs(file_handlers[0], opened[0])
self.assertEqual(opened[0].baseFilename, os.path.abspath(log_path))
# assertLogs() already restored the logger's own handler list by the
# time this runs, so this handler is not attached to anything --
# closing it is enough, with no removeHandler needed alongside it.
self.addCleanup(opened[0].close)
self.assertEqual(state.old_closed, 1)
self.assertEqual(state.new_closed, 0)
self.assertIs(state.alias, new_conn)
class TestUtilsGlobalsSnapshot(unittest.TestCase):
"""_utils_globals_snapshot() / _restore_utils_globals() cover every
module global the config loaders the SIGHUP reload calls assign:
``psl_overrides`` (load_psl_overrides), ``_IP_DB_PATH`` (load_ip_db) and
``_IPINFO_API_TOKEN`` (configure_ipinfo_api)."""
def setUp(self):
saved_overrides = list(parsedmarc.utils.psl_overrides)
saved_ip_db_path = parsedmarc.utils._IP_DB_PATH
saved_token = parsedmarc.utils._IPINFO_API_TOKEN
def _restore():
parsedmarc.utils.psl_overrides[:] = saved_overrides
parsedmarc.utils._IP_DB_PATH = saved_ip_db_path
parsedmarc.utils._IPINFO_API_TOKEN = saved_token
self.addCleanup(_restore)
def testRestoresEveryLoaderGlobalAfterTheLoadersRanAgain(self):
parsedmarc.utils.psl_overrides[:] = ["example.co.uk"]
parsedmarc.utils._IP_DB_PATH = "/before/ipinfo_lite.mmdb"
parsedmarc.utils._IPINFO_API_TOKEN = "before-token"
overrides_object = parsedmarc.utils.psl_overrides
snapshot = parsedmarc.cli._utils_globals_snapshot()
# What the loaders do: the overrides list is cleared and refilled in
# place, and the other two are rebound.
parsedmarc.utils.psl_overrides.clear()
parsedmarc.utils.psl_overrides.append("example.com.au")
parsedmarc.utils._IP_DB_PATH = "/after/ipinfo_lite.mmdb"
parsedmarc.utils._IPINFO_API_TOKEN = "after-token"
parsedmarc.cli._restore_utils_globals(snapshot)
self.assertEqual(parsedmarc.utils.psl_overrides, ["example.co.uk"])
self.assertEqual(parsedmarc.utils._IP_DB_PATH, "/before/ipinfo_lite.mmdb")
self.assertEqual(parsedmarc.utils._IPINFO_API_TOKEN, "before-token")
# In place, not rebound: get_base_domain() reads the list through
# the module global, but the snapshot must not hand back a
# different object to anything holding a reference to this one.
self.assertIs(parsedmarc.utils.psl_overrides, overrides_object)
def testSnapshotIsNotAliasedToTheLiveOverridesList(self):
"""The overrides list is snapshotted by value: load_psl_overrides()
mutates the live object, so a snapshot that aliased it would record
the failed reload's contents instead of the previous ones."""
parsedmarc.utils.psl_overrides[:] = ["example.co.uk"]
snapshot = parsedmarc.cli._utils_globals_snapshot()
parsedmarc.utils.psl_overrides.clear()
self.assertEqual(snapshot["psl_overrides"], ["example.co.uk"])
class TestSigtermShutdown(unittest.TestCase):
"""Tests for graceful SIGTERM/SIGINT shutdown."""
tests/test_init.py
+102 -4
View File
@@ -2083,6 +2083,39 @@ class Test(unittest.TestCase):
self.assertEqual(report["policy_published"]["domain"], "example.com")
print("Passed!")
def testParseAggregateReportFileLeavesCallerStreamOpen(self):
"""A caller-supplied BytesIO passed to parse_aggregate_report_file
is left open (never closed), both on a successful parse and when
unrecognized content raises InvalidAggregateReport.
parse_aggregate_report_file() forwards ``_input`` straight to
extract_report() (see both functions' docstrings), so it inherits
extract_report()'s contract of never closing a caller-supplied
file-like object, on either the success or the exception path.
This is a regression test for that contract reached through the
public parse_aggregate_report_file() entry point -- extract_report()
and parse_report_file() are covered by their own direct tests, but
neither exercises parse_aggregate_report_file() with a stream, only
with bytes (see testParseAggregateReportFile above). A real BytesIO
is used, not a mock, so the assertion is on real observable state
(``closed``), and because MagicMock auto-implements ``__fspath__``.
"""
sample_path = "samples/aggregate/rfc9990-sample.xml"
with open(sample_path, "rb") as f:
data = f.read()
success_stream = BytesIO(data)
report = parsedmarc.parse_aggregate_report_file(
success_stream, offline=True, always_use_local_files=True
)
self.assertEqual(report["report_metadata"]["org_name"], "Sample Reporter")
self.assertFalse(success_stream.closed)
garbage_stream = BytesIO(b"this is not a valid report")
with self.assertRaises(parsedmarc.InvalidAggregateReport):
parsedmarc.parse_aggregate_report_file(garbage_stream, offline=True)
self.assertFalse(garbage_stream.closed)
def testParseInvalidAggregateSample(self):
"""Test invalid aggregate samples are handled"""
print()
@@ -2216,6 +2249,47 @@ class TestExtractReport(unittest.TestCase):
result = parsedmarc.extract_report(bio)
self.assertIn("<feedback>", result)
def testExtractReportLeavesSeekableCallerStreamOpen(self):
"""A caller-supplied seekable stream is left open after a
successful call, and remains usable afterward.
Before this fix, extract_report's seekable-stream branch set
file_object = stream (aliasing the caller's own handle) and then
unconditionally closed file_object in its `finally` block on
every path, success included -- closing a handle it did not open.
This asserts the real observable state of a real BytesIO handle
(``closed`` and a post-seek re-read), not a mock's bookkeeping.
"""
xml = b'<?xml version="1.0"?><feedback></feedback>'
bio = BytesIO(xml)
result = parsedmarc.extract_report(bio)
self.assertIn("<feedback>", result)
self.assertFalse(bio.closed)
bio.seek(0)
self.assertEqual(bio.read(), xml)
def testExtractReportLeavesSeekableCallerStreamOpenOnError(self):
"""A caller-supplied seekable stream is left open when
extract_report raises ParserError, the other half of the "never
closed on success or failure" claim above.
Before this fix, extract_report's seekable-stream branch aliased
file_object to the caller's own stream and its `finally` block
closed file_object unconditionally -- including on this
exception path, since content matching no known format still
reaches the header sniff, `stream.seek(0)`, and the `finally`
close before raising. Uses a real BytesIO so the assertion is on
its own observable ``closed`` state, not a mock's bookkeeping.
"""
bio = BytesIO(b"this is not a valid archive")
with self.assertRaises(parsedmarc.ParserError):
parsedmarc.extract_report(bio)
self.assertFalse(bio.closed)
def testExtractReportFromNonSeekableStream(self):
"""extract_report handles non-seekable streams"""
xml = b'<?xml version="1.0"?><feedback></feedback>'
@@ -2799,13 +2873,13 @@ class TestParseReportFile(unittest.TestCase):
def testParseReportFileLeavesCallerHandleOpenOnReadError(self):
"""A caller-supplied file-like object is left open (not closed)
when reading it raises, preserving parse_report_file's
long-standing contract for handles it did not open itself.
when reading it raises.
Only a path input (opened internally by parse_report_file) is
closed on the exception path; a handle the caller passed in is
closed on success only, exactly as before the fix for the
internally-opened-path leak.
never closed by parse_report_file, on success or failure -- the
caller may want to seek(0) and retry, log tell(), or reuse the
handle otherwise.
A plain class is used instead of MagicMock because MagicMock
auto-implements ``__fspath__`` (supported since Python 3.8's
@@ -2831,6 +2905,30 @@ class TestParseReportFile(unittest.TestCase):
self.assertFalse(fake_handle.close_called)
def testParseReportFileLeavesCallerHandleOpenOnSuccess(self):
"""A caller-supplied file-like object is left open after a
successful parse, and remains usable afterward.
Before this fix, parse_report_file called .close() on any
caller-supplied file object once it had been read, on the success
path only. A function must not close a handle it did not open --
the caller may still want to seek(0) and re-read it, inspect
tell(), or otherwise reuse it. This asserts the real observable
state of a real BytesIO handle (``closed`` and post-seek
``read()``), not a mock's call-tracking.
"""
xml_path = "samples/aggregate/!example.com!1538204542!1538463818.xml"
with open(xml_path, "rb") as f:
data = f.read()
handle = BytesIO(data)
result = parsedmarc.parse_report_file(handle, offline=True)
self.assertEqual(result["report_type"], "aggregate")
self.assertFalse(handle.closed)
handle.seek(0)
self.assertEqual(handle.read(), data)
class TestParseReportEmail(unittest.TestCase):
"""Tests for parse_report_email edge cases"""