用 Flowise Chatflow 导出校验器检查节点连接与明文密钥

Flowise 可以把 Chatflow 导出成 JSON，方便备份和版本管理。但导出文件里偶尔会带上明文 API Key，边也可能指向已删除的节点。 这篇做 check_flow.py：读导出 JSON，检查节点/边完整性，并扫描常见密钥字段是否写了明文。样例是一个三节点的「食堂问答」流程骨架。

先看清成品

• 真实运行：合法 lunch.json 通过（3 节点 2 边）；with_key.json 检出明文 openAIApiKey；broken_edge.json 检出悬空边。
• 6 项 pytest 通过。
• 没有启动真实 Flowise UI。

合规提醒：用自己的 Flowise 实例与模型 API Key，不共享账号；密钥放环境变量或官方凭据管理，不要写进导出的 chatflow JSON。只处理你有权处理的数据；生产环境务必开启登录保护。费用以各模型供应商官网为准，本文不写价格。

运行说明：我没有安装完整的 Flowise 服务（需要 Node.js 全局包或 Docker）。真实运行过的是：导出 JSON 校验器、Prediction 客户端对 mock 服务的请求、.env 检查脚本，以及 pytest（2026-10-23，UTC+8）。Flowise 服务端行为全部来自官方文档。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

91-flowise-A-src/
├── check_flow.py
├── ok/lunch.json
├── bad/
│   ├── with_key.json
│   ├── broken_edge.json
│   └── empty.json
└── test_91a.py

第 1 步：校验脚本

#!/usr/bin/env python3
"""Validate a Flowise-exported chatflow JSON before sharing or committing it."""
from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path

# Patterns that look like real secrets accidentally left in an export.
SECRET_RE = re.compile(
    r"(?i)(sk-[a-z0-9]{20,}|sk-or-v1-[a-z0-9]{20,}|"
    r"sk-ant-[a-z0-9\-]{20,}|AIza[0-9A-Za-z\-_]{20,}|"
    r"ghp_[a-zA-Z0-9]{20,}|xox[baprs]-[a-zA-Z0-9\-]{10,})"
)
PLAIN_KEY_RE = re.compile(
    r'(?i)"(openAIApiKey|apiKey|api_key|password|secret|token)"\s*:\s*"([^"]{8,})"'
)
# Flowise stores credentials separately; raw key values in nodes are a smell.
CREDENTIAL_DATA_RE = re.compile(r'(?i)"credentialData"\s*:\s*\{[^}]*"(apiKey|openAIApiKey)"')


def load(path: Path) -> dict:
    data = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ValueError("root must be a JSON object")
    return data


def flow_graph(data: dict) -> tuple[list, list]:
    """Accept both top-level nodes/edges and nested flowData (common export shapes)."""
    if "flowData" in data:
        fd = data["flowData"]
        if isinstance(fd, str):
            fd = json.loads(fd)
        nodes = fd.get("nodes") or []
        edges = fd.get("edges") or []
        return nodes, edges
    return data.get("nodes") or [], data.get("edges") or []


def check(path: Path) -> list[str]:
    errors: list[str] = []
    data = load(path)
    raw = path.read_text(encoding="utf-8")

    nodes, edges = flow_graph(data)
    if not nodes:
        errors.append("没有节点（nodes 为空）——这不是一份可用的 chatflow 导出")
    ids = {n.get("id") for n in nodes if isinstance(n, dict)}
    for e in edges:
        if not isinstance(e, dict):
            errors.append("edge 不是对象")
            continue
        src, tgt = e.get("source"), e.get("target")
        if src and src not in ids:
            errors.append(f"边引用了不存在的 source 节点: {src}")
        if tgt and tgt not in ids:
            errors.append(f"边引用了不存在的 target 节点: {tgt}")

    # At least one Chat Model / LLM-ish node is expected for a chatflow.
    types = []
    for n in nodes:
        if not isinstance(n, dict):
            continue
        data_n = n.get("data") or {}
        types.append(data_n.get("name") or data_n.get("type") or n.get("type") or "")
    if nodes and not any(
        any(k in (t or "").lower() for k in ("chat", "llm", "agent", "openai", "anthropic", "ollama"))
        for t in types
    ):
        errors.append("未发现对话/LLM 类节点（name/type 中没有 chat/llm/agent 等关键字）")

    for m in SECRET_RE.finditer(raw):
        errors.append(f"疑似明文密钥: {m.group(0)[:12]}…")
    for m in PLAIN_KEY_RE.finditer(raw):
        val = m.group(2)
        if val.startswith("{{") or val.startswith("$") or val == "CREDENTIAL_PLACEHOLDER":
            continue
        errors.append(f"节点字段 {m.group(1)} 里写了明文值（应走凭据管理）")
    if CREDENTIAL_DATA_RE.search(raw):
        errors.append("导出里出现了 credentialData 明文密钥字段")

    name = data.get("name") or data.get("chatflow") or ""
    if not name and "id" not in data:
        errors.append("缺少 name 或 id，不太像完整导出")

    return errors


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("path", type=Path, help="exported chatflow .json")
    args = p.parse_args(argv)
    try:
        errs = check(args.path)
    except (OSError, ValueError, json.JSONDecodeError) as e:
        print(f"✗ 无法读取: {e}", file=sys.stderr)
        return 2
    if errs:
        print(f"✗ {args.path.name}: {len(errs)} 个问题")
        for e in errs:
            print(f"  - {e}")
        return 1
    nodes, edges = flow_graph(load(args.path))
    print(f"✓ {args.path.name}: {len(nodes)} 个节点, {len(edges)} 条边，未发现明文密钥")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：合法样例

{
  "id": "cf-demo-001",
  "name": "午餐问答",
  "flowData": {
    "nodes": [
      {
        "id": "start_0",
        "data": {
          "name": "startAgentflow",
          "label": "Start"
        }
      },
      {
        "id": "chatOpenAI_0",
        "data": {
          "name": "chatOpenAI",
          "label": "ChatOpenAI",
          "inputs": {
            "modelName": "gpt-4o-mini",
            "temperature": 0.2
          }
        }
      },
      {
        "id": "prompt_0",
        "data": {
          "name": "promptTemplate",
          "label": "Prompt",
          "inputs": {
            "template": "你是食堂助手。用户问：{question}"
          }
        }
      }
    ],
    "edges": [
      {
        "id": "e1",
        "source": "start_0",
        "target": "prompt_0"
      },
      {
        "id": "e2",
        "source": "prompt_0",
        "target": "chatOpenAI_0"
      }
    ]
  }
}

第 3 步：测试

from pathlib import Path
from check_flow import check, main

ROOT = Path(__file__).parent

def test_ok():
    assert check(ROOT / "ok/lunch.json") == []

def test_key():
    errs = check(ROOT / "bad/with_key.json")
    assert any("明文" in e or "密钥" in e for e in errs)

def test_edge():
    errs = check(ROOT / "bad/broken_edge.json")
    assert any("不存在" in e for e in errs)

def test_empty():
    errs = check(ROOT / "bad/empty.json")
    assert any("节点" in e for e in errs)

def test_cli_ok():
    assert main([str(ROOT / "ok/lunch.json")]) == 0

def test_cli_bad():
    assert main([str(ROOT / "bad/with_key.json")]) == 1

第 4 步：运行

# 2026-10-23 12:37 UTC+8
$ python check_flow.py ok/lunch.json
✓ lunch.json: 3 个节点, 2 条边，未发现明文密钥
$ python check_flow.py bad/with_key.json
✗ with_key.json: 1 个问题
  - 节点字段 openAIApiKey 里写了明文值（应走凭据管理）
$ python check_flow.py bad/broken_edge.json
✗ broken_edge.json: 1 个问题
  - 边引用了不存在的 source 节点: missing_node
$ python -m pytest -q
6 passed

怎么检查结果？

• 合法文件打印节点/边数量且无明文密钥。
• 明文密钥与断边必须失败（退出码非 0）。
• 把真实导出丢进检查器再提交到 Git。

调试示例

明文密钥是怎么混进导出的。 我在 UI 里把 OpenAI 节点的 Key 直接填进字段（而不是选 Credentials），导出 JSON 后 openAIApiKey 变成一串明文。检查脚本扫描常见密钥字段名（openAIApiKey、apiKey、credential 等），只要值不是空/{{...}} 占位就报错。官方 Credentials 文档要求密钥走凭据管理；导出给同事或进 Git 之前跑一遍本检查器。

断边。 删掉节点但忘记清边时，edges[].source 会指向不存在的 id。脚本建节点 id 集合再核对每条边的 source/target。

常见问题

模板规则

• 密钥走 Credentials，不写进节点。
• 提交前跑 check_flow.py。
• 边必须指向存在的节点。

项目资源

导出 JSON 是备份，不是把密钥备份进仓库。