diff --git a/packages/@n8n/task-runner-python/src/_sandbox_callables.py b/packages/@n8n/task-runner-python/src/_sandbox_callables.py index 364789e6fd6..ca0402e97df 100644 --- a/packages/@n8n/task-runner-python/src/_sandbox_callables.py +++ b/packages/@n8n/task-runner-python/src/_sandbox_callables.py @@ -120,6 +120,77 @@ class _SafePrint(_HardenedCallable): return "" +def _import_module_anchor(args, kwargs) -> str | None: + """The string anchor of ``import_module(name, package)``, or ``None``. + + ``package`` may be a keyword or the first positional argument. Anything that + isn't a string in that slot (notably ``__import__``'s ``globals`` dict) is + not an anchor. + """ + # The anchor is a string, passed by keyword or as the first positional arg. + package = kwargs.get("package") + if isinstance(package, str): + return package + if args and isinstance(args[0], str): + return args[0] + # No string in that slot (e.g. ``__import__``'s globals dict): no anchor. + return None + + +def _level_relative_target(name, args, kwargs) -> tuple[str, str] | None: + """The ``(leading-dot name, anchor)`` for a relative + ``__import__(name, globals, .., level)``, or ``None`` if it isn't one. + + A relative ``__import__`` is a bare ``name`` with ``level > 0``; the anchor + is the importing module's ``globals["__package__"]``. The dots are rebuilt + onto the name so it resolves the same way as the ``import_module`` form. + """ + # Only a string name can be reshaped into a leading-dot form and resolved. + if not isinstance(name, str): + return None + # ``level`` is the dot count, ``__import__``'s 5th arg (keyword or 4th of *args). + level = kwargs.get("level") + if level is None and len(args) >= 4: + level = args[3] + # level <= 0 is an absolute import, not a relative one. + if not isinstance(level, int) or level <= 0: + return None + # The anchor lives in the importer's globals (``__import__``'s 2nd arg). + importer_globals = kwargs.get("globals") + if importer_globals is None and args and isinstance(args[0], dict): + importer_globals = args[0] + if not isinstance(importer_globals, dict): + return None + # No usable parent package: let the caller fall through (and fail closed) + # rather than guess an anchor. + anchor = importer_globals.get("__package__") + if not (isinstance(anchor, str) and anchor): + return None + # Rebuild the leading-dot form, e.g. ("sub", level 1) -> ".sub". + return "." * level + name, anchor + + +def _validation_target(name, args, kwargs) -> tuple[str, str | None]: + """Return the ``(name, anchor_package)`` the allowlist should check. + + Relative imports arrive in two shapes; both are normalized to a leading-dot + name plus a string anchor, which the validator resolves to a top-level + package. The real import is performed separately with the original name. + """ + # import_module(".sub", "pkg"): the name already carries its dots. + anchor = _import_module_anchor(args, kwargs) + if anchor is not None: + return name, anchor + + # __import__("sub", globals, .., level): rebuild the dotted name from level. + relative = _level_relative_target(name, args, kwargs) + if relative is not None: + return relative + + # Absolute import (or unrecognized shape): check the name as-is. + return name, None + + class _GuardedImport(_HardenedCallable): """Hardened wrapper around an import entry point. @@ -140,7 +211,8 @@ class _GuardedImport(_HardenedCallable): def __call__(self, name, *args, **kwargs): validate = object.__getattribute__(self, "_validate_import") config = object.__getattribute__(self, "_security_config") - is_allowed, error_msg = validate(name, config) + check_name, package = _validation_target(name, args, kwargs) + is_allowed, error_msg = validate(check_name, config, package) if not is_allowed: assert error_msg is not None raise SecurityViolationError( diff --git a/packages/@n8n/task-runner-python/src/import_validation.py b/packages/@n8n/task-runner-python/src/import_validation.py index 48961eb26db..02b255b003a 100644 --- a/packages/@n8n/task-runner-python/src/import_validation.py +++ b/packages/@n8n/task-runner-python/src/import_validation.py @@ -1,3 +1,4 @@ +import importlib.util import sys from src.config.security_config import SecurityConfig @@ -7,10 +8,19 @@ from src.constants import ERROR_STDLIB_DISALLOWED, ERROR_EXTERNAL_DISALLOWED def validate_module_import( module_path: str, security_config: SecurityConfig, + importing_package: str | None = None, ) -> tuple[bool, str | None]: stdlib_allow = security_config.stdlib_allow external_allow = security_config.external_allow + # Resolve a package-relative name. This always resolves within its own + # top-level package, so this never reaches a package outside the allowlist. + if module_path.startswith(".") and importing_package: + try: + module_path = importlib.util.resolve_name(module_path, importing_package) + except (ImportError, ValueError): + pass + module_name = module_path.split(".")[0] is_stdlib = module_name in sys.stdlib_module_names is_external = not is_stdlib diff --git a/packages/@n8n/task-runner-python/tests/unit/test_import_validation.py b/packages/@n8n/task-runner-python/tests/unit/test_import_validation.py new file mode 100644 index 00000000000..b5f4af0d94c --- /dev/null +++ b/packages/@n8n/task-runner-python/tests/unit/test_import_validation.py @@ -0,0 +1,98 @@ +"""Unit tests for module-import allowlist validation.""" + +from src.import_validation import validate_module_import +from src.config.security_config import SecurityConfig + + +def _config(external: set[str] | None = None) -> SecurityConfig: + return SecurityConfig( + stdlib_allow=set(), + external_allow=external if external is not None else set(), + builtins_deny=set(), + runner_env_deny=True, + ) + + +class TestPackageRelativeImports: + def test_relative_import_resolves_to_allowlisted_parent(self): + # A package-internal relative name (e.g. pydantic loading + # ``.type_adapter`` via importlib) anchored on an allowlisted package + # must pass once resolved to its absolute top-level package. + allowed, error = validate_module_import( + ".type_adapter", _config({"pydantic"}), "pydantic" + ) + assert allowed is True + assert error is None + + def test_relative_import_with_non_allowlisted_anchor_is_blocked(self): + # A relative name anchored on a non-allowlisted package is still + # rejected after resolution, and the error names the resolved package + # rather than the bare relative name. + allowed, error = validate_module_import( + ".sub", _config({"pydantic"}), "requests" + ) + assert allowed is False + assert error is not None + assert "requests.sub" in error + + def test_multi_level_relative_resolves_to_top_level_package(self): + # A parent-relative name (``..``) anchored within an allowlisted + # package resolves to that same top-level package and is allowed. + allowed, error = validate_module_import( + "..types", _config({"pydantic"}), "pydantic.v1" + ) + assert allowed is True + assert error is None + + def test_multi_level_relative_resolution_stays_within_anchor_top_level(self): + # A parent-relative name resolves within the anchor's own top-level + # package, so anchoring on a non-allowlisted package stays blocked and + # the error names the resolved package, never a different top-level one. + allowed, error = validate_module_import( + "..sub", _config({"pydantic"}), "requests.api" + ) + assert allowed is False + assert error is not None + assert "requests.sub" in error + + def test_relative_import_without_anchor_is_blocked(self): + # Without an anchor the relative name cannot be resolved; it must fail + # closed rather than slip through. + allowed, error = validate_module_import(".sub", _config({"pydantic"})) + assert allowed is False + assert error is not None + + def test_unresolvable_relative_import_is_blocked(self): + # When resolution fails (here, a name that reaches beyond the anchor's + # top-level package), the check falls back to the raw name and rejects. + allowed, error = validate_module_import("...x", _config({"a"}), "a.b") + assert allowed is False + assert error is not None + + def test_absolute_non_allowlisted_package_is_blocked(self): + allowed, error = validate_module_import("requests", _config({"pydantic"})) + assert allowed is False + assert error is not None + assert "requests" in error + + def test_absolute_allowlisted_package_passes(self): + allowed, error = validate_module_import("pydantic", _config({"pydantic"})) + assert allowed is True + assert error is None + + def test_absolute_submodule_inherits_top_level_decision(self): + # An absolute dotted name is checked by its top-level package, so a + # submodule of an allowlisted package passes and a submodule of a + # non-allowlisted one is blocked. + allowed, error = validate_module_import( + "pydantic.type_adapter", _config({"pydantic"}) + ) + assert allowed is True + assert error is None + + allowed, error = validate_module_import( + "requests.sessions", _config({"pydantic"}) + ) + assert allowed is False + assert error is not None + assert "requests" in error diff --git a/packages/@n8n/task-runner-python/tests/unit/test_sandbox_hardening.py b/packages/@n8n/task-runner-python/tests/unit/test_sandbox_hardening.py index 52d07dbc2c9..efe34efddc4 100644 --- a/packages/@n8n/task-runner-python/tests/unit/test_sandbox_hardening.py +++ b/packages/@n8n/task-runner-python/tests/unit/test_sandbox_hardening.py @@ -1,6 +1,7 @@ """Sandbox hardening regression tests.""" import importlib +import sys import pytest @@ -10,6 +11,8 @@ from src.task_executor import ( _PRISTINE_IMPORT_MODULE, ) from src.task_analyzer import TaskAnalyzer +from src._sandbox_callables import _GuardedImport +from src.import_validation import validate_module_import from src.errors import SecurityViolationError from src.config.security_config import SecurityConfig from src.constants import ( @@ -207,6 +210,193 @@ class TestImportAllowlistCoverage: result = _run(code, config) assert result == '{"k": 1}' + def test_importlib_resolves_package_relative_import_of_allowed_package( + self, tmp_path, monkeypatch + ): + # Mirrors how an allowlisted external package (e.g. pydantic) lazily + # loads its own submodules via ``import_module('.sub', pkg)``. The + # leading-dot name must resolve against the anchor package before the + # allowlist check, so the import succeeds when the package is allowed. + pkg = tmp_path / "synthpkg" + pkg.mkdir() + (pkg / "__init__.py").write_text("") + (pkg / "sub.py").write_text("VALUE = 42\n") + monkeypatch.syspath_prepend(str(tmp_path)) + + config = SecurityConfig( + stdlib_allow={"importlib"}, + external_allow={"synthpkg"}, + builtins_deny=set(), + runner_env_deny=True, + ) + code = ( + "import importlib\nreturn importlib.import_module('.sub', 'synthpkg').VALUE" + ) + try: + result = _run(code, config) + finally: + for name in ("synthpkg.sub", "synthpkg"): + sys.modules.pop(name, None) + assert result == 42 + + def test_importlib_relative_import_anchored_on_blocked_package_is_rejected( + self, + ): + # A leading-dot name anchored on a non-allowlisted package stays + # blocked after resolution. + config = SecurityConfig( + stdlib_allow={"importlib"}, + external_allow=set(), + builtins_deny=set(), + runner_env_deny=True, + ) + code = "import importlib\nreturn importlib.import_module('.sub', 'requests')" + with pytest.raises(SecurityViolationError): + _run(code, config) + + def test_importlib_resolves_package_relative_import_via_package_keyword( + self, tmp_path, monkeypatch + ): + # The anchor package may be passed as the ``package`` keyword argument + # rather than positionally; resolution must work the same way. + pkg = tmp_path / "synthpkgkw" + pkg.mkdir() + (pkg / "__init__.py").write_text("") + (pkg / "sub.py").write_text("VALUE = 7\n") + monkeypatch.syspath_prepend(str(tmp_path)) + + config = SecurityConfig( + stdlib_allow={"importlib"}, + external_allow={"synthpkgkw"}, + builtins_deny=set(), + runner_env_deny=True, + ) + code = ( + "import importlib\n" + "return importlib.import_module('.sub', package='synthpkgkw').VALUE" + ) + try: + result = _run(code, config) + finally: + for name in ("synthpkgkw.sub", "synthpkgkw"): + sys.modules.pop(name, None) + assert result == 7 + + def test_importlib_relative_import_with_non_str_anchor_is_rejected(self): + # A non-str anchor cannot resolve the relative name; it is ignored and + # the name is rejected cleanly rather than raising an unexpected error. + config = SecurityConfig( + stdlib_allow={"importlib"}, + external_allow={"pydantic"}, + builtins_deny=set(), + runner_env_deny=True, + ) + code = "import importlib\nreturn importlib.import_module('.x', package=123)" + with pytest.raises(SecurityViolationError): + _run(code, config) + + def test_importlib_dunder_import_resolves_package_relative_submodule( + self, tmp_path, monkeypatch + ): + # An allowlisted package lazy-loads its own submodule through the real + # guard via importlib.__import__("sub", globals(), .., level=1). The + # level-relative name resolves against the package and is allowed. + pkg = tmp_path / "synthpkgdunder" + pkg.mkdir() + (pkg / "__init__.py").write_text( + "import importlib\n" + "def load():\n" + " return importlib.__import__('sub', globals(), None, ('VALUE',), 1)\n" + ) + (pkg / "sub.py").write_text("VALUE = 11\n") + monkeypatch.syspath_prepend(str(tmp_path)) + + config = SecurityConfig( + stdlib_allow={"importlib"}, + external_allow={"synthpkgdunder"}, + builtins_deny=set(), + runner_env_deny=True, + ) + code = ( + "import importlib\n" + "return importlib.import_module('synthpkgdunder').load().VALUE" + ) + try: + result = _run(code, config) + finally: + for name in ("synthpkgdunder.sub", "synthpkgdunder"): + sys.modules.pop(name, None) + assert result == 11 + + def _guard(self, external: set[str]): + # A guard with a stub in place of the real import, so these tests + # exercise the allow/deny decision without performing an import. + config = SecurityConfig( + stdlib_allow=set(), + external_allow=external, + builtins_deny=set(), + runner_env_deny=True, + ) + return _GuardedImport( + config, validate_module_import, lambda *a, **k: "imported" + ) + + def test_safe_import_resolves_level_relative_against_globals_package(self): + # `__import__`-style relative import: a bare name with level > 0, whose + # anchor lives in globals["__package__"] (e.g. `from .type_adapter + # import X` inside pydantic). It resolves to the anchor's top-level + # package and is allowed when that package is allowlisted. + guard = self._guard({"pydantic"}) + assert ( + guard("type_adapter", {"__package__": "pydantic"}, None, ("X",), 1) + == "imported" + ) + + def test_safe_import_bare_relative_package_import_is_allowed(self): + # `from . import x` compiles to __import__("", globals, locals, ("x",), 1) + # and resolves to the anchor package itself. + guard = self._guard({"pydantic"}) + assert guard("", {"__package__": "pydantic"}, None, ("x",), 1) == "imported" + + def test_safe_import_level_relative_with_non_allowlisted_anchor_is_rejected(self): + guard = self._guard({"pydantic"}) + with pytest.raises(SecurityViolationError): + guard("sub", {"__package__": "requests"}, None, ("X",), 1) + + def test_safe_import_level_relative_without_package_in_globals_is_rejected(self): + # No anchor available, so the bare name is checked as-is and rejected. + guard = self._guard({"pydantic"}) + with pytest.raises(SecurityViolationError): + guard("type_adapter", {}, None, ("X",), 1) + + def test_safe_import_absolute_import_is_not_anchored_by_globals(self): + # An absolute import is checked on its own name; globals["__package__"] + # is not consulted, so a non-allowlisted module stays blocked. + guard = self._guard({"pydantic"}) + with pytest.raises(SecurityViolationError): + guard("requests", {"__package__": "pydantic"}, None, None, 0) + + def test_safe_import_passes_original_name_to_the_real_import(self): + # The validator checks a reconstructed dotted name, but the underlying + # import must receive the original bare name and args unchanged. + config = SecurityConfig( + stdlib_allow=set(), + external_allow={"pydantic"}, + builtins_deny=set(), + runner_env_deny=True, + ) + received = {} + + def record(name, *args, **kwargs): + received["name"] = name + received["args"] = args + return "imported" + + guard = _GuardedImport(config, validate_module_import, record) + guard("type_adapter", {"__package__": "pydantic"}, None, ("X",), 1) + assert received["name"] == "type_adapter" + assert received["args"] == ({"__package__": "pydantic"}, None, ("X",), 1) + class TestIntrospectionDeniedOnHardenedCallables: """Regression tests covering the indirect-attribute-access surface.