182 lines
6.7 KiB
Python
182 lines
6.7 KiB
Python
#!/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())
|