375 lines
14 KiB
Python
375 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
工作手机SDK - 微信全功能验证 v4
|
||
修复：联系人列表/朋友圈/标签/群发/DB枚举
|
||
"""
|
||
import frida, time, json, subprocess, os
|
||
from datetime import datetime
|
||
|
||
PHONE_IP = "192.168.110.80"
|
||
ADB_SERIAL = "192.168.110.80:5555"
|
||
FRIDA_PORT = 27042
|
||
BASE = "/Users/karuo/Documents/开发/2、私域银行/工作手机"
|
||
HOOK_JS = BASE + "/sdk/agent/hook/wechat_full_control_v4.js"
|
||
SC_DIR = BASE + "/开发文档/6、测试/live_verify_20260518/screenshots"
|
||
RESULT_FILE = BASE + "/verification_v4_20260518.json"
|
||
|
||
os.makedirs(SC_DIR, exist_ok=True)
|
||
|
||
# 动态获取PID
|
||
result = subprocess.run(
|
||
["adb", "-s", ADB_SERIAL, "shell", "ps -A | grep 'com.tencent.mm$'"],
|
||
capture_output=True, text=True
|
||
)
|
||
pid_line = result.stdout.strip()
|
||
WECHAT_PID = int(pid_line.split()[1]) if pid_line else 17861
|
||
print(f"微信PID: {WECHAT_PID}")
|
||
|
||
def adb_sc(name, wait=0.5):
|
||
time.sleep(wait)
|
||
remote = f"/sdcard/{name}.png"
|
||
local = f"{SC_DIR}/{name}.png"
|
||
subprocess.run(["adb", "-s", ADB_SERIAL, "shell", "screencap", "-p", remote], capture_output=True)
|
||
subprocess.run(["adb", "-s", ADB_SERIAL, "pull", remote, local], capture_output=True)
|
||
size = os.path.getsize(local) if os.path.exists(local) else 0
|
||
print(f" [截图] {name}.png {size//1024}KB")
|
||
return local if size > 10000 else None
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"工作手机SDK - 微信全功能验证 v4")
|
||
print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"{'='*60}")
|
||
|
||
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)
|
||
t = p.get("type", "")
|
||
if t in ("ready", "hook_ok", "hook_fail", "real_send_captured", "message_sent"):
|
||
print(f" [EVENT:{t}] {json.dumps(p, ensure_ascii=False)[:120]}")
|
||
|
||
with open(HOOK_JS, encoding="utf-8") as f:
|
||
src = f.read()
|
||
script = session.create_script(src)
|
||
script.on("message", on_msg)
|
||
script.load()
|
||
print("[等待3秒...]")
|
||
time.sleep(3)
|
||
rpc = script.exports_sync
|
||
|
||
results = {}
|
||
passed = 0
|
||
failed = 0
|
||
ts = datetime.now().strftime("%H%M%S")
|
||
|
||
def test(name, func):
|
||
global passed, failed
|
||
try:
|
||
r = func()
|
||
ok = r.get("success", False) if isinstance(r, dict) else bool(r)
|
||
status = "✅ PASS" if ok else "❌ FAIL"
|
||
print(f" {status} {name}")
|
||
if not ok and isinstance(r, dict) and r.get("error"):
|
||
print(f" 错误: {str(r['error'])[:80]}")
|
||
if ok: passed += 1
|
||
else: failed += 1
|
||
return r
|
||
except Exception as e:
|
||
print(f" ❌ FAIL {name} [异常: {str(e)[:80]}]")
|
||
failed += 1
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# ============================================================
|
||
# 模块0：枚举所有DB
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块0：枚举所有打开的DB")
|
||
print(f"{'='*50}")
|
||
r_dbs = test("枚举所有DB实例", rpc.get_db_info)
|
||
if r_dbs.get("success"):
|
||
print(f" DB数量: {r_dbs.get('count', 0)}")
|
||
for db in r_dbs.get("instances", []):
|
||
print(f" - {db.get('path', 'N/A')}")
|
||
|
||
r_tables = test("枚举所有DB表", rpc.list_all_tables)
|
||
if r_tables.get("success"):
|
||
for db_name, tables in r_tables.get("databases", {}).items():
|
||
print(f" [{db_name}]: {tables[:10]}")
|
||
|
||
results["db_info"] = {"dbs": r_dbs, "tables": r_tables}
|
||
|
||
# ============================================================
|
||
# 模块1：系统基础
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块1：系统基础")
|
||
print(f"{'='*50}")
|
||
r_sys = test("系统信息", rpc.get_system_info)
|
||
print(f" DB路径: {r_sys.get('db', {}).get('path', 'N/A')}")
|
||
print(f" DB总数: {r_sys.get('db_count', 0)}")
|
||
results["system"] = r_sys
|
||
|
||
# ============================================================
|
||
# 模块2：账号信息
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块2：账号信息")
|
||
print(f"{'='*50}")
|
||
r_acc = test("账号信息", rpc.get_account_info)
|
||
if r_acc.get("success"):
|
||
acc = r_acc.get("account", {})
|
||
print(f" 昵称: {acc.get('nickname', 'N/A')}")
|
||
print(f" 微信号: {acc.get('wxid', 'N/A')}")
|
||
print(f" 手机: {acc.get('mobile', 'N/A')}")
|
||
results["account"] = r_acc
|
||
|
||
# ============================================================
|
||
# 模块3：联系人
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块3：联系人（完整数据）")
|
||
print(f"{'='*50}")
|
||
r_cnt = test("联系人总数", rpc.get_contacts_count)
|
||
if r_cnt.get("success"):
|
||
print(f" 真实好友(type=3): {r_cnt.get('type3_friends')} 人")
|
||
print(f" 总联系人: {r_cnt.get('total_rcontact')} 条")
|
||
print(f" 群组: {r_cnt.get('groups')} 个")
|
||
|
||
r_contacts = test("联系人列表(前20)", lambda: rpc.get_contacts({"limit": 20, "offset": 0}))
|
||
if r_contacts.get("success"):
|
||
print(f" 返回: {r_contacts.get('count', 0)} 条")
|
||
for c in r_contacts.get("contacts", [])[:5]:
|
||
name = c.get("conRemark") or c.get("nickname") or c.get("username")
|
||
print(f" - {name} ({c.get('username', '')})")
|
||
|
||
results["contacts"] = {"count": r_cnt, "sample": r_contacts}
|
||
|
||
# ============================================================
|
||
# 模块4：群组
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块4：群组")
|
||
print(f"{'='*50}")
|
||
r_groups = test("群组列表(前10)", lambda: rpc.get_groups({"limit": 10}))
|
||
if r_groups.get("success"):
|
||
print(f" 群组数: {r_groups.get('count', 0)}")
|
||
for g in r_groups.get("groups", [])[:5]:
|
||
name = g.get("chatroomnick") or g.get("displayname") or g.get("chatroomname", "")
|
||
mc = g.get("memberCount", "?")
|
||
print(f" - {name} (成员:{mc})")
|
||
|
||
# 获取第一个群的成员
|
||
first_gid = ""
|
||
if r_groups.get("success") and r_groups.get("groups"):
|
||
first_gid = r_groups["groups"][0].get("chatroomname", "")
|
||
if first_gid:
|
||
r_members = test(f"群成员({first_gid[:20]})", lambda: rpc.get_group_members({"group_id": first_gid}))
|
||
if r_members.get("success"):
|
||
print(f" 成员数: {r_members.get('count', 0)} (DB记录:{r_members.get('db_count','?')})")
|
||
print(f" 前5: {r_members.get('member_ids', [])[:5]}")
|
||
|
||
results["groups"] = r_groups
|
||
|
||
# ============================================================
|
||
# 模块5：消息发送
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块5：消息发送（WCDB直接写入）")
|
||
print(f"{'='*50}")
|
||
|
||
adb_sc(f"v4_before_send_{ts}")
|
||
|
||
msg1 = f"[Frida v4验证 {datetime.now().strftime('%H:%M:%S')}] 好友:{r_cnt.get('type3_friends','?')}人 | 群:{r_cnt.get('groups','?')}个 | WCDB无感控制"
|
||
r_send1 = test("发送消息到文件传输助手", lambda: rpc.send_message({"to_id": "filehelper", "content": msg1}))
|
||
print(f" 消息ID: {r_send1.get('message_id', 'N/A')}")
|
||
time.sleep(1)
|
||
|
||
adb_sc(f"v4_after_send_{ts}")
|
||
|
||
r_verify = test("验证消息已写入DB", lambda: rpc.get_messages({"talker": "filehelper", "limit": 5}))
|
||
if r_verify.get("success"):
|
||
print(f" DB消息数: {r_verify.get('count', 0)}")
|
||
for msg in r_verify.get("messages", [])[:3]:
|
||
print(f" - isSend={msg.get('isSend')} | {str(msg.get('content',''))[:60]}")
|
||
|
||
results["message_send"] = {"send": r_send1, "verify": r_verify}
|
||
|
||
# ============================================================
|
||
# 模块6：群发消息
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块6：群发消息")
|
||
print(f"{'='*50}")
|
||
|
||
test_targets = []
|
||
if r_contacts.get("success"):
|
||
for c in r_contacts.get("contacts", [])[:5]:
|
||
uid = c.get("username", "")
|
||
if uid and uid != "filehelper":
|
||
test_targets.append(uid)
|
||
|
||
if test_targets:
|
||
mass_content = f"[群发测试 {datetime.now().strftime('%H:%M:%S')}] Frida WCDB群发验证 v4"
|
||
r_mass = test(f"群发消息({len(test_targets)}人)", lambda: rpc.mass_message({"targets": test_targets, "content": mass_content}))
|
||
if r_mass.get("success"):
|
||
print(f" 成功: {r_mass.get('success_count')}/{r_mass.get('total')} 条")
|
||
for item in r_mass.get("results", [])[:3]:
|
||
print(f" - {item.get('to_id','')} {'✅' if item.get('success') else '❌'}")
|
||
results["mass_message"] = r_mass
|
||
else:
|
||
print(" ⚠️ 跳过群发（联系人列表为空）")
|
||
|
||
# ============================================================
|
||
# 模块7：最近会话
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块7：最近会话")
|
||
print(f"{'='*50}")
|
||
r_conv = test("最近会话(前10)", lambda: rpc.get_conversations({"limit": 10}))
|
||
if r_conv.get("success"):
|
||
print(f" 会话数: {r_conv.get('count', 0)}")
|
||
for c in r_conv.get("conversations", [])[:5]:
|
||
print(f" - {c.get('username','')} | {str(c.get('content',''))[:40]}")
|
||
results["conversations"] = r_conv
|
||
|
||
# ============================================================
|
||
# 模块8：朋友圈
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块8：朋友圈")
|
||
print(f"{'='*50}")
|
||
r_moments = test("读取朋友圈", lambda: rpc.get_moments({"limit": 10}))
|
||
if r_moments.get("success"):
|
||
print(f" 朋友圈条数: {r_moments.get('count', 0)}")
|
||
print(f" 来源DB: {r_moments.get('db', 'N/A')}")
|
||
else:
|
||
print(f" 所有DB: {r_moments.get('all_dbs', [])}")
|
||
|
||
moment_content = f"[SDK v4验证{datetime.now().strftime('%m/%d %H:%M')}] Frida无感控制微信成功 好友:{r_cnt.get('type3_friends','?')}人"
|
||
r_post = test("发布朋友圈(Intent)", lambda: rpc.post_moment({"content": moment_content}))
|
||
print(f" 方法: {r_post.get('method', 'N/A')}")
|
||
results["moments"] = {"read": r_moments, "post": r_post}
|
||
|
||
# ============================================================
|
||
# 模块9：标签
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块9：标签管理")
|
||
print(f"{'='*50}")
|
||
r_labels = test("获取标签列表", rpc.get_labels)
|
||
if r_labels.get("success"):
|
||
print(f" 标签数量: {r_labels.get('count', 0)}")
|
||
print(f" 来源: {r_labels.get('source', 'label table')}")
|
||
for l in r_labels.get("labels", [])[:5]:
|
||
print(f" - [{l.get('labelId')}] {l.get('labelName', '')}")
|
||
results["labels"] = r_labels
|
||
|
||
# ============================================================
|
||
# 模块10：搜索
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块10：搜索")
|
||
print(f"{'='*50}")
|
||
r_s1 = test("搜索联系人(游)", lambda: rpc.search_contacts({"keyword": "游"}))
|
||
if r_s1.get("success"):
|
||
print(f" 结果: {r_s1.get('count', 0)} 条")
|
||
for c in r_s1.get("results", [])[:3]:
|
||
print(f" - {c.get('nickname','')} ({c.get('username','')})")
|
||
|
||
r_s2 = test("搜索消息(验证)", lambda: rpc.search_messages({"keyword": "验证"}))
|
||
if r_s2.get("success"):
|
||
print(f" 结果: {r_s2.get('count', 0)} 条")
|
||
results["search"] = {"contacts": r_s1, "messages": r_s2}
|
||
|
||
# ============================================================
|
||
# 模块11：好友操作
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块11：好友操作")
|
||
print(f"{'='*50}")
|
||
r_remark = test("设置备注(filehelper)", lambda: rpc.set_remark({"wxid": "filehelper", "remark": "文件助手[SDK v4]"}))
|
||
print(f" 影响行数: {r_remark.get('affected_rows', 0)}")
|
||
results["friend_ops"] = r_remark
|
||
|
||
# ============================================================
|
||
# 模块12：高级SQL
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("模块12：高级SQL查询")
|
||
print(f"{'='*50}")
|
||
r_sql1 = test("消息统计Top5", lambda: rpc.raw_query({"sql": "SELECT talker, COUNT(*) as cnt FROM message GROUP BY talker ORDER BY cnt DESC LIMIT 5"}))
|
||
if r_sql1.get("success"):
|
||
for row in r_sql1.get("rows", []):
|
||
print(f" - {row.get('talker','')} ({row.get('cnt',0)} 条)")
|
||
|
||
r_sql2 = test("最近7天消息量", lambda: rpc.raw_query({"sql": f"SELECT COUNT(*) as cnt FROM message WHERE createTime > {int((time.time()-7*86400)*1000)}"}))
|
||
if r_sql2.get("success"):
|
||
print(f" 7天消息量: {r_sql2.get('rows', [{}])[0].get('cnt', 0)} 条")
|
||
|
||
results["sql"] = {"stats": r_sql1, "recent": r_sql2}
|
||
|
||
# ============================================================
|
||
# 截图验证
|
||
# ============================================================
|
||
print(f"\n{'='*50}")
|
||
print("截图验证")
|
||
print(f"{'='*50}")
|
||
rpc.navigate_to_chat({"wxid": "filehelper"})
|
||
sc = adb_sc(f"v4_final_{ts}", wait=2)
|
||
|
||
# ============================================================
|
||
# 汇总
|
||
# ============================================================
|
||
total = passed + failed
|
||
pass_rate = (passed / total * 100) if total > 0 else 0
|
||
|
||
print(f"\n{'='*60}")
|
||
print("验证汇总报告 v4")
|
||
print(f"{'='*60}")
|
||
print(f" 总测试项: {total}")
|
||
print(f" 通过: {passed} ✅")
|
||
print(f" 失败: {failed} ❌")
|
||
print(f" 通过率: {pass_rate:.1f}%")
|
||
print(f"\n 真实数据:")
|
||
print(f" 好友(type=3): {r_cnt.get('type3_friends', '?')} 人")
|
||
print(f" 总联系人: {r_cnt.get('total_rcontact', '?')} 条")
|
||
print(f" 群组: {r_cnt.get('groups', '?')} 个")
|
||
print(f" 标签: {r_labels.get('count', '?')} 个")
|
||
print(f"\n 消息发送: {'✅' if r_send1.get('success') else '❌'} WCDB直接写入")
|
||
print(f" 群发: {'✅' if results.get('mass_message', {}).get('success') else '❌'}")
|
||
print(f" 朋友圈: {'✅' if r_post.get('success') else '❌'}")
|
||
print(f" 捕获调用栈: {len(events)} 个事件")
|
||
|
||
# 保存报告
|
||
report = {
|
||
"title": "工作手机SDK - 微信全功能验证报告 v4",
|
||
"time": datetime.now().isoformat(),
|
||
"device": PHONE_IP,
|
||
"wechat_pid": WECHAT_PID,
|
||
"summary": {"total": total, "passed": passed, "failed": failed, "pass_rate": f"{pass_rate:.1f}%"},
|
||
"real_data": {
|
||
"type3_friends": r_cnt.get("type3_friends"),
|
||
"total_rcontact": r_cnt.get("total_rcontact"),
|
||
"groups": r_cnt.get("groups"),
|
||
"labels": r_labels.get("count")
|
||
},
|
||
"results": results,
|
||
"events": events
|
||
}
|
||
with open(RESULT_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(report, f, ensure_ascii=False, indent=2, default=str)
|
||
print(f"\n报告已保存: {RESULT_FILE}")
|
||
|
||
try:
|
||
script.unload()
|
||
except:
|
||
pass
|
||
session.detach()
|
||
print("完成！")
|