0ec156f303
补记 Phantom AF9 真机验证、矩阵 E2E 报告路径；catalog 探针补 batch_send/mass_send 参数。 Co-authored-by: Cursor <cursoragent@cursor.com>
273 lines
12 KiB
Python
273 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""矩阵补全：Hook catalog 全 action 真机 hook/execute 探针。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
|
||
BASE = os.environ.get("SDK_BASE_URL", "http://127.0.0.1:8899")
|
||
DEVICE = os.environ.get("SDK_DEVICE_ID", "xgfe65eimrrofyws")
|
||
TO = "filehelper"
|
||
OUT = Path(__file__).resolve().parents[1] / "tmp"
|
||
|
||
SAFE_PARAMS: dict[str, dict] = {
|
||
"send_message": {"to_id": TO, "content": "[catalog-probe]", "msg_type": "text"},
|
||
"batch_send": {"user_ids": [TO], "content": "[catalog-probe]"},
|
||
"mass_send": {"user_ids": [TO], "content": "[catalog-probe]"},
|
||
"get_messages": {"conversation_id": TO, "limit": 3},
|
||
"get_contacts": {"limit": 5},
|
||
"get_contact_info": {"wxid": TO},
|
||
"get_friend_info": {"user_id": TO},
|
||
"search_contacts": {"keyword": "文件", "limit": 5},
|
||
"search_contact": {"keyword": "文件", "limit": 5},
|
||
"search_messages": {"keyword": "test", "limit": 5},
|
||
"wechat_search": {"keyword": "微信", "limit": 5},
|
||
"get_groups": {"limit": 5},
|
||
"get_group_info": {"group_id": "test@chatroom"},
|
||
"get_group_members": {"group_id": "test@chatroom"},
|
||
"get_profile": {},
|
||
"check_account_status": {},
|
||
"get_favorites": {"limit": 5},
|
||
"get_labels": {},
|
||
"get_moments": {"limit": 3},
|
||
"browse_channels": {"limit": 3},
|
||
"global_search": {"keyword": "微信", "limit": 5},
|
||
"generate_my_qr_code": {},
|
||
"generate_group_qr_code": {"group_id": "test@chatroom"},
|
||
"get_wallet_balance": {},
|
||
"get_transaction_history": {"limit": 5},
|
||
"get_wechat_steps": {},
|
||
"get_sticker_list": {},
|
||
"get_top_stories": {},
|
||
"get_safety_center": {},
|
||
"check_restrictions": {},
|
||
"show_payment_code": {},
|
||
"pin_chat": {"wxid": TO, "pin": False},
|
||
"set_chat_top": {"user_id": TO, "enable": False},
|
||
"set_do_not_disturb": {"wxid": TO, "enable": False},
|
||
"add_favorite": {"content": "catalog-probe", "type": "text"},
|
||
"add_to_favorites": {"content": "catalog-probe", "type": "text"},
|
||
"set_privacy": {"setting": "moments_privacy", "value": "all"},
|
||
"set_moments_privacy": {"privacy_type": "all", "days": 180},
|
||
"set_moments_cover": {"image_path": "/sdcard/test.jpg"},
|
||
"open_mini_program": {"name": "微信支付"},
|
||
"open_miniprogram": {"name": "微信支付"},
|
||
"send_location": {"to_id": TO, "latitude": 24.48, "longitude": 118.08, "label": "probe"},
|
||
"send_emoji": {"to_id": TO, "emoji_name": "微笑"},
|
||
"send_voice": {"to_id": TO, "duration": 1},
|
||
"send_voice_message": {"to_id": TO, "duration": 1},
|
||
"send_file": {"to_id": TO, "file_path": "/sdcard/test.txt"},
|
||
"send_file_from_chat": {"to_id": TO, "file_path": "/sdcard/test.txt"},
|
||
"send_image": {"to_id": TO, "image_path": "/sdcard/test.jpg"},
|
||
"send_video": {"to_id": TO, "video_path": "/sdcard/test.mp4"},
|
||
"send_link": {"to_id": TO, "url": "https://example.com", "title": "probe"},
|
||
"send_card": {"to_id": TO, "card_wxid": TO},
|
||
"send_group_message": {"group_id": "test@chatroom", "content": "probe"},
|
||
"like_channel_video": {"index": 0},
|
||
"like_video": {"index": 0},
|
||
"like_wechat_steps": {"user_id": TO},
|
||
"like_moments": {"sns_id": "probe_sns"},
|
||
"comment_moments": {"sns_id": "probe_sns", "comment": "probe"},
|
||
"comment_channel_video": {"video_id": "index_0", "comment": "probe"},
|
||
"clear_cache": {},
|
||
"check_for_update": {},
|
||
"post_moments": {"content": "[catalog-probe]"},
|
||
"forward_moments_link": {"content": "https://example.com probe"},
|
||
"set_signature": {"signature": "catalog-probe"},
|
||
"set_nickname": {"nickname": "probe"},
|
||
"set_region": {"region": "福建 厦门"},
|
||
"set_avatar": {"image_path": "/sdcard/test.jpg"},
|
||
"set_remark": {"user_id": TO, "remark": "probe"},
|
||
"set_friend_remark": {"wxid": TO, "remark": "probe"},
|
||
"create_label": {"name": "catalog_probe"},
|
||
"create_tag": {"tag_name": "catalog_probe"},
|
||
"get_contacts_by_label": {"label_id": "1"},
|
||
"get_users_by_tag": {"tag_name": "catalog_probe"},
|
||
"add_friend": {"user_id": "28533368", "message": "catalog-probe"},
|
||
"accept_friend": {"encrypt_username": "probe", "ticket": "probe"},
|
||
"delete_friend": {"wxid": "invalid_wxid_probe"},
|
||
"invite_to_group": {"group_id": "test@chatroom", "member_ids": [TO]},
|
||
"remove_from_group": {"group_id": "test@chatroom", "member_ids": [TO]},
|
||
"create_group": {"member_ids": [TO, TO], "group_name": "catalog-probe"},
|
||
"set_group_announcement": {"group_id": "test@chatroom", "announcement": "probe"},
|
||
"set_group_notice": {"group_id": "test@chatroom", "notice": "probe"},
|
||
"set_group_name": {"group_id": "test@chatroom", "name": "probe"},
|
||
"quit_group": {"group_id": "test@chatroom"},
|
||
"set_contact_label": {"wxid": TO, "label_ids": []},
|
||
"add_tag": {"wxid": TO, "tag_name": "catalog_probe"},
|
||
"remove_tag": {"wxid": TO, "tag_name": "catalog_probe"},
|
||
"delete_label": {"label_id": "1"},
|
||
"delete_tag": {"label_id": "1"},
|
||
"delete_moments": {"sns_id": "1"},
|
||
"delete_favorite": {"local_id": "1"},
|
||
"forward_message": {"to_id": TO, "msg_svr_id": "1779123054724"},
|
||
"forward_multiple": {"to_id": TO, "msg_svr_ids": ["1779123054724"]},
|
||
"recall_message": {"msg_svr_id": "1779123054724"},
|
||
"revoke_message": {"msg_svr_id": "1779123054724"},
|
||
"send_red_packet": {"to_id": TO, "amount": "0.01", "message": "probe"},
|
||
"send_transfer": {"to_id": TO, "amount": "0.01", "description": "probe"},
|
||
"transfer": {"to_id": TO, "amount": "0.01", "description": "probe"},
|
||
"receive_red_packet": {"msg_svr_id": "1"},
|
||
"receive_transfer": {"msg_svr_id": "1"},
|
||
"receive_payment": {"msg_svr_id": "1"},
|
||
"follow_channel": {"channel_id": "probe"},
|
||
"unfollow_channel": {"channel_id": "probe"},
|
||
"follow_official_account": {"account_id": "probe"},
|
||
"unfollow_official_account": {"account_id": "probe"},
|
||
"get_official_account_articles": {"account_id": "probe", "limit": 5},
|
||
"share_channel_video": {"video_id": "index_0", "to_id": TO},
|
||
"share_mini_program": {"app_id": "wxprobe", "to_id": TO, "title": "probe"},
|
||
"add_friend_by_qr": {"qr_content": "probe"},
|
||
"add_custom_emoji": {"image_path": "/sdcard/test.jpg"},
|
||
"add_to_float": {"wxid": TO},
|
||
"remove_from_float": {"wxid": TO},
|
||
"clear_chat_history": {"wxid": TO},
|
||
"clear_history": {"user_id": TO},
|
||
"bind_phone": {"phone": "13800138000"},
|
||
"change_password": {"old_password": "old", "new_password": "new"},
|
||
"remove_login_device": {"device_id": "probe"},
|
||
"register_account": {"phone": "13800138000", "nickname": "probe"},
|
||
"login_by_password": {"phone": "13800138000", "password": "probe"},
|
||
"login_by_sms": {"phone": "13800138000"},
|
||
"switch_account": {"wxid": TO},
|
||
"batch_execute": {"actions": [{"action": "getProfile", "params": {}}]},
|
||
"get_hook_status": {},
|
||
"get_wechat_version": {},
|
||
"get_process_info": {},
|
||
}
|
||
|
||
|
||
def _load_actions():
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "app" / "agent"))
|
||
from hook.hook_executor import ACTION_TO_RPC, ACTION_ALIASES, resolve_action, normalize_params # type: ignore
|
||
actions = sorted(set(ACTION_TO_RPC.keys()) | set(ACTION_ALIASES.keys()))
|
||
return actions, resolve_action, normalize_params
|
||
|
||
|
||
async def probe_one(client: httpx.AsyncClient, device_id: str, action: str, resolve_action, normalize_params) -> dict:
|
||
canonical = resolve_action(action)
|
||
params = dict(SAFE_PARAMS.get(action) or SAFE_PARAMS.get(canonical, {}))
|
||
params = normalize_params(action, params)
|
||
t0 = time.time()
|
||
try:
|
||
resp = await client.post(
|
||
f"{BASE}/api/v3/hook/execute",
|
||
json={"device_id": device_id, "platform": "wechat", "action": action, "params": params, "hook_only": True},
|
||
timeout=60,
|
||
)
|
||
body = resp.json()
|
||
data = body.get("data") if isinstance(body.get("data"), dict) else body
|
||
ok = bool(data.get("success")) or body.get("code") == 200 and data.get("success") is not False and "不支持" not in str(data.get("message", ""))
|
||
if data.get("success") is False:
|
||
ok = False
|
||
err = "" if ok else str(data.get("message") or data.get("error") or body)[:120]
|
||
ch = body.get("channel_used") or data.get("_channel_used") or data.get("channel") or "hook"
|
||
except Exception as e:
|
||
ok, err, ch = False, str(e)[:120], "error"
|
||
if not ok and "Frida" in err:
|
||
await asyncio.sleep(1.5)
|
||
try:
|
||
resp = await client.post(
|
||
f"{BASE}/api/v3/hook/execute",
|
||
json={"device_id": device_id, "platform": "wechat", "action": action, "params": params, "hook_only": True},
|
||
timeout=60,
|
||
)
|
||
body = resp.json()
|
||
data = body.get("data") if isinstance(body.get("data"), dict) else body
|
||
ok = bool(data.get("success"))
|
||
err = "" if ok else str(data.get("message") or data.get("error") or body)[:120]
|
||
ch = body.get("channel_used") or data.get("_channel_used") or "hook"
|
||
except Exception as e2:
|
||
err = str(e2)[:120]
|
||
return {"action": action, "success": ok, "channel": str(ch), "elapsed_ms": int((time.time() - t0) * 1000), "error": err}
|
||
|
||
|
||
async def prefetch_context(client: httpx.AsyncClient, device_id: str) -> None:
|
||
"""从真机拉取 group_id / label_id 等，写入 SAFE_PARAMS。"""
|
||
try:
|
||
r = await client.post(
|
||
f"{BASE}/api/v3/hook/execute",
|
||
json={"device_id": device_id, "platform": "wechat", "action": "get_groups", "params": {"limit": 5}, "hook_only": True},
|
||
timeout=60,
|
||
)
|
||
groups = (r.json().get("data") or {}).get("groups") or []
|
||
gid = None
|
||
for g in groups:
|
||
candidate = g.get("group_id") or g.get("wxid") or g.get("username")
|
||
if not candidate:
|
||
continue
|
||
chk = await client.post(
|
||
f"{BASE}/api/v3/hook/execute",
|
||
json={"device_id": device_id, "platform": "wechat", "action": "get_group_info", "params": {"group_id": candidate}, "hook_only": True},
|
||
timeout=60,
|
||
)
|
||
if (chk.json().get("data") or {}).get("success"):
|
||
gid = candidate
|
||
break
|
||
if gid:
|
||
for k in (
|
||
"get_group_info", "get_group_members", "send_group_message",
|
||
"generate_group_qr_code", "invite_to_group", "remove_from_group",
|
||
"set_group_announcement", "set_group_notice", "set_group_name", "quit_group",
|
||
):
|
||
if k in SAFE_PARAMS:
|
||
SAFE_PARAMS[k]["group_id"] = gid
|
||
except Exception:
|
||
pass
|
||
try:
|
||
r = await client.post(
|
||
f"{BASE}/api/v3/hook/execute",
|
||
json={"device_id": device_id, "platform": "wechat", "action": "get_labels", "params": {}, "hook_only": True},
|
||
timeout=60,
|
||
)
|
||
labels = (r.json().get("data") or {}).get("labels") or []
|
||
if labels:
|
||
lid = str(labels[0].get("label_id") or labels[0].get("id") or "1")
|
||
SAFE_PARAMS["get_contacts_by_label"]["label_id"] = lid
|
||
SAFE_PARAMS["delete_tag"]["label_id"] = lid
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("-d", "--device-id", default=DEVICE)
|
||
args = parser.parse_args()
|
||
actions, resolve_action, normalize_params = _load_actions()
|
||
results = []
|
||
async with httpx.AsyncClient() as client:
|
||
await prefetch_context(client, args.device_id)
|
||
for action in actions:
|
||
row = await probe_one(client, args.device_id, action, resolve_action, normalize_params)
|
||
results.append(row)
|
||
print(f"{'✓' if row['success'] else '✗'} {action:<32} {row['elapsed_ms']:>5}ms")
|
||
await asyncio.sleep(0.25)
|
||
passed = sum(1 for x in results if x["success"])
|
||
report = {
|
||
"timestamp": datetime.now().isoformat(),
|
||
"device_id": args.device_id,
|
||
"total": len(results),
|
||
"passed": passed,
|
||
"failed": len(results) - passed,
|
||
"success_rate": f"{100 * passed / max(len(results), 1):.1f}%",
|
||
"results": results,
|
||
}
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
path = OUT / f"matrix_hook_catalog_{args.device_id}_{int(time.time())}.json"
|
||
path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"\nHook catalog: {passed}/{len(results)} → {path}")
|
||
return 0 if passed == len(results) else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(asyncio.run(main()))
|