import pytest from langchain_core.tools import ToolException from lfx.components.tools.python_repl import PythonREPLToolComponent from lfx.components.utilities.python_repl_core import PythonREPLComponent from tests.base import DID_NOT_EXIST, ComponentTestBaseWithoutClient DYNAMIC_FORMAT_TRAVERSAL_CODE = """ parts = ["__loader__", "find_spec", "__globals__"] field = "{0." + parts[0] + "." + parts[1] + "." + parts[2] + "[sys].modules[os].environ}" print("canary_present=" + str("PVR0800453_CANARY" in field.format(math))) """ # Sibling formatter sinks: the same __globals__ traversal lives in a *string* argument # (invisible to the AST attribute check) when fed through string.Formatter primitives or # operator.attrgetter. Each leaks os.environ unless the validator rejects the call. FORMATTER_VFORMAT_TRAVERSAL_CODE = """ f = lambda: 0 leaked = string.Formatter().vformat("{0.__globals__[os].environ[PVR0800453_CANARY]}", (f,), {}) print("leaked=" + str(leaked)) """ FORMATTER_GET_FIELD_TRAVERSAL_CODE = """ f = lambda: 0 obj, _ = string.Formatter().get_field("0.__globals__[os].environ[PVR0800453_CANARY]", (f,), {}) print("leaked=" + str(obj)) """ ATTRGETTER_TRAVERSAL_CODE = """ f = lambda: 0 leaked = operator.attrgetter("__globals__")(f)["os"].environ["PVR0800453_CANARY"] print("leaked=" + str(leaked)) """ class TestPythonREPLComponent(ComponentTestBaseWithoutClient): @pytest.fixture def component_class(self): return PythonREPLComponent @pytest.fixture def default_kwargs(self): return { "global_imports": "math", "python_code": "print('Hello, World!')", } @pytest.fixture def file_names_mapping(self): # Component not yet released, mark all versions as non-existent return [ {"version": "1.0.17", "module": "tools", "file_name": DID_NOT_EXIST}, {"version": "1.0.18", "module": "tools", "file_name": DID_NOT_EXIST}, {"version": "1.0.19", "module": "tools", "file_name": DID_NOT_EXIST}, {"version": "1.1.0", "module": "tools", "file_name": DID_NOT_EXIST}, {"version": "1.1.1", "module": "tools", "file_name": DID_NOT_EXIST}, ] def test_component_initialization(self, component_class, default_kwargs): component = component_class(**default_kwargs) frontend_node = component.to_frontend_node() node_data = frontend_node["data"]["node"] # Test template fields template = node_data["template"] assert "global_imports" in template assert "python_code" in template # Test global_imports configuration global_imports = template["global_imports"] assert global_imports["type"] == "str" assert global_imports["value"] == "math" assert global_imports["required"] is True # Test python_code configuration python_code = template["python_code"] # assert python_code["type"] == "code" # TODO: Restore when CodeInput is included in component assert python_code["type"] == "str" # TODO: Remove when CodeInput is included in component assert python_code["value"] == "print('Hello, World!')" assert python_code["required"] is True # Test base configuration - JSON is the new name (Data is alias for backward compatibility) assert "JSON" in node_data["base_classes"] class TestPythonREPLComponentSecurity: """Regression tests for the __builtins__ sandbox-bypass RCE in the Python Interpreter. The Global Imports allow-list must not be silently bypassable: dangerous builtins (__import__, open, eval, ...) and the builtin-free escape gadgets must be blocked, while ordinary code keeps working. """ def _run(self, code, global_imports="math"): return PythonREPLComponent(global_imports=global_imports, python_code=code).run_python_repl().data def test_legitimate_code_still_runs(self): """The default example (and print()) must keep working after restricting builtins.""" assert self._run("print('Hello, World!')").get("result") == "Hello, World!" def test_whitelisted_module_works(self): """A module from Global Imports remains usable.""" assert "4.0" in self._run("print(math.sqrt(16))").get("result", "") def test_import_builtin_is_blocked(self): """__import__ is not in the restricted builtins, so the headline RCE payload fails.""" rendered = str(self._run('print(__import__("subprocess").check_output(["echo", "PWNED"]))')) assert "PWNED" not in rendered assert "NameError" in rendered def test_open_builtin_is_blocked(self): """Filesystem access via open() is blocked.""" rendered = str(self._run('print(open("/etc/passwd").read())')) assert "root:" not in rendered assert "NameError" in rendered def test_eval_builtin_is_blocked(self): """eval() is not exposed.""" assert "NameError" in str(self._run('print(eval("1+1"))')) def test_subclasses_escape_is_blocked(self): """The classic builtin-free escape gadget is rejected before execution.""" data = self._run("print(().__class__.__bases__[0].__subclasses__())") assert "error" in data assert "not allowed" in data["error"] def test_inline_import_is_blocked(self): """Inline imports are rejected; modules must come from the Global Imports field.""" data = self._run("import os\nprint(os.getcwd())") assert "error" in data assert "not allowed" in data["error"] def test_format_string_dunder_escape_is_blocked(self): """str.format() dunder traversal (invisible to the AST attr check) is rejected.""" data = self._run('f = lambda: 0\nprint("{0.__globals__}".format(f))') assert "error" in data assert "not allowed" in data["error"] def test_dynamic_format_string_traversal_is_blocked(self, monkeypatch): """Dynamically-built str.format() fields cannot bypass the AST validator.""" monkeypatch.setenv("PVR0800453_CANARY", "SHOULD_NOT_LEAK") data = self._run(DYNAMIC_FORMAT_TRAVERSAL_CODE) assert "error" in data assert "not allowed" in data["error"] assert "SHOULD_NOT_LEAK" not in str(data) def test_formatter_vformat_traversal_is_blocked(self, monkeypatch): """string.Formatter().vformat carries the dunder chain in a string arg; blocked.""" monkeypatch.setenv("PVR0800453_CANARY", "SHOULD_NOT_LEAK") data = self._run(FORMATTER_VFORMAT_TRAVERSAL_CODE, global_imports="math,string") assert "error" in data assert "not allowed" in data["error"] assert "SHOULD_NOT_LEAK" not in str(data) def test_formatter_get_field_traversal_is_blocked(self, monkeypatch): """string.Formatter().get_field traverses a dotted string path; blocked.""" monkeypatch.setenv("PVR0800453_CANARY", "SHOULD_NOT_LEAK") data = self._run(FORMATTER_GET_FIELD_TRAVERSAL_CODE, global_imports="math,string") assert "error" in data assert "not allowed" in data["error"] assert "SHOULD_NOT_LEAK" not in str(data) def test_attrgetter_traversal_is_blocked(self, monkeypatch): """operator.attrgetter('__globals__') reaches os.environ via a string arg; blocked.""" monkeypatch.setenv("PVR0800453_CANARY", "SHOULD_NOT_LEAK") data = self._run(ATTRGETTER_TRAVERSAL_CODE, global_imports="math,operator") assert "error" in data assert "not allowed" in data["error"] assert "SHOULD_NOT_LEAK" not in str(data) def test_default_global_imports_excludes_pandas(self): """The default allow-list is minimal (no pandas) to avoid bundling a deserialization sink.""" template = PythonREPLComponent().to_frontend_node()["data"]["node"]["template"] assert template["global_imports"]["value"] == "math" def test_execution_refused_when_custom_components_disabled(self, monkeypatch): """GHSA-8qpj-27x8-pwpq: run_python_repl must consult the gate, not just exec. Locks in the wiring: a refactor dropping ensure_code_execution_enabled() from run_python_repl would make this fail, independent of the gate's own unit tests. """ from types import SimpleNamespace monkeypatch.setattr( "lfx.services.deps.get_settings_service", lambda: SimpleNamespace(settings=SimpleNamespace(allow_custom_components=False)), ) data = PythonREPLComponent(global_imports="math", python_code="print('SHOULD_NOT_RUN')").run_python_repl().data assert "error" in data assert "allow_custom_components" in data["error"] # The code must never have executed. assert "SHOULD_NOT_RUN" not in str(data) class TestPythonREPLToolComponentSecurity: """The same hardening applies to the legacy Python REPL *tool* component.""" def _func(self, global_imports="math"): component = PythonREPLToolComponent( name="python_repl", description="run code", global_imports=global_imports, code="print('x')", ) return component.build_tool().func def test_tool_runs_legitimate_code(self): assert "hi" in self._func()("print('hi')") def test_tool_blocks_inline_import(self): with pytest.raises(ToolException, match="not allowed"): self._func()("import os") def test_tool_blocks_subclasses_escape(self): with pytest.raises(ToolException, match="not allowed"): self._func()("().__class__.__bases__[0].__subclasses__()") def test_tool_blocks_dynamic_format_string_traversal(self, monkeypatch): monkeypatch.setenv("PVR0800453_CANARY", "SHOULD_NOT_LEAK") with pytest.raises(ToolException, match="not allowed") as exc_info: self._func()(DYNAMIC_FORMAT_TRAVERSAL_CODE) assert "SHOULD_NOT_LEAK" not in str(exc_info.value) def test_tool_blocks_import_builtin(self): """__import__ is a bare name; PythonREPL surfaces the NameError as a string.""" result = self._func()('__import__("subprocess").check_output(["echo", "PWNED"])') assert "PWNED" not in result assert "NameError" in result def test_tool_invocations_do_not_share_state(self): """Each tool call gets a fresh namespace, so variables do not leak across calls.""" func = self._func() func("leaked = 12345") result = func("print(leaked)") assert "NameError" in result def test_execution_refused_when_custom_components_disabled(self, monkeypatch): """GHSA-8qpj-27x8-pwpq: run_python_code must consult the gate, not just exec. Locks in the wiring: a refactor dropping ensure_code_execution_enabled() from run_python_code would make this fail, independent of the gate's own unit tests. """ from types import SimpleNamespace monkeypatch.setattr( "lfx.services.deps.get_settings_service", lambda: SimpleNamespace(settings=SimpleNamespace(allow_custom_components=False)), ) with pytest.raises(ToolException, match="allow_custom_components"): self._func()("print('SHOULD_NOT_RUN')")