Files
workphone-sdk/verify_v12.py

392 lines
15 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
工作手机 SDK - 完整验证脚本 v12
Frida 16.x: 驼峰转下划线 (getAccountInfo -> get_account_info)
"""
import frida, json, time, subprocess, sys, re
PHONE_IP = '192.168.110.80'
JS_FILE = '/Users/karuo/Documents/开发/2、私域银行/工作手机/sdk/agent/hook/wechat_api_v8_fix.js'
SCREENSHOT_DIR = '/Users/karuo/Documents/开发/2、私域银行/工作手机/开发文档/6、测试'
results = {}
errors = []
passed_count = 0
total_count = 0
def camel_to_snake(name):
"""驼峰转下划线: getAccountInfo -> get_account_info"""
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 screenshot(name):
path = f'{SCREENSHOT_DIR}/v12_{name}.png'
try:
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', f'screencap -p /sdcard/v12_{name}.png'], timeout=5, capture_output=True)
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'pull', f'/sdcard/v12_{name}.png', path], timeout=10, capture_output=True)
print(f" 📸 截图: v12_{name}.png")
except: pass
def wake_phone():
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'input keyevent 26'], timeout=5, capture_output=True)
time.sleep(0.5)
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'input swipe 540 1800 540 900'], timeout=5, capture_output=True)
def get_wechat_pid():
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 and ':tools' not in line and ':appbrand' not in line:
parts = line.split()
if len(parts) >= 2:
try: return int(parts[1])
except: pass
return None
def c(ex, method_camel, *args):
"""调用exports_sync方法自动转换为下划线格式"""
snake = camel_to_snake(method_camel)
try:
return getattr(ex, snake)(*args)
except Exception as e:
raise Exception(f"{snake}: {str(e)[:80]}")
def test(name, func):
global passed_count, total_count
total_count += 1
try:
r = func()
passed_count += 1
return r
except Exception as e:
errors.append(f"{name}: {str(e)[:100]}")
print(f"{name}: {str(e)[:100]}")
return None
def main():
global passed_count, total_count
print("=" * 70)
print("🔥 工作手机 SDK 完整验证 v12 (下划线方法名)")
print(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 70)
wake_phone()
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'forward', 'tcp:27042', 'tcp:27042'], timeout=5, capture_output=True)
pid = get_wechat_pid()
if not pid:
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)
pid = get_wechat_pid()
if not pid:
print("❌ 无法启动微信"); return 0, 0
print(f"✅ 微信PID: {pid}")
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()
script = session.create_script(js_code)
script.load()
time.sleep(3)
ex = script.exports_sync
# 验证方法名转换
methods = [m for m in dir(ex) if not m.startswith('_')]
print(f"✅ 可用方法: {len(methods)}")
print(f" 转换示例: getAccountInfo -> {camel_to_snake('getAccountInfo')}")
print(f" 转换示例: getAllStats -> {camel_to_snake('getAllStats')}")
print(f" 转换示例: getContacts -> {camel_to_snake('getContacts')}")
print(f" 转换示例: sendMessage -> {camel_to_snake('sendMessage')}")
print("\n" + "=" * 70)
print("📊 全量接口验证26项")
print("=" * 70)
# 1. ping
print("\n【1】系统ping")
r = test("ping", lambda: ex.ping())
if r: print(f"{r}")
results['ping'] = r
# 2. 账号信息
print("\n【2】账号信息")
r = test("account", lambda: c(ex, 'getAccountInfo'))
if r:
print(f" ✅ 账号: {r.get('nickname','?')} | wxid: {r.get('wxid','?')}")
print(f" 手机: {r.get('phone','?')} | 地区: {r.get('region','?')}")
results['account'] = r
# 3. 全量统计
print("\n【3】全量数据统计")
r = test("stats", lambda: c(ex, 'getAllStats'))
if r:
print(f" ✅ 好友:{r.get('friends',0)} 群:{r.get('groups',0)} 消息:{r.get('messages',0)}")
print(f" 图片:{r.get('images',0)} 视频:{r.get('videos',0)} 文件:{r.get('files',0)}")
results['stats'] = r
# 4. 联系人列表
print("\n【4】联系人列表前20个")
r = test("contacts", lambda: c(ex, 'getContacts', 20, 0))
if r:
rows = r.get('rows', [])
print(f" ✅ 联系人: {len(rows)}")
for contact in rows[:5]:
print(f" - {contact.get('nickName','?')} ({contact.get('username','?')})")
results['contacts'] = {'count': len(rows), 'sample': rows[:5]}
# 5. 联系人数量
print("\n【5】联系人数量")
r = test("contact_count", lambda: c(ex, 'getContactCount'))
if r:
print(f" ✅ 好友: {r.get('friends',0)} | 总计: {r.get('total',0)}")
results['contact_count'] = r
# 6. 群组列表
print("\n【6】群组列表")
groups_data = None
r = test("groups", lambda: c(ex, 'getGroups', 50))
if r:
rows = r.get('rows', [])
print(f" ✅ 群组: {len(rows)}")
for g in rows[:3]:
print(f" - {g.get('nickName', g.get('NickName','?'))} ({g.get('username','?')})")
results['groups'] = {'count': len(rows), 'sample': rows[:3]}
groups_data = rows
# 7. 标签
print("\n【7】标签列表")
r = test("labels", lambda: c(ex, 'getLabels'))
if r:
rows = r.get('rows', [])
print(f" ✅ 标签: {len(rows)}")
for l in rows[:5]:
print(f" - {l.get('labelName', l.get('name','?'))}")
results['labels'] = {'count': len(rows)}
# 8. 最近消息
print("\n【8】最近消息24小时")
r = test("recent_messages", lambda: c(ex, 'getRecentMessages', 24))
if r:
rows = r.get('rows', [])
print(f" ✅ 消息: {len(rows)}")
results['recent_messages'] = {'count': len(rows)}
# 9. 发送消息(最重要!)
print("\n【9】🔥 发送消息到文件传输助手")
msg_content = f'[工作手机SDK v12] 自动发送测试 {time.strftime("%H:%M:%S")}'
r = test("send_message", lambda: c(ex, 'sendMessage', 'filehelper', msg_content, 1))
if r:
print(f" ✅ 发送结果: {json.dumps(r, ensure_ascii=False)[:300]}")
results['send_message'] = r
time.sleep(2)
# 导航到文件传输助手截图
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell',
'am start -n com.tencent.mm/.ui.LauncherUI --ei "chat_type" 1 --es "username" "filehelper"'],
timeout=5, capture_output=True)
time.sleep(2)
screenshot('filehelper_after_send')
# 10. 群发消息
print("\n【10】群发消息前3个联系人")
try:
contacts_r = c(ex, 'getContacts', 5, 0)
rows = contacts_r.get('rows', [])
targets = [c_item.get('username','') for c_item in rows if c_item.get('username','') and not c_item.get('username','').startswith('gh_')][:3]
if targets:
r = test("mass_message", lambda: c(ex, 'massMessage', targets, f'[群发测试] {time.strftime("%H:%M:%S")}'))
if r:
print(f" ✅ 群发: 目标{len(targets)}人, 结果: {json.dumps(r, ensure_ascii=False)[:150]}")
results['mass_message'] = r
else:
print(f" ⚠️ 没有有效目标")
except Exception as e:
print(f" ❌ 群发: {e}")
errors.append(f"mass_message: {str(e)[:80]}")
total_count += 1
# 11. 群成员
print("\n【11】群成员第一个群")
if groups_data and len(groups_data) > 0:
chatroom_id = groups_data[0].get('username','')
r = test("group_members", lambda: c(ex, 'getGroupMembers', chatroom_id))
if r:
members = r.get('rows', [])
print(f" ✅ 群: {chatroom_id} | 成员: {len(members)}")
results['group_members'] = {'count': len(members), 'chatroom': chatroom_id}
else:
print(f" ⚠️ 没有群组数据")
# 12. 朋友圈DB信息
print("\n【12】朋友圈DB信息")
r = test("moments_db", lambda: c(ex, 'getSnsDbInfo'))
if r:
print(f" ✅ 朋友圈DB: {json.dumps(r, ensure_ascii=False)[:300]}")
results['moments_db'] = r
# 13. 朋友圈列表
print("\n【13】朋友圈列表")
r = test("moments", lambda: c(ex, 'getMoments', 10))
if r:
rows = r.get('rows', [])
print(f" ✅ 朋友圈: {len(rows)}")
for m in rows[:2]:
print(f" - {str(m)[:100]}")
results['moments'] = {'count': len(rows)}
# 14. 发布朋友圈(最重要!)
print("\n【14】🔥 发布朋友圈")
moment_content = f'[工作手机SDK v12] {time.strftime("%Y-%m-%d %H:%M:%S")}'
r = test("post_moment", lambda: c(ex, 'postMoment', moment_content))
if r:
print(f" ✅ 发布结果: {json.dumps(r, ensure_ascii=False)[:300]}")
results['post_moment'] = r
time.sleep(2)
screenshot('after_post_moment')
# 15. 视频号信息
print("\n【15】视频号信息")
r = test("finder_info", lambda: c(ex, 'getFinderInfo'))
if r:
print(f" ✅ 视频号: {json.dumps(r, ensure_ascii=False)[:300]}")
results['finder_info'] = r
# 16. 视频号联系人
print("\n【16】视频号联系人")
r = test("finder_contacts", lambda: c(ex, 'getFinderContacts', 10))
if r:
rows = r.get('rows', [])
print(f" ✅ 视频号联系人: {len(rows)}")
for fc in rows[:3]:
print(f" - {str(fc)[:80]}")
results['finder_contacts'] = {'count': len(rows)}
# 17. 视频号账号
print("\n【17】视频号账号")
r = test("finder_accounts", lambda: c(ex, 'getFinderAccounts'))
if r:
print(f" ✅ 视频号账号: {json.dumps(r, ensure_ascii=False)[:200]}")
results['finder_accounts'] = r
# 18. 收藏
print("\n【18】收藏列表")
r = test("favorites", lambda: c(ex, 'getFavorites', 10))
if r:
rows = r.get('rows', [])
print(f" ✅ 收藏: {len(rows)}")
for fav in rows[:3]:
print(f" - {str(fav)[:80]}")
results['favorites'] = {'count': len(rows)}
# 19. 钱包信息
print("\n【19】钱包信息")
r = test("wallet_info", lambda: c(ex, 'getWalletInfo'))
if r:
print(f" ✅ 钱包: {json.dumps(r, ensure_ascii=False)[:300]}")
results['wallet_info'] = r
# 20. 钱包流水
print("\n【20】钱包流水")
r = test("wallet_ledger", lambda: c(ex, 'getWalletLedger', 10))
if r:
rows = r.get('rows', [])
print(f" ✅ 钱包流水: {len(rows)}")
for row in rows[:3]:
print(f" - {str(row)[:100]}")
results['wallet_ledger'] = {'count': len(rows), 'sample': rows[:3]}
# 21. 红包
print("\n【21】红包记录")
r = test("lucky_money", lambda: c(ex, 'getLuckyMoney', 10))
if r:
rows = r.get('rows', [])
print(f" ✅ 红包: {len(rows)}")
for row in rows[:3]:
print(f" - {str(row)[:100]}")
results['lucky_money'] = {'count': len(rows), 'sample': rows[:3]}
# 22. 图片统计
print("\n【22】图片统计")
r = test("image_count", lambda: c(ex, 'getImageCount'))
if r:
print(f" ✅ 图片: {json.dumps(r, ensure_ascii=False)}")
results['image_count'] = r
# 23. 视频
print("\n【23】视频列表")
r = test("videos", lambda: c(ex, 'getVideoList', 5))
if r:
rows = r.get('rows', [])
print(f" ✅ 视频: {len(rows)}")
results['videos'] = {'count': len(rows)}
# 24. 文件
print("\n【24】文件列表")
r = test("files", lambda: c(ex, 'getFileList', 5))
if r:
rows = r.get('rows', [])
print(f" ✅ 文件: {len(rows)}")
results['files'] = {'count': len(rows)}
# 25. 小程序
print("\n【25】小程序信息")
r = test("appbrand", lambda: c(ex, 'getAppBrandList'))
if r:
print(f" ✅ 小程序: {json.dumps(r, ensure_ascii=False)[:200]}")
results['appbrand'] = r
# 26. 原始SQL
print("\n【26】原始SQL查询联系人类型分布")
r = test("raw_sql", lambda: c(ex, 'rawQuery', 'main', 'SELECT type, COUNT(*) as cnt FROM rcontact GROUP BY type ORDER BY cnt DESC LIMIT 10'))
if r:
print(f" ✅ SQL结果: {json.dumps(r, ensure_ascii=False)[:300]}")
results['raw_sql'] = r
# 最终截图
print("\n📸 最终截图...")
screenshot('final')
# 汇总
print("\n" + "=" * 70)
print(f"📊 验证结果: {passed_count}/{total_count} 通过 ({passed_count/total_count*100:.0f}%)")
print("=" * 70)
if errors:
print("\n❌ 失败项:")
for e in errors:
print(f" - {e}")
print("\n✅ 真实数据汇总:")
acc = results.get('account', {})
stats = results.get('stats', {})
cc = results.get('contact_count', {})
print(f" 📱 账号: {acc.get('nickname','?')} | wxid: {acc.get('wxid','?')}")
print(f" 👥 好友: {cc.get('friends', stats.get('friends','?'))} | 总联系人: {cc.get('total','?')}")
print(f" 💬 群组: {results.get('groups',{}).get('count', stats.get('groups','?'))}")
print(f" 🏷️ 标签: {results.get('labels',{}).get('count','?')}")
print(f" 📨 24h消息: {results.get('recent_messages',{}).get('count','?')}")
print(f" 🖼️ 图片: {stats.get('images','?')}")
print(f" 🎬 视频: {stats.get('videos','?')}")
print(f" 📁 文件: {stats.get('files','?')}")
print(f" ⭐ 收藏: {results.get('favorites',{}).get('count','?')}")
print(f" 💰 钱包流水: {results.get('wallet_ledger',{}).get('count','?')}")
print(f" 🧧 红包: {results.get('lucky_money',{}).get('count','?')}")
print(f" 📹 视频号联系人: {results.get('finder_contacts',{}).get('count','?')}")
result_file = f'{SCREENSHOT_DIR}/v12_verify_results.json'
with open(result_file, 'w', encoding='utf-8') as f:
json.dump({'timestamp': int(time.time()), 'time': time.strftime('%Y-%m-%d %H:%M:%S'),
'passed': passed_count, 'total': total_count,
'pass_rate': f"{passed_count/total_count*100:.0f}%",
'results': results, 'errors': errors}, f, ensure_ascii=False, indent=2)
print(f"\n💾 结果已保存: {result_file}")
return passed_count, total_count
if __name__ == '__main__':
passed, total = main()
print(f"\n🎯 最终: {passed}/{total} ({passed/total*100:.0f}%)")
sys.exit(0 if passed >= total * 0.8 else 1)