145 lines
5.7 KiB
Python
145 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
工作手机SDK - 真实数据统计验证
|
||
目标：查明联系人/群组真实数量，截图验证朋友圈和消息发送
|
||
"""
|
||
import frida, time, json, os
|
||
from datetime import datetime
|
||
|
||
PHONE_IP = "192.168.110.80"
|
||
FRIDA_PORT = 27042
|
||
WECHAT_PID = 16816
|
||
BASE = "/Users/karuo/Documents/开发/2、私域银行/工作手机"
|
||
HOOK_SCRIPT = BASE + "/sdk/agent/hook/wechat_hook_v2.js"
|
||
RESULT_FILE = BASE + "/verification_stats_20260518.json"
|
||
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] 连接Frida: {PHONE_IP}:{FRIDA_PORT}")
|
||
dm = frida.get_device_manager()
|
||
device = dm.add_remote_device(f"{PHONE_IP}:{FRIDA_PORT}")
|
||
session = device.attach(WECHAT_PID)
|
||
print(f"[OK] 附加微信 PID={WECHAT_PID}")
|
||
|
||
with open(HOOK_SCRIPT, encoding="utf-8") as f:
|
||
src = f.read()
|
||
script = session.create_script(src)
|
||
script.on("message", lambda m, d: None)
|
||
script.load()
|
||
print("[OK] Hook已加载，等待3秒...")
|
||
time.sleep(3)
|
||
rpc = script.exports_sync
|
||
|
||
results = {}
|
||
|
||
# === 1. 联系人/群组真实统计 ===
|
||
print("\n=== 1. 联系人/群组数据统计 ===")
|
||
stats = rpc.get_contact_stats()
|
||
print(json.dumps(stats, ensure_ascii=False, indent=2, default=str)[:3000])
|
||
results["contact_stats"] = stats
|
||
|
||
# === 2. 完整联系人列表（无过滤）===
|
||
print("\n=== 2. 完整联系人（type=3好友）===")
|
||
contacts_full = rpc.get_contacts_full({"limit": 500, "filter_type": 3})
|
||
print(f" type=3好友数量: {contacts_full.get('count', 0)}")
|
||
if contacts_full.get("contacts"):
|
||
for c in contacts_full["contacts"][:10]:
|
||
print(f" {c.get('username','')} | {c.get('nickname','')} | type={c.get('type','')} | verifyFlag={c.get('verifyFlag','')}")
|
||
results["contacts_type3"] = contacts_full
|
||
|
||
# === 3. 放宽过滤的联系人 ===
|
||
print("\n=== 3. 放宽过滤联系人（type!=0, 非群非公众号）===")
|
||
contacts_relaxed = rpc.get_contacts_full({"limit": 500})
|
||
print(f" 放宽过滤联系人数量: {contacts_relaxed.get('count', 0)}")
|
||
if contacts_relaxed.get("contacts"):
|
||
for c in contacts_relaxed["contacts"][:10]:
|
||
print(f" {c.get('username','')} | {c.get('nickname','')} | type={c.get('type','')} | verifyFlag={c.get('verifyFlag','')}")
|
||
results["contacts_relaxed"] = contacts_relaxed
|
||
|
||
# === 4. 完整群组列表 ===
|
||
print("\n=== 4. 完整群组列表 ===")
|
||
groups_full = rpc.get_groups_full({"limit": 500})
|
||
print(f" 群组数量: {groups_full.get('count', 0)} (来源: {groups_full.get('source', '?')})")
|
||
if groups_full.get("groups"):
|
||
for g in groups_full["groups"][:10]:
|
||
print(f" {g.get('group_id','')} | {g.get('name','')} | 成员:{g.get('member_count','?')}")
|
||
results["groups_full"] = groups_full
|
||
|
||
# === 5. 原始SQL查询验证 ===
|
||
print("\n=== 5. 原始SQL验证 ===")
|
||
# 5a. 总联系人数
|
||
r_total = rpc.raw_sql({"sql": "SELECT COUNT(*) as cnt FROM rcontact"})
|
||
print(f" rcontact总记录: {r_total.get('rows', [{}])[0].get('cnt', '?')}")
|
||
results["raw_total"] = r_total
|
||
|
||
# 5b. type分布
|
||
r_type = rpc.raw_sql({"sql": "SELECT type, COUNT(*) as cnt FROM rcontact GROUP BY type ORDER BY cnt DESC LIMIT 20"})
|
||
print(f" type分布: {json.dumps(r_type.get('rows', []), ensure_ascii=False)[:300]}")
|
||
results["raw_type_dist"] = r_type
|
||
|
||
# 5c. 真实好友（type=3）
|
||
r_t3 = rpc.raw_sql({"sql": "SELECT COUNT(*) as cnt FROM rcontact WHERE type=3"})
|
||
print(f" type=3好友数: {r_t3.get('rows', [{}])[0].get('cnt', '?')}")
|
||
|
||
# 5d. chatroom表群组数
|
||
r_chatroom = rpc.raw_sql({"sql": "SELECT COUNT(*) as cnt FROM chatroom"})
|
||
print(f" chatroom表群组数: {r_chatroom.get('rows', [{}])[0].get('cnt', '?')}")
|
||
results["raw_chatroom_count"] = r_chatroom
|
||
|
||
# 5e. 标签数
|
||
r_labels = rpc.raw_sql({"sql": "SELECT COUNT(*) as cnt FROM contactlabelinfo"})
|
||
print(f" 标签数: {r_labels.get('rows', [{}])[0].get('cnt', '?')}")
|
||
|
||
# === 6. 截图验证朋友圈发送前状态 ===
|
||
print("\n=== 6. 截图验证 ===")
|
||
# 截图当前状态（发送前）
|
||
rpc.take_screenshot({"path": "/sdcard/before_moment.png"})
|
||
print(" 截图已保存: /sdcard/before_moment.png")
|
||
|
||
# === 7. 发送朋友圈并截图 ===
|
||
print("\n=== 7. 发送朋友圈 ===")
|
||
moment_content = f"[SDK验证{datetime.now().strftime('%H:%M')}] 工作手机Frida真机验证-朋友圈功能正常"
|
||
moment_result = rpc.post_moments({"content": moment_content, "image_urls": []})
|
||
print(f" 发送结果: {json.dumps(moment_result, ensure_ascii=False)}")
|
||
results["moment_send"] = moment_result
|
||
time.sleep(2)
|
||
|
||
# 截图发送后
|
||
rpc.take_screenshot({"path": "/sdcard/after_moment.png"})
|
||
print(" 截图已保存: /sdcard/after_moment.png")
|
||
|
||
# === 8. 导航到朋友圈确认 ===
|
||
print("\n=== 8. 导航到朋友圈验证 ===")
|
||
# 先截图当前状态
|
||
rpc.navigate_to_main({})
|
||
time.sleep(1)
|
||
rpc.take_screenshot({"path": "/sdcard/main_screen.png"})
|
||
print(" 主界面截图: /sdcard/main_screen.png")
|
||
|
||
# === 9. 发送消息并截图 ===
|
||
print("\n=== 9. 消息发送验证 ===")
|
||
# 导航到文件传输助手
|
||
rpc.navigate_to_chat({"wxid": "filehelper"})
|
||
time.sleep(1)
|
||
rpc.take_screenshot({"path": "/sdcard/before_msg.png"})
|
||
print(" 发送前截图: /sdcard/before_msg.png")
|
||
|
||
# 发送消息
|
||
msg_content = f"[SDK验证{datetime.now().strftime('%H:%M:%S')}] 消息发送功能正常 ✓"
|
||
msg_result = rpc.send_message({"to_id": "filehelper", "content": msg_content, "msg_type": "text"})
|
||
print(f" 发送结果: {json.dumps(msg_result, ensure_ascii=False)}")
|
||
results["message_send"] = msg_result
|
||
time.sleep(1)
|
||
|
||
# 发送后截图
|
||
rpc.take_screenshot({"path": "/sdcard/after_msg.png"})
|
||
print(" 发送后截图: /sdcard/after_msg.png")
|
||
|
||
# === 保存结果 ===
|
||
with open(RESULT_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(results, f, ensure_ascii=False, indent=2, default=str)
|
||
print(f"\n结果已保存: {RESULT_FILE}")
|
||
|
||
script.unload()
|
||
session.detach()
|
||
print("完成！")
|