From c7a9d294b561abe7e84eff5a0186c48be6e33296 Mon Sep 17 00:00:00 2001 From: NefelibateLG <103415178+DreamchaserLG@users.noreply.github.com> Date: Tue, 27 Oct 2026 09:54:22 +0800 Subject: [PATCH] fix: restore parent IP-lookup state in pool workers under spawn/forkserver (#930) * fix: restore parent IP-lookup state in pool workers under spawn/forkserver parsedmarc.utils keeps the IPinfo Lite REST API token (_IPINFO_API_TOKEN, configure_ipinfo_api) and the selected IP database path (_IP_DB_PATH, load_ip_db) in process-global variables, and only the parent process sets them. Under the fork start method a pool worker inherits them ("All resources of the parent are inherited by the child process", https://docs.python.org/3/library/multiprocessing.html), but under spawn (macOS/Windows default) and forkserver (the Linux default from Python 3.14) it does not ("The child process will only inherit those resources necessary to run the process object's run() method"). A worker there started from scratch: its IPinfo API lookups were silently skipped, and _get_ip_database_path() fell back to the bundled MMDB instead of the downloaded or cached copy the parent selected, so enrichment differed from the operator's configuration and from the same run under fork. The CLI parses file inputs through the process pool even with -n 1, so every file-based CLI run on such platforms was affected. parallel_map() now snapshots both globals in the parent and restores them in every worker via its pool initializer, so IP enrichment matches the parent's configuration under every start method. The initializer previously only reconstructed logging and is renamed _init_worker accordingly. Fixes #928 Co-authored-by: ZCode Scout Signed-off-by: NefelibateLG <3453510037@qq.com> * Maintainer edits: Unreleased changelog, PSL overrides, forkserver coverage - CHANGELOG: move the entry under "## Unreleased" (feature/fix PRs never pick a version number) and describe the psl_overrides gap. - Restore parsedmarc.utils.psl_overrides in pool workers as a separate initarg, in place ([:] =), never rebound. A fresh spawn/forkserver worker loads the bundled psl_overrides.txt at import and only switches to the operator's path/URL on its first reverse DNS map load, so get_base_domain() calls before that load (the first aggregate report's org_name, the first source IP's reverse DNS name) used the bundled list. - Cross-reference _init_worker and cli._utils_globals_snapshot(), which enumerate the same three utils globals. - Extend TestWorkerIPState to forkserver (where available) and to the PSL overrides list; without the restore, both subtests fail with the worker reporting the bundled list. Co-Authored-By: Claude Fable 5.1 --------- Signed-off-by: NefelibateLG <3453510037@qq.com> Co-authored-by: ZCode Scout Co-authored-by: Sean Whalen <44679+seanthegeek@users.noreply.github.com> Co-authored-by: Claude Fable 5.1 --- CHANGELOG.md | 4 ++ parsedmarc/cli.py | 4 ++ parsedmarc/parallel.py | 56 +++++++++++++++++++++--- parsedmarc/utils.py | 34 +++++++++++++++ tests/test_parallel.py | 97 +++++++++++++++++++++++++++++++++++++++--- 5 files changed, 182 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cf48ed4..4299c44c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ - **CSV output now neutralizes spreadsheet formulas** (CWE-1236, [OWASP CSV Injection](https://owasp.org/www-community/attacks/CSV_Injection)). Every text field in the aggregate, failure, and SMTP TLS CSV output comes from a report that a third party sent, and a cell beginning with `=`, `+`, `-`, or `@` is evaluated as a formula by Excel, LibreOffice, and Google Sheets, so a reporter could set an `org_name` such as `=HYPERLINK("http://evil","x")` and have it run when an analyst opens the CSV. `_csv_safe()` now prefixes a string field that starts with `=`, `+`, `-`, `@`, a tab, or a carriage return with a single `'`, the OWASP-recommended fix, after the existing NUL strip; non-string values such as ASNs and counts are untouched. Compatibility: only cells starting with one of those six characters change, and real report fields almost never do. The JSON output and the data sent to the other output integrations are unchanged. (Closes #929) - **Control characters in sender-controlled strings are escaped before they reach logs and error messages.** An RFC 2047 encoded word in a `Subject` or `From` header can decode to a raw CR LF (`=?utf-8?q?line1=0D=0AINJECTED?=`), and `parse_report_email()` interpolated the decoded subject into its error messages and the `From` header into its `Parsing mail from` info line, so a sender could forge extra log lines in the mailbox watch loop's warnings. The new `parsedmarc.utils.sanitize_for_log()` replaces every non-printable character with its `unicode_escape` form (`\r\n`, `\x1b`, `\u2028`) rather than stripping it, so the log still shows that the characters were there; the ASCII space, letters in any script, and most emoji pass through, while non-ASCII spaces such as NBSP and U+3000 and format characters such as the zero-width joiner are escaped. It is applied to the subject and `From` header in `parse_report_email()`, the SMTP TLS `policy-type` in its "Invalid policy type" error, the RFC 9990 `np`, `testing`, and `discovery_method` values and the per-record parse error that the aggregate parser logs, and the `org_name` and `report_id` in the "Skipping duplicate aggregate report" debug lines in `__init__.py` and `cli.py`. (Closes #929) +### Bug fixes + +- **Pool workers now use the IPinfo API token and IP database the parent process selected, whatever the multiprocessing start method.** `parsedmarc.utils` keeps the IPinfo Lite REST API token (`_IPINFO_API_TOKEN`, set by `configure_ipinfo_api()`) and the selected IP database path (`_IP_DB_PATH`, set by `load_ip_db()`) in process-global variables, and only the parent process sets them. Under the `fork` start method (the Linux default through Python 3.13) a pool worker inherits them ("All resources of the parent are inherited by the child process", [Python multiprocessing docs](https://docs.python.org/3/library/multiprocessing.html)), but under `spawn` (the macOS and Windows default) and `forkserver` (the Linux default from Python 3.14) it does not ("The child process will only inherit those resources necessary to run the process object's `run()` method"). A worker there started from scratch: its IPinfo API lookups were silently skipped (`_ipinfo_api_lookup()` returns `None` when no token is set), and `_get_ip_database_path()` fell back to the bundled MMDB instead of the freshly downloaded or cached copy the parent had selected, so enrichment (country, ASN) could differ from what the operator configured and from the same run under `fork`. The CLI parses file inputs through the process pool even with `-n 1` (`cli.py`), so every file-based CLI run on such platforms was affected. `parallel_map()` now snapshots both globals in the parent and restores them in every worker via its pool initializer (which previously only reconstructed logging, and is renamed `_init_worker` accordingly), so workers make the same IPinfo API and MMDB selections the parent configured under every start method. The module-level `psl_overrides` list had a smaller gap: a fresh worker loads the bundled `psl_overrides.txt` at import time, and only switches to the operator's `psl_overrides_path` or URL when its first reverse DNS map load reloads the overrides from the config, so base-domain lookups made before that load (the first aggregate report's `org_name` and the first source IP's reverse DNS name) used the bundled list; the pool initializer now restores the parent's list in place as well. Reported by helixnebulatech. (Closes #928) + ## 11.0.5 ### Bug fixes diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index 21238bda..8a4de398 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -1843,6 +1843,10 @@ def _utils_globals_snapshot() -> dict[str, Any]: calls reach, and it only decides whether the selected database path is logged again. + ``parallel._init_worker()`` copies these same three globals into each + process-pool worker, since a ``spawn`` or ``forkserver`` worker starts + without them; keep the lists in sync. + Returns: dict: The values, to be handed to :func:`_restore_utils_globals`. """ diff --git a/parsedmarc/parallel.py b/parsedmarc/parallel.py index ec69de7e..02a7a14f 100644 --- a/parsedmarc/parallel.py +++ b/parsedmarc/parallel.py @@ -17,17 +17,51 @@ _J = TypeVar("_J") _R = TypeVar("_R") -def _init_worker_logging(log_level: int, log_files: list[str]) -> None: - """Pool initializer that reconstructs the parent's logging handlers. +def _init_worker( + log_level: int, + log_files: list[str], + ip_worker_state: tuple[str | None, str | None], + psl_overrides: list[str], +) -> None: + """Pool initializer that reconstructs the parent's lookup context. A spawned or forkserver worker process does not inherit the parent process's already-configured logging handlers, so without this the child's log records (including any that would surface a parsing bug) - are silently dropped. Called once per worker process by - ``ProcessPoolExecutor``'s ``initializer``/``initargs``. + are silently dropped. It also does not inherit the parent's IP-lookup + globals (``parsedmarc.utils._IPINFO_API_TOKEN`` and + ``parsedmarc.utils._IP_DB_PATH`` — under these start methods the child + "will only inherit those resources necessary to run the process + object's ``run()`` method", + https://docs.python.org/3/library/multiprocessing.html), so its IP + enrichment would silently fall back to the bundled MMDB and skip the + IPinfo API; the snapshot captured by the parent is restored here. + + The module-level ``parsedmarc.utils.psl_overrides`` list is restored + too. A fresh worker loads the bundled ``psl_overrides.txt`` when it + imports ``parsedmarc.utils``. Its reverse DNS map starts empty + (``ParserConfig.__setstate__`` rebinds it to the module default), and + its first map load calls ``load_psl_overrides()`` with the config's + ``psl_overrides_path``/``psl_overrides_url``, so the operator's + overrides do take effect, but only after that load: ``get_base_domain()`` + calls made before that load (the first aggregate report's ``org_name`` + and the first source IP's reverse DNS name) would use the bundled list. + It is restored in place (``[:] =``), never rebound, because + ``load_psl_overrides()``, ``get_base_domain()``, and + ``cli._restore_utils_globals()`` all share that one list object. + + These are the same three globals ``cli._utils_globals_snapshot()`` + enumerates for SIGHUP reload; keep the lists in sync. + + Called once per worker process by ``ProcessPoolExecutor``'s + ``initializer``/``initargs``. """ + from parsedmarc import utils from parsedmarc.log import configure_logging + utils._set_ip_worker_state(*ip_worker_state) + utils.psl_overrides[:] = psl_overrides + if not log_files: configure_logging(log_level, None) return @@ -128,6 +162,13 @@ def parallel_map( yielded before returning - no completed work is discarded, but the stop may block briefly until in-flight jobs finish. + Because ``spawn``/``forkserver`` workers do not inherit the parent's + process-global lookup state, the parent's + ``parsedmarc.utils._IPINFO_API_TOKEN``, ``_IP_DB_PATH``, and + ``psl_overrides`` are snapshotted here, once per pool, and restored in + every worker by the pool initializer, so IP enrichment and base-domain + lookups match what the parent configured under every start method. + Raises: ValueError: If ``n_procs`` is less than 1. Raised eagerly at the call itself (not on first iteration of the returned iterator). @@ -142,6 +183,7 @@ def parallel_map( except StopIteration: return + from parsedmarc import utils from parsedmarc.log import logger log_level = logger.getEffectiveLevel() @@ -150,13 +192,15 @@ def parallel_map( for h in logger.handlers if isinstance(h, logging.FileHandler) ] + ip_worker_state = utils._get_ip_worker_state() + psl_overrides = list(utils.psl_overrides) window_size = max(window_factor * n_procs, 1) with ProcessPoolExecutor( max_workers=n_procs, - initializer=_init_worker_logging, - initargs=(log_level, log_files), + initializer=_init_worker, + initargs=(log_level, log_files, ip_worker_state, psl_overrides), ) as executor: window: deque[Future[_R]] = deque() window.append(executor.submit(func, first_job)) diff --git a/parsedmarc/utils.py b/parsedmarc/utils.py index bcc79a00..2e1b9a3f 100644 --- a/parsedmarc/utils.py +++ b/parsedmarc/utils.py @@ -743,6 +743,40 @@ def configure_ipinfo_api( logger.info("IPinfo API configured") +def _get_ip_worker_state() -> tuple[str | None, str | None]: + """Snapshot the IP-lookup configuration a pool worker must reproduce. + + ``_IPINFO_API_TOKEN`` and ``_IP_DB_PATH`` are process globals set only + by the parent (via :func:`configure_ipinfo_api` and :func:`load_ip_db`). + Under the ``spawn`` and ``forkserver`` start methods a child process + "will only inherit those resources necessary to run the process + object's ``run()`` method" + (https://docs.python.org/3/library/multiprocessing.html), so a worker + without this snapshot would silently skip the IPinfo API and fall back + to the bundled MMDB instead of the database the parent selected. + ``parsedmarc.parallel.parallel_map`` captures this snapshot in the + parent and hands it to its pool initializer, which restores it with + :func:`_set_ip_worker_state`. + + A ``None`` element means "not configured in the parent": the worker + behaves exactly as the parent would for that element (no IPinfo API; + MMDB selected on demand at lookup time). + """ + return _IPINFO_API_TOKEN, _IP_DB_PATH + + +def _set_ip_worker_state(token: str | None, db_path: str | None) -> None: + """Restore a snapshot taken by :func:`_get_ip_worker_state`. + + ``_LAST_LOGGED_IP_DB_PATH`` is deliberately not restored: it only + deduplicates the database-selection log line within one process, and + each worker should log its own (restored) selection once. + """ + global _IPINFO_API_TOKEN, _IP_DB_PATH + _IPINFO_API_TOKEN = token + _IP_DB_PATH = db_path + + def _ipinfo_api_lookup(ip_address: str) -> _IPDatabaseRecord | None: """Look up an IP via the IPinfo Lite REST API. diff --git a/tests/test_parallel.py b/tests/test_parallel.py index e6b260ed..34d7be81 100644 --- a/tests/test_parallel.py +++ b/tests/test_parallel.py @@ -9,7 +9,7 @@ from unittest.mock import patch import parsedmarc from parsedmarc.parallel import ( - _init_worker_logging, + _init_worker, _parse_report_email_job, _parse_report_file_job, parallel_map, @@ -240,10 +240,11 @@ class TestWorkerLogging(_ParallelTestCase): class TestInitWorkerLogging(unittest.TestCase): - """_init_worker_logging must be usable directly as a pool initializer: - with no log files it sets the level (configure_logging still ensures - a console handler), and with log files it attaches a FileHandler per - path.""" + """_init_worker must be usable directly as a pool initializer: with no + log files it sets the level (configure_logging still ensures a console + handler), and with log files it attaches a FileHandler per path. The + IP-lookup state and PSL overrides it also restores are passed through + unchanged here so these tests exercise only the logging side.""" def setUp(self): from parsedmarc.log import logger as plog @@ -263,19 +264,31 @@ class TestInitWorkerLogging(unittest.TestCase): plog.setLevel(self._saved_level) def test_no_log_files_sets_level_only(self): + from parsedmarc import utils from parsedmarc.log import logger as plog - _init_worker_logging(logging.WARNING, []) + _init_worker( + logging.WARNING, + [], + utils._get_ip_worker_state(), + list(utils.psl_overrides), + ) self.assertEqual(plog.level, logging.WARNING) def test_log_files_attach_file_handlers(self): + from parsedmarc import utils from parsedmarc.log import logger as plog with tempfile.NamedTemporaryFile(suffix=".log", delete=False) as tf: log_path = tf.name self.addCleanup(lambda: os.path.exists(log_path) and os.remove(log_path)) - _init_worker_logging(logging.DEBUG, [log_path]) + _init_worker( + logging.DEBUG, + [log_path], + utils._get_ip_worker_state(), + list(utils.psl_overrides), + ) self.assertEqual(plog.level, logging.DEBUG) file_handlers = [h for h in plog.handlers if isinstance(h, logging.FileHandler)] @@ -284,6 +297,76 @@ class TestInitWorkerLogging(unittest.TestCase): h.close() +def _ip_worker_state_job(_): + """Read the worker process's IP-lookup state and PSL overrides so a + test can compare them with what the parent configured. Module-level and + picklable (a spawn child re-imports this module to unpickle it).""" + from parsedmarc import utils + + return (utils._IPINFO_API_TOKEN, utils._IP_DB_PATH, list(utils.psl_overrides)) + + +class TestWorkerIPState(unittest.TestCase): + """parallel_map workers must make the same IPinfo API / MMDB selections + and use the same PSL overrides the parent configured, whatever the + start method. + + parsedmarc.utils._IPINFO_API_TOKEN (configure_ipinfo_api), + parsedmarc.utils._IP_DB_PATH (load_ip_db), and the module-level + parsedmarc.utils.psl_overrides list (load_psl_overrides) are process + globals. Under ``fork`` a worker inherits them ("All resources of the + parent are inherited by the child process"), but under ``spawn`` and + ``forkserver`` it does not ("The child process will only inherit those + resources necessary to run the process object's ``run()`` method") - + https://docs.python.org/3/library/multiprocessing.html. The pool + initializer therefore restores the parent's snapshot in every worker. + This test forces each non-fork start method so it guards that path on + every Python version, not only the ones whose default start method + already exercises it (Linux defaults to forkserver from 3.14, macOS + and Windows to spawn). + + With the ``psl_overrides`` restore removed from ``_init_worker``, the + PSL assertion failed under both ``spawn`` and ``forkserver``: each + worker reported the bundled ``psl_overrides.txt`` list it loads at + import, not the parent's ``["example.test"]``. + """ + + def test_workers_see_parent_ip_state_under_spawn_and_forkserver(self): + import multiprocessing + + import parsedmarc.utils as utils + + methods = [ + m + for m in ("spawn", "forkserver") + if m in multiprocessing.get_all_start_methods() + ] + self.assertIn("spawn", methods) + + original_state = utils._get_ip_worker_state() + original_psl = list(utils.psl_overrides) + original_method = multiprocessing.get_start_method() + utils._set_ip_worker_state( + "worker-state-test-token", "/parent/only/ipinfo_lite.mmdb" + ) + utils.psl_overrides[:] = ["example.test"] + self.addCleanup(utils._set_ip_worker_state, *original_state) + self.addCleanup(utils.psl_overrides.__setitem__, slice(None), original_psl) + self.addCleanup(multiprocessing.set_start_method, original_method, force=True) + + for method in methods: + with self.subTest(start_method=method): + multiprocessing.set_start_method(method, force=True) + + results = list(parallel_map(_ip_worker_state_job, range(8), n_procs=2)) + + self.assertEqual(len(results), 8) + for token, db_path, psl in results: + self.assertEqual(token, "worker-state-test-token") + self.assertEqual(db_path, "/parent/only/ipinfo_lite.mmdb") + self.assertEqual(psl, ["example.test"]) + + class TestParseReportEmailJob(_ParallelTestCase): """_parse_report_email_job must return a ParserError as a value (never raise it) on an invalid message."""