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

fix(security): gate PythonREPL execution on allow_custom_components (GHSA-8qpj-27x8-pwpq) (#13700)

Browse Source 
* 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>
This commit is contained in:
Eric Hare
2026-06-19 10:29:17 -07:00
committed by GitHub
parent 11a18f7834
commit 2754c84aad
6 changed files with 161 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

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

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

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

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

57
src/lfx/src/lfx/utils/python_repl_security.py
View File

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

58
src/lfx/tests/unit/utils/test_python_repl_security.py
View File

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