From 2754c84aad1306f463db1c3bbb3e8ffbe85da77d Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Fri, 19 Jun 2026 10:29:17 -0700 Subject: [PATCH] fix(security): gate PythonREPL execution on allow_custom_components (GHSA-8qpj-27x8-pwpq) (#13700) * fix(security): gate PythonREPL execution on allow_custom_components (GHSA-8qpj-27x8-pwpq) The Python Interpreter component (PythonREPLComponent) and the legacy Python REPL tool execute arbitrary user-supplied Python. They are sanitized and blocked on the unauthenticated public path, but an authenticated user could run code even in deployments locked down with allow_custom_components=false. Add ensure_code_execution_enabled(), which refuses execution when allow_custom_components is False (consistent with the custom-component policy), and call it from both components before any sanitize/exec. When no settings service is present (lfx standalone CLI), execution is allowed as a local/trusted context. * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * Update src/lfx/src/lfx/components/utilities/python_repl_core.py Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> * fix: broad exception to import error * [autofix.ci] apply automated fixes * fix(security): fail closed on null settings service in code-exec gate Address review feedback on the PythonREPL exec gate (GHSA-8qpj-27x8-pwpq): - ensure_code_execution_enabled() now fails closed when get_settings_service() returns None (registered-but-failed stack), matching the sibling validate_flow_for_current_settings. get_service swallows init errors into None, so the previous return-allow could silently bypass the gate. - Add per-component tests asserting run_python_repl and run_python_code refuse execution when allow_custom_components=False, locking in the gate wiring. - Add a gate test that a non-ImportError from get_settings_service() propagates rather than failing open; flip the None case to assert fail-closed. - Drop a duplicated comment pair in python_repl_core.py and resync the embedded copy + sha256 in component_index.json. * [autofix.ci] apply automated fixes * Update component_index.json --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- .../components/tools/test_python_repl_tool.py | 33 +++++++++++ src/lfx/src/lfx/_assets/component_index.json | 10 ++-- .../src/lfx/components/tools/python_repl.py | 5 +- .../components/utilities/python_repl_core.py | 5 +- src/lfx/src/lfx/utils/python_repl_security.py | 57 ++++++++++++++++++ .../unit/utils/test_python_repl_security.py | 58 +++++++++++++++++++ 6 files changed, 161 insertions(+), 7 deletions(-) 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 7dc931a57a..746283c812 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 @@ -114,6 +114,24 @@ class TestPythonREPLComponentSecurity: 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.""" @@ -150,3 +168,18 @@ class TestPythonREPLToolComponentSecurity: 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')") diff --git a/src/lfx/src/lfx/_assets/component_index.json b/src/lfx/src/lfx/_assets/component_index.json index 957561e6df..6be1df8b23 100644 --- a/src/lfx/src/lfx/_assets/component_index.json +++ b/src/lfx/src/lfx/_assets/component_index.json @@ -108630,7 +108630,7 @@ "icon": "Python", "legacy": true, "metadata": { - "code_hash": "b15f1d388e03", + "code_hash": "8e9f47f2b23f", "dependencies": { "dependencies": [ { @@ -108708,7 +108708,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\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" + "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 ensure_code_execution_enabled, 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 # Refuse to run user code when allow_custom_components is disabled\n # (GHSA-8qpj-27x8-pwpq).\n ensure_code_execution_enabled()\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", @@ -112587,7 +112587,7 @@ "icon": "square-terminal", "legacy": false, "metadata": { - "code_hash": "b92cf9819227", + "code_hash": "d5130284a8c6", "dependencies": { "dependencies": [ { @@ -112640,7 +112640,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\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" + "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 ensure_code_execution_enabled, 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 # Refuse to run user code when allow_custom_components is disabled\n # (GHSA-8qpj-27x8-pwpq). Raised before any sanitize/exec.\n ensure_code_execution_enabled()\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", @@ -118473,6 +118473,6 @@ "num_components": 354, "num_modules": 95 }, - "sha256": "7554a49846c3c40bd9e724996aa996823ee543b9bbb856377b7354aa14097aec", + "sha256": "b88b98d25b576bcbab858bf13887856a0aeb099bf0330776679ba4a07666ae7b", "version": "1.10.1" } diff --git a/src/lfx/src/lfx/components/tools/python_repl.py b/src/lfx/src/lfx/components/tools/python_repl.py index 99d5c13d13..cdcf825ae8 100644 --- a/src/lfx/src/lfx/components/tools/python_repl.py +++ b/src/lfx/src/lfx/components/tools/python_repl.py @@ -9,7 +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 +from lfx.utils.python_repl_security import ensure_code_execution_enabled, safe_builtins, validate_code_safety class PythonREPLToolComponent(LCToolComponent): @@ -78,6 +78,9 @@ class PythonREPLToolComponent(LCToolComponent): def build_tool(self) -> Tool: def run_python_code(code: str) -> str: try: + # Refuse to run user code when allow_custom_components is disabled + # (GHSA-8qpj-27x8-pwpq). + ensure_code_execution_enabled() # 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 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 3ac8b7747c..e89b8895ce 100644 --- a/src/lfx/src/lfx/components/utilities/python_repl_core.py +++ b/src/lfx/src/lfx/components/utilities/python_repl_core.py @@ -5,7 +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 +from lfx.utils.python_repl_security import ensure_code_execution_enabled, safe_builtins, validate_code_safety class PythonREPLComponent(Component): @@ -78,6 +78,9 @@ class PythonREPLComponent(Component): def run_python_repl(self) -> Data: try: + # Refuse to run user code when allow_custom_components is disabled + # (GHSA-8qpj-27x8-pwpq). Raised before any sanitize/exec. + ensure_code_execution_enabled() # 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. diff --git a/src/lfx/src/lfx/utils/python_repl_security.py b/src/lfx/src/lfx/utils/python_repl_security.py index 46a26cab22..0ce9b52a8c 100644 --- a/src/lfx/src/lfx/utils/python_repl_security.py +++ b/src/lfx/src/lfx/utils/python_repl_security.py @@ -146,6 +146,63 @@ _BLOCKED_ATTRIBUTES = frozenset( _FORMAT_FIELD_DUNDER_RE = re.compile(r"\{[^{}]*__") +class CodeExecutionDisabledError(ValueError): + """Raised when code-execution components are disabled by policy. + + Subclasses ``ValueError`` so existing ``except ValueError``/``except Exception`` + handlers around the interpreter components surface the message gracefully. + """ + + +def ensure_code_execution_enabled() -> None: + """Refuse to run user code when ``allow_custom_components`` is disabled. + + Code-execution components (the Python Interpreter and the legacy Python REPL + tool) run arbitrary user-supplied Python. They honor the same + ``allow_custom_components`` switch as custom components: when an operator + locks a deployment down with ``LANGFLOW_ALLOW_CUSTOM_COMPONENTS=false``, + running arbitrary Python must be refused too — otherwise an authenticated + user can still execute code despite the policy (GHSA-8qpj-27x8-pwpq). + + Failure handling is deliberately asymmetric so the gate can never be + silently bypassed: + + * ``ImportError`` importing the services layer means there is no settings + stack at all (e.g. a stripped-down lfx embedding). That context is local + and trusted, so execution is allowed. + * Any other exception from ``get_settings_service()`` propagates — a stack + that exists but errors must not fail open. + * A ``None`` settings service means the stack is registered but failed to + initialise (``get_service`` swallows init errors into ``None``). Fail + closed, matching ``validate_flow_for_current_settings``, rather than + bypass the very control this gate enforces. + """ + try: + from lfx.services.deps import get_settings_service + + settings_service = get_settings_service() + except ImportError: + # Services layer absent (e.g. stripped-down lfx embed) -> local/trusted, allow. + # Only ImportError is treated as "no settings layer"; any other exception from + # get_settings_service() propagates rather than silently failing open and + # bypassing the allow_custom_components gate (GHSA-8qpj-27x8-pwpq). + return + if settings_service is None: + # Registered-but-failed settings stack (get_service swallows init errors into + # None). Fail closed rather than bypass the gate (GHSA-8qpj-27x8-pwpq). + msg = ( + "Python code execution is disabled because the settings service could not " + "be resolved. Check the server configuration and try again." + ) + raise CodeExecutionDisabledError(msg) + if not getattr(settings_service.settings, "allow_custom_components", True): + msg = ( + "Python code execution is disabled because allow_custom_components is False. " + "Set LANGFLOW_ALLOW_CUSTOM_COMPONENTS=true to enable this component." + ) + raise CodeExecutionDisabledError(msg) + + def safe_builtins() -> dict: """Return a fresh curated ``__builtins__`` mapping for interpreter globals. diff --git a/src/lfx/tests/unit/utils/test_python_repl_security.py b/src/lfx/tests/unit/utils/test_python_repl_security.py index 9a60724380..6306b64e40 100644 --- a/src/lfx/tests/unit/utils/test_python_repl_security.py +++ b/src/lfx/tests/unit/utils/test_python_repl_security.py @@ -110,3 +110,61 @@ class TestValidateCodeSafety: """Unparseable code surfaces a SyntaxError to the caller.""" with pytest.raises(SyntaxError): validate_code_safety("print('unterminated") + + +class TestEnsureCodeExecutionEnabled: + """GHSA-8qpj-27x8-pwpq: code execution honors allow_custom_components.""" + + def test_blocks_when_custom_components_disabled(self, monkeypatch): + from types import SimpleNamespace + + from lfx.utils.python_repl_security import CodeExecutionDisabledError, ensure_code_execution_enabled + + monkeypatch.setattr( + "lfx.services.deps.get_settings_service", + lambda: SimpleNamespace(settings=SimpleNamespace(allow_custom_components=False)), + ) + with pytest.raises(CodeExecutionDisabledError): + ensure_code_execution_enabled() + + def test_allows_when_custom_components_enabled(self, monkeypatch): + from types import SimpleNamespace + + from lfx.utils.python_repl_security import ensure_code_execution_enabled + + monkeypatch.setattr( + "lfx.services.deps.get_settings_service", + lambda: SimpleNamespace(settings=SimpleNamespace(allow_custom_components=True)), + ) + ensure_code_execution_enabled() # must not raise + + def test_allows_when_services_layer_absent(self, monkeypatch): + """An absent services layer (ImportError) is a local/trusted context -> allowed.""" + from lfx.utils.python_repl_security import ensure_code_execution_enabled + + # Deleting the imported name makes ``from lfx.services.deps import + # get_settings_service`` raise ImportError, simulating an absent services layer. + monkeypatch.delattr("lfx.services.deps.get_settings_service") + ensure_code_execution_enabled() # must not raise + + def test_blocks_when_settings_service_is_none(self, monkeypatch): + """A None settings service (registered-but-failed stack) fails closed, not open.""" + from lfx.utils.python_repl_security import CodeExecutionDisabledError, ensure_code_execution_enabled + + # get_service swallows init errors into None; the gate must refuse, not bypass. + monkeypatch.setattr("lfx.services.deps.get_settings_service", lambda: None) + with pytest.raises(CodeExecutionDisabledError): + ensure_code_execution_enabled() + + def test_non_import_error_propagates(self, monkeypatch): + """A non-ImportError from get_settings_service() propagates, never failing open.""" + from lfx.utils.python_repl_security import ensure_code_execution_enabled + + # Only ImportError means "no services layer"; anything else must surface. + def _boom(): + error = "settings stack exploded" + raise RuntimeError(error) + + monkeypatch.setattr("lfx.services.deps.get_settings_service", _boom) + with pytest.raises(RuntimeError, match="settings stack exploded"): + ensure_code_execution_enabled()