fix(security): 修复5.60安全用例JSON双重编码500错误 + 部署脚本防复发

- 新增 fix_560_json_double_encoding.py：修复5.60 MySQL中67条安全用例
  tags/steps/config 双重编码（sqlite3原生读取JSON列为文本字符串，
  部署脚本再次json.dumps导致MySQL JSON列存成字符串值而非数组/对象，
  Pydantic TestCaseResponse校验报 input_type=str）
- 验证：67条全部修复，JSON_TYPE均为ARRAY/OBJECT，API恢复200
- deploy_security_cases_560.py 新增 _parse_json_field()：加载本地用例
  时先json.loads还原JSON列，防止复发
- HANDOFF_安全测试.md：新增S18/S19踩坑 + 2026-09-10修复记录 + P5-P7待办
Co-Authored-By: Claude <noreply@anthropic.com>

HANDOFF_安全测试.md

@@ -80,6 +80,24 @@
 
 ---
 
+### 本次任务（2026-09-10）：5.60 部署后安全测试模块 500 修复（JSON 双重编码）+ 部署脚本防复发
+
+**背景**：标准版用例库补全（42→67）同步部署 5.60 后，用户访问安全测试模块报 500：`4 validation errors for TestCaseResponse — tags/steps/config: Input should be a valid list/dictionary (input_type=str)`。
+
+**根因**（已实测确认）：
+- `deploy_security_cases_560.py` 用 sqlite3 原生读取本地 SQLite 的 `tags/steps/config` 列，SQLAlchemy JSON 在 SQLite 上以 TEXT 存储 → **读到的是 JSON 文本字符串**；
+- 脚本又对其 `json.dumps()` → **双重编码**，MySQL JSON 列存成了"JSON 字符串值"（`"[\"...\"]"`）而非数组/对象；
+- API 读取后 Pydantic 收到 str 而非 list/dict → 4 个校验错误（恰巧 `depends_on/parameters/condition` 未被 dumps、直传文本被 MySQL 正常解析，所以只有 3 字段报错）。
+
+**修复**（全部完成并验证）：
+1. `backend/scripts/fix_560_json_double_encoding.py`（新增）— 对 5.60 MySQL 中 67 条安全用例的 tags/steps/config 逐层 `json.loads` 还原后 UPDATE 回写；运行结果：**修复 67 条，JSON_TYPE 验证全部 ARRAY/OBJECT**。
+2. `backend/scripts/deploy_security_cases_560.py` — 新增 `_parse_json_field()`，加载本地用例时对 6 个 JSON 列先 `json.loads` 还原，防止下次部署复发。
+3. **API 验证**：`GET http://192.168.5.60/api/cases?case_type=security` → 200，total=67，tags=list / steps=dict / config=dict 全部正常，前端安全测试模块恢复访问。
+
+**注意事项**：
+- 本地 `data/test_platform.db` 曾被并行窗口覆盖回退到 **42 条**安全用例（5.60 上是 67 条）；本地补录需重新运行 `create_security_cases.py`（注意与并行窗口错开数据库写入时间，见踩坑 S8）
+- 部署脚本与修复脚本在 `backend/scripts/` 下被 .gitignore 忽略，提交需 `git add -f`
+
 ### 本次任务（2026-08-19）：安全测试新增定时任务模块 + 部署 5.60 验证通过
 
 **背景**：安全测试模块补充定时任务能力（设置执行周期 interval/daily/weekly + 执行时间），参考 UI 自动化定时任务模块实现，复用同一张 `scheduled_tasks` 表和调度引擎，按 `case_type` 分流调度。
@@ -410,6 +428,9 @@ GET /api/security/executions/{id}/report/download → 下载 .md 文件
 | **P3** | ✅ ~~安全报告上传 ERP 流程对接~~ | ERP 配置入口已完成，ERP 上传流程已完成（2026-08-17） |
 | **P3.5** | ✅ ~~安全测试 ERP 任务创建对接~~ | 任务创建预览 + 创建接口 + 前端对话框已完成（2026-08-17）；**已部署 5.60 并实测 task-preview 通过**（真实执行 ID 返回正确的任务名/紧急程度/ERP 基础数据），create-task 待用户在界面实测 |
 | **P4** | ✅ ~~安全测试定时任务模块~~ | 周期调度（interval/daily/weekly）+ 自动报告可选 + 独立页面已完成（2026-08-19）；**已部署 5.60 并通过创建/删除冒烟测试**，定时调度触发待真实执行验证 |
+| **P5** | 基于主流漏洞库补充标准版新用例 | CWE Top 25 (2024) / OWASP API Top 10 (2023) / OWASP ASVS 4.0 三份调研已完成；**尚未真实写入** `create_security_cases.py`（此前会话曾误报完成，grep 核实未写入，见踩坑 S19）。需真实编写新用例定义并同步 5.60 |
+| **P6** | 修订 `Docs/PRD/需求文档/安全测试/` 两份草拟文档 | `_PRD_安全测试标准版用例补充_需求文档.md` 与 `_计划执行.md` 为早期草拟（API11-13/ASVS 方案），与实际执行情况有出入，待按 P5 落地后修订对齐 |
+| **P7** | 本地 DB 与 5.60 用例数对齐 | 本地 SQLite 安全用例 42 条（被并行窗口覆盖回退），5.60 为 67 条；需重新运行 `create_security_cases.py` 补齐（注意错开并行窗口数据库写入，见 S8） |
 
 ---
 
@@ -434,6 +455,8 @@ GET /api/security/executions/{id}/report/download → 下载 .md 文件
 | S15 | 前端构建时 `getSecurityConfigs` 的 `items` 类型错误 | Axios 类型声明是 `AxiosResponse`，运行时拦截器已解包为 body | 调用处使用 `const data = (res as any).data || res`，再访问 `data.items` |
 | S16 | 服务器远程命令通过 paramiko 读取超时 | 长时间 `curl`/MySQL 命令在 SSH channel 上未及时结束，PipeTimeout | 使用短命令 + `head -c`；健康检查优先使用容器内 `wget -qO- --timeout=10 http://127.0.0.1/health`；必要时对读取异常做容错 |
 | S17 | 安全定时任务接口未返回/表字段缺失 | 新字段未迁移或容器未加载新代码 | 上传绑定挂载目录后执行 `docker restart plat-auto-test-app`；启动时 `init_db()` 自动运行 `_ensure_columns()`，并验证 `case_type/config_id` |
+| **S18** | 部署同步报 `Unknown column 'project_type'` | 服务器 MySQL 表缺新列，先同步数据后迁移 | 部署脚本顺序：**先上传含 `_ensure_columns()` 的 database.py/模型 → 重启容器自动加列 → 再同步数据 → 再重启** |
+| **S19** | 部署后 API 500：`4 validation errors for TestCaseResponse (tags/steps/config input_type=str)` | sqlite3 原生读取 JSON 列得到的是 JSON 文本**字符串**（SQLAlchemy JSON 在 SQLite 上即 TEXT），脚本再 `json.dumps()` 造成**双重编码**，MySQL JSON 列存成了"字符串值"而非数组/对象 | 脚本加 `_parse_json_field()`：先 `json.loads` 还原为对象再写库；存量数据用 `scripts/fix_560_json_double_encoding.py` 逐层 json.loads 修复（2026-09-10 已修复 67 条并验证 API 恢复） |
 
 ---
 

backend/scripts/deploy_security_cases_560.py

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+部署脚本：同步安全测试标准版用例补充到 5.60 生产服务器
+
+同步内容：
+  1. backend/scripts/create_security_cases.py（新增用例定义）
+  2. 用例数据同步（本地 SQLite 67 个安全用例 → 服务器 MySQL）
+  3. 重启容器 + 冒烟验证
+
+用法:
+    cd backend && python scripts/deploy_security_cases_560.py
+"""
+import json
+import os
+import sys
+import time
+from pathlib import Path
+
+sys.stdout.reconfigure(encoding="utf-8")
+os.chdir(Path(__file__).parent.parent)  # 切到 backend/
+
+import paramiko
+import pymysql
+
+SERVER = ("192.168.5.60", "ubains", "Ubains@123")
+MYSQL_PORT = 3307
+MYSQL_USER = "platapp"
+MYSQL_PWD = "PlatApp2026"
+MYSQL_DB = "plat_auto_test"
+
+REMOTE_BASE = "/data/third_party/plat-auto-test"
+REMOTE_BACKEND = f"{REMOTE_BASE}/backend"
+CONTAINER = "plat-auto-test-app"
+
+# 需要同步的后端文件（本地路径 → 远程相对路径）
+BACKEND_FILES = {
+    "scripts/create_security_cases.py": "scripts/create_security_cases.py",
+    "app/database.py": "app/database.py",
+    "app/models/test_case.py": "app/models/test_case.py",
+}
+
+INSERT_SQL = (
+    "INSERT INTO test_cases "
+    "(id, module_id, name, description, status, priority, tags, steps, config, "
+    "depends_on, parameters, `condition`, `order`, version, created_at, updated_at, "
+    "case_type, project_type) "
+    "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
+    "ON DUPLICATE KEY UPDATE "
+    "name=VALUES(name), description=VALUES(description), priority=VALUES(priority), "
+    "tags=VALUES(tags), steps=VALUES(steps), config=VALUES(config), "
+    "depends_on=VALUES(depends_on), parameters=VALUES(parameters), "
+    "`condition`=VALUES(`condition`), `order`=VALUES(`order`), version=VALUES(version), "
+    "updated_at=VALUES(updated_at), case_type=VALUES(case_type), project_type=VALUES(project_type)"
+)
+
+
+def _parse_json_field(value):
+    """SQLite 原生读取的 JSON 列是 JSON 文本字符串，需还原为对象（防双重编码）"""
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except (json.JSONDecodeError, ValueError):
+            return value
+    return value
+
+
+def load_local_cases():
+    """从本地 SQLite 加载所有安全测试用例"""
+    import sqlite3
+    conn = sqlite3.connect("data/test_platform.db")
+    conn.row_factory = sqlite3.Row
+    cur = conn.cursor()
+    cur.execute(
+        "SELECT * FROM test_cases WHERE case_type='security' ORDER BY module_id, name"
+    )
+    rows = [dict(r) for r in cur.fetchall()]
+    conn.close()
+    # JSON 列（SQLite 存 TEXT）必须先解析，否则 json.dumps 会双重编码
+    for row in rows:
+        for col in ("tags", "steps", "config", "depends_on", "parameters", "condition"):
+            row[col] = _parse_json_field(row.get(col))
+    print(f"  本地安全用例: {len(rows)} 个")
+    return rows
+
+
+def load_local_file(path):
+    with open(path, "rb") as f:
+        return f.read()
+
+
+def remote_exec(cli, cmd, timeout=60):
+    stdin, stdout, stderr = cli.exec_command(cmd, timeout=timeout)
+    out = stdout.read().decode("utf-8", errors="replace").strip()
+    err = stderr.read().decode("utf-8", errors="replace").strip()
+    return out, err
+
+
+def sync_backend_files():
+    """通过 SFTP 上传后端文件到服务器"""
+    cli = paramiko.SSHClient()
+    cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    cli.connect(SERVER[0], username=SERVER[1], password=SERVER[2], timeout=15)
+    sftp = cli.open_sftp()
+    try:
+        for local_rel, remote_rel in BACKEND_FILES.items():
+            local_path = local_rel
+            remote_path = f"{REMOTE_BACKEND}/{remote_rel}"
+            data = load_local_file(local_path)
+            # 确保远程目录存在
+            remote_dir = "/".join(remote_path.split("/")[:-1])
+            try:
+                sftp.stat(remote_dir)
+            except FileNotFoundError:
+                parent = "/".join(remote_dir.split("/")[:-1])
+                try:
+                    sftp.mkdir(remote_dir)
+                except IOError:
+                    pass
+            with sftp.open(remote_path, "wb") as f:
+                f.write(data)
+            print(f"  ✅ 上传 {local_rel} → {remote_path} ({len(data)} bytes)")
+    finally:
+        sftp.close()
+        cli.close()
+
+
+def sync_cases_to_mysql(rows):
+    """将本地安全用例同步到服务器 MySQL"""
+    conn = pymysql.connect(
+        host=SERVER[0], port=MYSQL_PORT, user=MYSQL_USER,
+        password=MYSQL_PWD, database=MYSQL_DB, connect_timeout=15,
+        charset="utf8mb4",
+    )
+    cur = conn.cursor()
+    inserted = updated = 0
+    for row in rows:
+        cur.execute(
+            "SELECT id FROM test_cases WHERE id=%s", (row["id"],)
+        )
+        exists = cur.fetchone() is not None
+        cur.execute(
+            INSERT_SQL,
+            (
+                row["id"], row["module_id"], row["name"], row["description"],
+                row["status"], row["priority"], json.dumps(row["tags"], ensure_ascii=False),
+                json.dumps(row["steps"], ensure_ascii=False),
+                json.dumps(row["config"] or {}, ensure_ascii=False),
+                row["depends_on"], row["parameters"], row["condition"],
+                row["order"], row["version"],
+                row["created_at"], row["updated_at"],
+                row["case_type"], row["project_type"],
+            ),
+        )
+        if exists:
+            updated += 1
+        else:
+            inserted += 1
+    conn.commit()
+    cur.close()
+    conn.close()
+    print(f"  ✅ MySQL 同步完成: 新增 {inserted} / 更新 {updated}")
+
+
+def restart_container():
+    cli = paramiko.SSHClient()
+    cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    cli.connect(SERVER[0], username=SERVER[1], password=SERVER[2], timeout=15)
+    out, err = remote_exec(cli, f"docker restart {CONTAINER}")
+    print(f"  容器重启: {out or err or 'OK'}")
+    time.sleep(8)
+    # 等待容器 healthy
+    for _ in range(10):
+        out, _ = remote_exec(cli, "docker inspect -f '{{.State.Health.Status}}' " + CONTAINER)
+        status = out.strip()
+        if status == "healthy":
+            print("  ✅ 容器 healthy")
+            break
+        time.sleep(3)
+    cli.close()
+
+
+def verify_remote():
+    """验证服务器端用例总数与新增用例"""
+    INNER = """
+import pymysql
+c = pymysql.connect(host='mysql', user='platapp', password='PlatApp2026',
+                    database='plat_auto_test', connect_timeout=8)
+cur = c.cursor()
+cur.execute("SELECT COUNT(*) FROM test_cases WHERE case_type='security'")
+print("total security cases:", cur.fetchone()[0])
+cur.execute("SELECT COUNT(*) FROM test_cases WHERE case_type='security' AND id LIKE 'sec_2_8%'")
+print("sec_2_8 cases:", cur.fetchone()[0])
+c.close()
+"""
+    cli = paramiko.SSHClient()
+    cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    cli.connect(SERVER[0], username=SERVER[1], password=SERVER[2], timeout=20)
+    out, err = remote_exec(
+        cli,
+        f"docker exec -i {CONTAINER} python - <<'PYEOF'\n{INNER}\nPYEOF",
+        timeout=90,
+    )
+    print(out if out else f"(无输出) err: {err[:400]}")
+    cli.close()
+
+
+def main():
+    print("=" * 60)
+    print("安全测试用例同步部署到 5.60")
+    print("=" * 60)
+    rows = load_local_cases()
+    if not rows:
+        print("❌ 本地无安全用例")
+        return
+    print("\n[1/5] 上传后端文件...")
+    sync_backend_files()
+    print("\n[2/5] 重启容器（触发 _ensure_columns 自动迁移 project_type）...")
+    restart_container()
+    print("\n[3/5] 同步用例到 MySQL...")
+    sync_cases_to_mysql(rows)
+    print("\n[4/5] 再次重启容器（加载同步后的用例数据）...")
+    restart_container()
+    print("\n[5/5] 远程验证...")
+    verify_remote()
+    print("\n" + "=" * 60)
+    print("部署完成")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file

backend/scripts/fix_560_json_double_encoding.py

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+修复脚本：修复 5.60 服务器 test_cases 表 JSON 双重编码数据
+
+背景：
+  deploy_security_cases_560.py 从 SQLite 原生读取 JSON 列时得到的是 JSON 文本字符串，
+  脚本又对其 json.dumps() 造成双重编码，导致 MySQL JSON 列存储的是"字符串值"
+  而非数组/对象，API 读取后 Pydantic 校验报
+  "4 validation errors for TestCaseResponse (tags/steps/config: input_type=str)"。
+
+修复方式：
+  对所有 case_type='security' 行的 tags/steps/config 字段反复 json.loads
+  直到得到非字符串对象，再以 JSON 文本写回（MySQL JSON 列自动解析存储）。
+
+用法:
+    cd backend && python scripts/fix_560_json_double_encoding.py
+"""
+import json
+import sys
+
+sys.stdout.reconfigure(encoding="utf-8")
+
+import pymysql
+
+HOST, PORT, USER, PWD, DB = "192.168.5.60", 3307, "platapp", "PlatApp2026", "plat_auto_test"
+JSON_COLS = ("tags", "steps", "config")
+
+
+def normalize(value):
+    """反复解析嵌套编码的 JSON 文本，直到得到非字符串对象"""
+    if not isinstance(value, str):
+        return value, 0
+    depth = 0
+    current = value
+    while isinstance(current, str) and depth < 5:
+        try:
+            current = json.loads(current)
+        except (json.JSONDecodeError, ValueError):
+            break
+        depth += 1
+    return current, depth
+
+
+def main():
+    conn = pymysql.connect(host=HOST, port=PORT, user=USER, password=PWD,
+                           database=DB, connect_timeout=15, charset="utf8mb4")
+    cur = conn.cursor()
+    cur.execute(
+        "SELECT id, tags, steps, config FROM test_cases WHERE case_type='security'"
+    )
+    rows = cur.fetchall()
+    print(f"待检查安全用例: {len(rows)} 条")
+
+    fixed = 0
+    for case_id, tags, steps, config in rows:
+        updates, dirty = {}, False
+        for col, raw in zip(JSON_COLS, (tags, steps, config)):
+            obj, depth = normalize(raw)
+            if depth > 0:  # 至少解开了一层双重编码
+                updates[col] = json.dumps(obj, ensure_ascii=False)
+                dirty = True
+        if dirty:
+            set_clause = ", ".join(f"{c}=%s" for c in updates)
+            cur.execute(
+                f"UPDATE test_cases SET {set_clause} WHERE id=%s",
+                (*updates.values(), case_id),
+            )
+            fixed += 1
+
+    conn.commit()
+    print(f"✅ 修复双重编码: {fixed} 条")
+
+    # 验证：抽样确认字段类型
+    cur.execute(
+        "SELECT id, "
+        "JSON_TYPE(tags), JSON_TYPE(steps), JSON_TYPE(config) "
+        "FROM test_cases WHERE case_type='security'"
+    )
+    bad = 0
+    for case_id, t, s, c in cur.fetchall():
+        types = (t, s, c)
+        if any(x not in ("ARRAY", "OBJECT") for x in types):
+            print(f"  ⚠️ 仍异常: {case_id} tags={t} steps={s} config={c}")
+            bad += 1
+    print(f"验证: {'全部正常 (ARRAY/OBJECT)' if bad == 0 else f'{bad} 条仍异常'}")
+
+    cur.close()
+    conn.close()
+
+
+if __name__ == "__main__":
+    main()