2026-09-07 18:24:55 +08:00

32 KiB

Raw Permalink Blame History

Workflow Studio 后端审计整改方案

文档状态：已按 2026-08-29 决策部分实施（见「0. 整改决策记录」）
适用范围：offline-backend-20260512/backend
审核日期：2026-08-28（决策与实施：2026-08-29）
关联文档：工作流编排与流式执行后端开发方案
适用对象：后端开发、测试、安全评审、运维发布

0. 整改决策记录（2026-08-29）

本轮整改以功能健壮性为唯一目标（用户决策："咱们这次的整改主要是保证功能的健壮"），安全隔离类项目暂缓。逐项状态：

项目	状态	说明
P1-01 人工介入恢复原子性	✅ 已实施	`resume_with_payload` 单条件更新，并发恢复恰一个成功
P1-02 草稿乐观锁 + 发布版本号	✅ 已实施	草稿 CAS 条件更新；`next_version_number` 计数器分配版本号
P1-03 事件序号 + SSE 一致性	✅ 已实施	`next_event_seq` 计数器；落库失败丢弃不发布（无幽灵事件）
P1-06 测试纳入版本控制	✅ 已实施	`.gitignore` 白名单 + 新增 `tests/test_workflow_concurrency.py`（SQL 并发）
P2 §11.1 完整 JSON Schema 校验	✅ 已实施	`schema_validation.py`（Draft202012，中文字段路径错误）
P0-01 数据源越权 / HTTP 凭证外泄	❌ 不实施	内网可信环境，按 ID 共享数据源、HTTP 节点访问任意内网地址是既有使用方式，加 owner 校验 / host 白名单会直接破坏现网用法，风险由部署方接受
P0-02 子工作流跨用户执行	❌ 不实施	同上：内网用户间按 ID 引用子工作流是预期用法
P1-04 智能体/技能最小权限	⏸ 暂缓	安全隔离类，本轮不做（用户决策："安全性的问题咱们暂时不用做"）
P1-05 代码节点沙箱化	⏸ 暂缓	同上；`config.yaml` 中 `workflows.code.enabled` 保持默认 `false`，代码节点不开放
P2 §11.2 iframe Redis nonce	⏸ 暂缓	单实例部署，进程内 replay cache + 60s TTL 足够；多实例化时再引入 Redis

已实施项的回归测试集中在 tests/test_workflow_concurrency.py（内存 + 真实 SQLite 双轨），外加 tests/test_workflow_runs.py 的路由级 schema 校验回归；全部 88 个 workflow 测试通过。

1. 结论与整改目标

当前 Workflow Studio 的定义、发布、运行、SSE、SQL/HTTP、智能体、技能、人工介入等功能骨架已经具备，且基础测试可运行。但在共享资源授权、凭证使用边界、并发一致性、事件可靠性和代码执行隔离方面存在上线阻断问题。

本次整改的目标不是继续扩展节点功能，而是让现有节点满足以下底线：

一个用户不能通过资源 ID、流程 JSON 或子工作流引用使用别人的私有资源。
HTTP 凭证只能发送至管理员明确批准的目标域名，不能被流程编辑者导出到任意 URL。
人工介入恢复、草稿保存、发布版本和事件序号在多请求、多实例条件下保持一致。
SSE 中显示的业务事件必须能够从数据库重放；不能出现只在内存短暂存在的“幽灵事件”。
智能体和技能节点只能使用流程配置所授予的能力。
代码节点只有在隔离沙箱中才能开启，不能在 Gateway 宿主机执行用户代码。

在 P0 项全部完成前，生产环境不得向普通用户开放：共享 SQL 数据源、共享 HTTP 凭证、子工作流跨用户引用和工作流编辑/执行权限。

2. 总体整改策略

2.1 建立运行时资源授权边界

当前 API 层可以识别当前用户，但执行器通常只持有 source_id、workflow_id 或 agent_id。整改后必须将“谁在执行”作为运行时一等参数，从创建运行记录开始贯穿到每一个节点。

新增领域对象，建议放在 packages/harness/deerflow/workflows/：

from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class WorkflowExecutionPrincipal:
    """工作流运行时的可信执行身份，由 Gateway 创建，不接受前端传入。"""

    user_id: str
    is_admin: bool = False
    tenant_id: str | None = None

运行创建时将发起人固化在 workflow_runs.owner_id，恢复、重试和子工作流也沿用原始运行的 owner_id，不得使用前端请求中的任意 user_id。

principal = WorkflowExecutionPrincipal(
    user_id=run.owner_id,
    is_admin=await permission_service.is_admin(run.owner_id),
    tenant_id=run.tenant_id,
)

所有需要使用资源的运行时入口必须接收 principal：

data_source_store.resolve_for_execution(source_id, principal)
workflow_store.get_published_for_execution(workflow_id, version_id, principal)
agent_store.get_visible(agent_id, principal.user_id)
skill_registry.resolve_allowed(skill_id, principal, agent)

资源不存在、无权限、已禁用三种情况对普通用户应统一返回 RESOURCE_NOT_AVAILABLE，避免通过 UUID 枚举资源。

2.2 采用“发布校验 + 运行时校验”的双层防线

发布时校验是为了尽早提示编辑者；运行时校验是为了防止发布后授权被收回、资源被禁用、恶意请求绕过 UI 或历史版本继续引用资源。

阶段	应做的事
保存草稿	校验节点结构、字段格式，不因暂时无权限而破坏用户草稿
发布版本	校验每个外部资源对发布者是否可用，并将资源依赖写入版本元数据
创建运行	再次校验工作流版本和根工作流权限
节点执行	每次解析数据源、智能体、技能、子工作流时按 `principal` 校验
重试/恢复	使用原 run 的身份和已发布版本，不重新信任浏览器传参

3. P0-01：数据源越权与 HTTP 凭证外泄

决策（2026-08-29）：不实施。 本部署运行于内网可信环境，用户之间按数据源 ID 共享、HTTP 节点访问任意内网 URL 是既有且必需的使用方式；owner 校验、baseUrl+path 拆分、host 白名单会使现网流程全部失效。风险由部署方明确接受（"这个改了我们就没法用了"）。

3.1 问题说明

当前运行时在 app/gateway/workflow_executor.py 中按数据源 ID 直接解析 DSN/凭证；持久化层的 resolve_dsn 也没有传入 owner 或授权上下文。HTTP 节点可由流程 JSON 提供任意 URL，再把已解析的认证头附加到该 URL。

因此存在两类风险：

知道或获得他人私有数据源 ID 的用户，可能在自己的流程中引用该数据源。
可使用共享 HTTP 凭证的用户，可能把请求 URL 写为攻击者控制的公网域名，从而收到服务端附带的 API Key、Bearer Token 或 Basic Auth。

3.2 数据模型调整

在 workflow_data_sources 中增加显式共享与用途约束。字段命名可按现有 ORM 风格调整：

ALTER TABLE workflow_data_sources
  ADD COLUMN visibility VARCHAR(16) NOT NULL DEFAULT 'private',
  ADD COLUMN allowed_user_ids JSON NULL,
  ADD COLUMN allowed_tenant_ids JSON NULL,
  ADD COLUMN allowed_hosts JSON NULL,
  ADD COLUMN allowed_methods JSON NULL;

字段语义：

字段	含义
`visibility`	`private`、`shared`、`system`
`allowed_user_ids`	共享给指定用户；第一期可先支持此项
`allowed_tenant_ids`	多租户部署时按租户共享；单租户可暂不启用
`allowed_hosts`	HTTP 数据源允许请求的 hostname 白名单，必须填写
`allowed_methods`	HTTP 数据源可执行的方法白名单，例如 `GET`、`POST`

SQL 数据源的数据库账号仍必须由 DBA 创建为只读账号；应用层 SQL 解析和只读 SQL policy 是第二道防线，不能替代数据库权限。

3.3 Store 接口调整

将现有无身份的解析接口废弃：

# 禁止继续在节点执行路径使用
resolve_dsn(source_id: str)

替换为：

async def resolve_for_execution(
    self,
    source_id: str,
    principal: WorkflowExecutionPrincipal,
    expected_kind: Literal["sql", "http"],
) -> WorkflowDataSource:
    ...

实现规则：

资源必须存在且 enabled=True。
kind 必须与节点类型匹配。
允许 owner、被显式共享的用户、同租户授权用户或管理员访问。
禁止仅凭“已发布工作流”绕过资源权限；每一次运行均需检查。
无权限统一抛 ResourceNotAvailableError，日志中可记录真实原因和 source ID。

示例伪代码：

def can_execute(source: WorkflowDataSource, principal: WorkflowExecutionPrincipal) -> bool:
    if principal.is_admin:
        return True
    if source.owner_id == principal.user_id:
        return True
    if source.visibility == "shared" and principal.user_id in source.allowed_user_ids:
        return True
    if source.visibility == "system" and principal.tenant_id in source.allowed_tenant_ids:
        return True
    return False

需要同步修改：

packages/harness/deerflow/persistence/workflow_data_sources/store.py
packages/harness/deerflow/persistence/workflow_data_sources/sql.py
内存 store 实现与测试 fake
app/gateway/workflow_executor.py
使用数据源的 SQL / HTTP 节点执行器

3.4 HTTP 节点配置改造

不要继续允许流程节点同时填写任意完整 URL 与 credentialRef。推荐将 HTTP 数据源定义为受控目标，节点只填写相对路径：

{
  "type": "http",
  "dataSourceId": "crm-production-api",
  "method": "GET",
  "path": "/v1/customers",
  "query": {
    "page": "{{ inputs.page }}"
  }
}

数据源的机密配置中保存：

{
  "baseUrl": "https://api.company.com",
  "allowedHosts": ["api.company.com"],
  "allowedMethods": ["GET", "POST"],
  "auth": {
    "type": "bearer",
    "secret": "加密存储"
  }
}

执行顺序必须固定：

读取经授权的数据源。
变量渲染 path、query、body。
由 baseUrl + path 构造最终 URL，拒绝绝对 URL 和 //host/path 形式。
校验最终 host、端口、DNS 解析结果、方法、请求体上限。
最后才注入认证头。
每次重定向重新执行 host/SSRF 校验。
一旦发生跨 origin 重定向，移除 Authorization、Cookie、X-API-Key 等敏感头；建议默认拒绝跨 origin 重定向。

若业务确实需要任意 URL 调用，应提供另一种无凭证 HTTP 节点，并且其网络范围仍受全局 SSRF policy 约束。

3.5 验收测试

用户 A 创建私有 SQL 数据源；用户 B 使用其 ID 运行，返回 RESOURCE_NOT_AVAILABLE。
用户 A 发布引用私有数据源的流程后，用户 B 即使复制 JSON 也不能运行。
共享 HTTP 数据源只能请求 allowed_hosts 中的地址。
URL 模板渲染为不在白名单中的主机时被拒绝，且认证头没有发送。
302/307 跨域跳转时凭证不会跟随；推荐断言请求整体被拒绝。
SQL 节点使用只读账号执行 INSERT、CALL、多语句均被数据库和 SQL policy 双重拒绝。

4. P0-02：子工作流跨用户执行

决策（2026-08-29）：不实施。 内网可信环境下按 workflow/version ID 跨用户引用子工作流是预期用法，运行时沿用原 run 的 owner_id 已满足功能正确性；完整 ACL 见 P0-01 同批决策，一并接受风险。

4.1 工作流 ACL

在 workflow 主表增加与数据源一致的共享模型：

ALTER TABLE workflows
  ADD COLUMN visibility VARCHAR(16) NOT NULL DEFAULT 'private',
  ADD COLUMN allowed_user_ids JSON NULL,
  ADD COLUMN allowed_tenant_ids JSON NULL;

建议第一期仅实现：private、shared、system。不要把“已发布”当作“所有用户可执行”；发布只代表版本不可变并可被授权对象使用。

4.2 接口改动

以下接口均应由 principal 过滤：

接口能力	整改要求
已发布工作流目录	仅返回当前用户可执行的工作流
获取版本详情	owner / 被分享对象 / 管理员才可读
创建根运行	检查 workflow 与 version 的 ACL
子工作流节点	按父 run 的 principal 再次检查子流程 ACL
复制流程	允许复制公开定义不等于允许继承数据源和凭证引用

子工作流执行应改为：

child_version = await workflow_store.get_published_for_execution(
    workflow_id=node.workflow_id,
    version_id=node.version_id,
    principal=context.principal,
)
if child_version is None:
    raise ResourceNotAvailableError(node.workflow_id)

并在 child run 写入：

parent_run_id=parent_run.id
owner_id=parent_run.owner_id
tenant_id=parent_run.tenant_id

4.3 验收测试

用户 B 的资源目录不包含用户 A 的私有已发布流程。
用户 B 手工构造 A 的 workflow/version ID 创建运行，被拒绝。
用户 B 的父流程引用 A 的子流程，被拒绝。
A 共享子流程给 B 后，B 可运行；收回授权后，历史草稿仍可保存但新运行失败。

5. P1-01：人工介入恢复的原子性

状态（2026-08-29）：已实施。 Store 新增 resume_with_payload(run_id, resume_token, payload, *, context=None)（SQL/memory 双实现），暂停侧 set_awaiting_input 扩展 context 参数一次性写入 checkpoint。路由 resume_run 与执行器 _pause 均已切换到单条条件更新；consume_resume_token 仅为兼容保留。回归测试 tests/test_workflow_concurrency.py（内存 + SQLite 并发恢复、错 token、单 winner）。

5.1 问题与目标

当前恢复操作先写 resume_payload，再消费 token。两个并发请求可能先后覆盖 payload，最终执行的数据与收到成功响应的请求不一致。

目标是把“校验 token、写入输入、清除 token、变更状态”为单条条件更新。

5.2 Store 方法

新增方法，替代 update_run(...) + consume_resume_token(...) 组合调用：

async def resume_with_payload(
    self,
    run_id: str,
    resume_token: str,
    payload: dict[str, Any],
) -> WorkflowRun | None:
    """成功时返回已恢复的 run，token 无效或已消费时返回 None。"""

SQL 示例，JSON 写法按当前数据库方言封装：

UPDATE workflow_runs
SET
  context_json = :new_context_json,
  resume_token = NULL,
  pending_human_node_id = NULL,
  status = 'queued',
  updated_at = NOW()
WHERE id = :run_id
  AND status = 'awaiting_input'
  AND resume_token = :resume_token;

应用层规则：

rowcount != 1 时返回 409，且不得投递调度器。
rowcount == 1 后再投递调度器。
投递失败时保留 queued 状态，由扫描器补投；不要把 token 恢复。
对人工输入 payload 先完成 schema 校验、大小限制和脱敏审计，再进入事务。

暂停节点也应使用单次 store 调用，同时写入：状态、token、等待节点 ID 和上下文，避免执行器后续的独立 update_run 覆盖已恢复的数据。

5.3 验收测试

两个协程并发用同一 token 提交不同 payload，只有一个 200，另一个 409。
run context 中只存在 200 对应 payload。
worker 最终读取的 payload 与成功响应一致。
token 被消费后，多次重试均为 409。

6. P1-02：草稿与发布版本并发控制

状态（2026-08-29）：已实施。 草稿保存改为 UPDATE ... WHERE id AND draft_revision = :expected，rowcount=0 时回读当前 revision 抛 WorkflowDraftConflictError（409 带 currentRevision）。发布版本号改由 workflow_definitions.next_version_number 计数器 CAS 分配（max(counter, MAX(version)+1) 兼容存量行），唯一约束兜底、冲突重试而非 500。计数器列由迁移 20260828_04_workflow_counter_columns.py 增加（server_default，存量库由启动列同步自动补列）。回归测试：并发草稿双写单 winner、5 并发发布得相邻版本号、存量版本行计数器自愈。

6.1 草稿乐观锁

禁止“先读取 revision，再修改 ORM 对象，再 commit”的方式。改为带 revision 的条件更新：

UPDATE workflows
SET graph_json = :graph_json,
    draft_revision = draft_revision + 1,
    updated_at = NOW()
WHERE id = :workflow_id
  AND draft_revision = :expected_revision;

若受影响行数为零，则读取当前 revision，并返回：

{
  "code": "WORKFLOW_DRAFT_CONFLICT",
  "currentRevision": 18
}

前端收到 409 后应提示“流程已被其他编辑者保存”，支持刷新、比较或另存副本；不能静默覆盖。

6.2 发布版本号

禁止 MAX(version_number) + 1。建议在 workflow 行维护 next_version_number，在一个事务内锁定并递增：

SELECT next_version_number FROM workflows WHERE id = :id FOR UPDATE;
INSERT INTO workflow_versions (..., version_number) VALUES (..., :next_number);
UPDATE workflows
SET next_version_number = next_version_number + 1
WHERE id = :id;

若现有数据库不方便 FOR UPDATE，可采用具备唯一约束的 sequence/counter 表，捕获唯一冲突后重试；不要将冲突直接抛为 500。

6.3 验收测试

两个独立 SQLAlchemy session 同时保存同一 revision，严格只有一个成功。
两个发布请求同时执行，生成相邻但不同的版本号。
冲突被转换为稳定的 409 业务错误，而非数据库异常。

7. P1-03：事件持久化、序号和 SSE 一致性

状态（2026-08-29）：已实施（采用 7.2 计数器方案 + 7.3 的"丢弃不发布"策略）。 workflow_runs.next_event_seq 计数器在同一事务内 UPDATE ... SET next_event_seq = next_event_seq + 1 后读自增并插入事件，行写锁串行化并发追加（seq 严格递增、唯一、无空洞）；唯一约束冲突时 _resync_counter 以独立事务回写 MAX(seq) 自愈存量行后重试。落库失败的事件直接丢弃、绝不向 live hub 发布（原负序号幽灵事件实现已删除），信封带 eventPersistenceDegraded 标记、sink.degraded_count 计数并记结构化告警日志；SSE 端点对无终态事件的终态 run 合成关闭帧。回归测试：10 并发追加 seq 连续唯一、存量事件计数器自愈、落库失败零发布。

7.1 改造原则

业务事件必须遵循：获得持久序号 → 持久化成功 → 实时发布。不能生成负序号临时事件，也不能让前端看到无法通过重连补回的事件。

7.2 序号分配

不要使用 MAX(seq) + 1。在 workflow_runs 增加计数器：

ALTER TABLE workflow_runs
  ADD COLUMN next_event_seq BIGINT NOT NULL DEFAULT 1;

追加事件时在同一个数据库事务中：

对 run 原子递增 next_event_seq 并得到当前 seq。
以该 seq 插入 workflow_events。
提交事务。
仅提交成功后调用 live_hub.publish(event)。

不同数据库可选择 UPDATE ... RETURNING 或者用 run 行锁实现，但不可回退为 MAX(seq) 查询。

7.3 持久化失败处理

事件库短暂失败时：

在有限次数内重试；
仍失败则将运行标记为 retryable_failure 或 paused_system；
记录包含 run_id、node ID、异常分类的结构化日志和告警；
SSE 至少发送一个已成功持久化的终态或系统状态事件；
用户重连后只能从数据库事件日志恢复，不依赖当前进程内存。

不要使用“事件落库失败后仍向 live hub 发负数 seq”的实现，因为按 cursor 追尾的 SSE 客户端通常会忽略它。

7.4 验收测试

并行节点大量写事件，序号严格递增且无重复。
模拟一次插入事件失败，前端不会收到不可回放事件。
客户端从 Last-Event-ID=N 重连，得到所有 seq>N 的事件且顺序一致。
多实例下任一实例订阅 SSE，均可通过数据库补放完整事件。

8. P1-04：智能体和技能最小权限

决策（2026-08-29）：暂缓。 属安全隔离类项目，本轮整改只做功能性优化（用户决策："P1-04/P1-05 偏安全隔离不用做，安全性的问题咱们暂时不用做，咱们只改功能性的，优化的"）。后续若对外开放多租户再评估。

8.1 智能体可见性

agent 节点收到 agent_id 后，必须在 WorkflowAgentRunner 内部再次调用可见性查询：

agent = await agent_store.get_visible(
    agent_id=node.agent_id,
    user_id=context.principal.user_id,
)
if agent is None or not agent.enabled:
    raise ResourceNotAvailableError(node.agent_id)

不能仅依赖资源目录接口已过滤，也不能仅依赖发布阶段校验。

8.2 技能白名单

现有将 skill_names 传给 Lead Agent 的方式必须核实是否被真正消费。建议新增语义明确的运行时参数：

configurable["workflow_allowed_skill_ids"] = selected_skill_ids

并在 Lead Agent 加载技能、MCP 工具和工具调用路由之前强制过滤：

requested = set(configurable.get("workflow_allowed_skill_ids", []))
available = [skill for skill in all_enabled_skills if skill.id in requested]

还应校验：

选中的 skill 对当前用户可见。
skill 是否允许在 workflow 场景执行。
skill 的网络、数据库、文件权限是否与工作流 principal 相兼容。
系统级高风险 skill 是否只能由管理员在数据源/策略中启用。

8.3 验收测试

用户手工填写不可见 agent ID，运行被拒绝。
未选中的技能不出现在模型可调用工具列表中。
提示词诱导模型调用未选技能，调用被拒绝且记录审计事件。
选中的技能正常可执行，节点事件中仅暴露必要的脱敏参数。

9. P1-05：代码节点沙箱化

决策（2026-08-29）：暂缓，保持现状即安全。 本轮不迁移到容器沙箱；config.yaml 中 workflows.code.enabled 保持默认 false（代码节点全局关闭，tests/test_workflow_engine.py::test_code_node_disabled_by_default 已锁定该默认值），如需启用由运维在受限宿主上自行开启。现有 code_runner.py 子进程加固（净化环境、临时目录、rlimit、断网、超时杀进程组）在可信内网下作为过渡方案。

9.1 上线策略

在 DeerFlow 沙箱实现并通过安全验证前，保持：

workflows:
  code:
    enabled: false

不得以本地 exec、subprocess 或 monkey-patch socket 作为生产隔离方案。用户代码可借助原生扩展、子进程、文件系统或环境变量绕过此类限制。

9.2 目标接口

代码节点改为只调用现有 DeerFlow sandbox 抽象，而不自行启动宿主解释器：

result = await sandbox.run(
    code=rendered_code,
    language="python",
    timeout_seconds=30,
    memory_limit_mb=256,
    cpu_limit=1,
    max_processes=8,
    network_enabled=False,
    writable_paths=["/mnt/user-data/workflows/{run_id}"],
)

沙箱最低要求：

独立容器或受限命名空间，不能读取 Gateway 项目、环境变量与宿主文件。
默认无网络；若未来允许网络，也需复用 HTTP 节点的数据源、SSRF 和凭证策略。
CPU、内存、执行时长、磁盘、进程数和文件数限制。
stdout/stderr 使用流式读取并在读取中截断，例如各 1MB；禁止 communicate() 后再截断。
制品只能输出到工作流专属目录，并通过既有 artifact 机制登记。
记录代码 hash、镜像版本、资源消耗、退出码，日志不得泄露源代码中的密钥。

9.3 验收测试

代码读取 /etc、项目目录、环境变量失败。
导入 socket、创建子进程、连接外网失败。
死循环超时结束，运行正确失败或可重试。
大量 stdout/stderr 不造成 Gateway 内存膨胀。
制品只能写入限定目录。

10. P1-06：测试纳入版本控制与 CI

状态（2026-08-29）：已实施。 采用方案 1（白名单）：仓库根 .gitignore 对 tests/test_workflow_{schemas,definitions,engine,runs}.py 逐个 ! 白名单，并新增 tests/test_workflow_concurrency.py（本轮整改的内存 + 真实 SQLite 并发回归，含 P1-01/02/03 与 schema 校验）。CI 需在该清单基础上加 tests/test_workflow_concurrency.py。

当前工作流测试文件位于后端 tests/，但仓库根目录的 .gitignore 规则会忽略这一目录内容。必须在合并前处理，否则本地通过的测试不会进入提交和 CI。

推荐方案：

调整 .gitignore，仅忽略真正的生成物，不要忽略后端测试源码；或
若忽略规则短期不能改，使用 git add -f 纳入现有测试文件，但仍应补充注释说明原因；
CI 显式执行：

cd offline-backend-20260512/backend
PYTHONPATH=. uv run pytest \
  tests/test_workflow_schemas.py \
  tests/test_workflow_definitions.py \
  tests/test_workflow_engine.py \
  tests/test_workflow_runs.py -v

新增 SQL 集成测试 job，至少覆盖真实事务并发；不能只依赖 memory store。

11. P2：契约校验与 iframe 票据

状态（2026-08-29）：§11.1 已实施；§11.2 暂缓（单实例部署，进程内 replay cache + 60s TTL 足够，多实例化时再引入 Redis）。

11.1 完整 JSON Schema 校验

实施说明（2026-08-29）：新增纯函数模块 packages/harness/deerflow/workflows/schema_validation.py 的 schema_issues(value, schema)（Draft202012，中文短消息、字段路径、去重、上限 20 条、schema 本身非法时返回友好错误而非 500）。接入点：路由 start_run（启动前 400 WORKFLOW_INPUT_INVALID + details.errors）、引擎 Start/Output 节点（WORKFLOW_INPUT_INVALID / WORKFLOW_OUTPUT_SCHEMA_MISMATCH）。空 schema 不强制（未声明即不校验）。回归测试见 tests/test_workflow_concurrency.py 与 tests/test_workflow_runs.py::test_typed_input_schema_violations_return_structured_400。

输入节点和输出节点不应只校验 required 字段。应使用完整 JSON Schema validator，至少支持：type、enum、items、properties、required、additionalProperties、嵌套对象和数组。

from jsonschema import Draft202012Validator

validator = Draft202012Validator(schema)
errors = sorted(validator.iter_errors(value), key=lambda item: list(item.path))
if errors:
    raise WorkflowSchemaValidationError(
        errors=[{"path": list(err.path), "message": err.message} for err in errors]
    )

错误响应需要包含节点 ID、字段路径和可展示的消息；不应把内部 Python trace 返回给前端。

11.2 iframe 一次性票据

“一次性”语义不能使用单进程内存字典。若部署多实例，应使用 Redis 或数据库的原子写入：

SET workflow_embed_nonce:{jti} consumed NX EX 60

只有返回成功的实例可以继续验证。服务端还必须读取真实请求 Origin，并要求以下三者一致：

ticket 签发时绑定的 origin；
浏览器请求头中的 Origin；
后台 workflows.embed.allowed_origins 白名单。

不能只比较前端 body/query 参数中的 origin。

12. 建议实施顺序与分批发布

2026-08-29 实况：第二批全部落地；第四批仅 §11.1 落地（§11.2 暂缓）；第一批（P0-01/02）经部署方决策不实施；第三批（P1-04/05）暂缓。

第一批：安全阻断项

范围：资源 principal、数据源授权、HTTP host 绑定、子工作流 ACL。

发布策略：先迁移数据库字段，默认所有历史资源为 private；只允许管理员手工将必要资源改为 shared。此批完成并通过越权测试后，才允许开放 SQL/HTTP/子工作流节点。

第二批：一致性与可靠性

范围：人工恢复原子操作、草稿 CAS、发布版本锁、事件序号与持久化失败策略。

发布策略：需要停写窗口或兼容性迁移时，先双写/双读一版；对运行中的旧 run 保持旧 schema 可读，新增 run 使用新序号和新恢复逻辑。

第三批：能力隔离

范围：agent 可见性、skill 白名单、代码节点沙箱。

发布策略：代码节点继续全局关闭；先灰度给管理员和测试租户。技能节点在完成真实白名单前，不得宣称“只调用所选技能”。

第四批：契约与嵌入完善

范围：JSON Schema 完整校验、iframe Redis nonce、接口错误码统一、运维监控。

13. 发布准入清单

2026-08-29 更新：本清单按原始审计口径保留。结合「0. 整改决策记录」，P0 两项经部署方决策不实施（内网可信环境，风险已接受），与 P0 相关的准入项在内网部署中视为豁免；✅ 为本轮已达成项。

以下条件全部满足才可开放给普通用户：

所有资源解析接口都需要 WorkflowExecutionPrincipal。（❌ 不实施——内网可信环境，见 §3/§4 决策）
私有数据源、私有工作流、私有智能体均有跨用户拒绝测试。（❌ 不实施——同上）
HTTP 凭证绑定 host/method，跨域重定向不携带敏感认证头。（❌ 不实施——内网需访问任意地址，见 §3 决策）
SQL 数据源使用数据库只读账号，且连接/查询审计可追踪到 run ID。（部分：SQL 只读校验已有；账号隔离不在本轮范围）
子工作流按运行身份校验 ACL。（❌ 不实施——见 §4 决策）
人工输入恢复是单事务条件更新，并发测试通过。（✅ test_workflow_concurrency.py）
草稿保存和发布版本具备数据库级并发保护。（✅ 同上）
事件序号单调、唯一、可重放；没有负序号临时事件。（✅ 同上）
agent / skill 白名单在实际工具加载链路生效。（⏸ 暂缓——P1-04）
代码节点运行在受限 sandbox；否则保持关闭。（✅ 保持关闭：workflows.code.enabled: false 为默认且被测试锁定）
所有 workflow 测试文件已纳入 Git 与 CI。（✅ .gitignore 白名单 + 新增并发回归）
安全测试、SQL 并发集成测试、SSE 重连测试全部通过。（✅ 88 个 workflow 测试全绿；安全类测试随 P0/P1-04 决策豁免）

14. 相关代码入口

主题	当前主要代码入口
路由与运行 API	`app/gateway/routers/workflow_runs.py`
工作流执行与依赖装配	`app/gateway/workflow_executor.py`
智能体/技能运行	`app/gateway/workflow_agent_runner.py`
数据源 API	`app/gateway/routers/workflow_data_sources.py`
工作流资源目录	`app/gateway/routers/workflow_resources.py`
iframe ticket	`app/gateway/routers/workflow_embed.py`
数据源持久化	`packages/harness/deerflow/persistence/workflow_data_sources/`
工作流/版本持久化	`packages/harness/deerflow/persistence/workflows/`
运行与租约持久化	`packages/harness/deerflow/persistence/workflow_runs/`
事件持久化	`packages/harness/deerflow/persistence/workflow_events/`
SQL/HTTP 节点	`packages/harness/deerflow/workflows/nodes/integrations.py`
事件 sink	`packages/harness/deerflow/workflows/runtime/sink.py`
代码节点运行器	`packages/harness/deerflow/workflows/runtime/code_runner.py`
工作流安全策略	`packages/harness/deerflow/workflows/security/`
数据库迁移	`packages/harness/deerflow/persistence/migrations/`
现有工作流测试	`tests/test_workflow_*.py`

15. 非目标与注意事项

本整改不改变 Coze Studio 画布的基础交互，也不要求先改前端；前端仅需后续适配资源权限失败、409 冲突和结构化 schema 错误。
不要把资源密钥、数据库 DSN、人工输入 token 或工具原始参数写入前端 SSE 事件。
不要依赖“前端资源选择器只显示可见资源”作为授权机制；浏览器请求和画布 JSON 均视为不可信输入。
新字段上线后，历史资源默认按私有处理，避免一次迁移导致所有历史凭证意外共享。
每项整改完成后都要更新 WORKFLOW_STUDIO_BACKEND_DEV_ZH.md 中对应的“已完成”状态，避免文档声明与实际安全边界不一致。

32 KiB Raw Permalink Blame History Unescape Escape