pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-24 13:52:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: restrict builtins and validate code in Python Interpreter components (#13397)

Browse Source 
* 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) <noreply@anthropic.com>

* [autofix.ci] apply automated fixes

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Eric Hare
2026-06-01 14:17:28 -07:00
committed by GitHub
parent 75efd0a2a4
commit 181ed3771d
6 changed files with 438 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
src/backend/tests/unit/components/tools/test_python_repl_tool.py
View File

@ -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

12
src/lfx/src/lfx/_assets/component_index.json
View File

@ -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"
}

17
src/lfx/src/lfx/components/tools/python_repl.py
View File

@ -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

17
src/lfx/src/lfx/components/utilities/python_repl_core.py
View File

@ -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")

194
src/lfx/src/lfx/utils/python_repl_security.py Normal file
View File

@ -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)

112
src/lfx/tests/unit/utils/test_python_repl_security.py Normal file
View File

@ -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")