From 181ed3771de09a559ea9c5b5055df2acfc23aeb3 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Mon, 1 Jun 2026 14:17:28 -0700 Subject: [PATCH] fix: restrict builtins and validate code in Python Interpreter components (#13397) * fix: restrict builtins and validate code in Python Interpreter components PythonREPLComponent.get_globals() built the exec globals from the global_imports allow-list but never set __builtins__. CPython's exec() then auto-injects the full builtins module, so __import__, open, eval, exec and the whole import machinery stayed reachable regardless of the allow-list -- e.g. __import__("subprocess").check_output([...]) -- making the "Only modules specified in Global Imports can be used" claim false. The same get_globals() pattern existed in the legacy PythonREPLToolComponent. Add a shared lfx.utils.python_repl_security module: - safe_builtins(): a curated __builtins__ allow-list that keeps common safe builtins (print, len, range, container/number types, exceptions, ...) and removes __import__/eval/exec/compile/open/input/globals/locals/vars/getattr/setattr/... - validate_code_safety(): an AST check rejecting inline imports, dunder attribute access (the ().__class__.__subclasses__() escape), frame/traceback introspection, mro(), and str.format("{0.__globals__}") template traversal. Both components now set __builtins__ via safe_builtins() and validate the sanitized code (PythonREPL.sanitize_input) before exec. The tool component builds a fresh globals namespace per invocation so state cannot leak across tool calls. The core component's default Global Imports is reduced from "math,pandas" to "math" so the default no longer bundles a deserialization sink (pandas.read_pickle); pandas can be added back explicitly. This is defense-in-depth hardening, not a guaranteed sandbox -- Python sandboxing is hard; the primary control for untrusted access remains authentication. Adds unit tests for safe_builtins/validate_code_safety and component regression tests (import bypass, subclasses/frame/format escapes and inline imports blocked; legitimate code and whitelisted modules still work; per-invocation isolation; default excludes pandas). Co-Authored-By: Claude Opus 4.8 (1M context) * [autofix.ci] apply automated fixes --------- Co-authored-by: Claude Opus 4.8 (1M context) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- .../components/tools/test_python_repl_tool.py | 98 +++++++++ src/lfx/src/lfx/_assets/component_index.json | 12 +- .../src/lfx/components/tools/python_repl.py | 17 +- .../components/utilities/python_repl_core.py | 17 +- src/lfx/src/lfx/utils/python_repl_security.py | 194 ++++++++++++++++++ .../unit/utils/test_python_repl_security.py | 112 ++++++++++ 6 files changed, 438 insertions(+), 12 deletions(-) create mode 100644 src/lfx/src/lfx/utils/python_repl_security.py create mode 100644 src/lfx/tests/unit/utils/test_python_repl_security.py diff --git a/src/backend/tests/unit/components/tools/test_python_repl_tool.py b/src/backend/tests/unit/components/tools/test_python_repl_tool.py index 2243a4aca5..7dc931a57a 100644 --- a/src/backend/tests/unit/components/tools/test_python_repl_tool.py +++ b/src/backend/tests/unit/components/tools/test_python_repl_tool.py @@ -1,4 +1,6 @@ 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 @@ -52,3 +54,99 @@ class TestPythonREPLComponent(ComponentTestBaseWithoutClient): # 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_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" + + +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_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 diff --git a/src/lfx/src/lfx/_assets/component_index.json b/src/lfx/src/lfx/_assets/component_index.json index c80ac79403..a27cd7167f 100644 --- a/src/lfx/src/lfx/_assets/component_index.json +++ b/src/lfx/src/lfx/_assets/component_index.json @@ -110336,7 +110336,7 @@ "icon": "Python", "legacy": true, "metadata": { - "code_hash": "e5c7a6652f73", + "code_hash": "b15f1d388e03", "dependencies": { "dependencies": [ { @@ -110414,7 +110414,7 @@ "show": true, "title_case": false, "type": "code", - "value": "import importlib\n\nfrom langchain_core.tools import StructuredTool, ToolException\nfrom langchain_experimental.utilities import PythonREPL\nfrom pydantic import BaseModel, Field\n\nfrom lfx.base.langchain_utilities.model import LCToolComponent\nfrom lfx.field_typing import Tool\nfrom lfx.inputs.inputs import StrInput\nfrom lfx.log.logger import logger\nfrom lfx.schema.data import Data\n\n\nclass PythonREPLToolComponent(LCToolComponent):\n display_name = \"Python REPL\"\n description = \"A tool for running Python code in a REPL environment.\"\n name = \"PythonREPLTool\"\n icon = \"Python\"\n legacy = True\n replacement = [\"processing.PythonREPLComponent\"]\n\n inputs = [\n StrInput(\n name=\"name\",\n display_name=\"Tool Name\",\n info=\"The name of the tool.\",\n value=\"python_repl\",\n ),\n StrInput(\n name=\"description\",\n display_name=\"Tool Description\",\n info=\"A description of the tool.\",\n value=\"A Python shell. Use this to execute python commands. \"\n \"Input should be a valid python command. \"\n \"If you want to see the output of a value, you should print it out with `print(...)`.\",\n ),\n StrInput(\n name=\"global_imports\",\n display_name=\"Global Imports\",\n info=\"A comma-separated list of modules to import globally, e.g. 'math,numpy'.\",\n value=\"math\",\n ),\n StrInput(\n name=\"code\",\n display_name=\"Python Code\",\n info=\"The Python code to execute.\",\n value=\"print('Hello, World!')\",\n ),\n ]\n\n class PythonREPLSchema(BaseModel):\n code: str = Field(..., description=\"The Python code to execute.\")\n\n def get_globals(self, global_imports: str | list[str]) -> dict:\n global_dict = {}\n if isinstance(global_imports, str):\n modules = [module.strip() for module in global_imports.split(\",\")]\n elif isinstance(global_imports, list):\n modules = global_imports\n else:\n msg = \"global_imports must be either a string or a list\"\n raise TypeError(msg)\n\n for module in modules:\n try:\n imported_module = importlib.import_module(module)\n global_dict[imported_module.__name__] = imported_module\n except ImportError as e:\n msg = f\"Could not import module {module}\"\n raise ImportError(msg) from e\n return global_dict\n\n def build_tool(self) -> Tool:\n globals_ = self.get_globals(self.global_imports)\n python_repl = PythonREPL(_globals=globals_)\n\n def run_python_code(code: str) -> str:\n try:\n return python_repl.run(code)\n except Exception as e:\n logger.debug(\"Error running Python code\", exc_info=True)\n raise ToolException(str(e)) from e\n\n tool = StructuredTool.from_function(\n name=self.name,\n description=self.description,\n func=run_python_code,\n args_schema=self.PythonREPLSchema,\n )\n\n self.status = f\"Python REPL Tool created with global imports: {self.global_imports}\"\n return tool\n\n def run_model(self) -> list[Data]:\n tool = self.build_tool()\n result = tool.run(self.code)\n return [Data(data={\"result\": result})]\n" + "value": "import importlib\n\nfrom langchain_core.tools import StructuredTool, ToolException\nfrom langchain_experimental.utilities import PythonREPL\nfrom pydantic import BaseModel, Field\n\nfrom lfx.base.langchain_utilities.model import LCToolComponent\nfrom lfx.field_typing import Tool\nfrom lfx.inputs.inputs import StrInput\nfrom lfx.log.logger import logger\nfrom lfx.schema.data import Data\nfrom lfx.utils.python_repl_security import safe_builtins, validate_code_safety\n\n\nclass PythonREPLToolComponent(LCToolComponent):\n display_name = \"Python REPL\"\n description = \"A tool for running Python code in a REPL environment.\"\n name = \"PythonREPLTool\"\n icon = \"Python\"\n legacy = True\n replacement = [\"processing.PythonREPLComponent\"]\n\n inputs = [\n StrInput(\n name=\"name\",\n display_name=\"Tool Name\",\n info=\"The name of the tool.\",\n value=\"python_repl\",\n ),\n StrInput(\n name=\"description\",\n display_name=\"Tool Description\",\n info=\"A description of the tool.\",\n value=\"A Python shell. Use this to execute python commands. \"\n \"Input should be a valid python command. \"\n \"If you want to see the output of a value, you should print it out with `print(...)`.\",\n ),\n StrInput(\n name=\"global_imports\",\n display_name=\"Global Imports\",\n info=\"A comma-separated list of modules to import globally, e.g. 'math,numpy'.\",\n value=\"math\",\n ),\n StrInput(\n name=\"code\",\n display_name=\"Python Code\",\n info=\"The Python code to execute.\",\n value=\"print('Hello, World!')\",\n ),\n ]\n\n class PythonREPLSchema(BaseModel):\n code: str = Field(..., description=\"The Python code to execute.\")\n\n def get_globals(self, global_imports: str | list[str]) -> dict:\n global_dict = {}\n if isinstance(global_imports, str):\n modules = [module.strip() for module in global_imports.split(\",\")]\n elif isinstance(global_imports, list):\n modules = global_imports\n else:\n msg = \"global_imports must be either a string or a list\"\n raise TypeError(msg)\n\n for module in modules:\n try:\n imported_module = importlib.import_module(module)\n global_dict[imported_module.__name__] = imported_module\n except ImportError as e:\n msg = f\"Could not import module {module}\"\n raise ImportError(msg) from e\n # Restrict builtins so the import allow-list cannot be silently bypassed\n # (e.g. __import__(\"subprocess\")). Without this, exec() auto-injects the full\n # builtins module, leaving __import__/open/eval/exec reachable.\n global_dict[\"__builtins__\"] = safe_builtins()\n return global_dict\n\n def build_tool(self) -> Tool:\n def run_python_code(code: str) -> str:\n try:\n # Validate the exact (sanitized) code that will run, rejecting inline\n # imports and escape gadgets; combined with the restricted builtins in\n # get_globals(). A fresh globals namespace is built per invocation so\n # state does not leak across tool calls.\n cleaned_code = PythonREPL.sanitize_input(code)\n validate_code_safety(cleaned_code)\n python_repl = PythonREPL(_globals=self.get_globals(self.global_imports))\n return python_repl.run(cleaned_code)\n except Exception as e:\n logger.debug(\"Error running Python code\", exc_info=True)\n raise ToolException(str(e)) from e\n\n tool = StructuredTool.from_function(\n name=self.name,\n description=self.description,\n func=run_python_code,\n args_schema=self.PythonREPLSchema,\n )\n\n self.status = f\"Python REPL Tool created with global imports: {self.global_imports}\"\n return tool\n\n def run_model(self) -> list[Data]:\n tool = self.build_tool()\n result = tool.run(self.code)\n return [Data(data={\"result\": result})]\n" }, "description": { "_input_type": "StrInput", @@ -114297,7 +114297,7 @@ "icon": "square-terminal", "legacy": false, "metadata": { - "code_hash": "80eeaf032b83", + "code_hash": "b92cf9819227", "dependencies": { "dependencies": [ { @@ -114350,7 +114350,7 @@ "show": true, "title_case": false, "type": "code", - "value": "import importlib\n\nfrom langchain_experimental.utilities import PythonREPL\n\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.io import MultilineInput, Output, StrInput\nfrom lfx.schema.data import Data\n\n\nclass PythonREPLComponent(Component):\n display_name = \"Python Interpreter\"\n description = \"Run Python code with optional imports. Use print() to see the output.\"\n documentation: str = \"https://docs.langflow.org/python-interpreter\"\n icon = \"square-terminal\"\n\n inputs = [\n StrInput(\n name=\"global_imports\",\n display_name=\"Global Imports\",\n info=\"A comma-separated list of modules to import globally, e.g. 'math,numpy,pandas'.\",\n value=\"math,pandas\",\n required=True,\n ),\n MultilineInput(\n name=\"python_code\",\n display_name=\"Python Code\",\n info=\"The Python code to execute. Only modules specified in Global Imports can be used.\",\n value=\"print('Hello, World!')\",\n input_types=[\"Message\"],\n tool_mode=True,\n required=True,\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Results\",\n name=\"results\",\n type_=Data,\n method=\"run_python_repl\",\n ),\n ]\n\n def get_globals(self, global_imports: str | list[str]) -> dict:\n \"\"\"Create a globals dictionary with only the specified allowed imports.\"\"\"\n global_dict = {}\n\n try:\n if isinstance(global_imports, str):\n modules = [module.strip() for module in global_imports.split(\",\")]\n elif isinstance(global_imports, list):\n modules = global_imports\n else:\n msg = \"global_imports must be either a string or a list\"\n raise TypeError(msg)\n\n for module in modules:\n try:\n imported_module = importlib.import_module(module)\n global_dict[imported_module.__name__] = imported_module\n except ImportError as e:\n msg = f\"Could not import module {module}: {e!s}\"\n raise ImportError(msg) from e\n\n except Exception as e:\n self.log(f\"Error in global imports: {e!s}\")\n raise\n else:\n self.log(f\"Successfully imported modules: {list(global_dict.keys())}\")\n return global_dict\n\n def run_python_repl(self) -> Data:\n try:\n globals_ = self.get_globals(self.global_imports)\n python_repl = PythonREPL(_globals=globals_)\n result = python_repl.run(self.python_code)\n result = result.strip() if result else \"\"\n\n self.log(\"Code execution completed successfully\")\n return Data(data={\"result\": result})\n\n except ImportError as e:\n error_message = f\"Import Error: {e!s}\"\n self.log(error_message)\n return Data(data={\"error\": error_message})\n\n except SyntaxError as e:\n error_message = f\"Syntax Error: {e!s}\"\n self.log(error_message)\n return Data(data={\"error\": error_message})\n\n except (NameError, TypeError, ValueError) as e:\n error_message = f\"Error during execution: {e!s}\"\n self.log(error_message)\n return Data(data={\"error\": error_message})\n\n def build(self):\n return self.run_python_repl\n" + "value": "import importlib\n\nfrom langchain_experimental.utilities import PythonREPL\n\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.io import MultilineInput, Output, StrInput\nfrom lfx.schema.data import Data\nfrom lfx.utils.python_repl_security import safe_builtins, validate_code_safety\n\n\nclass PythonREPLComponent(Component):\n display_name = \"Python Interpreter\"\n description = \"Run Python code with optional imports. Use print() to see the output.\"\n documentation: str = \"https://docs.langflow.org/python-interpreter\"\n icon = \"square-terminal\"\n\n inputs = [\n StrInput(\n name=\"global_imports\",\n display_name=\"Global Imports\",\n info=\"A comma-separated list of modules to import globally, e.g. 'math,numpy,pandas'.\",\n # Default kept minimal: powerful modules (e.g. pandas, whose read_pickle/eval\n # are code-execution sinks) must be opted into explicitly via this field.\n value=\"math\",\n required=True,\n ),\n MultilineInput(\n name=\"python_code\",\n display_name=\"Python Code\",\n info=\"The Python code to execute. Only modules specified in Global Imports can be used.\",\n value=\"print('Hello, World!')\",\n input_types=[\"Message\"],\n tool_mode=True,\n required=True,\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Results\",\n name=\"results\",\n type_=Data,\n method=\"run_python_repl\",\n ),\n ]\n\n def get_globals(self, global_imports: str | list[str]) -> dict:\n \"\"\"Create a globals dictionary with only the specified allowed imports.\"\"\"\n global_dict = {}\n\n try:\n if isinstance(global_imports, str):\n modules = [module.strip() for module in global_imports.split(\",\")]\n elif isinstance(global_imports, list):\n modules = global_imports\n else:\n msg = \"global_imports must be either a string or a list\"\n raise TypeError(msg)\n\n for module in modules:\n try:\n imported_module = importlib.import_module(module)\n global_dict[imported_module.__name__] = imported_module\n except ImportError as e:\n msg = f\"Could not import module {module}: {e!s}\"\n raise ImportError(msg) from e\n\n except Exception as e:\n self.log(f\"Error in global imports: {e!s}\")\n raise\n else:\n self.log(f\"Successfully imported modules: {list(global_dict.keys())}\")\n # Restrict builtins so the import allow-list cannot be silently bypassed\n # (e.g. __import__(\"subprocess\")). Without this, exec() auto-injects the full\n # builtins module, leaving __import__/open/eval/exec reachable.\n global_dict[\"__builtins__\"] = safe_builtins()\n return global_dict\n\n def run_python_repl(self) -> Data:\n try:\n # Validate the exact code that will run: PythonREPL.run() strips a leading\n # \"python\"/backticks/whitespace prefix before exec, so validate the sanitized\n # form. Rejects inline imports and escape gadgets (e.g.\n # ().__class__.__subclasses__()); combined with restricted builtins in get_globals().\n code = PythonREPL.sanitize_input(self.python_code)\n validate_code_safety(code)\n globals_ = self.get_globals(self.global_imports)\n python_repl = PythonREPL(_globals=globals_)\n result = python_repl.run(code)\n result = result.strip() if result else \"\"\n\n self.log(\"Code execution completed successfully\")\n return Data(data={\"result\": result})\n\n except ImportError as e:\n error_message = f\"Import Error: {e!s}\"\n self.log(error_message)\n return Data(data={\"error\": error_message})\n\n except SyntaxError as e:\n error_message = f\"Syntax Error: {e!s}\"\n self.log(error_message)\n return Data(data={\"error\": error_message})\n\n except (NameError, TypeError, ValueError) as e:\n error_message = f\"Error during execution: {e!s}\"\n self.log(error_message)\n return Data(data={\"error\": error_message})\n\n def build(self):\n return self.run_python_repl\n" }, "global_imports": { "_input_type": "StrInput", @@ -114371,7 +114371,7 @@ "trace_as_metadata": true, "track_in_telemetry": false, "type": "str", - "value": "math,pandas" + "value": "math" }, "python_code": { "_input_type": "MultilineInput", @@ -120140,6 +120140,6 @@ "num_components": 360, "num_modules": 96 }, - "sha256": "34f5cff8250d410004c6f72b09bd84d6e221bd62a5ea890d7867bebe4b1a2993", + "sha256": "cc108c54bfd011c9a83d50a2424121f378db73710435686ea1e041879cf7b711", "version": "0.5.0" } diff --git a/src/lfx/src/lfx/components/tools/python_repl.py b/src/lfx/src/lfx/components/tools/python_repl.py index 23fc771618..99d5c13d13 100644 --- a/src/lfx/src/lfx/components/tools/python_repl.py +++ b/src/lfx/src/lfx/components/tools/python_repl.py @@ -9,6 +9,7 @@ from lfx.field_typing import Tool from lfx.inputs.inputs import StrInput from lfx.log.logger import logger from lfx.schema.data import Data +from lfx.utils.python_repl_security import safe_builtins, validate_code_safety class PythonREPLToolComponent(LCToolComponent): @@ -68,15 +69,23 @@ class PythonREPLToolComponent(LCToolComponent): except ImportError as e: msg = f"Could not import module {module}" raise ImportError(msg) from e + # Restrict builtins so the import allow-list cannot be silently bypassed + # (e.g. __import__("subprocess")). Without this, exec() auto-injects the full + # builtins module, leaving __import__/open/eval/exec reachable. + global_dict["__builtins__"] = safe_builtins() return global_dict def build_tool(self) -> Tool: - globals_ = self.get_globals(self.global_imports) - python_repl = PythonREPL(_globals=globals_) - def run_python_code(code: str) -> str: try: - return python_repl.run(code) + # Validate the exact (sanitized) code that will run, rejecting inline + # imports and escape gadgets; combined with the restricted builtins in + # get_globals(). A fresh globals namespace is built per invocation so + # state does not leak across tool calls. + cleaned_code = PythonREPL.sanitize_input(code) + validate_code_safety(cleaned_code) + python_repl = PythonREPL(_globals=self.get_globals(self.global_imports)) + return python_repl.run(cleaned_code) except Exception as e: logger.debug("Error running Python code", exc_info=True) raise ToolException(str(e)) from e diff --git a/src/lfx/src/lfx/components/utilities/python_repl_core.py b/src/lfx/src/lfx/components/utilities/python_repl_core.py index 87b506e9a5..3ac8b7747c 100644 --- a/src/lfx/src/lfx/components/utilities/python_repl_core.py +++ b/src/lfx/src/lfx/components/utilities/python_repl_core.py @@ -5,6 +5,7 @@ from langchain_experimental.utilities import PythonREPL from lfx.custom.custom_component.component import Component from lfx.io import MultilineInput, Output, StrInput from lfx.schema.data import Data +from lfx.utils.python_repl_security import safe_builtins, validate_code_safety class PythonREPLComponent(Component): @@ -18,7 +19,9 @@ class PythonREPLComponent(Component): name="global_imports", display_name="Global Imports", info="A comma-separated list of modules to import globally, e.g. 'math,numpy,pandas'.", - value="math,pandas", + # Default kept minimal: powerful modules (e.g. pandas, whose read_pickle/eval + # are code-execution sinks) must be opted into explicitly via this field. + value="math", required=True, ), MultilineInput( @@ -67,13 +70,23 @@ class PythonREPLComponent(Component): raise else: self.log(f"Successfully imported modules: {list(global_dict.keys())}") + # Restrict builtins so the import allow-list cannot be silently bypassed + # (e.g. __import__("subprocess")). Without this, exec() auto-injects the full + # builtins module, leaving __import__/open/eval/exec reachable. + global_dict["__builtins__"] = safe_builtins() return global_dict def run_python_repl(self) -> Data: try: + # Validate the exact code that will run: PythonREPL.run() strips a leading + # "python"/backticks/whitespace prefix before exec, so validate the sanitized + # form. Rejects inline imports and escape gadgets (e.g. + # ().__class__.__subclasses__()); combined with restricted builtins in get_globals(). + code = PythonREPL.sanitize_input(self.python_code) + validate_code_safety(code) globals_ = self.get_globals(self.global_imports) python_repl = PythonREPL(_globals=globals_) - result = python_repl.run(self.python_code) + result = python_repl.run(code) result = result.strip() if result else "" self.log("Code execution completed successfully") diff --git a/src/lfx/src/lfx/utils/python_repl_security.py b/src/lfx/src/lfx/utils/python_repl_security.py new file mode 100644 index 0000000000..46a26cab22 --- /dev/null +++ b/src/lfx/src/lfx/utils/python_repl_security.py @@ -0,0 +1,194 @@ +"""Best-effort hardening for the Python Interpreter / Python REPL components. + +These components execute user-supplied code via ``exec`` (through +``langchain_experimental``'s ``PythonREPL``). Historically the globals dict passed to +``exec`` contained only the allow-listed modules and never set ``__builtins__``; +CPython then auto-injects the *full* builtins module, so ``__import__``, ``open``, +``eval``, ``exec`` and the entire import machinery stayed reachable regardless of the +"Global Imports" allow-list. The allow-list looked like a sandbox but was silently +bypassable (e.g. ``__import__("subprocess").check_output(["id"])``). + +This module restricts that environment: + +* :func:`safe_builtins` returns a curated ``__builtins__`` mapping that keeps the common + safe builtins (``print``, ``len``, ``range``, container/number types, exceptions, ...) + but removes anything that can import modules, execute/compile code, read the + filesystem, or reach interpreter internals (``__import__``, ``eval``, ``exec``, + ``compile``, ``open``, ``input``, ``globals``/``locals``/``vars``, + ``getattr``/``setattr``/``delattr``, ``breakpoint``, ...). +* :func:`validate_code_safety` rejects code that bypasses the allow-list with an inline + ``import`` or reaches the classic builtin-free escape gadgets: dunder attribute access + (``().__class__.__subclasses__()``) and frame/traceback introspection + (``gen.gi_frame.f_back.f_globals``). + +This is defense-in-depth, NOT a guaranteed sandbox — Python sandboxing is notoriously +hard and determined attackers may still find gadgets. The primary control for untrusted +access is authentication: do not expose an instance with ``LANGFLOW_AUTO_LOGIN=true`` on +an untrusted network. +""" + +from __future__ import annotations + +import ast +import builtins +import re + +# Builtins considered safe to expose to interpreter code. Deliberately excludes anything +# that can import modules, execute/compile code, touch the filesystem, or reach +# interpreter internals. Names absent here resolve to NameError inside the interpreter, +# which is what makes the "Global Imports" allow-list the only way to bring in modules. +_SAFE_BUILTIN_NAMES = frozenset( + { + "abs", + "aiter", + "anext", + "all", + "any", + "ascii", + "bin", + "bool", + "bytearray", + "bytes", + "callable", + "chr", + "complex", + "dict", + "divmod", + "enumerate", + "filter", + "float", + "format", + "frozenset", + "hasattr", + "hash", + "hex", + "int", + "isinstance", + "issubclass", + "iter", + "len", + "list", + "map", + "max", + "min", + "next", + "oct", + "ord", + "pow", + "print", + "range", + "repr", + "reversed", + "round", + "set", + "slice", + "sorted", + "str", + "sum", + "tuple", + "type", + "zip", + # Exception hierarchy so try/except and explicit raises behave normally. + "ArithmeticError", + "AssertionError", + "AttributeError", + "BaseException", + "Exception", + "FloatingPointError", + "IndexError", + "KeyError", + "KeyboardInterrupt", + "LookupError", + "NameError", + "NotImplementedError", + "OverflowError", + "RecursionError", + "RuntimeError", + "StopIteration", + "TypeError", + "UnicodeDecodeError", + "UnicodeEncodeError", + "ValueError", + "ZeroDivisionError", + } +) + +# Attribute names that expose interpreter internals or sandbox-escape gadgets even +# without any dangerous builtin. Dunder attributes (``__class__``, ``__subclasses__``, +# ``__globals__``, ``__mro__``, ``__builtins__``, ...) are handled generically; this set +# covers the non-dunder frame / coroutine / traceback introspection attributes. +_BLOCKED_ATTRIBUTES = frozenset( + { + "gi_frame", + "gi_code", + "cr_frame", + "cr_code", + "ag_frame", + "ag_code", + "f_globals", + "f_locals", + "f_builtins", + "f_back", + "f_code", + "f_trace", + "tb_frame", + "tb_next", + "func_globals", + "func_code", + "__dict__", + "mro", # int.mro() reaches the object hierarchy without a dunder attribute + } +) + +# Matches a str.format()/format_map() replacement field that reaches into a dunder +# attribute or item (e.g. "{0.__globals__}", "{0[__builtins__]}"). Such traversals live +# inside the template string and are invisible to the AST attribute check below. +_FORMAT_FIELD_DUNDER_RE = re.compile(r"\{[^{}]*__") + + +def safe_builtins() -> dict: + """Return a fresh curated ``__builtins__`` mapping for interpreter globals. + + A new dict is returned on each call so callers cannot mutate shared state. + """ + return {name: getattr(builtins, name) for name in _SAFE_BUILTIN_NAMES if hasattr(builtins, name)} + + +def _is_blocked_attribute(attr: str) -> bool: + """Return True if attribute access to ``attr`` should be rejected.""" + # Dunder attributes are the classic escape gadgets; the explicit set covers + # frame/traceback introspection attributes that are not dunders. + return attr.startswith("__") or attr in _BLOCKED_ATTRIBUTES + + +def validate_code_safety(code: str) -> None: + """Reject code that bypasses the import allow-list or reaches escape gadgets. + + This complements :func:`safe_builtins`: restricted builtins stop ``__import__`` / + ``open`` / ``eval`` etc., while this AST check stops the builtin-free escapes + (dunder attribute traversal and frame introspection) and inline imports. + + Args: + code: The Python source to be executed. + + Raises: + ValueError: If the code performs an inline import or accesses a blocked attribute. + SyntaxError: If the code cannot be parsed (surfaced to the caller as-is). + """ + tree = ast.parse(code) + for node in ast.walk(tree): + if isinstance(node, (ast.Import, ast.ImportFrom)): + msg = "Imports are not allowed in the code; declare modules in the Global Imports field instead." + raise ValueError(msg) # noqa: TRY004 -- forbidden construct, not an argument type error + if isinstance(node, ast.Attribute) and _is_blocked_attribute(node.attr): + msg = f"Access to attribute '{node.attr}' is not allowed." + raise ValueError(msg) + if ( + isinstance(node, ast.Constant) + and isinstance(node.value, str) + and _FORMAT_FIELD_DUNDER_RE.search(node.value) + ): + # Blocks str.format("{0.__globals__}") style traversal that the AST attribute + # check cannot see because the attribute chain is inside a literal string. + msg = "Access to dunder attributes via format strings is not allowed." + raise ValueError(msg) diff --git a/src/lfx/tests/unit/utils/test_python_repl_security.py b/src/lfx/tests/unit/utils/test_python_repl_security.py new file mode 100644 index 0000000000..9a60724380 --- /dev/null +++ b/src/lfx/tests/unit/utils/test_python_repl_security.py @@ -0,0 +1,112 @@ +"""Unit tests for the Python REPL hardening helpers.""" + +import pytest +from lfx.utils.python_repl_security import safe_builtins, validate_code_safety + + +class TestSafeBuiltins: + def test_excludes_dangerous_builtins(self): + """Code-execution, import, filesystem and introspection builtins are removed.""" + builtins_map = safe_builtins() + for name in ( + "__import__", + "eval", + "exec", + "compile", + "open", + "input", + "globals", + "locals", + "vars", + "getattr", + "setattr", + "delattr", + "breakpoint", + "memoryview", + "exit", + "quit", + "help", + ): + assert name not in builtins_map, f"{name} must not be exposed" + + def test_includes_common_safe_builtins(self): + """Common, safe builtins remain available so normal code keeps working.""" + builtins_map = safe_builtins() + for name in ( + "print", + "len", + "range", + "int", + "float", + "str", + "list", + "dict", + "tuple", + "set", + "sum", + "min", + "max", + "sorted", + "enumerate", + "zip", + "isinstance", + "type", + "Exception", + "ValueError", + ): + assert name in builtins_map, f"{name} should be available" + + def test_returns_fresh_copy(self): + """Each call returns an independent dict so callers cannot mutate shared state.""" + first = safe_builtins() + first["injected"] = object() + assert "injected" not in safe_builtins() + + +class TestValidateCodeSafety: + @pytest.mark.parametrize( + "code", + [ + "import os", + "import subprocess", + "from os import system", + "().__class__", + "[].__class__.__bases__[0].__subclasses__()", + "(lambda: 0).__globals__", + "().__class__.__mro__", + "(x for x in []).gi_frame", + "object.__subclasses__()", + 'f"{().__class__}"', + "int.mro()", + '"{0.__globals__}".format(f)', + '"{0[__builtins__]}".format(f)', + '"{0.__class__}".format(obj)', + ], + ) + def test_blocks_escape_and_import(self, code): + """Inline imports, dunder/frame escape gadgets, and format-string dunder access are rejected.""" + with pytest.raises(ValueError, match="not allowed"): + validate_code_safety(code) + + @pytest.mark.parametrize( + "code", + [ + "print('Hello, World!')", + "print(len([1, 2, 3]))", + "x = 1\ny = x + 2\nprint(y)", + "math.sqrt(16)", + "result = [i * 2 for i in range(5)]\nprint(sum(result))", + "data = {'a': 1, 'b': 2}\nprint(sorted(data.items()))", + "'{0} and {1}'.format(1, 2)", + "'{name}'.format(name='x')", + "'{0:.2f}'.format(3.14159)", + ], + ) + def test_allows_safe_code(self, code): + """Ordinary code (no imports, no dunder/frame access) is allowed.""" + validate_code_safety(code) # should not raise + + def test_syntax_error_propagates(self): + """Unparseable code surfaces a SyntaxError to the caller.""" + with pytest.raises(SyntaxError): + validate_code_safety("print('unterminated")