|
1 | | -import ast |
2 | | -import asyncio |
3 | | -import builtins |
4 | | -import multiprocessing |
5 | | -import traceback |
6 | | -from typing import Any, Dict |
7 | | - |
8 | | -from pydantic import BaseModel |
| 1 | +from typing import Any |
9 | 2 |
|
10 | 3 | from workflow.engine.nodes.code.executor.base_executor import BaseExecutor |
11 | 4 | from workflow.exception.e import CustomException |
12 | 5 | from workflow.exception.errors.err_code import CodeEnum |
13 | 6 | from workflow.extensions.otlp.trace.span import Span |
14 | 7 |
|
15 | 8 |
|
16 | | -class Modules(BaseModel): |
17 | | - """ |
18 | | - Modules that are allowed to be imported for security reasons |
19 | | - """ |
20 | | - |
21 | | - imports: list[str] |
22 | | - from_imports: list[ast.ImportFrom] |
23 | | - |
24 | | - class Config: |
25 | | - arbitrary_types_allowed = True |
26 | | - |
27 | | - |
28 | 9 | class LocalExecutor(BaseExecutor): |
29 | | - """ |
30 | | - Local code executor using RestrictedPython for secure execution. |
| 10 | + """Compatibility shim for the removed in-process code executor. |
31 | 11 |
|
32 | | - Executes Python code in a restricted environment with limited built-ins |
33 | | - and forbidden modules to ensure security. Uses multiprocessing for isolation. |
| 12 | + User-provided code must never execute in the workflow service process. A |
| 13 | + child process and a timeout do not isolate the filesystem, credentials, |
| 14 | + network, or operating-system user from untrusted code. |
34 | 15 | """ |
35 | 16 |
|
36 | 17 | async def execute( |
37 | 18 | self, language: str, code: str, timeout: int, span: Span, **kwargs: Any |
38 | 19 | ) -> str: |
39 | | - """ |
40 | | - Execute code asynchronously using multiprocessing for isolation. |
41 | | -
|
42 | | - :param language: Programming language (currently only python supported) |
43 | | - :param code: Code string to execute |
44 | | - :param timeout: Maximum execution time in seconds |
45 | | - :param span: Tracing span for logging |
46 | | - :param kwargs: Additional execution parameters |
47 | | - :return: Execution result as string |
48 | | - """ |
49 | | - loop = asyncio.get_running_loop() |
50 | | - return await loop.run_in_executor( |
51 | | - None, # Use default thread pool |
52 | | - self._execute_in_process, # Wrapper for synchronous execution |
53 | | - code, |
54 | | - timeout, |
| 20 | + raise CustomException( |
| 21 | + err_code=CodeEnum.CODE_EXECUTION_ERROR, |
| 22 | + err_msg=( |
| 23 | + "The local code executor is disabled for security. " |
| 24 | + "Configure an isolated code executor before running code nodes." |
| 25 | + ), |
55 | 26 | ) |
56 | | - |
57 | | - def _execute_in_process(self, code: str, timeout: int) -> str: |
58 | | - """ |
59 | | - Execute code in a separate process with timeout control. |
60 | | -
|
61 | | - :param code: Code string to execute |
62 | | - :param timeout: Maximum execution time in seconds |
63 | | - :return: Execution result as string |
64 | | - :raises CustomException: If execution times out or fails |
65 | | - """ |
66 | | - with multiprocessing.Manager() as manager: |
67 | | - result_dict = manager.dict() |
68 | | - proc = multiprocessing.Process( |
69 | | - target=self._safe_exec, args=(code, result_dict) |
70 | | - ) |
71 | | - proc.start() |
72 | | - proc.join(timeout) |
73 | | - if proc.is_alive(): |
74 | | - proc.terminate() |
75 | | - raise CustomException(err_code=CodeEnum.CODE_EXECUTION_TIMEOUT_ERROR) |
76 | | - if "error" in result_dict: |
77 | | - raise CustomException( |
78 | | - err_code=CodeEnum.CODE_EXECUTION_ERROR, err_msg=result_dict["error"] |
79 | | - ) |
80 | | - return result_dict.get("output", "") |
81 | | - |
82 | | - def _safe_exec(self, code: str, result_dict: dict) -> None: |
83 | | - """ |
84 | | - Safely execute code using RestrictedPython with limited built-ins. |
85 | | -
|
86 | | - :param code: Code string to execute |
87 | | - :param result_dict: Shared dictionary to store execution results |
88 | | - """ |
89 | | - try: |
90 | | - locals_dict: Dict[str, Any] = {} |
91 | | - |
92 | | - modules = self._find_imports(code) |
93 | | - import_code_lines = [] |
94 | | - |
95 | | - for module in modules.imports: |
96 | | - import_code_lines.append(f"import {module}") |
97 | | - |
98 | | - for from_module in modules.from_imports: |
99 | | - imported_names = ", ".join(alias.name for alias in from_module.names) |
100 | | - import_code_lines.append( |
101 | | - f"from {from_module.module} import {imported_names}" |
102 | | - ) |
103 | | - |
104 | | - import_code = "\n".join(import_code_lines) |
105 | | - |
106 | | - sandbox_globals = {"__builtins__": builtins} |
107 | | - |
108 | | - exec(import_code, sandbox_globals) |
109 | | - exec(code, sandbox_globals, locals_dict) |
110 | | - |
111 | | - result_dict["output"] = locals_dict.get("output", "") |
112 | | - except Exception: |
113 | | - result_dict["error"] = traceback.format_exc() |
114 | | - |
115 | | - def _find_imports(self, code: str) -> Modules: |
116 | | - """ |
117 | | - Find imports and from imports in the code. |
118 | | -
|
119 | | - :param code: Code string to find imports and from imports |
120 | | - :return: Modules object containing imports and from_imports |
121 | | - """ |
122 | | - imports: list[str] = [] |
123 | | - from_imports: list[ast.ImportFrom] = [] |
124 | | - parsed_code = ast.parse(code) |
125 | | - for node in parsed_code.body: |
126 | | - if isinstance(node, ast.Import): |
127 | | - imports.extend(alias.name for alias in node.names) |
128 | | - elif isinstance(node, ast.ImportFrom): |
129 | | - from_imports.append(node) |
130 | | - return Modules(imports=imports, from_imports=from_imports) |
0 commit comments