diff --git a/app/application/workflow.py b/app/application/workflow.py index b06db6268..2c450be4b 100644 --- a/app/application/workflow.py +++ b/app/application/workflow.py @@ -659,7 +659,12 @@ class WorkflowDefinitionCommand: payload: Mapping[str, Any], share_id: Optional[int] = None, ) -> WorkflowMutationResult: - """解析共享工作流内容并在提交后更新远程复用次数。""" + """ + 解析共享工作流内容并在提交后更新远程复用次数。 + + 只复用动作、流程和触发配置；执行上下文属于分享者本机的运行状态，对新工作流没有意义， + 且来自外部不可信，因此忽略载荷中的 ``context``，新工作流固定从空上下文开始。 + """ values = dict(payload) if not values.get("name"): return WorkflowMutationResult(False, "工作流名称不能为空") @@ -667,7 +672,6 @@ class WorkflowDefinitionCommand: for field, default, error_message in ( ("actions", "[]", "actions字段JSON格式错误"), ("flows", "[]", "flows字段JSON格式错误"), - ("context", "{}", "context字段JSON格式错误"), ("event_conditions", "{}", "event_conditions字段JSON格式错误"), ): raw = values.get(field) @@ -684,7 +688,7 @@ class WorkflowDefinitionCommand: "event_conditions": parsed["event_conditions"], "actions": parsed["actions"], "flows": parsed["flows"], - "context": parsed["context"], + "context": {}, "state": "P", } if await self._repository.async_get_by_name(workflow_values["name"]): diff --git a/app/chain/workflow.py b/app/chain/workflow.py index 2067606dd..010f306a0 100644 --- a/app/chain/workflow.py +++ b/app/chain/workflow.py @@ -1,8 +1,6 @@ import ast -import base64 import copy import inspect -import pickle import threading from collections import defaultdict, deque from contextvars import Context, copy_context @@ -353,7 +351,10 @@ class WorkflowExecutor: def restore_context(self) -> ActionContext: """ - 恢复工作流上下文，兼容旧版 Base64 Pickle 存储格式。 + 恢复工作流上下文。 + + 只接受结构化 JSON 上下文；旧版 Base64 Pickle 格式（``content`` 字段）不再反序列化， + 按空上下文从头执行，避免执行数据库中来源不可信的序列化载荷。 """ context = ActionContext() if self.workflow.context: @@ -361,8 +362,7 @@ class WorkflowExecutor: logger.info(f"工作流已执行动作：{self.workflow.current_action}") try: if isinstance(self.workflow.context, dict) and self.workflow.context.get("content"): - decoded_data = base64.b64decode(self.workflow.context["content"]) - context = pickle.loads(decoded_data) + logger.warning(f"工作流 {self.workflow.name} 的上下文为已废弃的旧版格式，将从空上下文开始执行") elif isinstance(self.workflow.context, dict): context = ActionContext.model_validate(self.workflow.context) except Exception: diff --git a/tests/test_workflow_execution.py b/tests/test_workflow_execution.py index 0cfd69200..1f1aea26c 100644 --- a/tests/test_workflow_execution.py +++ b/tests/test_workflow_execution.py @@ -33,11 +33,19 @@ def _build_workflow(current_action=None, context=None, actions=None, flows=None, ) -def _encoded_context(context: ActionContext) -> dict: - """编码工作流恢复上下文。""" - return { - "content": base64.b64encode(pickle.dumps(context)).decode("utf-8"), - } +_LEGACY_PAYLOAD_CALLS = [] + + +def _record_legacy_payload_call(): + """记录旧版 pickle 载荷被反序列化时触发的调用。""" + _LEGACY_PAYLOAD_CALLS.append("loaded") + + +class _LegacyPayload: + """反序列化时会执行回调的旧版上下文载荷。""" + + def __reduce__(self): + return _record_legacy_payload_call, () class _FakeWorkflowManager: @@ -188,7 +196,7 @@ def test_workflow_executor_resumes_downstream_nodes(monkeypatch): fake_manager = _FakeWorkflowManager(calls) workflow = _build_workflow( current_action="A", - context=_encoded_context(ActionContext()), + context={"progress": 50}, ) monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager) @@ -203,6 +211,27 @@ def test_workflow_executor_resumes_downstream_nodes(monkeypatch): assert executor.context.progress == 100 +def test_workflow_executor_ignores_legacy_pickle_context(monkeypatch): + """旧版 Base64 Pickle 上下文不得被反序列化，应按空上下文继续执行。""" + _LEGACY_PAYLOAD_CALLS.clear() + calls = [] + fake_manager = _FakeWorkflowManager(calls) + workflow = _build_workflow( + current_action="A", + context={"content": base64.b64encode(pickle.dumps(_LegacyPayload())).decode("utf-8")}, + ) + + monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager) + monkeypatch.setattr(workflow_module.runtime_stop_state, "resume_workflow", lambda workflow_id: None) + monkeypatch.setattr(workflow_module.runtime_stop_state, "is_workflow_stopped", lambda workflow_id: False) + + executor = workflow_module.WorkflowExecutor(workflow) + + assert _LEGACY_PAYLOAD_CALLS == [] + assert executor.context.workflow_context == {} + assert executor.context.node_outputs == {} + + def test_workflow_executor_restores_structured_context(monkeypatch): """恢复执行时应兼容新版结构化上下文存储格式。""" calls = [] diff --git a/tests/test_workflow_mutation_command.py b/tests/test_workflow_mutation_command.py index 55f560af5..cf2bf109c 100644 --- a/tests/test_workflow_mutation_command.py +++ b/tests/test_workflow_mutation_command.py @@ -312,6 +312,36 @@ async def test_fork_commits_before_reporting_remote_count(): assert calls == ["commit", "report"] +@pytest.mark.asyncio +async def test_fork_ignores_shared_execution_context(): + """复用共享工作流时不得导入分享载荷中的执行上下文。""" + command, dependencies = _definition_command() + + result = await command.fork( + { + "name": "Forked", + "actions": "[]", + "flows": "[]", + "context": '{"content": "bGVnYWN5"}', + }, + ) + + assert result.success is True + created_values = dependencies["repository"].stage_create.await_args.args[0] + assert created_values["context"] == {} + + +@pytest.mark.asyncio +async def test_fork_accepts_malformed_shared_context(): + """分享载荷中的上下文不参与复用，其 JSON 格式错误也不应阻止创建。""" + command, dependencies = _definition_command() + + result = await command.fork({"name": "Forked", "context": "{"}) + + assert result.success is True + dependencies["repository"].stage_create.assert_awaited_once() + + @pytest.mark.asyncio async def test_fork_invalid_json_stops_before_database_write(): """共享内容 JSON 无效时不得创建半成品工作流。"""