fix(core): Let allowlisted Python packages import their own submodules via relative imports (#32772)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/@n8n/task-runner-python/src/_sandbox_callables.py

@@ -120,6 +120,77 @@ class _SafePrint(_HardenedCallable):
         return "<built-in function print>"
 
 
+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(

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

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

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.