fix: use jinja sandbox (#39609)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/core/helper/code_executor/jinja2/jinja2_transformer.py

@@ -39,15 +39,16 @@ class Jinja2TemplateTransformer(TemplateTransformer):
     @override
     def get_runner_script(cls) -> str:
         runner_script = dedent(f"""
-            import jinja2
             import json
             from base64 import b64decode
+            from jinja2.sandbox import SandboxedEnvironment
 
             # declare main function
             def main(**inputs):
                 # Decode base64-encoded template to handle special characters safely
                 template_code = b64decode('{cls._template_b64_placeholder}').decode('utf-8')
-                template = jinja2.Template(template_code)
+                env = SandboxedEnvironment()
+                template = env.from_string(template_code)
                 return template.render(**inputs)
 
             # decode and prepare input dict
@@ -67,12 +68,13 @@ class Jinja2TemplateTransformer(TemplateTransformer):
     @override
     def get_preload_script(cls) -> str:
         preload_script = dedent("""
-            import jinja2
+            from jinja2.sandbox import SandboxedEnvironment
             from base64 import b64decode
 
             def _jinja2_preload_():
-                # prepare jinja2 environment, load template and render before to avoid sandbox issue
-                template = jinja2.Template('{{s}}')
+                # prepare jinja2 sandboxed environment, load template and render
+                env = SandboxedEnvironment()
+                template = env.from_string('{{s}}')
                 template.render(s='a')
 
             if __name__ == '__main__':

api/tests/integration_tests/workflow/nodes/test_template_transform.py

@@ -17,10 +17,11 @@ class _SimpleJinja2Renderer:
     """Minimal Jinja2-based renderer for integration tests (no code executor)."""
 
     def render_template(self, template: str, variables: dict[str, object]) -> str:
-        from jinja2 import Template
+        from jinja2.sandbox import SandboxedEnvironment
 
         try:
-            return Template(template).render(**variables)
+            env = SandboxedEnvironment()
+            return env.from_string(template).render(**variables)
         except Exception as exc:
             raise TemplateRenderError(str(exc)) from exc
 

api/tests/unit_tests/core/helper/code_executor/jinja2/test_jinja2_sandbox.py

@@ -0,0 +1,67 @@
+"""Verify that Jinja2 transform_caller scripts block unsafe template attacks via SandboxedEnvironment."""
+
+import io
+import sys
+
+import pytest
+
+from core.helper.code_executor.jinja2.jinja2_transformer import Jinja2TemplateTransformer
+
+MALICIOUS_TEMPLATES = [
+    pytest.param(
+        "{{ lipsum.__globals__.__builtins__.__import__('os').popen('id').read() }}",
+        id="lipsum_globals_builtins",
+    ),
+    pytest.param(
+        "{{ ''.__class__.__mro__[1].__subclasses__() }}",
+        id="string_class_mro",
+    ),
+    pytest.param(
+        "{{ cycler.__init__.__globals__.os.popen('whoami').read() }}",
+        id="cycler_init_globals",
+    ),
+    pytest.param(
+        "{{ namespace.__init__.__globals__['__builtins__']['__import__']('os').system('id') }}",
+        id="namespace_init_globals",
+    ),
+]
+
+
+def _exec_scripts(runner: str, preload: str) -> str:
+    """Execute preload then runner in a shared namespace, return captured stdout."""
+    ns: dict = {}
+    exec(compile(preload, "<preload>", "exec"), ns)  # noqa: S102
+    captured = io.StringIO()
+    old_stdout = sys.stdout
+    sys.stdout = captured
+    try:
+        exec(compile(runner, "<runner>", "exec"), ns)  # noqa: S102
+    finally:
+        sys.stdout = old_stdout
+    return captured.getvalue()
+
+
+class TestJinja2TransformCallerSandbox:
+    """Test transform_caller output (runner + preload) blocks attacks and allows safe templates."""
+
+    @pytest.mark.parametrize("malicious_template", MALICIOUS_TEMPLATES)
+    def test_blocks_unsafe_template(self, malicious_template: str) -> None:
+        runner, preload = Jinja2TemplateTransformer.transform_caller(malicious_template, {})
+        ns: dict = {}
+        exec(compile(preload, "<preload>", "exec"), ns)  # noqa: S102
+        with pytest.raises(Exception) as exc_info:
+            exec(compile(runner, "<runner>", "exec"), ns)  # noqa: S102
+        assert "unsafe" in str(exc_info.value).lower() or "security" in str(exc_info.value).lower()
+
+    def test_renders_safe_template(self) -> None:
+        runner, preload = Jinja2TemplateTransformer.transform_caller(
+            "Hello {{ name }}, you are {{ age }} years old!",
+            {"name": "Alice", "age": 30},
+        )
+        output = _exec_scripts(runner, preload)
+        assert "Hello Alice, you are 30 years old!" in output
+
+    def test_scripts_use_sandboxed_environment(self) -> None:
+        runner, preload = Jinja2TemplateTransformer.transform_caller("{{ x }}", {"x": 1})
+        assert "SandboxedEnvironment" in runner
+        assert "SandboxedEnvironment" in preload