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): do not execute code in validate_code (GHSA-2wcq-pvw2-xh7v) (#13696)

Browse Source 
* fix(security): do not execute code in validate_code (GHSA-2wcq-pvw2-xh7v)

POST /api/v1/validate/code compiled and exec'd every function definition in the
submitted code as part of "validation". Executing a function definition
evaluates its decorators and default-argument expressions at definition time, so
a payload like

    def f(x=__import__("os").system("...")): ...

runs arbitrary code during validation without the function ever being called -
an authenticated RCE (effectively unauthenticated under the default AUTO_LOGIN
single-user setup). The same issue was also reported separately as a duplicate.

validate_code now compiles each function only to surface syntax/compile errors
and never exec()s it. The legitimate component-execution paths
(create_function/execute_function/eval_function), which are separately gated by
allow_custom_components, are unchanged.

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* refactor(security): drop dead _create_langflow_execution_context helper

Addresses review feedback on GHSA-2wcq-pvw2-xh7v fix. The helper only
existed to seed the exec() that validate_code no longer performs, so its
only remaining references were its own isolation tests. Remove both, and
assert function compile errors are empty in the non-execution regression
test.

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Eric Hare
2026-06-18 13:26:43 -07:00
committed by GitHub
parent 094694d3f2
commit e7c33dbaad
2 changed files with 32 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
src/backend/tests/unit/utils/test_validate.py
View File

@ -6,7 +6,6 @@ from unittest.mock import Mock, patch
import pytest
from lfx.custom.validate import (
_create_langflow_execution_context,
add_type_ignores,
build_class_constructor,
compile_class_code,
@ -119,6 +118,26 @@ def error_function():
assert result["imports"]["errors"] == []
assert result["function"]["errors"] == []
def test_validate_code_does_not_execute_default_args(self, tmp_path):
"""validate_code must not run function-definition-time code.
Regression for GHSA-2wcq-pvw2-xh7v: the
validator previously exec()'d each function definition, so a default
argument (or decorator) expression executed during validation. Here a
default arg would create a file as a side effect; it must NOT run.
"""
marker = tmp_path / "pwned.txt"
code = f'def exploit(x=open({str(marker)!r}, "w").write("pwned")):\n return x\n'
result = validate_code(code)
# No code executed -> the side-effect file was never created.
assert not marker.exists()
# Validation still returns the normal structure (valid syntax, no imports).
assert result["imports"]["errors"] == []
# Code is syntactically valid, so no compile errors expected.
assert result["function"]["errors"] == []
def test_code_with_multiple_imports(self):
"""Test validation handles multiple imports."""
code = """
@ -149,52 +168,6 @@ def test_func():
mock_logger.debug.assert_called_with("Error parsing code", exc_info=True)
class TestCreateLangflowExecutionContext:
"""Test cases for _create_langflow_execution_context function."""
def test_creates_context_with_langflow_imports(self):
"""Test that context includes langflow imports."""
# The function imports modules inside try/except blocks
# We don't need to patch anything, just test it works
context = _create_langflow_execution_context()
# Check that the context contains the expected keys
# The actual imports may succeed or fail, but the function should handle both cases
assert isinstance(context, dict)
# These keys should be present regardless of import success/failure
expected_keys = ["DataFrame", "Message", "Data", "Component", "HandleInput", "Output", "TabInput"]
for key in expected_keys:
assert key in context, f"Expected key '{key}' not found in context"
def test_creates_mock_classes_on_import_failure(self):
"""Test that mock classes are created when imports fail."""
# Test that the function handles import failures gracefully
# by checking the actual implementation behavior
with patch("builtins.__import__", side_effect=ImportError("Module not found")):
context = _create_langflow_execution_context()
# Even with import failures, the context should still be created
assert isinstance(context, dict)
# The function should create mock classes when imports fail
if "DataFrame" in context:
assert isinstance(context["DataFrame"], type)
def test_includes_typing_imports(self):
"""Test that typing imports are included."""
context = _create_langflow_execution_context()
assert "Any" in context
assert "Dict" in context
assert "List" in context
assert "Optional" in context
assert "Union" in context
def test_does_not_include_pandas(self):
"""Test that pandas is not included in the langflow execution context."""
context = _create_langflow_execution_context()
assert "pd" not in context
class TestEvalFunction:
"""Test cases for eval_function function."""

78
src/lfx/src/lfx/custom/validate.py
View File

@ -57,82 +57,28 @@ def validate_code(code):
except ModuleNotFoundError as e:
errors["imports"]["errors"].append(str(e))
# Evaluate the function definition with langflow context
# Validate each function definition WITHOUT executing it.
#
# Security (GHSA-2wcq-pvw2-xh7v): this endpoint only
# *validates* code, but it previously compiled and exec()'d every function
# definition. Executing a function definition evaluates its decorators and
# default-argument expressions at definition time, so a payload such as
# def f(x=__import__("os").system("...")): ...
# achieves arbitrary code execution during "validation" — the function never
# has to be called. Compile only, to surface syntax/compile errors; never
# exec untrusted code on the validation path.
for node in tree.body:
if isinstance(node, ast.FunctionDef):
code_obj = compile(ast.Module(body=[node], type_ignores=[]), "<string>", "exec")
try:
# Create execution context with common langflow imports
exec_globals = _create_langflow_execution_context()
exec(code_obj, exec_globals)
compile(ast.Module(body=[node], type_ignores=[]), "<string>", "exec")
except Exception as e: # noqa: BLE001
logger.debug("Error executing function code", exc_info=True)
logger.debug("Error compiling function code", exc_info=True)
errors["function"]["errors"].append(str(e))
# Return the errors dictionary
return errors
def _create_langflow_execution_context():
"""Create execution context with common langflow imports."""
context = {}
# Import common langflow types that are used in templates
try:
from lfx.schema.dataframe import DataFrame
context["DataFrame"] = DataFrame
except ImportError:
# Create a mock DataFrame if import fails
context["DataFrame"] = type("DataFrame", (), {})
try:
from lfx.schema.message import Message
context["Message"] = Message
except ImportError:
context["Message"] = type("Message", (), {})
try:
from lfx.schema.data import Data
context["Data"] = Data
except ImportError:
context["Data"] = type("Data", (), {})
try:
from lfx.custom import Component
context["Component"] = Component
except ImportError:
context["Component"] = type("Component", (), {})
try:
from lfx.io import HandleInput, Output, TabInput
context["HandleInput"] = HandleInput
context["Output"] = Output
context["TabInput"] = TabInput
except ImportError:
context["HandleInput"] = type("HandleInput", (), {})
context["Output"] = type("Output", (), {})
context["TabInput"] = type("TabInput", (), {})
# Add common Python typing imports
try:
from typing import Any, Optional, Union
context["Any"] = Any
context["Dict"] = dict
context["List"] = list
context["Optional"] = Optional
context["Union"] = Union
except ImportError:
pass
return context
def eval_function(function_string: str):
# Create an empty dictionary to serve as a separate namespace
namespace: dict = {}