cad-agent/verify_mcp_fault_injection.py

"""故障注入验证：畸形 / 类型错误的请求不得让 CAD 崩溃，应回协议错误或 result.isError。

背景
----
cad_mcp_frame 的工具执行发生在 AutoCAD 主线程的窗口消息处理链上
（TcpServerLoop -> PostMessage -> WndProc -> ExecuteTaskInMainThread -> mcp_Tools::Call）。
插件的 Execute 在参数类型不符时会抛 nlohmann::type_error，若异常逃逸出 WndProc
属未定义行为（实测表现为 CAD 崩溃），且会让 DLL 代理的文档写锁无法释放。

本脚本注入各类畸形参数，验证：
  1. 结构性错误（params/arguments/name 类型不对、请求不是对象、非法 JSON）
     -> 回协议错误（-32700 / -32600 / -32602），而不是崩溃；
  2. 工具内部因类型不符抛出的异常 -> 被兜住并映射成 result.isError = true；
  3. 全部注入之后 CAD 仍存活，且文档写锁未泄漏（有效写操作仍能成功）。

前置条件：CAD 已启动并加载 cad_mcp_frame.arx，监听 127.0.0.1:8080。
运行：python verify_mcp_fault_injection.py
"""

import json
import re
import socket
import sys

from dwg_review import config

PASS = 0
FAIL = 0

_MISSING = object()


def check(name, cond, detail=""):
    global PASS, FAIL
    if cond:
        PASS += 1
        print(f"[PASS] {name}")
    else:
        FAIL += 1
        print(f"[FAIL] {name}  {detail!r}")


def send_raw(line, expect_reply=True):
    """向基座发送一帧原始字节，返回解析后的响应 dict（无响应返回 None）。"""
    sock = socket.create_connection((config.HOST, config.PORT), timeout=15)
    try:
        sock.sendall(line)
        if not expect_reply:
            return None
        buf = b""
        while b"\n" not in buf:
            chunk = sock.recv(65536)
            if not chunk:
                break
            buf += chunk
        text = buf.decode("utf-8", errors="replace").strip()
        return json.loads(text) if text else None
    finally:
        sock.close()


def rpc(method, params=_MISSING, rid=1):
    msg = {"jsonrpc": "2.0", "id": rid, "method": method}
    if params is not _MISSING:
        msg["params"] = params
    return send_raw(json.dumps(msg, ensure_ascii=False).encode("utf-8") + b"\n")


def call_tool(name, arguments, rid=1):
    return rpc("tools/call", {"name": name, "arguments": arguments}, rid=rid)


def err_code(resp):
    return ((resp or {}).get("error") or {}).get("code")


def err_msg(resp):
    return ((resp or {}).get("error") or {}).get("message", "")


def is_tool_error(resp):
    result = (resp or {}).get("result") or {}
    return result.get("isError") is True


def result_text(resp):
    content = ((resp or {}).get("result") or {}).get("content") or []
    for block in content:
        if block.get("type") == "text":
            return block.get("text", "")
    return ""


def main():
    print("=" * 68)
    print(" 故障注入验证：畸形请求不得崩溃 CAD")
    print("=" * 68)

    # ---- A. 结构性错误：应为顶层协议错误 -------------------------------
    r = rpc("tools/call", "abc")
    check("A1 params 非对象 -> -32602", err_code(r) == -32602, r)
    check("A1 消息说明 params 必须为对象", "params must be an object" in err_msg(r), r)

    r = rpc("tools/call", {"name": "get_drawing_info", "arguments": "abc"})
    check("A2 arguments 非对象 -> -32602", err_code(r) == -32602, r)
    check("A2 消息说明 arguments 必须为对象", "arguments must be an object" in err_msg(r), r)

    r = rpc("tools/call", {"name": 123, "arguments": {}})
    check("A3 name 非字符串 -> -32602", err_code(r) == -32602, r)
    check("A3 消息说明 name 必须为字符串", "name must be a string" in err_msg(r), r)

    r = rpc("tools/call", {"name": "__no_such_tool__", "arguments": {}})
    check("A4 未知工具 -> -32602", err_code(r) == -32602, r)
    check("A4 消息为 Unknown tool", err_msg(r).startswith("Unknown tool"), r)

    r = rpc("tools/call", {"arguments": {}})
    check("A5 缺 name -> -32602", err_code(r) == -32602, r)

    r = rpc("tools/call", {"name": "get_drawing_info", "arguments": []})
    check("A6 arguments 为数组 -> -32602", err_code(r) == -32602, r)

    r = send_raw(b"[1,2,3]\n")
    check("A7 请求不是对象 -> -32600", err_code(r) == -32600, r)

    r = send_raw(b"{not valid json\n")
    check("A8 非法 JSON -> -32700", err_code(r) == -32700, r)

    # ---- B. 工具内部类型异常：应被兜住为 result.isError -----------------
    # B1 bulge 传字符串：args.value("bulge", 0.5) 在类型不符时抛 type_error
    r = call_tool("draw_revision_cloud", {
        "shape": "rect", "bulge": "x", "min_x": 0, "min_y": 0, "max_x": 10, "max_y": 10,
    }, rid=20)
    check("B1 draw_revision_cloud bulge='x' -> isError", is_tool_error(r), r)
    check("B1 错误信息含类型提示", "type" in result_text(r).lower() or "execution failed" in result_text(r).lower(), result_text(r))

    # B2 points 内层元素非数字：parsePoint 只查 is_array/size，取 get<double>() 时抛异常
    r = call_tool("add_leader_annotation", {
        "text": "t", "points": [["a", "b"], ["c", "d"]],
    }, rid=21)
    check("B2 add_leader_annotation points 内层非数字 -> isError", is_tool_error(r), r)

    # B3 add_line 端点元素非数字
    r = call_tool("add_line", {"start": ["a", "b"], "end": [0, 0]}, rid=22)
    check("B3 add_line start 元素非数字 -> isError", is_tool_error(r), r)

    # B4 对照：缺参数的优雅处理（本就走 kErrBadParam，非异常路径）
    r = call_tool("add_leader_annotation", {"text": "t", "points": ["a", "b"]}, rid=23)
    check("B4 对照：points 元素非数组 -> isError（优雅处理）", is_tool_error(r), r)

    # B5 非对象 arguments 直接指向会 value() 的工具（校验层已挡下，应回 -32602）
    r = call_tool("draw_revision_cloud", "abc", rid=24)
    check("B5 会抛异常的工具 + arguments 非对象 -> -32602", err_code(r) == -32602, r)

    # ---- C. 反复注入后再确认存活与写锁 ---------------------------------
    for i in range(5):
        call_tool("draw_revision_cloud", {"shape": "rect", "bulge": "x",
                                          "min_x": 0, "min_y": 0, "max_x": 1, "max_y": 1}, rid=100 + i)
        call_tool("add_line", {"start": ["a", "b"], "end": [0, 0]}, rid=200 + i)

    r = rpc("tools/list", {}, rid=300)
    tools = ((r or {}).get("result") or {}).get("tools") or []
    check("C1 反复注入后 tools/list 仍正常（26 个工具）", len(tools) == 26, f"got {len(tools)}")

    r = call_tool("get_drawing_info", {}, rid=301)
    check("C2 反复注入后只读工具仍正常（CAD 存活）",
          (r or {}).get("result") is not None and not is_tool_error(r), r)

    # C3 写锁完整性：若前面某次异常泄漏了文档写锁，写操作会拿不到模型空间而失败
    r = call_tool("add_line", {"start": [0.0, 0.0], "end": [10.0, 10.0]}, rid=302)
    check("C3 反复注入后写操作仍成功（写锁未泄漏）",
          (r or {}).get("result") is not None and not is_tool_error(r), r)

    # 从返回文本中取出本次创建的实体 handle 并清理
    created_handles = []
    text = result_text(r)
    try:
        payload = json.loads(text)
        created_handles = [item["handle"] for item in payload.get("created", []) if item.get("handle")]
    except (ValueError, AttributeError, TypeError):
        m = (re.search(r'"handle"\s*:\s*"([0-9A-Fa-f]+)"', text)
             or re.search(r"handle[=:\s]+([0-9A-Fa-f]+)", text))
        if m:
            created_handles = [m.group(1)]

    if created_handles:
        cleanup = call_tool("delete_entities", {"handles": created_handles}, rid=303)
        check(f"C4 清理测试实体 {created_handles}",
              (cleanup or {}).get("result") is not None and not is_tool_error(cleanup), cleanup)
    else:
        print(f"[INFO] 未能从返回文本解析 handle，跳过清理：{text!r}")

    # ---- 汇总 ----------------------------------------------------------
    print("-" * 68)
    print(f"通过 {PASS} / 失败 {FAIL}")
    if FAIL:
        print("结果：FAIL")
        return 1
    print("结果：全部通过")
    return 0


if __name__ == "__main__":
    sys.exit(main())