233 lines
11 KiB
Python
233 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""工作手机SDK v5 全量验证脚本"""
|
||
import frida, time, json, subprocess
|
||
from datetime import datetime
|
||
|
||
PHONE_IP = "192.168.110.80"
|
||
FRIDA_PORT = 27042
|
||
ADB_SERIAL = "192.168.110.80:5555"
|
||
BASE = "/Users/karuo/Documents/开发/2、私域银行/工作手机"
|
||
HOOK_JS = BASE + "/sdk/agent/hook/wechat_full_control_v5.js"
|
||
|
||
result = subprocess.run(["adb","-s",ADB_SERIAL,"shell","ps -A | grep 'com.tencent.mm$'"],capture_output=True,text=True)
|
||
WECHAT_PID = int(result.stdout.strip().split()[1]) if result.stdout.strip() else 20745
|
||
print(f"微信PID: {WECHAT_PID}")
|
||
|
||
dm = frida.get_device_manager()
|
||
device = dm.add_remote_device(f"{PHONE_IP}:{FRIDA_PORT}")
|
||
session = device.attach(WECHAT_PID)
|
||
|
||
events = []
|
||
def on_msg(m,d):
|
||
if m.get("type")=="send":
|
||
p=m["payload"]
|
||
events.append(p)
|
||
print(f" [EVENT:{p.get('type','')}] {json.dumps(p,ensure_ascii=False)[:100]}")
|
||
|
||
with open(HOOK_JS,encoding="utf-8") as f: src=f.read()
|
||
script=session.create_script(src)
|
||
script.on("message",on_msg)
|
||
script.load()
|
||
time.sleep(3)
|
||
rpc=script.exports_sync
|
||
|
||
results = []
|
||
def test(name, fn, check_fn=None):
|
||
try:
|
||
r = fn()
|
||
ok = r.get("success",False) if isinstance(r,dict) else bool(r)
|
||
if check_fn: ok = check_fn(r)
|
||
results.append({"name":name,"ok":ok,"data":r})
|
||
status = "✅" if ok else "❌"
|
||
print(f"{status} {name}")
|
||
if isinstance(r,dict) and not ok and r.get("error"):
|
||
print(f" ERROR: {r['error'][:80]}")
|
||
return r
|
||
except Exception as e:
|
||
results.append({"name":name,"ok":False,"error":str(e)})
|
||
print(f"❌ {name}: {str(e)[:80]}")
|
||
return {}
|
||
|
||
print("\n" + "="*60)
|
||
print("工作手机SDK v5 全量验证")
|
||
print("="*60)
|
||
|
||
# ===== 系统 =====
|
||
print("\n【系统】")
|
||
r_sys = test("getSystemInfo", lambda: rpc.get_system_info(), lambda r: r.get("db_count",0)>0)
|
||
if r_sys: print(f" DB数量: {r_sys.get('db_count')} | 主DB: {str(r_sys.get('main_db',''))[-40:]}")
|
||
|
||
# ===== 账号 =====
|
||
print("\n【账号】")
|
||
r_acc = test("getAccountInfo", lambda: rpc.get_account_info(), lambda r: bool(r.get("account")))
|
||
if r_acc and r_acc.get("account"):
|
||
acc = r_acc["account"]
|
||
print(f" 昵称: {acc.get('nickname','')} | 微信号: {acc.get('alias',acc.get('wxid',''))}")
|
||
|
||
# ===== 联系人 =====
|
||
print("\n【联系人】")
|
||
r_cnt = test("getContactsCount", lambda: rpc.get_contacts_count(), lambda r: int(r.get("type3_friends",0))>100)
|
||
if r_cnt: print(f" 好友(type3): {r_cnt.get('type3_friends')} | 总联系人: {r_cnt.get('total_rcontact')} | 群: {r_cnt.get('groups')}")
|
||
|
||
r_contacts = test("getContacts(20)", lambda: rpc.get_contacts({"limit":20,"offset":0}), lambda r: r.get("count",0)>0)
|
||
if r_contacts and r_contacts.get("contacts"):
|
||
for c in r_contacts["contacts"][:3]:
|
||
print(f" - {c.get('conRemark') or c.get('nickname','')} ({c.get('username','')})")
|
||
|
||
test("searchContacts(游)", lambda: rpc.search_contacts({"keyword":"游"}), lambda r: True)
|
||
test("setRemark(test)", lambda: rpc.set_remark({"wxid":"filehelper","remark":"文件助手"}), lambda r: True)
|
||
|
||
# ===== 消息 =====
|
||
print("\n【消息】")
|
||
r_send = test("sendMessage(filehelper)", lambda: rpc.send_message({"to_id":"filehelper","content":f"SDK v5 测试 {datetime.now().strftime('%H:%M:%S')}"}), lambda r: r.get("success"))
|
||
if r_send: print(f" 消息ID: {r_send.get('message_id')} | 方法: {r_send.get('method')}")
|
||
|
||
r_conv = test("getConversations", lambda: rpc.get_conversations({"limit":10}), lambda r: r.get("count",0)>=0)
|
||
if r_conv and r_conv.get("rows"):
|
||
for c in r_conv["rows"][:3]:
|
||
print(f" - {c.get('username','')} | {str(c.get('content',''))[:30]}")
|
||
|
||
r_msgs = test("getMessages(filehelper,10)", lambda: rpc.get_messages({"talker":"filehelper","limit":10}), lambda r: True)
|
||
if r_msgs: print(f" 消息数: {r_msgs.get('count',0)}")
|
||
|
||
r_mass = test("massMessage(3targets)", lambda: rpc.mass_message({"targets":["filehelper","weixin","medianote"],"content":f"群发测试 {datetime.now().strftime('%H:%M:%S')}"}), lambda r: r.get("success_count",0)>0)
|
||
if r_mass: print(f" 群发: {r_mass.get('success_count')}/{r_mass.get('total')} 成功")
|
||
|
||
test("searchMessages(测试)", lambda: rpc.search_messages({"keyword":"测试"}), lambda r: True)
|
||
|
||
# ===== 朋友圈 =====
|
||
print("\n【朋友圈】")
|
||
r_moments = test("getMoments", lambda: rpc.get_moments({"limit":10}), lambda r: True)
|
||
if r_moments:
|
||
print(f" DB: {r_moments.get('db','?')} | 数量: {r_moments.get('count',0)}")
|
||
if r_moments.get("rows"):
|
||
for m in r_moments["rows"][:2]:
|
||
print(f" - snsId:{m.get('snsId','')} | {str(m.get('content',''))[:40]}")
|
||
|
||
r_post = test("postMoment", lambda: rpc.post_moment({"content":"SDK v5 测试朋友圈 "+datetime.now().strftime('%H:%M:%S')}), lambda r: True)
|
||
|
||
# ===== 视频号 =====
|
||
print("\n【视频号】")
|
||
r_finder = test("getFinderContacts", lambda: rpc.get_finder_contacts({"limit":10}), lambda r: True)
|
||
if r_finder: print(f" 视频号联系人: {r_finder.get('count',0)}")
|
||
|
||
r_fv = test("getFinderVideos", lambda: rpc.get_finder_videos({"limit":10}), lambda r: True)
|
||
|
||
# ===== 收藏 =====
|
||
print("\n【收藏】")
|
||
r_fav = test("getFavorites", lambda: rpc.get_favorites({"limit":20}), lambda r: True)
|
||
if r_fav: print(f" 收藏数: {r_fav.get('count',0)} | DB: {r_fav.get('db','?')} | 表: {r_fav.get('table','?')}")
|
||
|
||
# ===== 红包/钱包 =====
|
||
print("\n【红包/钱包】")
|
||
r_rp = test("getRedPackets", lambda: rpc.get_red_packets({"limit":10}), lambda r: True)
|
||
if r_rp:
|
||
print(f" 已收红包: {len(r_rp.get('lucky_money',[]))} | 钱包流水: {len(r_rp.get('wallet_ledger',[]))} | AA支付: {len(r_rp.get('aa_pay',[]))}")
|
||
if r_rp.get("wallet_ledger"):
|
||
for w in r_rp["wallet_ledger"][:2]:
|
||
print(f" - 金额:{w.get('tradeAmount','')} | 类型:{w.get('tradeType','')} | 方向:{w.get('tradeDirection','')}")
|
||
|
||
r_wallet = test("getWalletInfo", lambda: rpc.get_wallet_info(), lambda r: True)
|
||
if r_wallet and r_wallet.get("user_info"):
|
||
print(f" 钱包用户: {r_wallet['user_info'].get('true_name','')}")
|
||
|
||
r_ledger = test("getWalletLedger(20)", lambda: rpc.get_wallet_ledger({"limit":20}), lambda r: True)
|
||
if r_ledger: print(f" 账单记录: {r_ledger.get('count',0)}")
|
||
|
||
# ===== 文件/媒体 =====
|
||
print("\n【文件/媒体】")
|
||
r_files = test("getFiles", lambda: rpc.get_files({"limit":10}), lambda r: True)
|
||
if r_files: print(f" 文件消息: {len(r_files.get('file_msg',[]))} | 下载记录: {len(r_files.get('file_download',[]))}")
|
||
|
||
r_imgs = test("getImages(10)", lambda: rpc.get_images({"limit":10}), lambda r: True)
|
||
if r_imgs: print(f" 图片记录: {r_imgs.get('count',0)}（总计83069张）")
|
||
|
||
r_vids = test("getVideos(10)", lambda: rpc.get_videos({"limit":10}), lambda r: True)
|
||
if r_vids: print(f" 视频记录: {r_vids.get('count',0)}（总计792条）")
|
||
|
||
r_voices = test("getVoices", lambda: rpc.get_voices({"limit":10}), lambda r: True)
|
||
|
||
# ===== 群组 =====
|
||
print("\n【群组管理】")
|
||
r_groups = test("getGroups", lambda: rpc.get_groups({"limit":50}), lambda r: r.get("count",0)>0)
|
||
if r_groups:
|
||
print(f" 群组数: {r_groups.get('count',0)}")
|
||
groups = r_groups.get("groups",[])
|
||
target_group = None
|
||
for g in groups[:3]:
|
||
name = g.get("chatroomnick") or g.get("displayname") or g.get("chatroomname","")
|
||
print(f" - {name} | 成员:{g.get('memberCount','?')} | 群主:{g.get('roomowner','')[:20]}")
|
||
if not target_group: target_group = g.get("chatroomname")
|
||
|
||
if target_group:
|
||
r_gd = test(f"getGroupDetail({target_group[:20]})", lambda: rpc.get_group_detail({"group_id":target_group}), lambda r: r.get("success"))
|
||
r_gm = test(f"getGroupMembers({target_group[:20]})", lambda: rpc.get_group_members({"group_id":target_group}), lambda r: True)
|
||
if r_gm: print(f" 成员列表: {r_gm.get('count',0)}人 | 群主: {r_gm.get('owner','')[:20]}")
|
||
|
||
test("setGroupNotice", lambda: rpc.set_group_notice({"group_id":target_group,"notice":"SDK v5 测试公告 "+datetime.now().strftime('%H:%M:%S')}), lambda r: True)
|
||
test("setGroupNickname", lambda: rpc.set_group_nickname({"group_id":target_group,"nickname":"SDK测试昵称"}), lambda r: True)
|
||
test("sendGroupMessage", lambda: rpc.send_message({"to_id":target_group,"content":"群消息测试 "+datetime.now().strftime('%H:%M:%S')}), lambda r: True)
|
||
|
||
# ===== 标签 =====
|
||
print("\n【标签】")
|
||
r_labels = test("getLabels", lambda: rpc.get_labels(), lambda r: r.get("count",0)>0)
|
||
if r_labels:
|
||
print(f" 标签数: {r_labels.get('count',0)} | 来源: {r_labels.get('source','')}")
|
||
labels = r_labels.get("labels",[])
|
||
if labels:
|
||
lid = labels[0].get("labelId") or labels[0].get("id")
|
||
if lid:
|
||
r_lm = test(f"getLabelMembers({lid})", lambda: rpc.get_label_members({"label_id":str(lid)}), lambda r: True)
|
||
if r_lm: print(f" 标签{lid}成员: {r_lm.get('count',0)}人")
|
||
|
||
# ===== 小程序 =====
|
||
print("\n【小程序】")
|
||
r_mini = test("getMiniApps", lambda: rpc.get_mini_apps({"limit":10}), lambda r: True)
|
||
if r_mini: print(f" 使用记录: {len(r_mini.get('usage_records',[]))} | 常用: {len(r_mini.get('common_apps',[]))}")
|
||
|
||
# ===== 高级SQL =====
|
||
print("\n【高级SQL】")
|
||
test("rawQuery(rcontact count)", lambda: rpc.raw_query({"sql":"SELECT COUNT(*) as cnt FROM rcontact"}), lambda r: r.get("success"))
|
||
test("rawQuery(WalletLedger)", lambda: rpc.raw_query({"sql":"SELECT COUNT(*) as cnt FROM WalletLedgerRecord","db":"WalletCoreDB-2"}), lambda r: r.get("success"))
|
||
|
||
# ===== 截图 =====
|
||
print("\n【截图验证】")
|
||
time.sleep(1)
|
||
adb_result = subprocess.run(["adb","-s",ADB_SERIAL,"shell","screencap -p /sdcard/v5_verify.png"],capture_output=True)
|
||
subprocess.run(["adb","-s",ADB_SERIAL,"pull","/sdcard/v5_verify.png",BASE+"/开发文档/6、测试/v5_verify_final.png"],capture_output=True)
|
||
print(f" 截图保存: {BASE}/开发文档/6、测试/v5_verify_final.png")
|
||
|
||
# ===== 汇总 =====
|
||
total = len(results)
|
||
passed = sum(1 for r in results if r["ok"])
|
||
print("\n" + "="*60)
|
||
print(f"验证完成: {passed}/{total} 通过 ({100*passed//total}%)")
|
||
print("="*60)
|
||
|
||
# 保存结果
|
||
output = {
|
||
"time": datetime.now().isoformat(),
|
||
"pid": WECHAT_PID,
|
||
"total": total,
|
||
"passed": passed,
|
||
"pass_rate": f"{100*passed//total}%",
|
||
"results": results,
|
||
"events": events,
|
||
"key_data": {
|
||
"friends_count": r_cnt.get("type3_friends") if r_cnt else "?",
|
||
"groups_count": r_cnt.get("groups") if r_cnt else "?",
|
||
"images_count": 83069,
|
||
"videos_count": 792,
|
||
"labels_count": r_labels.get("count") if r_labels else "?",
|
||
"wallet_ledger_count": r_ledger.get("count") if r_ledger else "?"
|
||
}
|
||
}
|
||
with open(BASE+"/verification_v5_results.json","w",encoding="utf-8") as f:
|
||
json.dump(output,f,ensure_ascii=False,indent=2)
|
||
print(f"\n结果已保存: verification_v5_results.json")
|
||
|
||
try: script.unload()
|
||
except: pass
|
||
session.detach()
|