From de6e1fee1942de1f46ff465fd7ec9ec39efa8c6e Mon Sep 17 00:00:00 2001 From: "R. Garcia-Dias" Date: Fri, 11 Sep 2026 17:35:16 +0100 Subject: [PATCH 1/3] fix: screen logging.conf class=/args= before fileConfig (GHSA-wvpx) logging.config.fileConfig resolves each handler/formatter class= field through eval() in a namespace holding the logging module, and evaluates args=/kwargs= the same way. A bundle ships configs/logging.conf and it is applied before any bundle config is parsed, so an untrusted bundle can run arbitrary code from the logging file alone (see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3). Replace the warn-and-execute behaviour of #9078 with a hard allowlist: - class= must live in the logging or logging.handlers namespace; - args=/kwargs= must be literal expressions, optionally naming sys.stdout / sys.stderr, enforced by an AST walk that rejects calls, attribute traversal, subscripts, comprehensions and any other executable form. Benign bundle configs (including the repo's own StreamHandler fixture with args=(sys.stdout,)) continue to apply; anything executable is refused with a ValueError that names the offending section and field. Signed-off-by: R. Garcia-Dias --- monai/bundle/workflows.py | 160 ++++++++++++++++++---- tests/bundle/test_bundle_workflow.py | 53 ++++++- tests/fl/monai_algo/test_fl_monai_algo.py | 27 ++-- 3 files changed, 199 insertions(+), 41 deletions(-) diff --git a/monai/bundle/workflows.py b/monai/bundle/workflows.py index 3d8637cb34..22ccd30000 100644 --- a/monai/bundle/workflows.py +++ b/monai/bundle/workflows.py @@ -11,11 +11,12 @@ from __future__ import annotations +import ast +import configparser import json import os import sys import time -import warnings from abc import ABC, abstractmethod from collections.abc import Sequence from copy import copy @@ -34,22 +35,134 @@ logger = get_logger(module_name=__name__) +# `class=` values accepted in a bundle's logging INI. `fileConfig` eval()s this field, so the +# allowlist is by module: stdlib logging handlers/formatters cover every legitimate bundle. +_ALLOWED_LOGGING_CLASS_MODULES = {"logging", "logging.handlers"} -def _warn_logging_file_execution(logging_file: str) -> None: + +def _reject_executable_logging_config(logging_file: str) -> None: + """ + Reject a logging INI whose ``class=``/``args=`` fields would execute arbitrary code. + + `logging.config.fileConfig` resolves each handler/formatter ``class=`` through ``eval()`` in a + namespace containing the ``logging`` module, and evaluates ``args=``/``kwargs=`` the same way. + A bundle ships ``configs/logging.conf`` and it is applied before any of the bundle's own config + is parsed, so an untrusted bundle gets code execution from the logging file alone. + + Legitimate bundles only ever name stdlib logging classes, so ``class=`` is restricted to the + ``logging`` / ``logging.handlers`` namespaces and ``args=``/``kwargs=`` must be literals. + + Args: + logging_file: path to the INI file that is about to be passed to `fileConfig`. + + Raises: + ValueError: if a field would run code that this allowlist does not cover. + """ + parser = configparser.RawConfigParser() + try: + parser.read(logging_file, encoding="utf-8") + except configparser.Error as e: + raise ValueError(f"cannot parse logging config file {logging_file}: {e}") from e + + for section in parser.sections(): + for field in ("class", "args", "kwargs"): + if not parser.has_option(section, field): + continue + value = parser.get(section, field, raw=True).strip() + if not value: + continue + if field == "class": + # `fileConfig` eval()s this name; a bare identifier is resolved against the + # `logging` module, a dotted path against the eval namespace. + qualified = value if "." in value else f"logging.{value}" + if qualified.rsplit(".", 1)[0] not in _ALLOWED_LOGGING_CLASS_MODULES: + raise ValueError( + f"refusing to apply logging config {logging_file}: section [{section}] sets " + f"class={value!r}, which `logging.config.fileConfig` would pass to `eval()`. Only " + f"classes from {sorted(_ALLOWED_LOGGING_CLASS_MODULES)} are allowed " + "(see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3)." + ) + else: + # `args`/`kwargs` are eval()ed in a namespace holding `logging`, `os` and `sys`. + # Literals alone are too strict -- `args=(sys.stdout,)` is the standard + # StreamHandler form -- so allow literals plus a fixed set of safe stream names, + # and reject calls, attribute traversal and comprehensions. + _reject_non_literal_expression(value, logging_file, section, field) + + +_ALLOWED_LOGGING_ARG_NAMES = {"sys.stdout", "sys.stderr"} + + +def _reject_non_literal_expression(value: str, logging_file: str, section: str, field: str) -> None: + """ + Require ``value`` to be a literal expression, optionally naming a safe stream. + + `fileConfig` evaluates ``args=``/``kwargs=`` with ``eval()``. Constants, tuples, lists, dicts + and sets are inert; the only non-literal forms a real logging INI needs are ``sys.stdout`` and + ``sys.stderr``. Everything else -- calls, subscripts, arbitrary attribute chains, comprehensions + -- can execute code and is rejected. + + Args: + value: the raw field value from the INI. + logging_file: path of the file, used in the error message. + section: INI section name, used in the error message. + field: field name, used in the error message. + + Raises: + ValueError: if the expression is not in the allowed subset. """ - Warn that ``logging_file`` is about to be executed by `logging.config.fileConfig`. - Called immediately before every `fileConfig` invocation in this module, so the warning is only - raised when the file is really executed -- not when it is missing or logging is disabled. + def _fail(reason: str) -> None: + raise ValueError( + f"refusing to apply logging config {logging_file}: section [{section}] sets " + f"{field}={value!r}, which {reason} and would execute code through " + "`logging.config.fileConfig`'s `eval()` " + "(see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3)." + ) + + def _dotted_name(node: ast.AST) -> str | None: + """Render an attribute/name chain such as ``sys.stdout``, or None if it is not one.""" + parts: list[str] = [] + current = node + while isinstance(current, ast.Attribute): + parts.append(current.attr) + current = current.value + if not isinstance(current, ast.Name): + return None + parts.append(current.id) + return ".".join(reversed(parts)) + + try: + tree = ast.parse(value, mode="eval") + except SyntaxError as e: + _fail(f"is not a parsable expression ({e.msg})") + + for node in ast.walk(tree.body): # type: ignore[union-attr] + if isinstance(node, (ast.Constant, ast.Tuple, ast.List, ast.Dict, ast.Set, ast.Load)): + continue + if isinstance(node, (ast.Attribute, ast.Name)): + name = _dotted_name(node) + if name is None: + _fail("uses a name this allowlist does not cover") + # Sub-nodes of an allowed chain (e.g. the `sys` of `sys.stdout`) are reached by the + # walk too; accept any prefix of a permitted name. + if name not in _ALLOWED_LOGGING_ARG_NAMES and not any( + allowed.startswith(f"{name}.") for allowed in _ALLOWED_LOGGING_ARG_NAMES + ): + _fail(f"references {name!r}, which is not in {sorted(_ALLOWED_LOGGING_ARG_NAMES)}") + continue + _fail(f"contains a {type(node).__name__} node") + + +def _apply_logging_file(logging_file: str) -> None: + """ + Validate ``logging_file`` and apply it with `logging.config.fileConfig`. + + Args: + logging_file: path to the logging INI file. """ - warnings.warn( - f"applying logging config {logging_file}: `logging.config.fileConfig` passes the `class=` and " - "`args=` fields of the INI's handler and formatter sections to Python `eval()`, so this file " - "runs as code. A bundle ships its own `configs/logging.conf` and it is applied by default, " - "before any of the bundle's config is parsed. Only proceed if this file is from a source you " - "trust (see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3).", - stacklevel=3, - ) + _reject_executable_logging_config(logging_file) + fileConfig(logging_file, disable_existing_loggers=False) class BundleWorkflow(ABC): @@ -74,8 +187,9 @@ class BundleWorkflow(ABC): logging_file: config file for `logging` module in the program. for more details: https://docs.python.org/3/library/logging.config.html#logging.config.fileConfig. Security note: `fileConfig` passes the INI's `class=` and `args=` fields to Python - `eval()`, so this file runs as code and applying it raises a warning -- once per call - site, as Python's default warning filter suppresses repeats + `eval()`, so this file runs as code. `class=` is restricted to the `logging` and + `logging.handlers` namespaces and `args=`/`kwargs=` to literals plus `sys.stdout` / + `sys.stderr`; anything else raises `ValueError` (see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3). """ @@ -94,8 +208,7 @@ def __init__( if not os.path.isfile(logging_file): raise FileNotFoundError(f"Cannot find the logging config file: {logging_file}.") logger.info(f"Setting logging properties based on config: {logging_file}.") - _warn_logging_file_execution(logging_file) - fileConfig(logging_file, disable_existing_loggers=False) + _apply_logging_file(logging_file) if meta_file is not None: if isinstance(meta_file, str) and not os.path.isfile(meta_file): @@ -297,8 +410,9 @@ class PythonicWorkflow(BundleWorkflow): logging_file: config file for `logging` module in the program. for more details: https://docs.python.org/3/library/logging.config.html#logging.config.fileConfig. Security note: `fileConfig` passes the INI's `class=` and `args=` fields to Python - `eval()`, so this file runs as code and applying it raises a warning -- once per call - site, as Python's default warning filter suppresses repeats + `eval()`, so this file runs as code. `class=` is restricted to the `logging` and + `logging.handlers` namespaces and `args=`/`kwargs=` to literals plus `sys.stdout` / + `sys.stderr`; anything else raises `ValueError` (see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3). """ @@ -403,8 +517,9 @@ class ConfigWorkflow(BundleWorkflow): If None, default to "configs/logging.conf", which is commonly used for bundles in MONAI model zoo. If False, the logging logic for the bundle will not be modified. Security note: `fileConfig` passes the INI's `class=` and `args=` fields to Python - `eval()`, so this file runs as code and applying it raises a warning -- once per call - site, as Python's default warning filter suppresses repeats + `eval()`, so this file runs as code. `class=` is restricted to the `logging` and + `logging.handlers` namespaces and `args=`/`kwargs=` to literals plus `sys.stdout` / + `sys.stderr`; anything else raises `ValueError` (see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3). init_id: ID name of the expected config expression to initialize before running, default to "initialize". allow a config to have no `initialize` logic and the ID. @@ -475,8 +590,7 @@ def __init__( else: raise FileNotFoundError(f"Cannot find the logging config file: {logging_file}.") else: - _warn_logging_file_execution(str(logging_file)) - fileConfig(str(logging_file), disable_existing_loggers=False) + _apply_logging_file(str(logging_file)) logger.info(f"Setting logging properties based on config: {logging_file}.") self.parser = ConfigParser() diff --git a/tests/bundle/test_bundle_workflow.py b/tests/bundle/test_bundle_workflow.py index 2fa9b8eac4..e686e7fd23 100644 --- a/tests/bundle/test_bundle_workflow.py +++ b/tests/bundle/test_bundle_workflow.py @@ -276,8 +276,9 @@ class TestConfigWorkflowWarnsOnLoggingConf(unittest.TestCase): bundle's own "configs/logging.conf" and hands it to `logging.config.fileConfig`, which `eval()`s the INI's `class=`/`args=` fields. It fires in `__init__`, before `initialize()` or `run()`, and lives in a plain INI rather than the MONAI `$`-DSL, so it is easy to miss when reviewing a - bundle. Applying it is still not blocked -- as for GHSA-873f-pvrv-4x83, MONAI has no way to - establish whether a bundle is trustworthy -- but applying it now raises a `UserWarning`.""" + bundle. `class=` is now restricted to the stdlib logging namespaces and `args=`/`kwargs=` to + literals plus `sys.stdout`/`sys.stderr`, so a config that would execute code is rejected before + `fileConfig` ever sees it.""" def setUp(self): # `fileConfig` reconfigures logging process-wide. Snapshot the root logger and restore it @@ -303,15 +304,16 @@ def _restore(): self.addCleanup(_restore) - def test_default_logging_conf_warns_and_executes(self): + def test_default_logging_conf_payload_is_rejected(self): + """The `class=` payload is refused and never runs.""" with tempfile.TemporaryDirectory() as tempdir: configs = os.path.join(tempdir, "configs") os.makedirs(configs) marker = os.path.join(tempdir, "PWNED") with open(os.path.join(configs, "train.json"), "w") as f: json.dump({"initialize": []}, f) - # `fileConfig` eval()s the `class=` field, so the tuple subscript runs the payload and - # still yields a usable handler class. + # `fileConfig` would eval() the `class=` field, so the tuple subscript runs the payload + # and still yields a usable handler class -- unless the config is rejected first. with open(os.path.join(configs, "logging.conf"), "w") as f: f.write( "[loggers]\nkeys=root\n[handlers]\nkeys=h\n[formatters]\nkeys=f\n" @@ -321,9 +323,46 @@ def test_default_logging_conf_warns_and_executes(self): "__import__('logging').StreamHandler)[1]\nargs=()\nformatter=f\n" "[formatter_f]\nformat=%(message)s\n" ) - with self.assertWarnsRegex(UserWarning, r"GHSA-wvpx-5qmp-46g3"): + with self.assertRaisesRegex(ValueError, r"GHSA-wvpx-5qmp-46g3"): ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") - self.assertTrue(os.path.exists(marker)) + self.assertFalse(os.path.exists(marker), "the logging.conf payload executed") + + def test_args_payload_is_rejected(self): + """A payload hidden in `args=` is refused even with an allowlisted `class=`.""" + with tempfile.TemporaryDirectory() as tempdir: + configs = os.path.join(tempdir, "configs") + os.makedirs(configs) + marker = os.path.join(tempdir, "PWNED") + with open(os.path.join(configs, "train.json"), "w") as f: + json.dump({"initialize": []}, f) + with open(os.path.join(configs, "logging.conf"), "w") as f: + f.write( + "[loggers]\nkeys=root\n[handlers]\nkeys=h\n[formatters]\nkeys=f\n" + "[logger_root]\nlevel=NOTSET\nhandlers=h\n" + "[handler_h]\nclass=StreamHandler\n" + f"args=(__import__('pathlib').Path({marker!r}).write_text('pwned'),)\nformatter=f\n" + "[formatter_f]\nformat=%(message)s\n" + ) + with self.assertRaisesRegex(ValueError, r"GHSA-wvpx-5qmp-46g3"): + ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") + self.assertFalse(os.path.exists(marker), "the logging.conf payload executed") + + def test_benign_logging_conf_still_applies(self): + """The standard `class=StreamHandler` / `args=(sys.stdout,)` form keeps working.""" + with tempfile.TemporaryDirectory() as tempdir: + configs = os.path.join(tempdir, "configs") + os.makedirs(configs) + with open(os.path.join(configs, "train.json"), "w") as f: + json.dump({"initialize": []}, f) + with open(os.path.join(configs, "logging.conf"), "w") as f: + f.write( + "[loggers]\nkeys=root\n[handlers]\nkeys=h\n[formatters]\nkeys=f\n" + "[logger_root]\nlevel=INFO\nhandlers=h\n" + "[handler_h]\nclass=StreamHandler\nlevel=INFO\nformatter=f\nargs=(sys.stdout,)\n" + "[formatter_f]\nformat=%(asctime)s - %(message)s\n" + ) + # Must not raise. + ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") def test_no_warning_when_logging_disabled(self): """No warning when `fileConfig` is never reached -- the file exists but is opted out of.""" diff --git a/tests/fl/monai_algo/test_fl_monai_algo.py b/tests/fl/monai_algo/test_fl_monai_algo.py index 06374e4af6..68eb7bf1c0 100644 --- a/tests/fl/monai_algo/test_fl_monai_algo.py +++ b/tests/fl/monai_algo/test_fl_monai_algo.py @@ -399,20 +399,25 @@ def test_explicit_none_logging_file_does_not_apply_provisioned_conf(self, algo_c self.assertFalse(os.path.exists(logging_marker)) @parameterized.expand([[MonaiAlgoStats], [MonaiAlgo]]) - def test_logging_file_opt_in_applies_provisioned_conf(self, algo_class): + def test_logging_file_opt_in_still_rejects_executable_conf(self, algo_class): + """Opting back in to logging does not opt in to code execution. + + Passing `ExtraItems.LOGGING_FILE` explicitly re-enables `fileConfig`, but the INI is still + screened: a `class=`/`args=` payload is refused and never runs + (see GHSA-wvpx-5qmp-46g3). + """ with tempfile.TemporaryDirectory() as tempdir: app_root, _, logging_marker = self._stage_malicious_app(tempdir) algo = self._algo(algo_class) - with self.assertWarnsRegex(UserWarning, r"GHSA-wvpx-5qmp-46g3"): - with self.assertRaises(KeyError): - algo.initialize( - extra={ - ExtraItems.CLIENT_NAME: "test_fl", - ExtraItems.APP_ROOT: app_root, - ExtraItems.LOGGING_FILE: os.path.join(app_root, "configs", "logging.conf"), - } - ) - self.assertTrue(os.path.exists(logging_marker)) + with self.assertRaisesRegex(ValueError, r"GHSA-wvpx-5qmp-46g3"): + algo.initialize( + extra={ + ExtraItems.CLIENT_NAME: "test_fl", + ExtraItems.APP_ROOT: app_root, + ExtraItems.LOGGING_FILE: os.path.join(app_root, "configs", "logging.conf"), + } + ) + self.assertFalse(os.path.exists(logging_marker), "the provisioned logging.conf payload executed") def test_no_logging_warning_when_logging_disabled(self): """The `fileConfig` warning must not fire when nothing is actually executed.""" From ff407ebfbd55303c1ec82b1404a34765e6db3fef Mon Sep 17 00:00:00 2001 From: "R. Garcia-Dias" Date: Fri, 11 Sep 2026 19:36:50 +0100 Subject: [PATCH 2/3] fix: address PR #9114 review feedback - monai/bundle/workflows.py: parse class= with ast.parse and allow only a bare name or an attribute chain rooted in an allowlisted logging module; reject calls, subscripts, operators, and all other expression nodes so an expression without a period (e.g. __builtins__.eval) cannot slip past a string-prefix check - monai/bundle/workflows.py: read the logging file once, validate the captured text, and pass that same text through io.StringIO to fileConfig, so a file swapped in between the two reads cannot bypass the validation - tests/bundle/test_bundle_workflow.py: add regression tests for the executable class= expression shapes Signed-off-by: R. Garcia-Dias --- monai/bundle/workflows.py | 95 +++++++++++++++++++++++----- tests/bundle/test_bundle_workflow.py | 50 +++++++++++++++ 2 files changed, 129 insertions(+), 16 deletions(-) diff --git a/monai/bundle/workflows.py b/monai/bundle/workflows.py index 22ccd30000..ca8a184104 100644 --- a/monai/bundle/workflows.py +++ b/monai/bundle/workflows.py @@ -13,6 +13,7 @@ import ast import configparser +import io import json import os import sys @@ -22,7 +23,7 @@ from copy import copy from logging.config import fileConfig from pathlib import Path -from typing import Any +from typing import Any, cast from monai.apps.utils import get_logger from monai.bundle.config_parser import ConfigParser @@ -40,7 +41,7 @@ _ALLOWED_LOGGING_CLASS_MODULES = {"logging", "logging.handlers"} -def _reject_executable_logging_config(logging_file: str) -> None: +def _reject_executable_logging_config(logging_file: str) -> str: """ Reject a logging INI whose ``class=``/``args=`` fields would execute arbitrary code. @@ -55,12 +56,22 @@ def _reject_executable_logging_config(logging_file: str) -> None: Args: logging_file: path to the INI file that is about to be passed to `fileConfig`. + Returns: + The validated file content, so the caller can apply exactly this text without re-reading + the path (a second read could see swapped-in content and bypass this check). + Raises: ValueError: if a field would run code that this allowlist does not cover. """ + try: + with open(logging_file, encoding="utf-8") as f: + content = f.read() + except OSError as e: + raise ValueError(f"cannot read logging config file {logging_file}: {e}") from e + parser = configparser.RawConfigParser() try: - parser.read(logging_file, encoding="utf-8") + parser.read_string(content) except configparser.Error as e: raise ValueError(f"cannot parse logging config file {logging_file}: {e}") from e @@ -72,27 +83,75 @@ def _reject_executable_logging_config(logging_file: str) -> None: if not value: continue if field == "class": - # `fileConfig` eval()s this name; a bare identifier is resolved against the - # `logging` module, a dotted path against the eval namespace. - qualified = value if "." in value else f"logging.{value}" - if qualified.rsplit(".", 1)[0] not in _ALLOWED_LOGGING_CLASS_MODULES: - raise ValueError( - f"refusing to apply logging config {logging_file}: section [{section}] sets " - f"class={value!r}, which `logging.config.fileConfig` would pass to `eval()`. Only " - f"classes from {sorted(_ALLOWED_LOGGING_CLASS_MODULES)} are allowed " - "(see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3)." - ) + # `fileConfig` eval()s this expression, so a bare string prefix is not a safe boundary: + # a call or subscript without a period (e.g. `__builtins__.eval`) would slip past it. + # Parse and allow only a bare name or an attribute chain whose root is allowlisted. + _reject_non_logging_class(value, logging_file, section) else: # `args`/`kwargs` are eval()ed in a namespace holding `logging`, `os` and `sys`. # Literals alone are too strict -- `args=(sys.stdout,)` is the standard # StreamHandler form -- so allow literals plus a fixed set of safe stream names, # and reject calls, attribute traversal and comprehensions. _reject_non_literal_expression(value, logging_file, section, field) + return content _ALLOWED_LOGGING_ARG_NAMES = {"sys.stdout", "sys.stderr"} +def _reject_non_logging_class(value: str, logging_file: str, section: str) -> None: + """ + Require ``value`` to be a bare name or an attribute chain rooted in an allowlisted module. + + `fileConfig` passes ``class=`` to ``eval()``, so the whole expression -- not just a string + prefix -- must be inert. A bare identifier is resolved against the ``logging`` module; a + dotted path is resolved against the ``eval`` namespace. Calls, subscripts, operators and every + other expression node are rejected. + + Args: + value: the raw ``class=`` value from the INI. + logging_file: path of the file, used in the error message. + section: INI section name, used in the error message. + + Raises: + ValueError: if the expression is not a name or an attribute chain rooted in an allowlisted module. + """ + + def _fail(reason: str) -> None: + raise ValueError( + f"refusing to apply logging config {logging_file}: section [{section}] sets " + f"class={value!r}, which {reason} and would execute code through " + f"`logging.config.fileConfig`'s `eval()`. Only classes from " + f"{sorted(_ALLOWED_LOGGING_CLASS_MODULES)} are allowed " + "(see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-wvpx-5qmp-46g3)." + ) + + try: + tree = ast.parse(value, mode="eval") + except SyntaxError as e: + _fail(f"is not a parsable expression ({e.msg})") + + node = tree.body # type: ignore[union-attr] + # A bare name resolves against the `logging` module in fileConfig's eval namespace. + if isinstance(node, ast.Name): + return + # An attribute chain must be rooted in an allowlisted module. + if isinstance(node, ast.Attribute): + parts: list[str] = [] + current: ast.AST = node + while isinstance(current, ast.Attribute): + parts.append(current.attr) + current = current.value + if not isinstance(current, ast.Name): + _fail("is not a simple dotted name") + root = cast(ast.Name, current).id + parts.append(root) + if parts[-1] not in _ALLOWED_LOGGING_CLASS_MODULES: + _fail(f"references {'.'.join(reversed(parts))!r}, which is not in {sorted(_ALLOWED_LOGGING_CLASS_MODULES)}") + return + _fail("is not a name or an attribute chain") + + def _reject_non_literal_expression(value: str, logging_file: str, section: str, field: str) -> None: """ Require ``value`` to be a literal expression, optionally naming a safe stream. @@ -156,13 +215,17 @@ def _dotted_name(node: ast.AST) -> str | None: def _apply_logging_file(logging_file: str) -> None: """ - Validate ``logging_file`` and apply it with `logging.config.fileConfig`. + Validate ``logging_file`` and apply it, using exactly the content that was validated. + + The single read performed by the validator is what `fileConfig` applies: re-opening the path + in between would let a swapped-in file bypass the allowlist and reach `fileConfig`'s `eval()` + calls. Args: logging_file: path to the logging INI file. """ - _reject_executable_logging_config(logging_file) - fileConfig(logging_file, disable_existing_loggers=False) + content = _reject_executable_logging_config(logging_file) + fileConfig(io.StringIO(content), disable_existing_loggers=False) class BundleWorkflow(ABC): diff --git a/tests/bundle/test_bundle_workflow.py b/tests/bundle/test_bundle_workflow.py index e686e7fd23..f5299df0df 100644 --- a/tests/bundle/test_bundle_workflow.py +++ b/tests/bundle/test_bundle_workflow.py @@ -304,6 +304,56 @@ def _restore(): self.addCleanup(_restore) + def test_class_call_expr_is_rejected(self): + """A ``class=`` call expression (no dot, no tuple subscript) is refused. + + This is the exact shape the previous string-prefix check could not see: ``rsplit('.', 1)[0]`` + on a call with no period returns the whole expression, and `class=` values in real configs + are dotted logging names, so such a call would never have been allowed -- it is rejected + here by the AST parser rather than the prefix test. + """ + with tempfile.TemporaryDirectory() as tempdir: + configs = os.path.join(tempdir, "configs") + os.makedirs(configs) + marker = os.path.join(tempdir, "PWNED") + with open(os.path.join(configs, "train.json"), "w") as f: + json.dump({"initialize": []}, f) + payload = f"__import__('pathlib').Path({marker!r}).write_text('pwned')" + with open(os.path.join(configs, "logging.conf"), "w") as f: + f.write( + "[loggers]\nkeys=root\n[handlers]\nkeys=h\n[formatters]\nkeys=f\n" + "[logger_root]\nlevel=NOTSET\nhandlers=h\n" + f"[handler_h]\nclass={payload}()\nargs=()\nformatter=f\n" + "[formatter_f]\nformat=%(message)s\n" + ) + with self.assertRaisesRegex(ValueError, r"GHSA-wvpx-5qmp-46g3"): + ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") + self.assertFalse(os.path.exists(marker), "the logging.conf payload executed") + + def test_class_attribute_call_is_rejected(self): + """A ``class=`` attribute-chain expression is refused even when it is not a bare call. + + The root module is not on the `logging` allowlist, so the attribute chain is rejected by + the AST check regardless of the trailing call. + """ + with tempfile.TemporaryDirectory() as tempdir: + configs = os.path.join(tempdir, "configs") + os.makedirs(configs) + marker = os.path.join(tempdir, "PWNED") + with open(os.path.join(configs, "train.json"), "w") as f: + json.dump({"initialize": []}, f) + payload = f"__import__('pathlib').Path({marker!r}).write_text" + with open(os.path.join(configs, "logging.conf"), "w") as f: + f.write( + "[loggers]\nkeys=root\n[handlers]\nkeys=h\n[formatters]\nkeys=f\n" + "[logger_root]\nlevel=NOTSET\nhandlers=h\n" + f"[handler_h]\nclass={payload}\nargs=()\nformatter=f\n" + "[formatter_f]\nformat=%(message)s\n" + ) + with self.assertRaisesRegex(ValueError, r"GHSA-wvpx-5qmp-46g3"): + ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") + self.assertFalse(os.path.exists(marker), "the logging.conf payload executed") + def test_default_logging_conf_payload_is_rejected(self): """The `class=` payload is refused and never runs.""" with tempfile.TemporaryDirectory() as tempdir: From a2693dac8a9c0c5482c267abe48c62ceef29c80b Mon Sep 17 00:00:00 2001 From: "R. Garcia-Dias" Date: Fri, 11 Sep 2026 21:13:55 +0100 Subject: [PATCH 3/3] fix: reject class= names that are not logging handlers (GHSA-wvpx) The previous check accepted any bare `class=` name, so `class=eval` passed validation. `fileConfig` resolves a dotless class against the logging namespace with `eval()`, so pairing it with a literal `args=` tuple -- which the literal check permits, the tuple itself being inert -- passed an attacker-controlled string to `eval` and executed code before handler setup failed. - monai/bundle/workflows.py: resolve `class=` against the allowlisted modules and require a logging.Handler/Formatter subclass; deriving this from the module keeps the allowlist in step with the standard library - tests/bundle/test_bundle_workflow.py: regression tests for `class=eval`, for a non-handler logging attribute, and for RotatingFileHandler still being accepted Signed-off-by: R. Garcia-Dias --- monai/bundle/workflows.py | 37 ++++++++++++++-- tests/bundle/test_bundle_workflow.py | 66 ++++++++++++++++++++++++++++ 2 files changed, 99 insertions(+), 4 deletions(-) diff --git a/monai/bundle/workflows.py b/monai/bundle/workflows.py index ca8a184104..968c80d729 100644 --- a/monai/bundle/workflows.py +++ b/monai/bundle/workflows.py @@ -15,6 +15,8 @@ import configparser import io import json +import logging +import logging.handlers # ensures `logging.handlers` is importable for the `class=` allowlist check import os import sys import time @@ -99,6 +101,25 @@ def _reject_executable_logging_config(logging_file: str) -> str: _ALLOWED_LOGGING_ARG_NAMES = {"sys.stdout", "sys.stderr"} +def _is_allowed_logging_class(module: str, attribute: str) -> bool: + """ + Return whether ``module.attribute`` names a logging handler or formatter class. + + The check resolves the attribute on the already-imported allowlisted module and requires the + object to be a ``logging.Handler`` or ``logging.Formatter`` subclass. Deriving the answer from + the module keeps the allowlist in step with the standard library instead of hard-coding a name + list, and it rejects callables such as ``eval`` that are not logging classes at all. + + Args: + module: dotted module name; must be one of ``_ALLOWED_LOGGING_CLASS_MODULES``. + attribute: attribute looked up on that module. + """ + if module not in _ALLOWED_LOGGING_CLASS_MODULES: + return False + resolved = getattr(sys.modules.get(module), attribute, None) + return isinstance(resolved, type) and issubclass(resolved, (logging.Handler, logging.Formatter)) + + def _reject_non_logging_class(value: str, logging_file: str, section: str) -> None: """ Require ``value`` to be a bare name or an attribute chain rooted in an allowlisted module. @@ -132,10 +153,14 @@ def _fail(reason: str) -> None: _fail(f"is not a parsable expression ({e.msg})") node = tree.body # type: ignore[union-attr] - # A bare name resolves against the `logging` module in fileConfig's eval namespace. + # A bare name is resolved against the `logging` module in fileConfig's eval namespace, so it + # must name a real logging handler/formatter there -- `class=eval` is a bare name too, and + # would hand an attacker-controlled `args=` literal straight to `eval()`. if isinstance(node, ast.Name): + if not _is_allowed_logging_class("logging", node.id): + _fail(f"does not name a handler or formatter in {sorted(_ALLOWED_LOGGING_CLASS_MODULES)}") return - # An attribute chain must be rooted in an allowlisted module. + # An attribute chain must be rooted in an allowlisted module and end at a handler/formatter. if isinstance(node, ast.Attribute): parts: list[str] = [] current: ast.AST = node @@ -146,8 +171,12 @@ def _fail(reason: str) -> None: _fail("is not a simple dotted name") root = cast(ast.Name, current).id parts.append(root) - if parts[-1] not in _ALLOWED_LOGGING_CLASS_MODULES: - _fail(f"references {'.'.join(reversed(parts))!r}, which is not in {sorted(_ALLOWED_LOGGING_CLASS_MODULES)}") + dotted = ".".join(reversed(parts)) + module, _, attribute = dotted.rpartition(".") + if module not in _ALLOWED_LOGGING_CLASS_MODULES: + _fail(f"references {dotted!r}, which is not in {sorted(_ALLOWED_LOGGING_CLASS_MODULES)}") + if not _is_allowed_logging_class(module, attribute): + _fail(f"references {dotted!r}, which is not a handler or formatter class") return _fail("is not a name or an attribute chain") diff --git a/tests/bundle/test_bundle_workflow.py b/tests/bundle/test_bundle_workflow.py index f5299df0df..7b3f365d01 100644 --- a/tests/bundle/test_bundle_workflow.py +++ b/tests/bundle/test_bundle_workflow.py @@ -354,6 +354,72 @@ def test_class_attribute_call_is_rejected(self): ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") self.assertFalse(os.path.exists(marker), "the logging.conf payload executed") + def test_bare_eval_class_is_rejected(self): + """A bare ``class=eval`` is refused before it can run an ``args=`` payload. + + `fileConfig` resolves a dotless ``class=`` against the `logging` module's namespace via + ``eval()``, so ``class=eval`` paired with a literal ``args=`` tuple -- which the literal + check permits, because the tuple itself is inert -- would hand an attacker-controlled + string straight to ``eval``. Accepting any bare name is therefore not safe: the name must + resolve to a real handler or formatter class. + """ + with tempfile.TemporaryDirectory() as tempdir: + configs = os.path.join(tempdir, "configs") + os.makedirs(configs) + marker = os.path.join(tempdir, "PWNED") + with open(os.path.join(configs, "train.json"), "w") as f: + json.dump({"initialize": []}, f) + payload = f"__import__('pathlib').Path({marker!r}).write_text('pwned')" + with open(os.path.join(configs, "logging.conf"), "w") as f: + f.write( + "[loggers]\nkeys=root\n[handlers]\nkeys=h\n[formatters]\nkeys=f\n" + "[logger_root]\nlevel=NOTSET\nhandlers=h\n" + f"[handler_h]\nclass=eval\nargs=({payload!r},)\nformatter=f\n" + "[formatter_f]\nformat=%(message)s\n" + ) + with self.assertRaisesRegex(ValueError, r"GHSA-wvpx-5qmp-46g3"): + ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") + self.assertFalse(os.path.exists(marker), "the logging.conf payload executed") + + def test_non_handler_logging_attribute_is_rejected(self): + """A real `logging` attribute that is not a handler or formatter is refused. + + ``logging.Logger`` lives in an allowlisted module but is not a `Handler`/`Formatter` + subclass, so allowing the module alone is too coarse a boundary. + """ + with tempfile.TemporaryDirectory() as tempdir: + configs = os.path.join(tempdir, "configs") + os.makedirs(configs) + with open(os.path.join(configs, "train.json"), "w") as f: + json.dump({"initialize": []}, f) + with open(os.path.join(configs, "logging.conf"), "w") as f: + f.write( + "[loggers]\nkeys=root\n[handlers]\nkeys=h\n[formatters]\nkeys=f\n" + "[logger_root]\nlevel=NOTSET\nhandlers=h\n" + "[handler_h]\nclass=logging.Logger\nargs=()\nformatter=f\n" + "[formatter_f]\nformat=%(message)s\n" + ) + with self.assertRaisesRegex(ValueError, r"GHSA-wvpx-5qmp-46g3"): + ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") + + def test_rotating_file_handler_is_accepted(self): + """A dotted `logging.handlers` handler keeps working under the class allowlist.""" + with tempfile.TemporaryDirectory() as tempdir: + configs = os.path.join(tempdir, "configs") + os.makedirs(configs) + logfile = os.path.join(tempdir, "run.log") + with open(os.path.join(configs, "train.json"), "w") as f: + json.dump({"initialize": []}, f) + with open(os.path.join(configs, "logging.conf"), "w") as f: + f.write( + "[loggers]\nkeys=root\n[handlers]\nkeys=h\n[formatters]\nkeys=f\n" + "[logger_root]\nlevel=INFO\nhandlers=h\n" + "[handler_h]\nclass=logging.handlers.RotatingFileHandler\n" + f"level=INFO\nformatter=f\nargs=({logfile!r},)\n" + "[formatter_f]\nformat=%(message)s\n" + ) + ConfigWorkflow(config_file=os.path.join(configs, "train.json"), workflow_type="train") + def test_default_logging_conf_payload_is_rejected(self): """The `class=` payload is refused and never runs.""" with tempfile.TemporaryDirectory() as tempdir: