diff --git a/sdk/tests/run_live_verification.py b/sdk/tests/run_live_verification.py new file mode 100644 index 0000000000..8972c99129 --- /dev/null +++ b/sdk/tests/run_live_verification.py @@ -0,0 +1,359 @@ +#!/usr/bin/env python3 +""" +工作手机SDK - 真机验证脚本 +通过WiFi连接frida-server,加载wechat_hook_v2.js,逐个验证所有RPC方法 +""" +import frida +import json +import time +import os +import sys +import subprocess +from datetime import datetime +from typing import Dict, Any, List, Tuple + +# 配置 +DEVICE_IP = "192.168.0.12" +FRIDA_PORT = 27042 +WECHAT_PACKAGE = "com.tencent.mm" +DEVICE_SERIAL = "xgfe65eimrrofyws" + +# Hook脚本路径 +SCRIPT_PATH = os.path.join(os.path.dirname(__file__), '..', 'agent', 'hook', 'wechat_hook_v2.js') +if not os.path.exists(SCRIPT_PATH): + # 尝试从GitHub仓库路径 + SCRIPT_PATH = os.path.expanduser("~/Documents/GitHub/workphone-sdk/sdk/agent/hook/wechat_hook_v2.js") + +# 截图保存目录 +SCREENSHOT_DIR = os.path.join(os.path.dirname(__file__), '..', '..', 'verification_screenshots') +os.makedirs(SCREENSHOT_DIR, exist_ok=True) + +# 结果文件 +RESULT_FILE = os.path.join(os.path.dirname(__file__), '..', '..', 'verification_results.json') + + +def take_screenshot(name: str) -> str: + """截取手机屏幕""" + filename = f"{name}_{int(time.time())}.png" + filepath = os.path.join(SCREENSHOT_DIR, filename) + try: + subprocess.run( + ["adb", "-s", DEVICE_SERIAL, "shell", "screencap", "-p", f"/sdcard/screenshot.png"], + capture_output=True, timeout=10 + ) + subprocess.run( + ["adb", "-s", DEVICE_SERIAL, "pull", "/sdcard/screenshot.png", filepath], + capture_output=True, timeout=10 + ) + if os.path.exists(filepath): + return filepath + except Exception as e: + print(f" 截图失败: {e}") + return "" + + +def connect_frida(): + """连接Frida并attach微信""" + print(f"[1/4] 连接 Frida Server ({DEVICE_IP}:{FRIDA_PORT})...") + mgr = frida.get_device_manager() + device = mgr.add_remote_device(f"{DEVICE_IP}:{FRIDA_PORT}") + print(f" 设备: {device.name}") + + # 找微信主进程 + print("[2/4] 查找微信主进程...") + procs = [p for p in device.enumerate_processes() if p.name == WECHAT_PACKAGE] + if not procs: + # 尝试通过应用列表 + apps = [a for a in device.enumerate_applications() if a.identifier == WECHAT_PACKAGE] + if apps and apps[0].pid > 0: + pid = apps[0].pid + else: + print(" 微信主进程未找到!尝试启动微信...") + subprocess.run( + ["adb", "-s", DEVICE_SERIAL, "shell", "am", "start", "-n", + "com.tencent.mm/.ui.LauncherUI"], + capture_output=True, timeout=10 + ) + time.sleep(3) + procs = [p for p in device.enumerate_processes() if p.name == WECHAT_PACKAGE] + if not procs: + raise Exception("微信主进程无法启动") + pid = procs[0].pid + else: + pid = procs[0].pid + + print(f" 微信 PID: {pid}") + + # Attach + print("[3/4] Attach 微信进程...") + session = device.attach(pid) + print(" Attach 成功!") + + # 加载脚本 + print(f"[4/4] 加载 Hook 脚本...") + if not os.path.exists(SCRIPT_PATH): + raise FileNotFoundError(f"Hook脚本不存在: {SCRIPT_PATH}") + + with open(SCRIPT_PATH, 'r', encoding='utf-8') as f: + source = f.read() + + script = session.create_script(source) + + messages = [] + def on_message(message, data): + if message.get('type') == 'send': + messages.append(message.get('payload', {})) + elif message.get('type') == 'error': + print(f" [脚本错误] {message.get('description', '')}") + + script.on('message', on_message) + script.load() + + rpc = script.exports_sync + + # 验证连接 + try: + pong = rpc.ping() + print(f" RPC ping: {pong}") + except Exception as e: + print(f" RPC ping 失败: {e}") + + return device, session, script, rpc + + +# 所有需要验证的操作(按模块分组) +VERIFICATION_PLAN = { + "H15_消息接收": { + "get_messages": {"conversation_id": "", "limit": 3}, + "get_recent_messages": {"limit": 5}, + "search_messages": {"keyword": "你好", "limit": 3}, + }, + "H16_联系人": { + "get_contacts": {"limit": 10}, + "get_contact_info": {"wxid": "filehelper"}, + "search_contacts": {"keyword": "文件", "limit": 5}, + }, + "H17_消息发送": { + "send_message": {"to_id": "filehelper", "content": "[SDK验证] 消息发送测试 " + datetime.now().strftime("%H:%M:%S"), "msg_type": "text"}, + }, + "H18_好友请求": { + "get_friend_requests": {"limit": 5}, + }, + "H19_好友管理": { + "set_friend_remark": {"wxid": "filehelper", "remark": "文件传输助手"}, + }, + "H20_朋友圈发布": { + # 跳过实际发布,只验证接口存在 + }, + "H21_朋友圈浏览": { + "get_moments": {"wxid": "", "limit": 3}, + }, + "H22_群管理": { + "get_groups": {"limit": 10}, + }, + "H23_账号管理": { + "get_profile": {}, + "check_account_status": {}, + }, + "H24_账号安全": { + "get_login_devices": {}, + }, + "H25_支付": { + "get_wallet_balance": {}, + }, + "H26_二维码": { + "generate_my_qr_code": {}, + }, + "H27_视频号": { + "browse_channels": {"limit": 3}, + }, + "H28_标签": { + "get_labels": {}, + }, + "H29_收藏": { + "get_favorites": {"limit": 5}, + }, + "H30_设置": { + # 只读操作 + }, + "H31_搜索": { + "global_search": {"keyword": "微信", "limit": 5}, + }, + "H32_小程序": { + "get_recent_mini_programs": {}, + }, + "H33_文件传输": { + "send_image": {"to_id": "filehelper", "image_path": "/sdcard/DCIM/test.jpg"}, + }, + "H34_消息转发": { + # 需要msg_id,跳过 + }, + "H35_注册登录": { + "check_login_state": {}, + "get_sim_phone": {}, + }, + "H36_公众号": { + "get_official_accounts": {"limit": 5}, + }, + "H37_表情": { + # 需要emoji_md5,跳过 + }, + "H38_浮窗": { + # 需要特定条件,跳过 + }, + "H39_设备信息": { + "get_device_info": {}, + "get_storage_info": {}, + "get_network_info": {}, + }, + "系统_Hook状态": { + "get_hook_status": {}, + "get_process_info": {}, + "get_wechat_version": {}, + }, +} + +# RPC方法名映射(snake_case -> camelCase) +def to_camel(name: str) -> str: + parts = name.split('_') + return parts[0] + ''.join(p.capitalize() for p in parts[1:]) + + +def run_verification(): + """运行完整验证""" + print("=" * 60) + print(" 工作手机SDK - Frida无线真机验证") + print(f" 设备: {DEVICE_IP}:{FRIDA_PORT}") + print(f" 时间: {datetime.now().isoformat()}") + print("=" * 60) + print() + + # 连接 + device, session, script, rpc = connect_frida() + print("\n✅ Frida 连接成功!开始验证...\n") + + # 初始截图 + take_screenshot("00_initial_state") + + results = [] + total = 0 + passed = 0 + failed = 0 + + for module_name, actions in VERIFICATION_PLAN.items(): + print(f"\n{'='*50}") + print(f" 模块: {module_name}") + print(f"{'='*50}") + + for action, params in actions.items(): + total += 1 + rpc_method = to_camel(action) + print(f"\n [{total}] {action} -> rpc.{rpc_method}()") + print(f" 参数: {json.dumps(params, ensure_ascii=False)[:80]}") + + start_time = time.time() + try: + fn = getattr(rpc, rpc_method, None) + if fn is None: + # 尝试下划线版本 + fn = getattr(rpc, action, None) + + if fn is None: + result = {"success": False, "error": f"RPC方法不存在: {rpc_method}"} + status = "FAIL" + failed += 1 + else: + result = fn(params) if params else fn({}) + if isinstance(result, dict) and result.get("success", False): + status = "PASS" + passed += 1 + elif isinstance(result, dict) and "error" in result: + status = "FAIL" + failed += 1 + else: + # 有返回就算通过 + status = "PASS" + passed += 1 + + except Exception as e: + result = {"success": False, "error": str(e)} + status = "FAIL" + failed += 1 + + latency = int((time.time() - start_time) * 1000) + + icon = "✅" if status == "PASS" else "❌" + print(f" {icon} {status} ({latency}ms)") + if isinstance(result, dict): + # 打印简要结果 + preview = json.dumps(result, ensure_ascii=False)[:120] + print(f" 结果: {preview}") + + # 截图 + screenshot = take_screenshot(f"{module_name}_{action}") + + results.append({ + "module": module_name, + "action": action, + "rpc_method": rpc_method, + "params": params, + "status": status, + "latency_ms": latency, + "result": result if isinstance(result, dict) else str(result), + "screenshot": screenshot, + "timestamp": datetime.now().isoformat(), + }) + + # 每个操作间隔一下,避免触发风控 + time.sleep(0.5) + + # 最终截图 + take_screenshot("99_final_state") + + # 保存结果 + report = { + "title": "工作手机SDK - Frida无线真机验证报告", + "device": { + "ip": DEVICE_IP, + "port": FRIDA_PORT, + "serial": DEVICE_SERIAL, + "model": "2312DRAABC", + "android": "13", + }, + "test_time": datetime.now().isoformat(), + "total": total, + "passed": passed, + "failed": failed, + "pass_rate": f"{passed/total*100:.1f}%" if total > 0 else "0%", + "results": results, + } + + with open(RESULT_FILE, 'w', encoding='utf-8') as f: + json.dump(report, f, ensure_ascii=False, indent=2) + + print(f"\n{'='*60}") + print(f" 验证完成!") + print(f" 总计: {total} | 通过: {passed} | 失败: {failed}") + print(f" 通过率: {passed/total*100:.1f}%" if total > 0 else " 无测试") + print(f" 结果: {RESULT_FILE}") + print(f" 截图: {SCREENSHOT_DIR}") + print(f"{'='*60}") + + # 清理 + try: + script.unload() + session.detach() + except: + pass + + return report + + +if __name__ == "__main__": + try: + report = run_verification() + except Exception as e: + print(f"\n❌ 验证失败: {e}") + import traceback + traceback.print_exc() + sys.exit(1)