diff --git a/CHANGELOG.rst b/CHANGELOG.rst index a94455f8..27c838ca 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -12,6 +12,40 @@ The semantic versioning only considers the public API as described in paths are considered internals and can change in minor and patch releases. +v4.52.0 (unreleased) +-------------------- + +Added +^^^^^ +- New ``import_path_denylist`` and ``import_path_allowlist`` settings in + ``set_parsing_settings`` that limit which import paths a value is allowed to + name, so that configs from an untrusted source can't reach arbitrary code. A + set of standard library paths that give code execution, e.g. ``os``, + ``subprocess`` and ``pickle``, is denied by default, see + :ref:`untrusted-configs` (`#959 + `__). + +Fixed +^^^^^ +- ``Callable`` types that have a class as return type, e.g. ``Callable[..., + Model]``, and instance factory protocols, accepted any class whose instances + are callable and any function, instead of only subclasses of the return type + and functions that return it (`#959 + `__). +- Instance factory protocols with a ``__call__`` that takes no parameters + instantiated the class instead of giving a factory (`#959 + `__). + +Deprecated +^^^^^^^^^^ +- Values that name a denied import path, e.g. a ``class_path`` of + ``subprocess.Popen``, currently only emit a deprecation warning and the import + proceeds. From v5.0.0 they will fail. Give a value to ``import_path_denylist`` + or ``import_path_allowlist`` in ``set_parsing_settings``, an empty list + included, to get the future behavior now and silence the warning (`#959 + `__). + + v4.51.0 (2026-08-20) -------------------- diff --git a/DOCUMENTATION.rst b/DOCUMENTATION.rst index 4a152139..39d5b3b1 100644 --- a/DOCUMENTATION.rst +++ b/DOCUMENTATION.rst @@ -2352,9 +2352,9 @@ be accepted. In this case the config would be like: .. note:: - It is also possible to provide to ``class_path`` a function that has as return - type a class. The accepted ``init_args`` would be the parameters of that - function. + It is also possible to provide to ``class_path`` a function that has as + return type a class. The accepted ``init_args`` would be the parameters of + that function. .. note:: @@ -2363,6 +2363,85 @@ be accepted. In this case the config would be like: reason they are not included in the known subclasses shown in the help. +.. _untrusted-configs: + +Untrusted configs +----------------- + +Resolving a ``class_path`` imports the named module and instantiates the named +class with the given ``init_args``, so a config decides what code runs. When the +configs come from a trusted source, e.g. the same repository as the code, this +is not a concern. When they don't, e.g. a config uploaded by a user of a +service, an import path denylist limits what a config can reach. + +Import paths that come from a value, i.e. a ``class_path``, a ``Callable``, a +``type[...]`` or a ``types.ModuleType`` given in a config file, the command line +or an environment variable, are checked against a denylist before the import +happens. Paths that come from code, e.g. type annotations and defaults, are +never checked. jsonargparse denies a set of paths by default, mostly standard +library modules that give arbitrary code execution, e.g. ``os``, ``subprocess``, +``pickle`` and ``importlib``. Two settings adjust the list: + +.. testsetup:: import_paths + + saved_import_path_settings = dict(_common.parsing_settings) + +.. testcode:: import_paths + + from jsonargparse import set_parsing_settings + + set_parsing_settings( + import_path_denylist=["mypackage._internal"], + import_path_allowlist=["functools.partial"], + ) + +An entry denies or allows a dot import path and everything under it, so ``os`` +also denies ``os.system``. The most specific entry decides, which is why +``functools.partial`` above is allowed even though ``functools`` is denied by +default. An entry given in both lists is allowed, so naming a default entry in +``import_path_allowlist`` is how to stop denying it. + +An object is denied by where it is defined, not only by the path used to reach +it. Modules commonly import others, e.g. ``import os``, so without this +``some.module.os.system`` would give the same object as the denied +``os.system``. This second check can only happen once the object is resolved, so +it prevents the object from being used, unlike the check on the given path, +which prevents the import from happening at all. + +Entries given are added to the ones denied by default, they don't replace them. +For configs that are entirely untrusted, prefer denying everything and allowing +only what the application expects. The ``*`` entry is only accepted in +``import_path_denylist``: + +.. testcode:: import_paths + + set_parsing_settings( + import_path_denylist=["*"], + import_path_allowlist=["mypackage.tools"], + ) + +.. testcleanup:: import_paths + + _common.parsing_settings.clear() + _common.parsing_settings.update(saved_import_path_settings) + +.. note:: + + A denylist is a mitigation, not a sandbox. A large enough set of installed + dependencies is likely to contain something that reaches a denied capability + without naming a denied path, e.g. a class that runs a command given to it. + Only ``*`` plus a narrow allowlist gives a bound on what a config can + import. + +.. note:: + + Until v5.0.0 a denied import path only gives a deprecation warning and the + import proceeds, so that existing configs don't break. Giving a value to + ``import_path_denylist`` or ``import_path_allowlist``, an empty list + included, makes denied import paths fail instead. From v5.0.0 they always + fail. + + .. _sub-config-files: Sub-config files diff --git a/jsonargparse/_actions.py b/jsonargparse/_actions.py index 62d3a55f..8e38cd26 100644 --- a/jsonargparse/_actions.py +++ b/jsonargparse/_actions.py @@ -9,7 +9,15 @@ from contextvars import ContextVar from typing import Any -from ._common import Action, NonParsingAction, get_parsing_setting, is_subclass, is_subclasses_disabled, parser_context +from ._common import ( + Action, + ImportDenied, + NonParsingAction, + get_parsing_setting, + is_subclass, + is_subclasses_disabled, + parser_context, +) from ._loaders_dumpers import get_loader_exceptions, load_value from ._namespace import Namespace from ._optionals import _get_config_read_mode, ruamel_support @@ -288,6 +296,8 @@ def resolve_subclass_spec(self, value): def resolve_class(class_path): try: return import_object(resolve_class_path_by_name(self.basetype, class_path)) + except ImportDenied: + raise except Exception: return None @@ -321,7 +331,7 @@ def _load_config(self, value, parser): with load_config_path_context(cfg_path), change_to_path_dir(cfg_path): cfg = parser._apply_actions(cfg, parent_key=self.dest) return cfg - except SubclassesDisabledError as ex: + except (SubclassesDisabledError, ImportDenied) as ex: raise TypeError(f'Parser key "{self.dest}":\n{indent_text(str(ex))}') from ex except (TypeError,) + get_loader_exceptions() as ex: str_ex = indent_text(f"- {ex}") @@ -560,7 +570,7 @@ def __init__( ValueError: If the parser parameter is invalid. """ self._parser = parser - if not isinstance(self._parser, import_object("jsonargparse.ArgumentParser")): + if not isinstance(self._parser, import_object("jsonargparse.ArgumentParser", check_path=False)): raise ValueError("Expected parser keyword argument to be an ArgumentParser.") @staticmethod diff --git a/jsonargparse/_common.py b/jsonargparse/_common.py index 16ba2fa0..b71d075e 100644 --- a/jsonargparse/_common.py +++ b/jsonargparse/_common.py @@ -117,6 +117,131 @@ def parser_context(**kwargs): context_var.reset(token) +class ImportDenied(ImportError, ValueError): + """Raised when an import path given as a value is not allowed. + + Both bases are needed so that the existing handlers report it as a parsing + error: subclass specs catch ``ImportError`` and typehint checking catches + ``ValueError``. + """ + + +# Import paths denied by default. Only checked for paths that come from parsed +# values, i.e. class paths, callables, types and modules given in configs, the +# command line or environment variables. Paths that come from code, e.g. type +# annotations and defaults, are never checked. Groups from higher to lower risk. +# Entries such as posix and _pickle are the C implementations that objects are +# defined in, e.g. os.system is defined in posix. They are next to the module +# that exposes them and both are needed, since a pure Python implementation is +# used when the C one is unavailable. +default_import_path_denylist = ( + # Command and code execution. Instantiation is the execution, so a class + # path plus init_args suffices to run arbitrary code or shell commands. + "builtins.eval", + "builtins.exec", + "builtins.compile", + "builtins.__import__", + "os", + "posix", + "nt", + "subprocess", + "_posixsubprocess", + "_winapi", + "multiprocessing", + "_multiprocessing", + "ctypes", + "_ctypes", + "runpy", + "code", + "codeop", + "pty", + "timeit", + "pdb", + "bdb", + "trace", + # Untrusted deserialization. Loading attacker influenced bytes is execution, + # and only a loader plus a file path is needed, not the payload itself. + "pickle", + "_pickle", + "marshal", + "shelve", + "dbm", + # Import machinery and package installation. Resolve or install code by name + # at runtime, which would reopen everything the other groups deny. + "importlib", + "_imp", + "pkgutil", + "zipimport", + "pydoc", + "sys", + "site", + "pip", + "setuptools", + "distutils", + "venv", + "sysconfig", + # Callable adapters and reflection. Wrap or synthesize a callable so that the + # call happens later in code that receives it, where no type check applies. + "functools", + "_functools", + "operator", + "_operator", + "inspect", + "types", + "ast", + "py_compile", + "compileall", + # Filesystem and process lifetime. Destructive or disruptive instead of + # executing, e.g. deleting trees, signals or deferring a call to exit. + "shutil", + "tempfile", + "pathlib", + "signal", + "_signal", + "atexit", + "gc", + # File access. Opening a path for writing truncates or creates it, and the + # archive and database openers do the same, so only a path is needed to + # destroy or plant a file, not any payload. builtins.open and the io classes + # are the primitives, the rest wrap them. + "builtins.open", + "io", + "_io", + "fileinput", + "mmap", + "fcntl", + "gzip", + "bz2", + "lzma", + "zipfile", + "tarfile", + "sqlite3", + "_sqlite3", + "logging.config", + "logging.handlers", + "logging.FileHandler", + # Network and external launch. Exfiltration primitives and launching an + # external program with an argument that the config decides. The client and + # server modules connect or listen with a destination the config decides. + "socket", + "_socket", + "ssl", + "webbrowser", + "asyncio", + "urllib", + "http", + "ftplib", + "smtplib", + "poplib", + "imaplib", + "nntplib", + "telnetlib", + "socketserver", + "xmlrpc", + "wsgiref", +) + + parsing_settings: dict = { "validate_defaults": False, "validate_subclass_spec_in_any": False, @@ -126,9 +251,66 @@ def parser_context(**kwargs): "stubs_resolver_allow_py_files": False, "omegaconf_absolute_to_relative_paths": False, "unset_sentinel": None, + "import_path_verdicts": {entry: False for entry in default_import_path_denylist}, + "import_paths_enforced": False, # v5.0.0: change default to True } +def get_settings_logger() -> logging.Logger: + """Returns the logger for parsing settings, which being global have no parser to take one from.""" + return parse_logger({"level": "DEBUG"} if debug_mode_active() else False, "set_parsing_settings") + + +def set_import_path_verdicts(denylist: list[str] | None, allowlist: list[str] | None) -> None: + """Adds entries to the import path policy, allowlist taking precedence.""" + logger = get_settings_logger() + verdicts = dict(parsing_settings["import_path_verdicts"]) + for entries, allowed in [(denylist, False), (allowlist, True)]: + state = "allowed" if allowed else "denied" + for entry in entries or []: + if entry == "*" and allowed: + raise ValueError("'*' is only accepted in import_path_denylist, to deny all not allowed paths.") + if not isinstance(entry, str) or (entry != "*" and not all(p.isidentifier() for p in entry.split("."))): + raise ValueError(f"Expected import path entries to be dot import paths or '*', but got {entry!r}.") + previous = verdicts.get(entry) + verdicts[entry] = allowed + if previous is None: + logger.debug(f"Import path {entry!r} added as {state}") + elif previous == allowed: + logger.debug(f"Import path {entry!r} already {state}") + else: + logger.debug(f"Import path {entry!r} changed to {state}") + parsing_settings["import_path_verdicts"] = verdicts + parsing_settings["import_paths_enforced"] = True + + +def denying_import_path_entry(path: str) -> str | None: + """Returns the most specific entry that denies an import path, if any.""" + verdicts = get_parsing_setting("import_path_verdicts") + parts = path.split(".") + for num in range(len(parts), 0, -1): + entry = ".".join(parts[:num]) + if entry in verdicts: + return None if verdicts[entry] else entry + return "*" if "*" in verdicts else None # '*' is only accepted as a denylist entry + + +def check_import_path(path: str) -> None: + """Fails or warns when an import path given as a value is denied.""" + entry = denying_import_path_entry(path) + if entry is None: + return + message = ( + f"Importing '{path}' is not allowed, denied by the {entry!r} entry of the import path denylist. " + "Add the import path to import_path_allowlist in set_parsing_settings to allow it." + ) + if get_parsing_setting("import_paths_enforced"): + raise ImportDenied(message) + from ._deprecated import import_paths_not_enforced + + import_paths_not_enforced(path, message) + + def get_env_var_bool(name: str) -> bool: raw_value = os.getenv(name, "") value = raw_value.lower() @@ -153,6 +335,8 @@ def set_parsing_settings( unset_sentinel: bool | None = None, subclasses_disabled: list[type | Callable[[type], bool]] | None = None, subclasses_enabled: list[type | str] | None = None, + import_path_denylist: list[str] | None = None, + import_path_allowlist: list[str] | None = None, ) -> None: """ Modify global parser settings that affect parser creation and parsing behavior. @@ -217,6 +401,15 @@ class when a value for a type that accepts any value, i.e. ``Any``, corresponding function from ``subclasses_disabled``. By default, the following disable functions are registered: ``is_pure_dataclass``, ``is_pydantic_model``, ``is_attrs_class`` and ``is_final_class``. + import_path_denylist: Import paths that a value is not allowed to name, + added to the ones denied by default. An entry denies a dot import + path and everything under it, e.g. ``os`` also denies ``os.system``. + The entry ``*`` denies everything, so that only what the allowlist + permits is importable. + import_path_allowlist: Import paths that a value is allowed to name, + taking precedence over the denylist for the same entry. The most + specific entry decides, so ``functools.partial`` here allows only + that path out of a denied ``functools``. """ # validate_defaults if isinstance(validate_defaults, bool): @@ -272,6 +465,9 @@ class when a value for a type that accepts any value, i.e. ``Any``, parsing_settings["unset_sentinel"] = Unset if unset_sentinel else None elif unset_sentinel is not None: raise ValueError(f"unset_sentinel must be a boolean, but got {unset_sentinel}.") + # import paths + if import_path_denylist is not None or import_path_allowlist is not None: + set_import_path_verdicts(import_path_denylist, import_path_allowlist) # subclass behavior if subclasses_disabled or subclasses_enabled: subclass_type_behavior( diff --git a/jsonargparse/_completions.py b/jsonargparse/_completions.py index e7647801..2ba4e0d1 100644 --- a/jsonargparse/_completions.py +++ b/jsonargparse/_completions.py @@ -398,7 +398,7 @@ def add_subactions_and_get_subclass_choices( if isinstance(class_or_path, str): choices.append(class_or_path) try: - cls = import_object(class_or_path) if isinstance(class_or_path, str) else class_or_path + cls = import_object(class_or_path, check_path=False) if isinstance(class_or_path, str) else class_or_path params = get_signature_parameters(cls, None, parser._logger) except Exception as ex: parser._logger.debug(f"Unable to get signature parameters for '{name}': {ex}") diff --git a/jsonargparse/_deprecated.py b/jsonargparse/_deprecated.py index 6dbd4336..3105ef28 100644 --- a/jsonargparse/_deprecated.py +++ b/jsonargparse/_deprecated.py @@ -1090,6 +1090,23 @@ def unset_instantiate_subclass_spec_in_any() -> bool: return True +def import_paths_not_enforced(path: str, message: str) -> None: + """Warns instead of failing when an import path given as a value is denied. + + Remove in v5.0.0, changing the default of the ``import_paths_enforced`` + setting from ``False`` to ``True``, and thus making + ``_common.check_import_path`` always raise. + """ + deprecation_warning( + f"import_path_denied:{path}", + f""" + {message} Currently this only warns, but from v5.0.0 it will fail. To get the future + behavior now and silence this warning, call set_parsing_settings with + import_path_denylist and/or import_path_allowlist. + """, + ) + + def renamed_parameter_warning(renames: dict[str, str], stacklevel: int = 1): def decorator(func): diff --git a/jsonargparse/_typehints.py b/jsonargparse/_typehints.py index b8b5cd71..8dba9342 100644 --- a/jsonargparse/_typehints.py +++ b/jsonargparse/_typehints.py @@ -57,6 +57,8 @@ remove_actions, ) from ._common import ( + ImportDenied, + check_import_path, get_generic_origin, get_parsing_setting, get_unaliased_type, @@ -1360,10 +1362,13 @@ def adapt_typehints( if isinstance(val, ModuleType): if serialize: val = val.__name__ - elif not is_importable_module_path(val): - raise_unexpected_value("Expected an import path corresponding to a module", val) - elif instantiate_classes: - val = import_module(val) + else: + if isinstance(val, str): + check_import_path(val) + if not is_importable_module_path(val): + raise_unexpected_value("Expected an import path corresponding to a module", val) + elif instantiate_classes: + val = import_module(val) # UnionType and GenericAlias elif typehint in type_expression_types: @@ -1374,6 +1379,8 @@ def adapt_typehints( expected = f"Expected a string with a {type_expression_types[typehint]} type expression" try: type_expression = str_to_type_expression(val) + except ImportDenied: + raise except Exception as ex: raise_unexpected_value(expected, val, ex) if not isinstance(type_expression, typehint): @@ -1548,6 +1555,11 @@ def adapt_typehints( if inspect.isclass(val_obj): val = Namespace(class_path=class_path) elif callable(val_obj): + subclass_types = get_subclass_types(return_type) + if subclass_types and not function_returns_subclass(val_obj, subclass_types, logger): + raise ImportError( + f"Expected '{class_path}' to be a function that returns {type_to_str(return_type)}." + ) val = val_obj else: raise ImportError(f"Unexpected import object {val_obj}") @@ -1564,12 +1576,18 @@ def adapt_typehints( ) val, partial_skip_args = adapt_partial_callable_class(typehint, val) val_class = import_object(val["class_path"]) - if inspect.isclass(val_class) and not (partial_skip_args or callable_instances(val_class)): - base_type = get_callable_return_type(typehint) or typehint - raise ImportError( - f"Expected '{val['class_path']}' to be a class that instantiates into callable " - f"or a subclass of {base_type}." - ) + if inspect.isclass(val_class) and partial_skip_args is None: + # partial_skip_args is only None when not a subclass of the return type + return_type = get_callable_return_type(typehint) + if get_subclass_types(return_type): + raise ImportError( + f"Expected '{val['class_path']}' to be a subclass of {type_to_str(return_type)}." + ) + if not callable_instances(val_class): + raise ImportError( + f"Expected '{val['class_path']}' to be a class that instantiates into callable " + f"or a subclass of {return_type or typehint}." + ) val["class_path"] = get_import_path(val_class) val = adapt_class_type( val, @@ -2080,6 +2098,17 @@ def get_subclass_names(typehint, callable_return=False): ) +def function_returns_subclass(function, subclass_types, logger) -> bool: + """Whether the return type of a function is a subclass of the given types.""" + from ._postponed_annotations import get_return_type + + try: + return_type = get_return_type(function, logger) + except ValueError: + return False # e.g. a builtin that doesn't have an inspectable signature + return is_subclass(return_type, subclass_types) + + def adapt_partial_callable_class(callable_type, subclass_spec): partial_skip_args = None return_type = get_callable_return_type(callable_type) @@ -2267,7 +2296,7 @@ def adapt_class_type( # kept as is for an Any typed parameter, which must not be expanded into kwargs init_kwargs = dict(init_args.items(branches=True, nested=False)) - if partial_skip_args: + if partial_skip_args is not None: # an empty set for a factory that takes no arguments return partial( instantiator_fn, val_class, @@ -2720,7 +2749,7 @@ def typehint_metavar(typehint): def serialize_class_instance(val): with suppress(Exception): import_path = get_import_path(val) - if import_path and import_object(import_path) is val: + if import_path and import_object(import_path, check_path=False) is val: return import_path val = f"Unable to serialize instance {val}" warning(val) diff --git a/jsonargparse/_util.py b/jsonargparse/_util.py index 18a41633..59c2605a 100644 --- a/jsonargparse/_util.py +++ b/jsonargparse/_util.py @@ -1,5 +1,6 @@ """Collection of general functions and classes.""" +import functools import inspect import os import textwrap @@ -19,6 +20,7 @@ ) from ._common import ( + check_import_path, get_generic_origin, parser_capture, parser_context, @@ -184,12 +186,18 @@ def parse_value_or_config(value: Any, enable_path: bool = True, simple_types: bo return value, cfg_path -def import_object(name: str): - """Returns an object in a module given its dot import path.""" +def import_object(name: str, check_path: bool = True): + """Returns an object in a module given its dot import path. + + ``check_path`` must only be false when the path comes from code, e.g. a type + annotation, instead of from a parsed value. + """ if not isinstance(name, str) or "." not in name: raise ValueError(f"Expected a dot import path string: {name}") if not all(x.isidentifier() for x in name.split(".")): raise ValueError(f"Unexpected import path format: {name}") + if check_path: + check_import_path(name) name_module, name_object = name.rsplit(".", 1) try: parent = __import__(name_module, fromlist=[name_object]) @@ -199,12 +207,57 @@ def import_object(name: str): name_module, name_object1 = name_module.rsplit(".", 1) parent = getattr(__import__(name_module, fromlist=[name_object1]), name_object1) obj = getattr(parent, name_object) + if check_path: + for canonical in canonical_import_paths(obj): + if canonical != name: + check_import_path(canonical) if not (inspect.isclass(obj) or inspect.ismodule(obj)): # an instance doesn't know where it was imported from, so it is remembered to make it serializable resolved_import_paths.add(obj, name) return obj +def canonical_import_path(obj) -> str | None: + """Returns where an object is defined, which can differ from the path used to import it. + + Modules commonly import others, e.g. ``import os``, so ``some.module.os.system`` + imports and gives the same object as ``os.system``. Not ``get_import_path`` + because that gives the shortest path, which can be a re-export that hides where + the object is defined, and it fails for objects that have no import path. + """ + if inspect.ismodule(obj): + return obj.__name__ + if not (inspect.isclass(obj) or inspect.isroutine(obj)): + return None + module = getattr(obj, "__module__", None) + qualname = getattr(obj, "__qualname__", None) + return f"{module}.{qualname}" if module and qualname else None + + +def canonical_import_paths(obj) -> set: + """Returns the canonical paths that must be allowed for an object to be usable. + + A class, routine or module has a single defining path. An object that instead + wraps or exposes a callable without an import path of its own, e.g. a + ``functools.partial`` or a callable instance, is denied by the callable it + reaches: the bound function for a partial and the defining class for an + instance. Otherwise binding or instancing a denied callable under an allowed + name would evade the denylist. + """ + paths: set = set() + stack = [obj] + while stack: + current = stack.pop() + canonical = canonical_import_path(current) + if canonical: + paths.add(canonical) + if isinstance(current, functools.partial): + stack.append(current.func) # a partial is denied by the callable it binds + elif not (inspect.isclass(current) or inspect.ismodule(current) or inspect.isroutine(current)): + stack.append(type(current)) # a callable instance is denied by its class + return paths + + unresolvable_import_paths: dict[Any, str] = {} @@ -313,7 +366,7 @@ def get_import_path(value: Any) -> str | None: def object_path_serializer(value): try: path = get_import_path(value) - reimported = import_object(path) + reimported = import_object(path, check_path=False) if value is not reimported: raise ValueError return path diff --git a/jsonargparse/typing.py b/jsonargparse/typing.py index f19ef029..3547a082 100644 --- a/jsonargparse/typing.py +++ b/jsonargparse/typing.py @@ -510,7 +510,7 @@ def register_type( def register_type_on_first_use(import_path: str, *args, **kwargs): registration_pending[import_path] = lambda: register_type( - import_object(import_path), + import_object(import_path, check_path=False), *args, **kwargs, ) diff --git a/jsonargparse_tests/test_deprecated.py b/jsonargparse_tests/test_deprecated.py index b488e7d7..cceeb283 100644 --- a/jsonargparse_tests/test_deprecated.py +++ b/jsonargparse_tests/test_deprecated.py @@ -30,8 +30,10 @@ get_config_read_mode, set_config_read_mode, set_docstring_parse_options, + set_parsing_settings, set_url_support, ) +from jsonargparse._common import ImportDenied from jsonargparse._deprecated import ( ActionEnum, ActionJsonnetExtVars, @@ -57,11 +59,12 @@ ruamel_support, url_support, ) -from jsonargparse._util import argument_error +from jsonargparse._util import argument_error, import_object from jsonargparse.typing import Path from jsonargparse_tests.conftest import ( get_parser_help, is_posix, + patch_parsing_settings, responses_activate, skip_if_docstring_parser_unavailable, skip_if_fsspec_unavailable, @@ -1448,3 +1451,44 @@ def test_instantiate_subclass_spec_in_any_deprecation(parser): init = parser.instantiate(cfg) assert isinstance(init.any, Calendar) assert w == [] + + +# denied import paths only warn deprecation tests + + +@patch_parsing_settings +def test_denied_import_path_not_enforced_warns_and_imports(): + shown_deprecation_warnings.clear() + with catch_warnings(record=True) as w: + assert import_object("subprocess.Popen") is not None + assert len(w) == 2 + assert "subprocess.Popen" in str(w[-1].message) + assert "from v5.0.0 it will fail" in str(w[-1].message) + + +@patch_parsing_settings +def test_denied_import_path_warns_once_per_path(): + shown_deprecation_warnings.clear() + with catch_warnings(record=True) as w: + import_object("subprocess.Popen") + import_object("subprocess.Popen") + import_object("subprocess.run") + assert len(w) == 3 + assert "subprocess.Popen" in str(w[1].message) + assert "subprocess.run" in str(w[2].message) + + +@pytest.mark.parametrize( + "settings", + [ + {"import_path_denylist": []}, + {"import_path_allowlist": []}, + {"import_path_denylist": ["calendar"]}, + {"import_path_allowlist": ["calendar"]}, + ], +) +@patch_parsing_settings +def test_denied_import_path_enforced_when_setting_given(settings): + set_parsing_settings(**settings) + with pytest.raises(ImportDenied): + import_object("subprocess.Popen") diff --git a/jsonargparse_tests/test_import_paths.py b/jsonargparse_tests/test_import_paths.py new file mode 100644 index 00000000..5a9417a1 --- /dev/null +++ b/jsonargparse_tests/test_import_paths.py @@ -0,0 +1,362 @@ +from __future__ import annotations + +import calendar +import functools +import json +import logging +import operator +import os +import pickle +import sys +from calendar import Calendar +from types import ModuleType, UnionType +from typing import Any, Callable, Optional +from unittest.mock import patch + +import pytest + +from jsonargparse import ArgumentError, FromConfigMixin, set_parsing_settings +from jsonargparse._common import ImportDenied, check_import_path, get_settings_logger, null_logger +from jsonargparse._util import import_object +from jsonargparse_tests.conftest import capture_logs, get_parser_help, json_or_yaml_dump + + +@pytest.fixture(autouse=True) +def import_paths_settings(parsing_settings_patch): + yield + + +# settings validation + + +def test_invalid_entry_not_a_string(): + with pytest.raises(ValueError, match="import path"): + set_parsing_settings(import_path_denylist=[123]) + + +@pytest.mark.parametrize("entry", ["my-pkg", "os.", "", "mypkg.*", "a b"]) +def test_invalid_entry_not_an_import_path(entry): + with pytest.raises(ValueError, match="import path"): + set_parsing_settings(import_path_allowlist=[entry]) + + +def test_star_only_accepted_in_denylist(): + with pytest.raises(ValueError, match="only accepted in import_path_denylist"): + set_parsing_settings(import_path_allowlist=["*"]) + + +def test_invalid_entry_leaves_settings_unchanged(): + with pytest.raises(ValueError): + set_parsing_settings(import_path_denylist=["calendar", "not valid"]) + check_import_path("calendar.TextCalendar") + + +# policy resolution + + +def test_denied_by_default_entry(): + set_parsing_settings(import_path_denylist=[]) + with pytest.raises(ImportDenied, match="'subprocess'"): + check_import_path("subprocess.Popen") + + +def test_allowed_when_no_entry_matches(): + set_parsing_settings(import_path_denylist=[]) + check_import_path("calendar.TextCalendar") + + +def test_prefix_matches_only_on_component_boundary(): + set_parsing_settings(import_path_denylist=["cal"]) + check_import_path("calendar.TextCalendar") + with pytest.raises(ImportDenied): + check_import_path("cal.sub.Thing") + + +def test_more_specific_allow_wins_over_broader_deny(): + set_parsing_settings(import_path_allowlist=["functools.partial"]) + check_import_path("functools.partial") + with pytest.raises(ImportDenied, match="'functools'"): + check_import_path("functools.reduce") + + +def test_more_specific_deny_wins_over_broader_allow(): + set_parsing_settings(import_path_denylist=["calendar.TextCalendar"]) + check_import_path("calendar.Calendar") + with pytest.raises(ImportDenied, match="'calendar.TextCalendar'"): + check_import_path("calendar.TextCalendar") + + +def test_allow_wins_tie_removing_a_default_entry(): + set_parsing_settings(import_path_allowlist=["functools"]) + check_import_path("functools.reduce") + + +def test_entries_add_to_the_defaults(): + set_parsing_settings(import_path_denylist=["calendar"]) + with pytest.raises(ImportDenied): + check_import_path("calendar.TextCalendar") + with pytest.raises(ImportDenied): + check_import_path("subprocess.Popen") + + +def test_entries_accumulate_across_calls(): + set_parsing_settings(import_path_denylist=["calendar"]) + set_parsing_settings(import_path_allowlist=["calendar.Calendar"]) + check_import_path("calendar.Calendar") + with pytest.raises(ImportDenied): + check_import_path("calendar.TextCalendar") + + +def test_star_denies_everything_not_allowed(): + set_parsing_settings(import_path_denylist=["*"], import_path_allowlist=["calendar"]) + check_import_path("calendar.TextCalendar") + with pytest.raises(ImportDenied, match=r"'\*'"): + check_import_path("json.JSONEncoder") + + +def test_only_builtins_code_execution_names_denied(): + set_parsing_settings(import_path_denylist=[]) + check_import_path("builtins.print") + for name in ["eval", "exec", "compile", "__import__"]: + with pytest.raises(ImportDenied): + check_import_path(f"builtins.{name}") + + +@pytest.mark.parametrize( + "path", + [ + "builtins.open", + "io.FileIO", + "logging.FileHandler", + "logging.handlers.SocketHandler", + "sqlite3.connect", + "gzip.GzipFile", + "tarfile.TarFile", + "urllib.request.urlopen", + "http.client.HTTPConnection", + "ftplib.FTP", + "smtplib.SMTP", + "socketserver.TCPServer", + "xmlrpc.client.ServerProxy", + "asyncio.create_subprocess_shell", + ], +) +def test_file_and_network_paths_denied_by_default(path): + set_parsing_settings(import_path_denylist=[]) + with pytest.raises(ImportDenied): + check_import_path(path) + + +# debug logging + + +def test_debug_log_disabled_by_default(): + assert get_settings_logger() is null_logger + + +@patch.dict(os.environ, {"JSONARGPARSE_DEBUG": "true"}) +def test_debug_log_entries_set(): + with capture_logs(get_settings_logger()) as logs: + set_parsing_settings( + import_path_denylist=["mypackage.internal", "subprocess"], + import_path_allowlist=["functools"], + ) + assert "Import path 'mypackage.internal' added as denied" in logs.getvalue() + assert "Import path 'subprocess' already denied" in logs.getvalue() + assert "Import path 'functools' changed to allowed" in logs.getvalue() + + +# import_object + + +def test_denied_before_the_module_is_imported(): + set_parsing_settings(import_path_denylist=["unimportable_denied_module"]) + with pytest.raises(ImportDenied): + import_object("unimportable_denied_module.Thing") + assert "unimportable_denied_module" not in sys.modules + + +def test_denied_module_reexported_by_another_module(): + set_parsing_settings(import_path_denylist=[]) + with pytest.raises(ImportDenied, match="'os'"): + import_object("jsonargparse._util.os") + with pytest.raises(ImportDenied, match=f"'{os.system.__module__}'"): + import_object("jsonargparse._util.os.system") # os.system is defined in posix, nt on Windows + + +def test_denied_object_reexported_under_another_name(): + set_parsing_settings(import_path_denylist=[]) + with pytest.raises(ImportDenied, match="'_pickle'"): + import_object(f"{__name__}.load_bytes") + + +def test_denied_callable_bound_by_a_partial(): + set_parsing_settings(import_path_denylist=[]) + with pytest.raises(ImportDenied, match=f"'{os.system.__module__}'"): + import_object(f"{__name__}.system_partial") # partial bound to os.system, defined in posix, nt on Windows + + +def test_denied_callable_exposed_by_an_instance(): + set_parsing_settings(import_path_denylist=[]) + with pytest.raises(ImportDenied, match="'operator'"): + import_object(f"{__name__}.attr_getter") # instance of the denied operator.attrgetter + + +def test_object_without_canonical_path_is_not_rechecked(): + set_parsing_settings(import_path_denylist=[]) + assert import_object("calendar.day_name") is calendar.day_name + + +def test_object_without_module_is_not_rechecked(): + set_parsing_settings(import_path_denylist=[]) + assert import_object(f"{__name__}.no_module_function") is no_module_function + + +def test_import_object_unaffected_when_allowed(): + set_parsing_settings(import_path_denylist=[]) + assert import_object("calendar.TextCalendar") is calendar.TextCalendar + + +# objects imported by the tests above + + +load_bytes = pickle.loads # defined in _pickle, reachable here under a different module + +system_partial = functools.partial(os.system, "echo test") # binds a denied callable, defined in posix + +attr_getter = operator.attrgetter("__globals__") # instance of the denied operator.attrgetter + + +def no_module_function(): + """Mimics extension functions that have __module__ set to None.""" + + +no_module_function.__module__ = None # type: ignore[assignment] + + +# parsing + + +class Data: + def __init__(self, cal: Optional[Calendar] = None): + self.cal = cal # pragma: no cover + + +def test_subclass_class_path_denied(parser): + set_parsing_settings(import_path_denylist=["calendar"]) + parser.add_argument("--cal", type=Calendar) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args(["--cal=calendar.TextCalendar"]) + + +def test_subclass_class_path_allowed_by_exception(parser): + set_parsing_settings(import_path_denylist=["calendar"], import_path_allowlist=["calendar.TextCalendar"]) + parser.add_argument("--cal", type=Calendar) + cfg = parser.parse_args(["--cal=calendar.TextCalendar"]) + assert cfg.cal.class_path == "calendar.TextCalendar" + + +def test_subclass_class_path_denied_by_default(parser): + set_parsing_settings(import_path_denylist=[]) + parser.add_argument("--handler", type=logging.Handler) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args(["--handler=logging.FileHandler"]) + + +def test_star_callable_bound_to_denied_callable_denied(parser): + set_parsing_settings(import_path_denylist=["*"], import_path_allowlist=["jsonargparse_tests"]) + parser.add_argument("--fn", type=Callable) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args([f"--fn={__name__}.system_partial"]) + + +def test_any_type_class_path_denied(parser): + set_parsing_settings(import_path_denylist=[], validate_subclass_spec_in_any=True) + parser.add_argument("--any", type=Any) + spec = json.dumps({"class_path": "subprocess.Popen", "init_args": {"args": ["ls"]}}) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args([f"--any={spec}"]) + + +def test_callable_type_denied(parser): + set_parsing_settings(import_path_denylist=[]) + parser.add_argument("--fn", type=Callable) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args(["--fn=os.getcwd"]) + + +def test_type_type_denied(parser): + set_parsing_settings(import_path_denylist=[]) + parser.add_argument("--cls", type=type) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args(["--cls=subprocess.Popen"]) + + +def test_module_type_denied(parser): + set_parsing_settings(import_path_denylist=[]) + parser.add_argument("--mod", type=ModuleType) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args(["--mod=subprocess"]) + + +def test_module_type_non_string_value(parser): + set_parsing_settings(import_path_denylist=[]) + parser.add_argument("--mod", type=ModuleType) + with pytest.raises(ArgumentError, match="import path corresponding to a module"): + parser.parse_args(["--mod=1"]) + + +def test_module_type_allowed(parser): + set_parsing_settings(import_path_denylist=[]) + parser.add_argument("--mod", type=ModuleType) + cfg = parser.parse_args(["--mod=calendar"]) + assert parser.instantiate(cfg).mod is calendar + + +def test_type_expression_denied(parser): + set_parsing_settings(import_path_denylist=[]) + parser.add_argument("--expr", type=UnionType) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args(["--expr=subprocess.Popen | int"]) + + +def test_subclasses_disabled_class_path_denied(parser, subclass_behavior): + set_parsing_settings(import_path_denylist=["calendar"], subclasses_disabled=[Calendar]) + parser.add_argument("--cal", type=Calendar) + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args(["--cal=calendar.TextCalendar"]) + + +def test_help_class_path_denied(parser): + set_parsing_settings(import_path_denylist=["calendar"]) + parser.add_subclass_arguments(Calendar, "cal") + with pytest.raises(ArgumentError, match="not allowed"): + parser.parse_args(["--cal.help=calendar.TextCalendar"]) + + +def test_from_config_class_path_denied(tmp_cwd): + class Base(FromConfigMixin): + def __init__(self, p: int = 1): + self.p = p # pragma: no cover + + set_parsing_settings(import_path_denylist=["calendar"]) + config = json_or_yaml_dump({"class_path": "calendar.TextCalendar"}) + (tmp_cwd / "config.yaml").write_text(config) + with pytest.raises(ImportDenied): + Base.from_config("config.yaml") + + +# not applied to code given import paths + + +def test_dump_of_denied_default_is_not_checked(parser): + parser.add_argument("--cls", type=type, default=calendar.TextCalendar) + set_parsing_settings(import_path_denylist=["calendar"]) + assert "calendar.TextCalendar" in parser.dump(parser.get_defaults()) + + +def test_annotation_of_denied_type_not_checked(parser): + set_parsing_settings(import_path_denylist=["calendar"]) + parser.add_class_arguments(Data, "data") + help_str = get_parser_help(parser) + assert "--data.cal" in help_str diff --git a/jsonargparse_tests/test_subclasses.py b/jsonargparse_tests/test_subclasses.py index d0afbe48..04da0842 100644 --- a/jsonargparse_tests/test_subclasses.py +++ b/jsonargparse_tests/test_subclasses.py @@ -53,6 +53,13 @@ ) +@pytest.fixture +def allow_gzip_import_path(): + with patch.dict("jsonargparse._common.parsing_settings"): + set_parsing_settings(import_path_allowlist=["gzip"]) + yield + + class BaseC: def __init__(self, p: int = 0): self.p = p @@ -235,7 +242,7 @@ class LocalSubC(BaseC): assert "LocalSubC" not in help_str -def test_subclass_known_subclasses_multiple_bases(parser): +def test_subclass_known_subclasses_multiple_bases(parser, allow_gzip_import_path): parser.add_argument("--op", type=Union[BaseC, GzipFile, None]) help_str = get_parser_help(parser) for class_path in [f"{__name__}.BaseC", f"{__name__}.SubA", f"{__name__}.SubB", "gzip.GzipFile"]: @@ -733,7 +740,7 @@ def test_subclass_class_name_parse(parser): assert cfg.op.class_path == f"{__name__}.SubA" -def test_subclass_class_name_help(parser): +def test_subclass_class_name_help(parser, allow_gzip_import_path): parser.add_argument("--op", type=Union[BaseC, GzipFile, None]) help_str = get_parse_args_stdout(parser, ["--op.help=GzipFile"]) assert "Help for --op.help=gzip.GzipFile" in help_str @@ -774,7 +781,7 @@ def test_subclass_invalid_class_name(parser): ctx.match("NotASubclass") -def test_subclass_class_name_then_invalid_init_args(parser): +def test_subclass_class_name_then_invalid_init_args(parser, allow_gzip_import_path): parser.add_argument("--op", type=Union[BaseC, GzipFile]) with pytest.raises(ArgumentError) as ctx: parser.parse_args(["--op=SubA", "--op=GzipFile", "--op.p=2"]) diff --git a/jsonargparse_tests/test_typehints.py b/jsonargparse_tests/test_typehints.py index f016106d..82d6833e 100644 --- a/jsonargparse_tests/test_typehints.py +++ b/jsonargparse_tests/test_typehints.py @@ -2576,6 +2576,66 @@ def test_callable_protocol_instance_factory(parser, subtests): assert "--optimizer.params" not in help_str +@pytest.mark.parametrize( + "typehint", + [ + Callable[[List[float]], Optimizer], + Callable[..., Optimizer], + OptimizerFactory, + ], + ids=["callable_args", "callable_ellipsis", "protocol"], +) +def test_callable_return_type_bounds_the_accepted_class(parser, typehint): + parser.add_argument("--optimizer", type=typehint) + with pytest.raises(ArgumentError) as ctx: + parser.parse_args([f"--optimizer={__name__}.CallableClassPath"]) + ctx.match(f"Expected '{__name__}.CallableClassPath' to be a subclass of .*Optimizer") + + +class NoArgsOptimizerFactory(Protocol): + def __call__(self) -> Optimizer: ... + + +def test_callable_protocol_instance_factory_without_parameters(parser): + parser.add_argument("--optimizer", type=NoArgsOptimizerFactory) + cfg = parser.parse_args(["--optimizer=Adam", "--optimizer.params=[1.2]", "--optimizer.lr=0.01"]) + assert cfg.optimizer.class_path == f"{__name__}.Adam" + init = parser.instantiate(cfg) + # nothing to skip, so instantiate still gives a factory, not the instance + optimizer = init.optimizer() + assert isinstance(optimizer, Adam) + assert optimizer.params == [1.2] + assert optimizer.lr == 0.01 + + +def optimizer_factory(params: List[float]) -> Optimizer: + return SGD(params) # pragma: no cover + + +def test_callable_return_type_bounds_the_accepted_function(parser): + parser.add_argument("--optimizer", type=Callable[[List[float]], Optimizer]) + cfg = parser.parse_args([f"--optimizer={__name__}.optimizer_factory"]) + assert cfg.optimizer is optimizer_factory + + +@pytest.mark.parametrize( + "path", + ["calendar.month", "time.time"], + ids=["return type not annotated", "signature not inspectable"], +) +def test_callable_return_type_rejects_function(parser, path): + parser.add_argument("--optimizer", type=Callable[[List[float]], Optimizer]) + with pytest.raises(ArgumentError) as ctx: + parser.parse_args([f"--optimizer={path}"]) + ctx.match(f"Expected '{path}' to be a function that returns .*Optimizer") + + +def test_callable_without_return_type_accepts_any_function(parser): + parser.add_argument("--optimizer", type=Callable) + cfg = parser.parse_args(["--optimizer=calendar.month"]) + assert cfg.optimizer is calendar.month + + OptimizerVar = TypeVar("OptimizerVar", covariant=True)