323 lines
12 KiB
Python
323 lines
12 KiB
Python
#!/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}")
|