diff --git a/verify_v9_final.py b/verify_v9_final.py new file mode 100644 index 0000000000..4739da982e --- /dev/null +++ b/verify_v9_final.py @@ -0,0 +1,322 @@ +#!/usr/bin/env python3 +"""工作手机 SDK v9-final 完整验证脚本""" +import frida, json, time, subprocess, re, sys, os + +PHONE_IP = '192.168.110.80' +JS_FILE = '/Users/karuo/Documents/开发/2、私域银行/工作手机/sdk/agent/hook/wechat_api_v9_final.js' +SCREENSHOT_DIR = '/Users/karuo/Documents/开发/2、私域银行/工作手机/开发文档/6、测试' + +def camel_to_snake(name): + s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) + return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower() + +def call(ex, method, *args): + snake = camel_to_snake(method) + return getattr(ex, snake)(*args) + +def screenshot(name): + ts = int(time.time()) + remote = f'/sdcard/{name}_{ts}.png' + local = f'{SCREENSHOT_DIR}/{name}_{ts}.png' + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', f'screencap -p {remote}'], timeout=5, capture_output=True) + time.sleep(0.5) + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'pull', remote, local], timeout=10, capture_output=True) + return local + +def p(label, data, show_rows=3): + print(f"\n{'='*50}") + print(f" {label}") + print(f"{'='*50}") + if isinstance(data, dict): + if 'error' in data: + print(f" ❌ 错误: {data['error']}") + elif 'rows' in data: + print(f" ✅ 返回 {data.get('count', len(data['rows']))} 条") + for row in data['rows'][:show_rows]: + print(f" {json.dumps(row, ensure_ascii=False)}") + else: + for k, v in data.items(): + if isinstance(v, (dict, list)): + print(f" {k}: {json.dumps(v, ensure_ascii=False)[:100]}") + else: + print(f" {k}: {v}") + else: + print(f" {data}") + +print("\n" + "="*60) +print(" 工作手机 SDK v9-final 完整验证") +print("="*60) + +# 设置ADB端口转发 +subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'forward', 'tcp:27042', 'tcp:27042'], timeout=5, capture_output=True) + +# 获取微信PID +result = subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'ps -A'], capture_output=True, text=True, timeout=5) +pid = None +for line in result.stdout.split('\n'): + if 'com.tencent.mm' in line and ':push' not in line and ':sandbox' not in line and ':tools' not in line and ':appbrand' not in line and ':xweb' not in line: + parts = line.split() + if len(parts) >= 2: + try: + pid = int(parts[1]) + print(f"\n 微信主进程 PID: {pid}") + break + except: pass + +if not pid: + print("❌ 微信主进程未运行,正在启动...") + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'monkey -p com.tencent.mm -c android.intent.category.LAUNCHER 1'], timeout=10, capture_output=True) + time.sleep(8) + result = subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'ps -A'], capture_output=True, text=True, timeout=5) + for line in result.stdout.split('\n'): + if 'com.tencent.mm' in line and ':push' not in line and ':sandbox' not in line: + parts = line.split() + if len(parts) >= 2: + try: pid = int(parts[1]); break + except: pass + +# 连接Frida +dm = frida.get_device_manager() +device = dm.add_remote_device('localhost:27042') +session = device.attach(pid) + +with open(JS_FILE, 'r', encoding='utf-8') as f: + js_code = f.read() + +messages = [] +def on_message(msg, data): + if msg.get('type') == 'send': + messages.append(msg['payload']) + print(f" [Frida] {msg['payload']}") + +script = session.create_script(js_code) +script.on('message', on_message) +script.load() +print(" 等待DB初始化...") +time.sleep(4) # 等待Java.choose完成 + +ex = script.exports_sync + +# ===== 测试开始 ===== +results = {} +passed = 0 +failed = 0 + +def test(name, func): + global passed, failed + try: + r = func() + if isinstance(r, dict) and 'error' in r: + print(f" ❌ [{name}] {r['error'][:80]}") + failed += 1 + results[name] = {'status': 'FAIL', 'error': r['error'][:80]} + else: + count = r.get('count', '-') if isinstance(r, dict) else '-' + print(f" ✅ [{name}] count={count}") + passed += 1 + results[name] = {'status': 'PASS', 'data': r} + return r + except Exception as e: + print(f" ❌ [{name}] Exception: {str(e)[:80]}") + failed += 1 + results[name] = {'status': 'FAIL', 'error': str(e)[:80]} + return None + +print("\n--- 1. 系统初始化 ---") +r = test('ping', lambda: call(ex, 'ping')) +if r: + print(f" DB状态: {json.dumps(r.get('dbs', {}), ensure_ascii=False)}") + +r = test('initDbs', lambda: call(ex, 'initDbs')) +if r: + print(f" DB状态: {json.dumps(r.get('status', {}), ensure_ascii=False)}") + +print("\n--- 2. 账号信息 ---") +r = test('getAccountInfo', lambda: call(ex, 'getAccountInfo')) +if r and not r.get('error'): + print(f" 账号: {r.get('wxid')} | 昵称: {r.get('nickname')} | 手机: {r.get('phone')}") + +print("\n--- 3. 联系人(修复版)---") +r = test('getContactCount', lambda: call(ex, 'getContactCount')) +if r: + print(f" 好友: {r.get('friends')} | 总计: {r.get('total')}") + +r = test('getContacts(100)', lambda: call(ex, 'getContacts', 100, 0)) +if r and r.get('count', 0) > 0: + print(f" 前3条: {json.dumps(r['rows'][:3], ensure_ascii=False)[:200]}") + +r = test('searchContacts', lambda: call(ex, 'searchContacts', '游')) +if r: + print(f" 搜索"游": {r.get('count')} 条") + +print("\n--- 4. 标签(修复版)---") +r = test('getLabels', lambda: call(ex, 'getLabels')) +if r and r.get('count', 0) > 0: + print(f" 前5个标签: {json.dumps(r['rows'][:5], ensure_ascii=False)[:200]}") + +print("\n--- 5. 统计 ---") +r = test('getAllStats', lambda: call(ex, 'getAllStats')) +if r and not r.get('error'): + print(f" 好友:{r.get('friends')} 群:{r.get('groups')} 消息:{r.get('messages')} 图片:{r.get('images')} 视频:{r.get('videos')} 标签:{r.get('labels')}") + +print("\n--- 6. 消息 ---") +r = test('getRecentMessages(24h)', lambda: call(ex, 'getRecentMessages', 24)) +if r: + print(f" 24小时消息: {r.get('count')} 条") + if r.get('rows'): + print(f" 最新: {json.dumps(r['rows'][0], ensure_ascii=False)[:150]}") + +# 截图(发送前) +subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'input keyevent 26'], timeout=3, capture_output=True) +time.sleep(0.5) +subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'input swipe 540 1200 540 600'], timeout=3, capture_output=True) +time.sleep(0.5) +sc_before = screenshot('v9_before_send') +print(f"\n 📸 发送前截图: {sc_before}") + +# 发送消息 +print("\n--- 7. 发送消息 ---") +r = test('sendMessage', lambda: call(ex, 'sendMessage', 'filehelper', f'[工作手机SDK v9-final] 测试消息 {time.strftime("%H:%M:%S")}', 1)) +if r: + print(f" msgId: {r.get('msgId')} | DB验证: {r.get('db_verified')}") + print(f" 说明: {r.get('note', '')}") + +# 截图(发送后) +time.sleep(1) +sc_after = screenshot('v9_after_send') +print(f" 📸 发送后截图: {sc_after}") + +print("\n--- 8. 群发消息 ---") +r = test('massMessage', lambda: call(ex, 'massMessage', ['filehelper', 'wxid_xvlpukwd3i8r12'], f'[群发测试] {time.strftime("%H:%M:%S")}')) +if r: + print(f" 总计:{r.get('total')} 成功:{r.get('success')}") + +print("\n--- 9. 群组(修复版)---") +r = test('getGroupCount', lambda: call(ex, 'getGroupCount')) +if r: + print(f" 群组总数: {r.get('rows', [{}])[0].get('total', '?')}") + +r = test('getGroups(10)', lambda: call(ex, 'getGroups', 10)) +if r and r.get('count', 0) > 0: + print(f" 前3个群: {json.dumps(r['rows'][:3], ensure_ascii=False)[:300]}") + # 获取第一个群的成员 + first_group = r['rows'][0].get('chatroomname') + if first_group: + r2 = test('getGroupMembers', lambda: call(ex, 'getGroupMembers', first_group)) + if r2: + print(f" 群{first_group}: {r2.get('memberCount')}人, 群主:{r2.get('owner')}") + +print("\n--- 10. 朋友圈 ---") +# 先触发朋友圈DB加载 +subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', + 'am start -n com.tencent.mm/.plugin.sns.ui.SnsTimeLineUI'], timeout=5, capture_output=True) +time.sleep(3) +r = test('getSnsDbInfo', lambda: call(ex, 'getSnsDbInfo')) +if r: + print(f" SNS DB: {json.dumps(r, ensure_ascii=False)}") + +r = test('getMoments(5)', lambda: call(ex, 'getMoments', 5)) +if r and r.get('count', 0) > 0: + print(f" 朋友圈: {r.get('count')} 条") + print(f" 最新: {json.dumps(r['rows'][0], ensure_ascii=False)[:200]}") + +# 截图朋友圈 +time.sleep(1) +sc_sns = screenshot('v9_moments') +print(f" 📸 朋友圈截图: {sc_sns}") + +print("\n--- 11. 视频号 ---") +r = test('getFinderInfo', lambda: call(ex, 'getFinderInfo')) +if r: + print(f" 视频号表: {r.get('tables', [])[:5]}") + +r = test('getFinderContacts', lambda: call(ex, 'getFinderContacts', 5)) +if r: + print(f" 视频号联系人: {r.get('count')} 条") + +r = test('getFinderAccounts', lambda: call(ex, 'getFinderAccounts')) +if r: + print(f" 视频号账号: {json.dumps(r, ensure_ascii=False)[:200]}") + +print("\n--- 12. 收藏 ---") +r = test('getFavorites(10)', lambda: call(ex, 'getFavorites', 10)) +if r: + if r.get('count', 0) > 0: + print(f" 收藏: {r.get('count')} 条") + elif 'available_tables' in r: + print(f" 可用表: {r.get('available_tables', [])[:5]}") + +print("\n--- 13. 钱包 ---") +r = test('getWalletInfo', lambda: call(ex, 'getWalletInfo')) +if r: + print(f" 钱包表: {r.get('tables', [])[:5]}") + +r = test('getWalletLedger(5)', lambda: call(ex, 'getWalletLedger', 5)) +if r and r.get('count', 0) > 0: + print(f" 钱包流水: {r.get('count')} 条") + print(f" 最新: {json.dumps(r['rows'][0], ensure_ascii=False)[:200]}") + +r = test('getLuckyMoney', lambda: call(ex, 'getLuckyMoney', 5)) +if r: + print(f" 红包记录: {r.get('count')} 条") + +print("\n--- 14. 媒体 ---") +r = test('getImageCount', lambda: call(ex, 'getImageCount')) +if r: + print(f" 图片总数: {r.get('rows', [{}])[0].get('total', '?')}") + +r = test('getVideoList(3)', lambda: call(ex, 'getVideoList', 3)) +if r: + print(f" 视频列表: {r.get('count')} 条") + +r = test('getFileList(3)', lambda: call(ex, 'getFileList', 3)) +if r: + print(f" 文件列表: {r.get('count')} 条") + +print("\n--- 15. 小程序 ---") +r = test('getAppBrandList', lambda: call(ex, 'getAppBrandList')) +if r: + print(f" 小程序表: {r.get('tables', [])[:5]}") + if r.get('common_apps'): + print(f" 常用小程序: {r['common_apps'].get('count')} 条") + +print("\n--- 16. 全局搜索 ---") +r = test('globalSearch', lambda: call(ex, 'globalSearch', '游')) +if r: + print(f" 联系人:{r.get('contacts',{}).get('count')} 群:{r.get('groups',{}).get('count')} 消息:{r.get('messages',{}).get('count')}") + +print("\n--- 17. 添加好友 ---") +r = test('addFriend', lambda: call(ex, 'addFriend', 'wxid_test123456', '你好,我是游条姐')) +if r: + print(f" 结果: {json.dumps(r, ensure_ascii=False)[:200]}") + +print("\n--- 18. 导航 ---") +r = test('navigate_main', lambda: call(ex, 'navigate', 'main')) +if r: + print(f" 导航到主界面: {r}") + +# 最终截图 +time.sleep(2) +sc_final = screenshot('v9_final') +print(f"\n 📸 最终截图: {sc_final}") + +# ===== 汇总 ===== +total = passed + failed +print(f"\n{'='*60}") +print(f" 验证结果: {passed}/{total} 通过 ({100*passed//total if total>0 else 0}%)") +print(f"{'='*60}") + +# 保存结果 +result_file = f'/Users/karuo/Documents/开发/2、私域银行/工作手机/开发文档/6、测试/v9_verify_result_{int(time.time())}.json' +with open(result_file, 'w', encoding='utf-8') as f: + json.dump({ + 'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'), + 'version': 'v9-final', + 'passed': passed, + 'failed': failed, + 'total': total, + 'pass_rate': f'{100*passed//total if total>0 else 0}%', + 'results': results + }, f, ensure_ascii=False, indent=2) +print(f"\n 结果已保存: {result_file}") +print(f" 截图: {sc_before}, {sc_after}, {sc_sns}, {sc_final}")