workphone-sdk/sdk/scripts/wechat_interface_audit.py

#!/usr/bin/env python3
"""
微信 Hook 接口完整性审计 — 禁止空接口/占位冒充

检查项：
1. WECHAT_ACTIONS ↔ ACTION_TO_RPC ↔ wechat_hook_v2.js rpc.exports 三方对齐
2. ACTION_ALIASES（173 catalog）→ canonical 均在 WECHAT_ACTIONS
3. （可选）真机 hook/execute 探针 / 全量 128 探针

用法:
  python3 sdk/scripts/wechat_interface_audit.py
  python3 sdk/scripts/wechat_interface_audit.py --probe -d xgfe65eimrrofyws
  python3 sdk/scripts/wechat_interface_audit.py --probe-full -d xgfe65eimrrofyws
"""

from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[2]
JS = ROOT / "sdk/agent/hook/wechat_hook_v2.js"
HE = ROOT / "sdk/agent/hook/hook_executor.py"
SKILL = ROOT / "sdk/app/skills/wechat/skill_v2.py"
OUT = ROOT / "sdk/tmp"


def _load_rpc_keys() -> set[str]:
    text = JS.read_text(encoding="utf-8", errors="ignore")
    return set(re.findall(r"^\s+(\w+)\s*:\s*function", text, re.M))


def _load_action_rpc() -> dict[str, str]:
    text = HE.read_text(encoding="utf-8", errors="ignore")
    block = text.split("ACTION_TO_RPC:", 1)[1].split("ACTION_ALIASES:", 1)[0]
    return dict(re.findall(r'"([a-z_]+)"\s*:\s*"(\w+)"', block))


def _load_action_aliases() -> dict[str, str]:
    text = HE.read_text(encoding="utf-8", errors="ignore")
    block = text.split("ACTION_ALIASES:", 1)[1].split("\n\n", 1)[0]
    return dict(re.findall(r'"([a-z_]+)"\s*:\s*"([a-z_]+)"', block))


def _load_wechat_actions() -> dict[str, str]:
    text = SKILL.read_text(encoding="utf-8", errors="ignore")
    block = text.split("WECHAT_ACTIONS = {", 1)[1].split("\n}\n", 1)[0]
    return dict(re.findall(r'"([a-z_]+)"\s*:\s*"(\w+)"', block))


def audit() -> dict:
    rpc = _load_rpc_keys()
    action_rpc = _load_action_rpc()
    aliases = _load_action_aliases()
    skill_actions = _load_wechat_actions()
    catalog = sorted(set(action_rpc) | set(aliases))

    missing_rpc = sorted(set(action_rpc.values()) - rpc)
    skill_not_in_executor = sorted(set(skill_actions) - set(action_rpc))
    # diag_* 为 Hook 内部联调诊断项（JS 有真实现），不进对外 130 action 清单，豁免对齐
    executor_not_in_skill = sorted(
        a for a in set(action_rpc) - set(skill_actions)
        if not a.startswith("diag_")
    )
    skill_rpc_mismatch = [
        a for a, r in skill_actions.items()
        if a in action_rpc and action_rpc[a] != r
    ]
    alias_unresolved = sorted(
        a for a, target in aliases.items()
        if target not in skill_actions
    )

    ok = not (
        missing_rpc
        or skill_not_in_executor
        or skill_rpc_mismatch
        or executor_not_in_skill
        or alias_unresolved
    )

    return {
        "timestamp": datetime.now().isoformat(),
        "rpc_exports": len(rpc),
        "action_to_rpc": len(action_rpc),
        "wechat_actions": len(skill_actions),
        "action_aliases": len(aliases),
        "catalog_actions": len(catalog),
        "missing_rpc_in_js": missing_rpc,
        "skill_not_in_executor": skill_not_in_executor,
        "executor_not_in_skill": executor_not_in_skill,
        "skill_rpc_mismatch": skill_rpc_mismatch,
        "alias_unresolved": alias_unresolved,
        "complete": ok,
    }


def _probe_actions(device_id: str, base: str, actions: list[str], params_map: dict) -> list[dict]:
    import httpx

    rows = []
    with httpx.Client(timeout=60) as c:
        for action in actions:
            params = dict(params_map.get(action, {}))
            r = c.post(
                f"{base}/api/v3/hook/execute",
                json={
                    "device_id": device_id,
                    "platform": "wechat",
                    "action": action,
                    "params": params,
                    "hook_only": True,
                },
            )
            data = r.json()
            ch = data.get("channel_used") or data.get("_channel_used") or data.get("channel") or ""
            inner = data.get("data") if isinstance(data.get("data"), dict) else {}
            ok = bool(inner.get("success") or data.get("success")) and "offline" not in str(ch).lower()
            if inner.get("success") is False:
                ok = False
            rows.append({"action": action, "ok": ok, "channel": ch, "error": inner.get("error") or inner.get("message")})
    return rows


def probe(device_id: str, base: str, limit: int = 20) -> list[dict]:
    action_rpc = _load_action_rpc()
    safe = ["ping", "get_wechat_version", "get_hook_status", "get_contacts", "get_profile", "get_messages"]
    actions = [a for a in safe if a in action_rpc][:limit]
    params = {
        "get_messages": {"conversation_id": "filehelper", "limit": 3},
        "get_contacts": {"limit": 5},
    }
    return _probe_actions(device_id, base, actions, params)


def probe_full(device_id: str, base: str) -> list[dict]:
    sys.path.insert(0, str(ROOT / "sdk" / "scripts"))
    from matrix_hook_catalog_verify import SAFE_PARAMS  # type: ignore

    actions = sorted(_load_wechat_actions().keys())
    return _probe_actions(device_id, base, actions, SAFE_PARAMS)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--probe", action="store_true", help="6 项安全探针")
    parser.add_argument("--probe-full", action="store_true", help="128 canonical 全量探针")
    parser.add_argument("-d", "--device-id", default="xgfe65eimrrofyws")
    parser.add_argument("--base", default="http://127.0.0.1:8899")
    args = parser.parse_args()

    report = audit()
    OUT.mkdir(parents=True, exist_ok=True)
    out = OUT / f"wechat_interface_audit_{int(datetime.now().timestamp())}.json"
    if args.probe or args.probe_full:
        rows = probe_full(args.device_id, args.base) if args.probe_full else probe(args.device_id, args.base)
        report["probe"] = rows
        report["probe_pass"] = sum(1 for x in rows if x["ok"])
        report["probe_total"] = len(rows)
    out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")

    print(f"rpc.exports: {report['rpc_exports']}")
    print(f"ACTION_TO_RPC: {report['action_to_rpc']}")
    print(f"WECHAT_ACTIONS: {report['wechat_actions']}")
    print(f"Catalog(含alias): {report['catalog_actions']}")
    print(f"缺失 RPC: {len(report['missing_rpc_in_js'])}")
    print(f"Skill 未映射 executor: {len(report['skill_not_in_executor'])}")
    print(f"Alias 未解析: {len(report.get('alias_unresolved', []))}")
    print(f"完整对齐: {'✅' if report['complete'] else '❌'}")
    print(f"报告: {out}")
    if args.probe or args.probe_full:
        print(f"真机探针: {report.get('probe_pass', 0)}/{report.get('probe_total', 0)}")
    return 0 if report["complete"] else 1


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