From 5f120b49decab494672c9b33fbdf8e2369fb042d Mon Sep 17 00:00:00 2001 From: fnvtk Date: Mon, 18 May 2026 20:37:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=95=B4Frida=E6=8E=A7?= =?UTF-8?q?=E5=88=B6=E8=84=9A=E6=9C=ACv3=20verify=5Fall=5Ffeatures.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- verify_all_features.py | 377 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 377 insertions(+) create mode 100644 verify_all_features.py diff --git a/verify_all_features.py b/verify_all_features.py new file mode 100644 index 0000000000..a491f80e7c --- /dev/null +++ b/verify_all_features.py @@ -0,0 +1,377 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +工作手机SDK - 微信全功能Frida验证脚本 v3 +基于WCDB直接操作，无感控制微信所有功能 +""" +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 +WECHAT_PID = 9462 +BASE = "/Users/karuo/Documents/开发/2、私域银行/工作手机" +HOOK_JS = BASE + "/sdk/agent/hook/wechat_full_control.js" +SC_DIR = BASE + "/开发文档/6、测试/live_verify_20260518/screenshots" +RESULT_FILE = BASE + "/verification_full_v3_20260518.json" + +os.makedirs(SC_DIR, exist_ok=True) + +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) + r = subprocess.run(["adb", "-s", ADB_SERIAL, "pull", remote, local], capture_output=True) + size = os.path.getsize(local) if os.path.exists(local) else 0 + status = "OK" if size > 10000 else "FAIL(small)" + print(f" [截图] {name}.png {size//1024}KB [{status}]") + return local if size > 10000 else None + +print(f"\n{'='*60}") +print(f"工作手机SDK - 微信全功能验证 v3") +print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") +print(f"{'='*60}") + +# 连接Frida +print(f"\n[连接] Frida {PHONE_IP}:{FRIDA_PORT} 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) + t = p.get("type", "") + if t in ("ready", "hook_ok", "hook_fail", "real_send_captured", "message_sent", "db_query_ok", "message_inserted"): + print(f" [EVENT:{t}] {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() +print("[等待3秒让Hook初始化...]") +time.sleep(3) +rpc = script.exports_sync + +results = {} +passed = 0 +failed = 0 +ts = datetime.now().strftime("%H%M%S") + +def test(name, func, *args): + global passed, failed + try: + r = func(*args) if args else 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" 错误: {r['error'][:80]}") + if ok: passed += 1 + else: failed += 1 + return r + except Exception as e: + print(f" ❌ FAIL {name} [异常: {str(e)[:60]}]") + failed += 1 + return {"success": False, "error": str(e)} + +# ============================================================ +# 模块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" ClassLoader: {r_sys.get('loader', 'N/A')}") + +r_db = test("DB实例枚举", rpc.get_db_info) +print(f" DB实例数: {r_db.get('count', 0)}") +for inst in r_db.get("instances", []): + print(f" - {inst.get('path', 'N/A')} [open={inst.get('open')}]") + +results["system"] = {"sys_info": r_sys, "db_info": r_db} + +# ============================================================ +# 模块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')}") + print(f" 别名: {acc.get('alias', '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("群组列表", lambda: rpc.get_groups({"limit": 50})) +if r_groups.get("success"): + ct = r_groups.get("chatroom_table", []) + rc = r_groups.get("rcontact_groups", []) + print(f" chatroom表: {len(ct)} 个") + print(f" rcontact群: {len(rc)} 个") + for g in ct[:5]: + name = g.get("chatroomnick") or g.get("chatroomname", "") + mc = g.get("membercount", "?") + print(f" - {name} (成员:{mc})") + +# 获取第一个群的成员 +if r_groups.get("success") and r_groups.get("chatroom_table"): + first_gid = r_groups["chatroom_table"][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)}") + print(f" 前5成员: {r_members.get('member_ids', [])[:5]}") + +results["groups"] = r_groups + +# ============================================================ +# 模块5：消息发送（WCDB直接写入） +# ============================================================ +print(f"\n{'='*50}") +print("模块5：消息发送（WCDB直接写入）") +print(f"{'='*50}") + +# 5a. 截图发送前 +adb_sc(f"v3_before_send_{ts}") + +# 5b. 发送消息 +msg1 = f"[Frida WCDB验证 {datetime.now().strftime('%H:%M:%S')}] 好友:{r_cnt.get('type3_friends','?')}人 | 群:{r_cnt.get('groups','?')}个 | 直接DB写入" +r_send1 = test("发送消息到文件传输助手", lambda: rpc.send_message({"to_id": "filehelper", "content": msg1, "msg_type": "text"})) +print(f" 消息ID: {r_send1.get('message_id', 'N/A')}") +print(f" 方法: {r_send1.get('method', 'N/A')}") +time.sleep(1) + +# 5c. 截图发送后 +adb_sc(f"v3_after_send_{ts}") + +# 5d. 验证DB中是否有消息 +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}") + +# 获取3个联系人ID用于测试 +test_targets = [] +if r_contacts.get("success"): + for c in r_contacts.get("contacts", [])[:3]: + test_targets.append(c.get("username", "")) +test_targets = [t for t in test_targets if t] + +if test_targets: + mass_content = f"[群发测试 {datetime.now().strftime('%H:%M:%S')}] Frida WCDB群发验证" + 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')} 条") + results["mass_message"] = r_mass +else: + print(" ⚠️ 跳过群发（无联系人）") + +# ============================================================ +# 模块7：朋友圈 +# ============================================================ +print(f"\n{'='*50}") +print("模块7：朋友圈") +print(f"{'='*50}") + +r_moments = test("读取朋友圈", lambda: rpc.get_moments({"limit": 10})) +if r_moments.get("success"): + print(f" 朋友圈条数: {r_moments.get('count', 0)}") + for m in r_moments.get("moments", [])[:3]: + print(f" - {str(m.get('content',''))[:60]}") + +moment_content = f"[SDK验证{datetime.now().strftime('%m/%d %H:%M')}] Frida WCDB无感控制微信成功 好友:{r_cnt.get('type3_friends','?')}人" +r_post = test("发布朋友圈(WCDB)", lambda: rpc.post_moment({"content": moment_content})) +print(f" SNS ID: {r_post.get('sns_id', 'N/A')}") + +results["moments"] = {"read": r_moments, "post": r_post} + +# ============================================================ +# 模块8：标签管理 +# ============================================================ +print(f"\n{'='*50}") +print("模块8：标签管理") +print(f"{'='*50}") + +r_labels = test("获取标签列表", rpc.get_labels) +if r_labels.get("success"): + print(f" 标签数量: {r_labels.get('count', 0)}") + for l in r_labels.get("labels", [])[:5]: + print(f" - [{l.get('labelId')}] {l.get('labelName')} (成员:{l.get('memberCount','?')})") + +# 获取第一个标签的成员 +if r_labels.get("success") and r_labels.get("labels"): + first_label = r_labels["labels"][0] + lid = first_label.get("labelId") + if lid: + r_lm = test(f"标签成员(ID={lid})", lambda: rpc.get_label_members({"label_id": lid})) + if r_lm.get("success"): + print(f" 成员数: {r_lm.get('count', 0)}") + +results["labels"] = r_labels + +# ============================================================ +# 模块9：搜索 +# ============================================================ +print(f"\n{'='*50}") +print("模块9：搜索") +print(f"{'='*50}") + +r_search1 = test("搜索联系人(关键词:游)", lambda: rpc.search_contacts({"keyword": "游"})) +if r_search1.get("success"): + print(f" 结果: {r_search1.get('count', 0)} 条") + for c in r_search1.get("results", [])[:3]: + print(f" - {c.get('nickname','')} ({c.get('username','')})") + +r_search2 = test("搜索消息(关键词:验证)", lambda: rpc.search_messages({"keyword": "验证"})) +if r_search2.get("success"): + print(f" 结果: {r_search2.get('count', 0)} 条") + +results["search"] = {"contacts": r_search1, "messages": r_search2} + +# ============================================================ +# 模块10：好友操作 +# ============================================================ +print(f"\n{'='*50}") +print("模块10：好友操作") +print(f"{'='*50}") + +r_remark = test("设置好友备注(filehelper)", lambda: rpc.set_remark({"wxid": "filehelper", "remark": "文件助手[SDK测试]"})) +print(f" 影响行数: {r_remark.get('affected_rows', 0)}") + +results["friend_ops"] = {"remark": r_remark} + +# ============================================================ +# 模块11：原始SQL查询 +# ============================================================ +print(f"\n{'='*50}") +print("模块11：原始SQL查询（高级）") +print(f"{'='*50}") + +r_sql1 = test("查询最近会话", lambda: rpc.raw_query({"sql": "SELECT username, content, createTime FROM conversation ORDER BY createTime DESC LIMIT 10"})) +if r_sql1.get("success"): + print(f" 会话数: {r_sql1.get('count', 0)}") + for row in r_sql1.get("rows", [])[:3]: + print(f" - {row.get('username','')} | {str(row.get('content',''))[:40]}") + +r_sql2 = test("查询消息统计", lambda: rpc.raw_query({"sql": "SELECT talker, COUNT(*) as cnt FROM message GROUP BY talker ORDER BY cnt DESC LIMIT 10"})) +if r_sql2.get("success"): + print(f" 活跃会话Top5:") + for row in r_sql2.get("rows", [])[:5]: + print(f" - {row.get('talker','')} ({row.get('cnt',0)} 条)") + +results["raw_sql"] = {"sessions": r_sql1, "stats": r_sql2} + +# ============================================================ +# 截图验证 +# ============================================================ +print(f"\n{'='*50}") +print("截图验证") +print(f"{'='*50}") + +# 导航到文件传输助手截图 +r_nav = rpc.navigate_to_chat({"wxid": "filehelper"}) +print(f" 导航: {r_nav}") +sc = adb_sc(f"v3_final_chat_{ts}", wait=2) + +# ============================================================ +# 汇总报告 +# ============================================================ +total = passed + failed +pass_rate = (passed / total * 100) if total > 0 else 0 + +print(f"\n{'='*60}") +print("验证汇总报告") +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 消息发送: {'✅ WCDB直接写入' if r_send1.get('success') else '❌ 失败'}") +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 - 微信全功能Frida验证报告 v3", + "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}") + +script.unload() +session.detach() +print("完成！")