190 lines
7.8 KiB
Python
Executable File
190 lines
7.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""无线微信安全能力矩阵验收。
|
||
|
||
默认只执行低风险读类与 filehelper 白名单消息；高风险、资金、账号破坏类动作只登记 gated。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import requests
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
EVID_ROOT = ROOT / "开发文档" / "8、部署" / "05-测试验收"
|
||
|
||
READ_ACTIONS: dict[str, dict[str, Any]] = {
|
||
"ping": {},
|
||
"get_hook_status": {},
|
||
"get_wechat_version": {},
|
||
"get_profile": {},
|
||
"get_device_info": {},
|
||
"get_storage_info": {},
|
||
"get_network_info": {},
|
||
"get_process_info": {},
|
||
"check_login_state": {},
|
||
"get_contacts": {"limit": 10},
|
||
"search_contacts": {"keyword": "文件", "limit": 5},
|
||
"get_messages": {"conversation_id": "filehelper", "limit": 5},
|
||
"get_recent_messages": {"limit": 5},
|
||
"search_messages": {"keyword": "验收", "limit": 5},
|
||
"get_groups": {"limit": 5},
|
||
"get_labels": {},
|
||
"get_moments": {"limit": 5},
|
||
"get_friend_requests": {},
|
||
"get_favorites": {"limit": 5},
|
||
"get_official_accounts": {"limit": 5},
|
||
}
|
||
|
||
SAFE_WRITE_ACTIONS: dict[str, dict[str, Any]] = {
|
||
"send_message": {
|
||
"to_id": "filehelper",
|
||
"content": f"[工作手机安全矩阵] {datetime.now().strftime('%Y%m%d-%H%M%S')}",
|
||
"msg_type": "text",
|
||
},
|
||
}
|
||
|
||
GATED_ACTIONS: dict[str, str] = {
|
||
"batch_send": "批量触达：需要 WORKPHONE_TEST_TO_IDS 白名单，默认 dry-run，真实发送按 65s+。",
|
||
"mass_send": "群发：需要白名单测试联系人、内容去重、限流与停止条件。",
|
||
"add_friend": "主动加好友：需要测试 wxid/手机号白名单，默认 180s+，新号限制更严。",
|
||
"batch_add_friend": "批量加好友：需要白名单与用户确认，不自动对真实客户执行。",
|
||
"accept_friend": "通过好友：需要真实 pending 请求样本。",
|
||
"create_group": "建群：需要白名单成员，避免真实客户被拉群。",
|
||
"invite_to_group": "邀请入群：需要测试群与测试成员。",
|
||
"send_group_message": "群消息：需要测试群 group_id。",
|
||
"post_moments": "发朋友圈：受 2h+ 限流，本轮已有 P0 验收，不重复发布。",
|
||
"like_moments": "朋友圈点赞：需要真实 sns_id，按 15-45s 节律。",
|
||
"comment_moments": "朋友圈评论：需要真实 sns_id，按 60-180s 节律。",
|
||
"send_red_packet": "资金类：测试号、小额、二次确认、审计；默认禁自动跑。",
|
||
"send_transfer": "资金类：测试号、小额、二次确认、审计；默认禁自动跑。",
|
||
"receive_red_packet": "资金类：需要真实红包消息样本与人工确认。",
|
||
"receive_transfer": "资金类：需要真实转账消息样本与人工确认。",
|
||
"receive_payment": "资金类：需要测试场景与人工确认。",
|
||
"show_payment_code": "付款码：敏感 UI，默认不自动展示。",
|
||
"logout": "破坏性账号动作：默认不执行。",
|
||
"switch_account": "账号切换：默认不执行。",
|
||
"login_by_password": "登录动作：需要测试账号凭证与人工确认。",
|
||
}
|
||
|
||
|
||
def hook_execute(base: str, device: str, action: str, params: dict[str, Any]) -> dict[str, Any]:
|
||
started = time.time()
|
||
try:
|
||
resp = requests.post(
|
||
f"{base}/api/v3/hook/execute",
|
||
json={
|
||
"device_id": device,
|
||
"platform": "wechat",
|
||
"action": action,
|
||
"params": params,
|
||
"hook_only": True,
|
||
},
|
||
timeout=90,
|
||
)
|
||
body = resp.json()
|
||
except Exception as exc:
|
||
return {
|
||
"action": action,
|
||
"status": "failed",
|
||
"elapsed_ms": int((time.time() - started) * 1000),
|
||
"error": str(exc),
|
||
}
|
||
|
||
data = body.get("data") if isinstance(body.get("data"), dict) else body
|
||
success = bool(data.get("success")) if isinstance(data, dict) and "success" in data else body.get("code") == 200
|
||
return {
|
||
"action": action,
|
||
"status": "passed" if success else "failed",
|
||
"elapsed_ms": int((time.time() - started) * 1000),
|
||
"channel": body.get("channel_used") or data.get("channel") if isinstance(data, dict) else body.get("channel_used"),
|
||
"code": body.get("code"),
|
||
"error": "" if success else str(data.get("error") or data.get("message") or body)[:300],
|
||
"summary": summarize_data(data),
|
||
}
|
||
|
||
|
||
def summarize_data(data: Any) -> dict[str, Any]:
|
||
if not isinstance(data, dict):
|
||
return {"type": type(data).__name__}
|
||
summary: dict[str, Any] = {}
|
||
for key in ("success", "wechat_version", "version", "channel", "action_resolved"):
|
||
if key in data:
|
||
summary[key] = data[key]
|
||
for key in ("contacts", "messages", "groups", "labels", "moments", "requests", "favorites", "accounts"):
|
||
val = data.get(key)
|
||
if isinstance(val, list):
|
||
summary[f"{key}_count"] = len(val)
|
||
if isinstance(data.get("profile"), dict):
|
||
summary["profile_present"] = True
|
||
if data.get("message_id"):
|
||
summary["message_id_present"] = True
|
||
return summary
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("-d", "--device-id", default="xgfe65eimrrofyws")
|
||
parser.add_argument("--base", default="http://127.0.0.1:8899")
|
||
parser.add_argument("--include-safe-write", action="store_true")
|
||
args = parser.parse_args()
|
||
|
||
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||
out_dir = EVID_ROOT / f"{datetime.now().strftime('%Y%m%d')}_微信安全矩阵验收"
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
probe = requests.get(f"{args.base}/api/v3/hook/probe/{args.device_id}", timeout=60).json()
|
||
rows: list[dict[str, Any]] = []
|
||
if not probe.get("supports_hook"):
|
||
report = {"timestamp": stamp, "device_id": args.device_id, "probe": probe, "results": []}
|
||
path = out_dir / f"safe_matrix_{stamp}.json"
|
||
path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"Hook 未就绪，报告 {path}")
|
||
return 2
|
||
|
||
for action, params in READ_ACTIONS.items():
|
||
row = hook_execute(args.base, args.device_id, action, params)
|
||
rows.append(row)
|
||
print(f"{'✓' if row['status'] == 'passed' else '✗'} {action:<24} {row['status']} {row['elapsed_ms']}ms")
|
||
time.sleep(0.8)
|
||
|
||
if args.include_safe_write:
|
||
for action, params in SAFE_WRITE_ACTIONS.items():
|
||
row = hook_execute(args.base, args.device_id, action, params)
|
||
rows.append(row)
|
||
print(f"{'✓' if row['status'] == 'passed' else '✗'} {action:<24} {row['status']} {row['elapsed_ms']}ms")
|
||
time.sleep(1.5)
|
||
|
||
for action, reason in GATED_ACTIONS.items():
|
||
rows.append({"action": action, "status": "gated", "reason": reason})
|
||
|
||
passed = sum(1 for row in rows if row["status"] == "passed")
|
||
failed = sum(1 for row in rows if row["status"] == "failed")
|
||
gated = sum(1 for row in rows if row["status"] == "gated")
|
||
report = {
|
||
"timestamp": datetime.now().isoformat(),
|
||
"device_id": args.device_id,
|
||
"base": args.base,
|
||
"probe": {
|
||
"supports_hook": probe.get("supports_hook"),
|
||
"wechat_version": probe.get("wechat_version"),
|
||
"transport": probe.get("transport"),
|
||
},
|
||
"summary": {"passed": passed, "failed": failed, "gated": gated, "total": len(rows)},
|
||
"results": rows,
|
||
}
|
||
path = out_dir / f"safe_matrix_{stamp}.json"
|
||
path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"\n安全矩阵: passed={passed} failed={failed} gated={gated} → {path}")
|
||
return 0 if failed == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|