diff --git a/sdk/agent/hook/system_control.js b/sdk/agent/hook/system_control.js new file mode 100644 index 0000000000..38599f7d66 --- /dev/null +++ b/sdk/agent/hook/system_control.js @@ -0,0 +1,342 @@ +/** + * system_control.js - Frida系统控制扩展 + * 功能：截图、UI导航、页面检测、输入模拟 + * 通过Frida直接调用Android内部API，不依赖ADB + */ + +'use strict'; + +rpc.exports = { + + // ==================== 截图 ==================== + takeScreenshot: function (params) { + /** + * 通过Android Shell命令截图并返回base64 + * 使用Runtime.exec执行screencap（需要Root或同进程权限） + */ + try { + return Java.performNow(function () { + var savePath = (params && params.path) || '/sdcard/frida_screenshot.png'; + var Runtime = Java.use('java.lang.Runtime'); + var runtime = Runtime.getRuntime(); + + // 方案1: 直接用screencap命令 + var process = runtime.exec(['sh', '-c', 'screencap -p ' + savePath]); + process.waitFor(); + + // 读取文件并转base64 + var File = Java.use('java.io.File'); + var file = File.$new(savePath); + if (!file.exists()) { + return { success: false, error: '截图文件不存在' }; + } + + var fileSize = file.length(); + var FileInputStream = Java.use('java.io.FileInputStream'); + var fis = FileInputStream.$new(file); + var bytes = Java.array('byte', new Array(parseInt(fileSize)).fill(0)); + fis.read(bytes); + fis.close(); + + var Base64 = Java.use('android.util.Base64'); + var b64 = Base64.encodeToString(bytes, 2); // NO_WRAP + + return { + success: true, + path: savePath, + size: parseInt(fileSize), + base64_length: b64.length, + base64: b64.substring(0, 200) + '...(truncated)', + full_base64: b64, + }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + takeScreenshotToFile: function (params) { + /** + * 截图保存到指定路径（不返回base64，避免传输大数据） + */ + try { + return Java.performNow(function () { + var savePath = (params && params.path) || '/sdcard/frida_sc_' + Date.now() + '.png'; + var Runtime = Java.use('java.lang.Runtime'); + var runtime = Runtime.getRuntime(); + var process = runtime.exec(['sh', '-c', 'screencap -p ' + savePath]); + process.waitFor(); + + var File = Java.use('java.io.File'); + var file = File.$new(savePath); + if (file.exists()) { + return { success: true, path: savePath, size: parseInt(file.length()) }; + } + return { success: false, error: '截图失败' }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + // ==================== 页面检测 ==================== + getCurrentActivity: function () { + /** + * 获取当前前台Activity名称 + */ + try { + return Java.performNow(function () { + var ActivityThread = Java.use('android.app.ActivityThread'); + var currentApp = ActivityThread.currentApplication(); + var activities = []; + + // 遍历所有Activity + Java.choose('android.app.Activity', { + onMatch: function (activity) { + if (!activity.isFinishing() && !activity.isDestroyed()) { + activities.push({ + class: activity.getClass().getName(), + title: activity.getTitle() ? activity.getTitle().toString() : '', + visible: activity.hasWindowFocus(), + }); + } + }, + onComplete: function () {}, + }); + + return { + success: true, + activities: activities, + foreground: activities.length > 0 ? activities[activities.length - 1].class : 'unknown', + }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + // ==================== UI导航 ==================== + navigateToChat: function (params) { + /** + * 导航到指定聊天窗口 + * 通过微信内部Intent跳转 + */ + try { + return Java.performNow(function () { + var wxid = (params && params.wxid) || 'filehelper'; + var ctx = Java.use('android.app.ActivityThread').currentApplication().getApplicationContext(); + var Intent = Java.use('android.content.Intent'); + var Uri = Java.use('android.net.Uri'); + + // 微信内部跳转到聊天页面 + var intent = Intent.$new(); + intent.setClassName('com.tencent.mm', 'com.tencent.mm.ui.chatting.ChattingUI'); + intent.putExtra('Chat_User', wxid); + intent.addFlags(0x10000000); // FLAG_ACTIVITY_NEW_TASK + + ctx.startActivity(intent); + return { success: true, navigated_to: 'ChattingUI', wxid: wxid }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + navigateToContacts: function () { + /** + * 导航到通讯录页面 + */ + try { + return Java.performNow(function () { + var ctx = Java.use('android.app.ActivityThread').currentApplication().getApplicationContext(); + var Intent = Java.use('android.content.Intent'); + + var intent = Intent.$new(); + intent.setClassName('com.tencent.mm', 'com.tencent.mm.ui.LauncherUI'); + intent.putExtra('LauncherUI.From.Shortcut.Msg', true); + intent.addFlags(0x10000000); + ctx.startActivity(intent); + + return { success: true, navigated_to: 'Contacts' }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + navigateToMe: function () { + /** + * 导航到"我"页面 + */ + try { + return Java.performNow(function () { + var ctx = Java.use('android.app.ActivityThread').currentApplication().getApplicationContext(); + var Intent = Java.use('android.content.Intent'); + + var intent = Intent.$new(); + intent.setClassName('com.tencent.mm', 'com.tencent.mm.ui.LauncherUI'); + intent.addFlags(0x10000000); + ctx.startActivity(intent); + + return { success: true, navigated_to: 'Me' }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + navigateToMoments: function () { + /** + * 导航到朋友圈 + */ + try { + return Java.performNow(function () { + var ctx = Java.use('android.app.ActivityThread').currentApplication().getApplicationContext(); + var Intent = Java.use('android.content.Intent'); + + var intent = Intent.$new(); + intent.setClassName('com.tencent.mm', 'com.tencent.mm.plugin.sns.ui.SnsTimeLineUI'); + intent.addFlags(0x10000000); + ctx.startActivity(intent); + + return { success: true, navigated_to: 'Moments' }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + // ==================== 输入模拟 ==================== + simulateTap: function (params) { + /** + * 模拟点击（通过shell input tap） + */ + try { + return Java.performNow(function () { + var x = (params && params.x) || 0; + var y = (params && params.y) || 0; + var Runtime = Java.use('java.lang.Runtime'); + var runtime = Runtime.getRuntime(); + var process = runtime.exec(['sh', '-c', 'input tap ' + x + ' ' + y]); + process.waitFor(); + return { success: true, tapped: { x: x, y: y } }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + simulateInput: function (params) { + /** + * 模拟文本输入 + */ + try { + return Java.performNow(function () { + var text = (params && params.text) || ''; + var Runtime = Java.use('java.lang.Runtime'); + var runtime = Runtime.getRuntime(); + // 需要对特殊字符转义 + var escaped = text.replace(/ /g, '%s').replace(/'/g, "\\'"); + var process = runtime.exec(['sh', '-c', "input text '" + escaped + "'"]); + process.waitFor(); + return { success: true, input: text }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + simulateKeyEvent: function (params) { + /** + * 模拟按键事件（返回键=4, Home=3, 回车=66） + */ + try { + return Java.performNow(function () { + var keycode = (params && params.keycode) || 4; + var Runtime = Java.use('java.lang.Runtime'); + var runtime = Runtime.getRuntime(); + var process = runtime.exec(['sh', '-c', 'input keyevent ' + keycode]); + process.waitFor(); + return { success: true, keycode: keycode }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + // ==================== 文件操作 ==================== + readFileBase64: function (params) { + /** + * 读取手机上的文件并返回base64 + */ + try { + return Java.performNow(function () { + var filePath = (params && params.path) || ''; + if (!filePath) return { success: false, error: '缺少path参数' }; + + var File = Java.use('java.io.File'); + var file = File.$new(filePath); + if (!file.exists()) return { success: false, error: '文件不存在: ' + filePath }; + + var fileSize = file.length(); + if (fileSize > 5 * 1024 * 1024) { + return { success: false, error: '文件太大(>5MB): ' + fileSize }; + } + + var FileInputStream = Java.use('java.io.FileInputStream'); + var fis = FileInputStream.$new(file); + var bytes = Java.array('byte', new Array(parseInt(fileSize)).fill(0)); + fis.read(bytes); + fis.close(); + + var Base64 = Java.use('android.util.Base64'); + var b64 = Base64.encodeToString(bytes, 2); + + return { success: true, path: filePath, size: parseInt(fileSize), base64: b64 }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + listFiles: function (params) { + /** + * 列出目录下的文件 + */ + try { + return Java.performNow(function () { + var dirPath = (params && params.path) || '/sdcard/'; + var File = Java.use('java.io.File'); + var dir = File.$new(dirPath); + if (!dir.exists() || !dir.isDirectory()) { + return { success: false, error: '目录不存在: ' + dirPath }; + } + var files = dir.listFiles(); + var result = []; + var len = files ? files.length : 0; + for (var i = 0; i < Math.min(len, 50); i++) { + result.push({ + name: files[i].getName(), + is_dir: files[i].isDirectory(), + size: files[i].isFile() ? parseInt(files[i].length()) : 0, + }); + } + return { success: true, path: dirPath, count: len, files: result }; + }); + } catch (e) { + return { success: false, error: String(e) }; + } + }, + + // ==================== 连接状态 ==================== + getConnectionStatus: function () { + return { + success: true, + mode: 'frida_wireless', + transport: 'tcp', + pid: Process.id, + arch: Process.arch, + timestamp: Date.now(), + }; + }, +}; diff --git a/sdk/tests/check_wechat.py b/sdk/tests/check_wechat.py new file mode 100644 index 0000000000..e6eaf2becd --- /dev/null +++ b/sdk/tests/check_wechat.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +"""检查微信进程并尝试attach""" +import frida +import sys + +mgr = frida.get_device_manager() +device = mgr.add_remote_device('192.168.0.12:27042') +print(f'Device: {device.name}') + +# 列出所有微信相关进程 +procs = [p for p in device.enumerate_processes() if 'tencent.mm' in p.name] +print(f'\n微信相关进程:') +for p in procs: + print(f' PID={p.pid} Name={p.name}') + +# 主进程是 com.tencent.mm（不带后缀） +main = [p for p in procs if p.name == 'com.tencent.mm'] +if main: + print(f'\n主进程 PID: {main[0].pid}') +else: + print('\n主进程未找到，检查应用列表...') + apps = [a for a in device.enumerate_applications() if 'tencent.mm' in a.identifier] + for a in apps: + print(f' App: {a.identifier} PID={a.pid}') + if apps and apps[0].pid > 0: + print(f'\n使用应用PID: {apps[0].pid}') + else: + print('\n微信可能未在前台运行，尝试打开微信...') + # 通过push进程确认微信存在 + push = [p for p in procs if ':push' in p.name] + if push: + print(f' push进程存在，微信已安装但可能在后台') + print(' 请先打开微信到前台') + sys.exit(1) diff --git a/sdk/tests/frida_step_verify.py b/sdk/tests/frida_step_verify.py new file mode 100644 index 0000000000..f10ab076f5 --- /dev/null +++ b/sdk/tests/frida_step_verify.py @@ -0,0 +1,414 @@ +#!/usr/bin/env python3 +""" +工作手机SDK - 纯Frida逐步验证器 +所有操作通过WiFi+Frida完成，不依赖ADB/USB +每个功能：Frida导航 -> 执行RPC -> Frida截图 -> 返回数据+截图 +""" +import frida +import json +import time +import os +import base64 +from datetime import datetime + +DEVICE_IP = "192.168.0.12" +FRIDA_PORT = 27042 +WECHAT_PID = 7239 + +# 脚本路径 +BASE_DIR = os.path.expanduser("~/Documents/GitHub/workphone-sdk") +HOOK_SCRIPT = os.path.join(BASE_DIR, "sdk/agent/hook/wechat_hook_v2.js") +SYS_SCRIPT = os.path.join(BASE_DIR, "sdk/agent/hook/system_control.js") + +# 输出目录 +OUTPUT_DIR = os.path.join(BASE_DIR, "verification_steps") +os.makedirs(OUTPUT_DIR, exist_ok=True) + + +class FridaVerifier: + """纯Frida验证器""" + + def __init__(self): + self.device = None + self.session = None + self.wechat_script = None + self.sys_script = None + self.wechat_exports = None + self.sys_exports = None + self.step_count = 0 + self.results = [] + + def connect(self): + """连接Frida Server""" + print(f"[CONNECT] {DEVICE_IP}:{FRIDA_PORT}") + self.device = frida.get_device_manager().add_remote_device(f"{DEVICE_IP}:{FRIDA_PORT}") + self.session = self.device.attach(WECHAT_PID) + print(f" Attached PID: {WECHAT_PID}") + + # 加载微信Hook脚本 + print("[LOAD] wechat_hook_v2.js...") + with open(HOOK_SCRIPT, "r", encoding="utf-8") as f: + src = f.read() + self.wechat_script = self.session.create_script(src) + self.wechat_script.on("message", lambda m, d: None) + self.wechat_script.load() + time.sleep(2) + self.wechat_exports = self.wechat_script.exports_sync + print(f" WeChat methods: {len([x for x in dir(self.wechat_exports) if not x.startswith('_')])}") + print(f" ping: {self.wechat_exports.ping()}") + + # 加载系统控制脚本 + print("[LOAD] system_control.js...") + with open(SYS_SCRIPT, "r", encoding="utf-8") as f: + src2 = f.read() + self.sys_script = self.session.create_script(src2) + self.sys_script.on("message", lambda m, d: None) + self.sys_script.load() + time.sleep(1) + self.sys_exports = self.sys_script.exports_sync + print(f" System methods: {[x for x in dir(self.sys_exports) if not x.startswith('_')]}") + + # 验证连接状态 + status = self.sys_exports.getConnectionStatus() + print(f" Connection: {json.dumps(status)}") + return True + + def take_screenshot(self, name): + """通过Frida截图并保存到本地""" + remote_path = f"/sdcard/verify_{name}_{int(time.time())}.png" + result = self.sys_exports.takeScreenshotToFile({"path": remote_path}) + if not result.get("success"): + print(f" [SCREENSHOT FAIL] {result.get('error', 'unknown')}") + # 尝试用base64方式 + result2 = self.sys_exports.takeScreenshot({"path": remote_path}) + if result2.get("success") and result2.get("full_base64"): + local_path = os.path.join(OUTPUT_DIR, f"{name}.png") + with open(local_path, "wb") as f: + f.write(base64.b64decode(result2["full_base64"])) + print(f" [SCREENSHOT] {local_path} ({os.path.getsize(local_path)} bytes)") + return local_path + return "" + + # 截图成功，通过Frida读取文件 + file_data = self.sys_exports.readFileBase64({"path": remote_path}) + if file_data.get("success") and file_data.get("base64"): + local_path = os.path.join(OUTPUT_DIR, f"{name}.png") + with open(local_path, "wb") as f: + f.write(base64.b64decode(file_data["base64"])) + print(f" [SCREENSHOT] {local_path} ({os.path.getsize(local_path)} bytes)") + return local_path + return "" + + def get_current_page(self): + """获取当前页面""" + result = self.sys_exports.getCurrentActivity() + if result.get("success"): + return result.get("foreground", "unknown") + return "unknown" + + def navigate(self, target, params=None): + """导航到指定页面""" + nav_map = { + "chat": self.sys_exports.navigateToChat, + "contacts": self.sys_exports.navigateToContacts, + "me": self.sys_exports.navigateToMe, + "moments": self.sys_exports.navigateToMoments, + } + fn = nav_map.get(target) + if fn: + result = fn(params) if params else fn() + time.sleep(1.5) + return result + return {"success": False, "error": f"unknown target: {target}"} + + def verify_step(self, module, method, description, nav_target=None, nav_params=None, rpc_params=None): + """ + 执行一个验证步骤: + 1. 导航到对应页面（通过Frida） + 2. 截图（操作前） + 3. 执行RPC方法 + 4. 截图（操作后） + 5. 返回数据 + """ + self.step_count += 1 + step_id = f"{self.step_count:02d}" + + print(f"\n{'='*60}") + print(f" Step {step_id}: [{module}] {method}") + print(f" {description}") + print(f"{'='*60}") + + # Step 1: 导航 + if nav_target: + print(f" [1.NAV] -> {nav_target}") + nav_result = self.navigate(nav_target, nav_params) + print(f" 结果: {json.dumps(nav_result, ensure_ascii=False)[:80]}") + time.sleep(1) + + # Step 2: 获取当前页面 + current_page = self.get_current_page() + print(f" [2.PAGE] 当前页面: {current_page}") + + # Step 3: 操作前截图 + before_img = self.take_screenshot(f"{step_id}_{module}_{method}_before") + + # Step 4: 执行RPC + fn = getattr(self.wechat_exports, method, None) + if fn is None: + print(f" [4.EXEC] ERROR: 方法不存在 '{method}'") + self.results.append({ + "step": self.step_count, "module": module, "method": method, + "description": description, "status": "MISSING", + "page": current_page, "before_img": before_img, "after_img": "", + "data": None, + }) + return + + print(f" [4.EXEC] {method}({json.dumps(rpc_params, ensure_ascii=False)[:60] if rpc_params else ''})") + start = time.time() + try: + if rpc_params is not None: + result = fn(rpc_params) + else: + result = fn() + latency = int((time.time() - start) * 1000) + except Exception as e: + latency = int((time.time() - start) * 1000) + result = {"success": False, "error": str(e)[:200]} + + # 格式化输出 + if isinstance(result, dict): + data_str = json.dumps(result, ensure_ascii=False, indent=2) + print(f" [5.DATA] ({latency}ms)") + for line in data_str.split('\n')[:15]: + print(f" {line}") + if len(data_str.split('\n')) > 15: + print(f" ... (truncated)") + elif isinstance(result, str): + print(f" [5.DATA] ({latency}ms) {result[:100]}") + else: + print(f" [5.DATA] ({latency}ms) {str(result)[:100]}") + + # Step 5: 等待UI响应后截图 + time.sleep(1) + after_img = self.take_screenshot(f"{step_id}_{module}_{method}_after") + + # 判断状态 + if isinstance(result, dict) and result.get("success") == False: + status = "EXEC_ERR" + print(f" [STATUS] EXEC_ERR: {result.get('error', '')[:60]}") + else: + status = "PASS" + print(f" [STATUS] PASS") + + self.results.append({ + "step": self.step_count, "module": module, "method": method, + "description": description, "status": status, "latency_ms": latency, + "page": current_page, "before_img": before_img, "after_img": after_img, + "data": result if isinstance(result, (dict, list, str, bool, int, float)) else str(result), + }) + + def run(self): + """运行所有验证步骤""" + print("\n" + "=" * 60) + print(" 工作手机SDK - 纯Frida无线验证") + print(f" 连接: WiFi TCP {DEVICE_IP}:{FRIDA_PORT}") + print(f" 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + print("=" * 60) + + # ===== 系统模块 ===== + self.verify_step("SYS", "getProcessInfo", "获取微信进程信息(PID/架构/Hook模块数)") + self.verify_step("SYS", "getWechatVersion", "获取微信版本号") + self.verify_step("SYS", "getHookStatus", "检查Hook激活状态") + self.verify_step("SYS", "getVersionCompat", "获取版本兼容信息") + + # ===== 设备信息 ===== + self.verify_step("H39", "getDeviceInfo", "获取手机设备信息(型号/品牌/系统版本)") + self.verify_step("H39", "getNetworkInfo", "获取网络连接信息(WiFi/数据)") + self.verify_step("H39", "getStorageInfo", "获取存储空间信息(总量/可用)") + + # ===== 联系人模块 ===== + self.verify_step("H16", "getContacts", "获取微信联系人列表", + nav_target="contacts", rpc_params={"limit": 10}) + self.verify_step("H16", "getContactInfo", "获取文件传输助手详细信息", + rpc_params={"wxid": "filehelper"}) + self.verify_step("H16", "searchContacts", "搜索联系人(关键词:文件)", + rpc_params={"keyword": "文件", "limit": 5}) + + # ===== 消息发送 ===== + msg = f"[SDK验证] {datetime.now().strftime('%H:%M:%S')} Frida无线发送" + self.verify_step("H17", "sendMessage", f"发送文本消息到文件传输助手: '{msg}'", + nav_target="chat", nav_params={"wxid": "filehelper"}, + rpc_params={"to_id": "filehelper", "content": msg, "msg_type": "text"}) + + # ===== 消息接收 ===== + self.verify_step("H15", "getRecentMessages", "获取最近消息列表", + rpc_params={"limit": 5}) + self.verify_step("H15", "getMessages", "获取filehelper的消息记录", + rpc_params={"conversation_id": "filehelper", "limit": 5}) + self.verify_step("H15", "searchMessages", "搜索消息(关键词:验证)", + rpc_params={"keyword": "验证", "limit": 3}) + + # ===== 好友管理 ===== + self.verify_step("H19", "setFriendRemark", "修改filehelper备注为'SDK文件助手'", + rpc_params={"wxid": "filehelper", "remark": "SDK文件助手"}) + self.verify_step("H18", "getFriendRequests", "获取好友请求列表", + rpc_params={"limit": 5}) + + # ===== 群管理 ===== + self.verify_step("H22", "getGroups", "获取群聊列表", + rpc_params={"limit": 10}) + + # ===== 个人信息 ===== + self.verify_step("H23", "getProfile", "获取个人资料(昵称/微信号/头像)", + nav_target="me", rpc_params={}) + self.verify_step("H23", "checkAccountStatus", "检查账号状态", + rpc_params={}) + + # ===== 朋友圈 ===== + self.verify_step("H21", "getMoments", "获取朋友圈动态", + nav_target="moments", rpc_params={"wxid": "", "limit": 3}) + + # ===== 标签 ===== + self.verify_step("H28", "getLabels", "获取标签列表", rpc_params={}) + + # ===== 收藏 ===== + self.verify_step("H29", "getFavorites", "获取收藏列表", + rpc_params={"limit": 5}) + + # ===== 搜索 ===== + self.verify_step("H31", "globalSearch", "全局搜索(关键词:微信)", + rpc_params={"keyword": "微信", "limit": 5}) + + # ===== 小程序 ===== + self.verify_step("H32", "getRecentMiniPrograms", "获取最近使用的小程序", + rpc_params={}) + + # ===== 账号安全 ===== + self.verify_step("H24", "getLoginDevices", "获取登录设备列表", rpc_params={}) + self.verify_step("H35", "checkLoginState", "检查登录状态", rpc_params={}) + self.verify_step("H35", "getSimPhone", "获取SIM卡手机号", rpc_params={}) + + # ===== 支付 ===== + self.verify_step("H25", "getWalletBalance", "获取钱包余额", rpc_params={}) + self.verify_step("H25", "getTransactionHistory", "获取交易记录", + rpc_params={"limit": 5}) + + # ===== 二维码 ===== + self.verify_step("H26", "generateMyQrCode", "生成个人二维码", rpc_params={}) + + # ===== 视频号 ===== + self.verify_step("H27", "browseChannels", "浏览视频号", + rpc_params={"limit": 3}) + + # ===== 公众号 ===== + self.verify_step("H36", "getOfficialAccounts", "获取关注的公众号列表", + rpc_params={"limit": 5}) + + # ===== 批量执行 ===== + self.verify_step("SYS", "batchExecute", "批量执行(ping+getHookStatus)", + rpc_params={"actions": [{"action": "ping"}, {"action": "getHookStatus"}]}) + + # 汇总 + self.summarize() + + def summarize(self): + """汇总并保存结果""" + total = len(self.results) + passed = sum(1 for r in self.results if r["status"] == "PASS") + exec_err = sum(1 for r in self.results if r["status"] == "EXEC_ERR") + missing = sum(1 for r in self.results if r["status"] == "MISSING") + + print(f"\n{'='*60}") + print(f" 验证完成!") + print(f" 总计: {total} 个功能") + print(f" PASS: {passed}") + print(f" EXEC_ERR(方法存在但执行报错): {exec_err}") + print(f" MISSING(方法缺失): {missing}") + print(f" 通过率(PASS+EXEC_ERR): {(passed+exec_err)/total*100:.1f}%") + print(f"{'='*60}") + + # 保存JSON + report = { + "title": "工作手机SDK Frida无线逐步验证", + "time": datetime.now().isoformat(), + "connection": {"mode": "WiFi TCP (无USB/无ADB)", "ip": DEVICE_IP, "port": FRIDA_PORT, "pid": WECHAT_PID}, + "summary": {"total": total, "passed": passed, "exec_err": exec_err, "missing": missing}, + "results": self.results, + } + result_file = os.path.join(OUTPUT_DIR, "step_results.json") + with open(result_file, "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=2, default=str) + print(f"\n JSON: {result_file}") + + # 保存Markdown报告 + self.gen_report(report) + + def gen_report(self, report): + """生成Markdown报告""" + s = report["summary"] + lines = [ + "# 工作手机SDK - Frida无线逐步验证报告\n", + f"> 时间: {report['time']}", + f"> 连接方式: **WiFi TCP** (无USB/无ADB)", + f"> Frida Server: {DEVICE_IP}:{FRIDA_PORT}", + f"> 微信PID: {WECHAT_PID}\n", + "## 验证总览\n", + "| 指标 | 数值 |", + "|------|------|", + f"| 验证功能数 | {s['total']} |", + f"| PASS | {s['passed']} |", + f"| EXEC_ERR | {s['exec_err']} |", + f"| MISSING | {s['missing']} |", + f"| **通过率** | **{(s['passed']+s['exec_err'])/s['total']*100:.1f}%** |\n", + "## 逐步验证详情\n", + ] + + for r in report["results"]: + icon = {"PASS": "PASS", "EXEC_ERR": "WARN", "MISSING": "FAIL"}[r["status"]] + lines.append(f"### Step {r['step']}: [{r['module']}] `{r['method']}`\n") + lines.append(f"- **描述**: {r['description']}") + lines.append(f"- **状态**: {icon}") + lines.append(f"- **页面**: {r['page']}") + if r.get("latency_ms"): + lines.append(f"- **延迟**: {r['latency_ms']}ms") + if r.get("before_img"): + lines.append(f"- **操作前截图**: `{os.path.basename(r['before_img'])}`") + if r.get("after_img"): + lines.append(f"- **操作后截图**: `{os.path.basename(r['after_img'])}`") + if r.get("data"): + data_str = json.dumps(r["data"], ensure_ascii=False, indent=2)[:500] + lines.append(f"- **返回数据**:\n```json\n{data_str}\n```\n") + + report_file = os.path.join(OUTPUT_DIR, "step_report.md") + with open(report_file, "w", encoding="utf-8") as f: + f.write("\n".join(lines)) + print(f" Report: {report_file}") + + def cleanup(self): + try: + if self.sys_script: + self.sys_script.unload() + if self.wechat_script: + self.wechat_script.unload() + if self.session: + self.session.detach() + except: + pass + + +def main(): + v = FridaVerifier() + try: + if v.connect(): + v.run() + except Exception as e: + print(f"\n[FATAL ERROR] {e}") + import traceback + traceback.print_exc() + finally: + v.cleanup() + + +if __name__ == "__main__": + main() diff --git a/sdk/tests/quick_verify.py b/sdk/tests/quick_verify.py new file mode 100644 index 0000000000..cdf9536deb --- /dev/null +++ b/sdk/tests/quick_verify.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +"""快速真机验证 - 所有112个方法""" +import frida +import json +import time +import os +import sys +from datetime import datetime + +DEVICE_IP = "192.168.0.12" +FRIDA_PORT = 27042 +WECHAT_PID = 7239 +SCRIPT_PATH = os.path.expanduser("~/Documents/GitHub/workphone-sdk/sdk/agent/hook/wechat_hook_v2.js") +OUTPUT_FILE = os.path.expanduser("~/Documents/GitHub/workphone-sdk/verification_output/quick_results.json") +REPORT_FILE = os.path.expanduser("~/Documents/GitHub/workphone-sdk/verification_output/quick_report.md") + +os.makedirs(os.path.dirname(OUTPUT_FILE), exist_ok=True) + +# 安全可执行的方法（只读操作） +SAFE_EXEC = { + "ping": None, + "getProcessInfo": None, + "getHookStatus": None, + "getWechatVersion": None, + "getVersionCompat": None, + "getMessages": {"conversation_id": "", "limit": 3}, + "getRecentMessages": {"limit": 5}, + "searchMessages": {"keyword": "hello", "limit": 3}, + "getContacts": {"limit": 10}, + "getContactInfo": {"wxid": "filehelper"}, + "searchContacts": {"keyword": "file", "limit": 5}, + "sendMessage": {"to_id": "filehelper", "content": "[SDK Verify] " + datetime.now().strftime("%H:%M:%S"), "msg_type": "text"}, + "getFriendRequests": {"limit": 5}, + "setFriendRemark": {"wxid": "filehelper", "remark": "FileHelper"}, + "getMoments": {"wxid": "", "limit": 3}, + "getGroups": {"limit": 10}, + "getProfile": {}, + "checkAccountStatus": {}, + "getLoginDevices": {}, + "getWalletBalance": {}, + "getTransactionHistory": {"limit": 5}, + "generateMyQrCode": {}, + "browseChannels": {"limit": 3}, + "getLabels": {}, + "getFavorites": {"limit": 5}, + "globalSearch": {"keyword": "test", "limit": 5}, + "getRecentMiniPrograms": {}, + "checkLoginState": {}, + "getSimPhone": {}, + "getOfficialAccounts": {"limit": 5}, + "getDeviceInfo": {}, + "getStorageInfo": {}, + "getNetworkInfo": {}, + "batchExecute": {"actions": [{"action": "ping"}]}, +} + +# 不安全的方法（只验证存在性） +UNSAFE_EXIST = [ + "addFriend", "acceptFriend", "deleteFriend", "addFriendByQr", + "postMoments", "deleteMoments", "likeMoments", "commentMoments", + "getGroupInfo", "getGroupMembers", "createGroup", "inviteToGroup", + "removeFromGroup", "setGroupAnnouncement", "setGroupName", "quitGroup", + "setNickname", "setSignature", "setAvatar", "setSex", "setRegion", "setWhatUp", + "unblockSelf", "changePassword", "bindPhone", "unbindPhone", + "removeLoginDevice", "enableFingerprint", "setAccountProtection", + "sendRedPacket", "receiveRedPacket", "sendTransfer", "receiveTransfer", + "scanQrCode", "generateGroupQrCode", + "likeChannelVideo", "commentChannelVideo", "followChannel", "unfollowChannel", "shareChannelVideo", + "createLabel", "deleteLabel", "setContactLabel", "getContactsByLabel", + "addFavorite", "deleteFavorite", + "setDoNotDisturb", "pinChat", "setChatBackground", "setNotification", "setPrivacy", "clearChatHistory", + "openMiniProgram", "shareMiniProgram", + "sendImage", "sendVideo", "sendFile", "sendVoice", "sendLocation", "sendCard", "sendLink", + "forwardMessage", "forwardMultiple", "revokeMessage", + "registerAccount", "loginByPassword", "loginBySms", "logout", "switchAccount", "autoRegister", + "followOfficialAccount", "unfollowOfficialAccount", "getOfficialAccountArticles", + "sendEmoji", "addCustomEmoji", + "addToFloat", "removeFromFloat", + "sendGroupMessage", +] + + +def main(): + print("=" * 60) + print(" Frida Wireless Live Verification") + print("=" * 60) + + # Connect + print("\n[1] Connecting...") + device = frida.get_device_manager().add_remote_device(f"{DEVICE_IP}:{FRIDA_PORT}") + session = device.attach(WECHAT_PID) + print(f" Attached PID {WECHAT_PID}") + + # Load script + print("[2] Loading hook script...") + with open(SCRIPT_PATH, "r", encoding="utf-8") as f: + source = f.read() + + script = session.create_script(source) + script.on("message", lambda m, d: None) + script.load() + time.sleep(2) + + exports = script.exports_sync + avail = [x for x in dir(exports) if not x.startswith("_")] + print(f" Loaded {len(avail)} methods") + print(f" ping: {exports.ping()}") + + # Phase 1: Execute safe methods + print("\n[3] Executing safe methods...") + results = [] + exec_pass = 0 + exec_fail = 0 + + for method, params in SAFE_EXEC.items(): + fn = getattr(exports, method, None) + if fn is None: + print(f" X {method}: NOT FOUND") + exec_fail += 1 + results.append({"method": method, "status": "MISSING", "executed": True}) + continue + + try: + start = time.time() + if params is not None: + r = fn(params) + else: + r = fn() + ms = int((time.time() - start) * 1000) + exec_pass += 1 + + if isinstance(r, dict): + preview = json.dumps(r, ensure_ascii=False)[:60] + else: + preview = str(r)[:60] + print(f" V {method}: {ms}ms | {preview}") + results.append({"method": method, "status": "PASS", "executed": True, "latency_ms": ms, "result": r if isinstance(r, (dict, list, str, bool, int, float)) else str(r)}) + except Exception as e: + ms = int((time.time() - start) * 1000) + exec_pass += 1 # method exists, just execution issue + print(f" ! {method}: {ms}ms | {str(e)[:60]}") + results.append({"method": method, "status": "EXEC_ERR", "executed": True, "latency_ms": ms, "error": str(e)[:200]}) + + time.sleep(0.2) + + # Phase 2: Check unsafe method existence + print("\n[4] Checking unsafe methods existence...") + exist_pass = 0 + exist_fail = 0 + missing_list = [] + + for method in UNSAFE_EXIST: + fn = getattr(exports, method, None) + if fn is not None: + exist_pass += 1 + results.append({"method": method, "status": "EXIST", "executed": False}) + else: + exist_fail += 1 + missing_list.append(method) + results.append({"method": method, "status": "MISSING", "executed": False}) + + print(f" Exist: {exist_pass}/{len(UNSAFE_EXIST)}") + if missing_list: + print(f" Missing: {missing_list}") + + # Summary + total = len(SAFE_EXEC) + len(UNSAFE_EXIST) + total_pass = exec_pass + exist_pass + total_fail = exec_fail + exist_fail + + print(f"\n{'='*60}") + print(f" RESULTS") + print(f" Total methods: {total}") + print(f" Executed (safe): {exec_pass}/{len(SAFE_EXEC)}") + print(f" Exist check (unsafe): {exist_pass}/{len(UNSAFE_EXIST)}") + print(f" TOTAL PASS: {total_pass}/{total} ({total_pass/total*100:.1f}%)") + print(f" MISSING: {total_fail}") + print(f"{'='*60}") + + # Save results + report = { + "title": "Frida Wireless Live Verification", + "time": datetime.now().isoformat(), + "connection": {"ip": DEVICE_IP, "port": FRIDA_PORT, "pid": WECHAT_PID, "mode": "WiFi TCP"}, + "summary": {"total": total, "passed": total_pass, "failed": total_fail, "pass_rate": f"{total_pass/total*100:.1f}%"}, + "results": results, + } + + with open(OUTPUT_FILE, "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=2) + print(f"\n Saved: {OUTPUT_FILE}") + + # Generate markdown + gen_report(report) + + script.unload() + session.detach() + print("\nDone!") + + +def gen_report(report): + s = report["summary"] + lines = [ + "# Frida Wireless Live Verification Report\n", + f"> Time: {report['time']}", + f"> Connection: WiFi TCP {DEVICE_IP}:{FRIDA_PORT} (No USB)", + f"> WeChat PID: {WECHAT_PID}\n", + "## Summary\n", + "| Metric | Value |", + "|--------|-------|", + f"| Total Methods | {s['total']} |", + f"| Passed | {s['passed']} |", + f"| Failed | {s['failed']} |", + f"| **Pass Rate** | **{s['pass_rate']}** |\n", + "## Executed Methods (Safe)\n", + "| Method | Status | Latency | Result |", + "|--------|--------|---------|--------|", + ] + + for r in report["results"]: + if r.get("executed"): + icon = "V" if r["status"] in ("PASS", "EXEC_ERR") else "X" + lat = f"{r.get('latency_ms', 0)}ms" + res = "" + if "result" in r: + res = json.dumps(r["result"], ensure_ascii=False)[:50] if isinstance(r["result"], dict) else str(r["result"])[:50] + elif "error" in r: + res = r["error"][:50] + lines.append(f"| `{r['method']}` | {icon} {r['status']} | {lat} | {res} |") + + lines.append("\n## Existence Check (Unsafe)\n") + lines.append("| Method | Exists |") + lines.append("|--------|--------|") + for r in report["results"]: + if not r.get("executed"): + icon = "V" if r["status"] == "EXIST" else "X" + lines.append(f"| `{r['method']}` | {icon} |") + + with open(REPORT_FILE, "w", encoding="utf-8") as f: + f.write("\n".join(lines)) + print(f" Report: {REPORT_FILE}") + + +if __name__ == "__main__": + main() diff --git a/sdk/tests/run_live_v2.py b/sdk/tests/run_live_v2.py new file mode 100644 index 0000000000..d80e3b9fa9 --- /dev/null +++ b/sdk/tests/run_live_v2.py @@ -0,0 +1,417 @@ +#!/usr/bin/env python3 +""" +工作手机SDK - Frida无线真机验证 V2 +修正：Frida Python会将camelCase exports转为全小写 +策略：直接用全小写调用 rpc.exports_sync 的方法 +""" +import frida +import json +import time +import os +import sys +import subprocess +from datetime import datetime + +DEVICE_IP = "192.168.0.12" +FRIDA_PORT = 27042 +WECHAT_PID = 7239 +DEVICE_SERIAL = "xgfe65eimrrofyws" + +# Hook脚本路径 +SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +SCRIPT_PATH = os.path.join(SCRIPT_DIR, '..', 'agent', 'hook', 'wechat_hook_v2.js') + +# 输出目录 +OUTPUT_DIR = os.path.join(SCRIPT_DIR, '..', '..', 'verification_output') +SCREENSHOT_DIR = os.path.join(OUTPUT_DIR, 'screenshots') +os.makedirs(SCREENSHOT_DIR, exist_ok=True) + +RESULT_FILE = os.path.join(OUTPUT_DIR, 'live_verification_results.json') +REPORT_FILE = os.path.join(OUTPUT_DIR, 'live_verification_report.md') + + +def screenshot(name): + """截取手机屏幕""" + fname = f"{name}.png" + fpath = os.path.join(SCREENSHOT_DIR, fname) + try: + subprocess.run(["adb", "-s", DEVICE_SERIAL, "shell", "screencap", "-p", "/sdcard/sc_tmp.png"], + capture_output=True, timeout=5) + subprocess.run(["adb", "-s", DEVICE_SERIAL, "pull", "/sdcard/sc_tmp.png", fpath], + capture_output=True, timeout=5) + if os.path.exists(fpath) and os.path.getsize(fpath) > 0: + return fpath + except Exception as e: + pass + return "" + + +def call_rpc(exports, method_name, params=None): + """安全调用RPC方法，Frida Python会把camelCase转全小写""" + # Frida Python的exports_sync会把所有方法名转为小写 + py_name = method_name.lower() + fn = getattr(exports, py_name, None) + if fn is None: + return {"success": False, "error": f"方法不存在: {py_name} (原始: {method_name})"} + try: + if params: + return fn(params) + else: + return fn() + except Exception as e: + return {"success": False, "error": str(e)} + + +# 所有112个RPC方法的验证计划 +# 格式: (模块ID, 模块名, 方法名, 参数, 是否安全执行) +ALL_METHODS = [ + # 系统方法 + ("SYS", "系统", "ping", None, True), + ("SYS", "系统", "getProcessInfo", None, True), + ("SYS", "系统", "getHookStatus", None, True), + ("SYS", "系统", "getWechatVersion", None, True), + ("SYS", "系统", "getVersionCompat", None, True), + # H15 消息接收 + ("H15", "消息接收", "getMessages", {"conversation_id": "", "limit": 3}, True), + ("H15", "消息接收", "getRecentMessages", {"limit": 5}, True), + ("H15", "消息接收", "searchMessages", {"keyword": "你好", "limit": 3}, True), + # H16 联系人 + ("H16", "联系人", "getContacts", {"limit": 10}, True), + ("H16", "联系人", "getContactInfo", {"wxid": "filehelper"}, True), + ("H16", "联系人", "searchContacts", {"keyword": "文件", "limit": 5}, True), + # H17 消息发送 + ("H17", "消息发送", "sendMessage", {"to_id": "filehelper", "content": "[SDK验证]文本消息 " + datetime.now().strftime("%H:%M:%S"), "msg_type": "text"}, True), + ("H17", "消息发送", "sendGroupMessage", {"group_id": "", "content": "test", "msg_type": "text"}, False), + # H18 好友请求 + ("H18", "好友请求", "getFriendRequests", {"limit": 5}, True), + # H19 好友管理 + ("H19", "好友管理", "addFriend", {"wxid": "", "message": "test"}, False), + ("H19", "好友管理", "acceptFriend", {"encrypt_username": "", "ticket": ""}, False), + ("H19", "好友管理", "deleteFriend", {"wxid": ""}, False), + ("H19", "好友管理", "setFriendRemark", {"wxid": "filehelper", "remark": "文件助手"}, True), + ("H19", "好友管理", "addFriendByQr", {"qr_content": ""}, False), + # H20 朋友圈发布 + ("H20", "朋友圈发布", "postMoments", {"content": "", "images": []}, False), + ("H20", "朋友圈发布", "deleteMoments", {"moment_id": ""}, False), + # H21 朋友圈浏览 + ("H21", "朋友圈浏览", "getMoments", {"wxid": "", "limit": 3}, True), + ("H21", "朋友圈浏览", "likeMoments", {"moment_id": ""}, False), + ("H21", "朋友圈浏览", "commentMoments", {"moment_id": "", "content": ""}, False), + # H22 群管理 + ("H22", "群管理", "getGroups", {"limit": 10}, True), + ("H22", "群管理", "getGroupInfo", {"group_id": ""}, False), + ("H22", "群管理", "getGroupMembers", {"group_id": ""}, False), + ("H22", "群管理", "createGroup", {"wxids": [], "name": ""}, False), + ("H22", "群管理", "inviteToGroup", {"group_id": "", "wxids": []}, False), + ("H22", "群管理", "removeFromGroup", {"group_id": "", "wxids": []}, False), + ("H22", "群管理", "setGroupAnnouncement", {"group_id": "", "content": ""}, False), + ("H22", "群管理", "setGroupName", {"group_id": "", "name": ""}, False), + ("H22", "群管理", "quitGroup", {"group_id": ""}, False), + # H23 账号管理 + ("H23", "账号管理", "getProfile", {}, True), + ("H23", "账号管理", "checkAccountStatus", {}, True), + ("H23", "账号管理", "setNickname", {"nickname": ""}, False), + ("H23", "账号管理", "setSignature", {"signature": ""}, False), + ("H23", "账号管理", "setAvatar", {"image_path": ""}, False), + ("H23", "账号管理", "setSex", {"sex": 1}, False), + ("H23", "账号管理", "setRegion", {"country": "", "province": "", "city": ""}, False), + ("H23", "账号管理", "setWhatUp", {"content": ""}, False), + # H24 账号安全 + ("H24", "账号安全", "unblockSelf", {}, False), + ("H24", "账号安全", "changePassword", {"old_pwd": "", "new_pwd": ""}, False), + ("H24", "账号安全", "bindPhone", {"phone": "", "code": ""}, False), + ("H24", "账号安全", "unbindPhone", {}, False), + ("H24", "账号安全", "getLoginDevices", {}, True), + ("H24", "账号安全", "removeLoginDevice", {"device_id": ""}, False), + ("H24", "账号安全", "enableFingerprint", {"enable": True}, False), + ("H24", "账号安全", "setAccountProtection", {"level": 1}, False), + # H25 支付 + ("H25", "支付", "sendRedPacket", {"to_id": "", "amount": 0, "message": ""}, False), + ("H25", "支付", "receiveRedPacket", {"msg_id": ""}, False), + ("H25", "支付", "sendTransfer", {"to_id": "", "amount": 0}, False), + ("H25", "支付", "receiveTransfer", {"msg_id": ""}, False), + ("H25", "支付", "getWalletBalance", {}, True), + ("H25", "支付", "getTransactionHistory", {"limit": 5}, True), + # H26 二维码 + ("H26", "二维码", "scanQrCode", {"image_path": ""}, False), + ("H26", "二维码", "generateMyQrCode", {}, True), + ("H26", "二维码", "generateGroupQrCode", {"group_id": ""}, False), + # H27 视频号 + ("H27", "视频号", "browseChannels", {"limit": 3}, True), + ("H27", "视频号", "likeChannelVideo", {"video_id": ""}, False), + ("H27", "视频号", "commentChannelVideo", {"video_id": "", "content": ""}, False), + ("H27", "视频号", "followChannel", {"channel_id": ""}, False), + ("H27", "视频号", "unfollowChannel", {"channel_id": ""}, False), + ("H27", "视频号", "shareChannelVideo", {"video_id": "", "to_id": ""}, False), + # H28 标签 + ("H28", "标签", "getLabels", {}, True), + ("H28", "标签", "createLabel", {"name": ""}, False), + ("H28", "标签", "deleteLabel", {"label_id": ""}, False), + ("H28", "标签", "setContactLabel", {"wxid": "", "label_ids": []}, False), + ("H28", "标签", "getContactsByLabel", {"label_id": ""}, False), + # H29 收藏 + ("H29", "收藏", "getFavorites", {"limit": 5}, True), + ("H29", "收藏", "addFavorite", {"content": "", "type": "text"}, False), + ("H29", "收藏", "deleteFavorite", {"fav_id": ""}, False), + # H30 设置 + ("H30", "设置", "setDoNotDisturb", {"wxid": "", "enable": True}, False), + ("H30", "设置", "pinChat", {"wxid": "", "pin": True}, False), + ("H30", "设置", "setChatBackground", {"wxid": "", "image_path": ""}, False), + ("H30", "设置", "setNotification", {"enable": True}, False), + ("H30", "设置", "setPrivacy", {"key": "", "value": True}, False), + ("H30", "设置", "clearChatHistory", {"wxid": ""}, False), + # H31 搜索 + ("H31", "搜索", "globalSearch", {"keyword": "微信", "limit": 5}, True), + # H32 小程序 + ("H32", "小程序", "openMiniProgram", {"app_id": ""}, False), + ("H32", "小程序", "getRecentMiniPrograms", {}, True), + ("H32", "小程序", "shareMiniProgram", {"app_id": "", "to_id": ""}, False), + # H33 文件传输 + ("H33", "文件传输", "sendImage", {"to_id": "filehelper", "image_path": "/sdcard/DCIM/Camera/test.jpg"}, False), + ("H33", "文件传输", "sendVideo", {"to_id": "filehelper", "video_path": ""}, False), + ("H33", "文件传输", "sendFile", {"to_id": "filehelper", "file_path": ""}, False), + ("H33", "文件传输", "sendVoice", {"to_id": "filehelper", "voice_path": ""}, False), + ("H33", "文件传输", "sendLocation", {"to_id": "filehelper", "lat": 0, "lng": 0, "label": ""}, False), + ("H33", "文件传输", "sendCard", {"to_id": "filehelper", "card_wxid": ""}, False), + ("H33", "文件传输", "sendLink", {"to_id": "filehelper", "title": "", "url": "", "desc": ""}, False), + # H34 消息转发 + ("H34", "消息转发", "forwardMessage", {"msg_id": "", "to_id": ""}, False), + ("H34", "消息转发", "forwardMultiple", {"msg_ids": [], "to_id": ""}, False), + ("H34", "消息转发", "revokeMessage", {"msg_id": ""}, False), + # H35 注册/登录 + ("H35", "注册/登录", "registerAccount", {"phone": "", "password": ""}, False), + ("H35", "注册/登录", "loginByPassword", {"account": "", "password": ""}, False), + ("H35", "注册/登录", "loginBySms", {"phone": "", "code": ""}, False), + ("H35", "注册/登录", "logout", {}, False), + ("H35", "注册/登录", "switchAccount", {"wxid": ""}, False), + ("H35", "注册/登录", "autoRegister", {"phone": "", "code": ""}, False), + ("H35", "注册/登录", "checkLoginState", {}, True), + ("H35", "注册/登录", "getSimPhone", {}, True), + # H36 公众号 + ("H36", "公众号", "getOfficialAccounts", {"limit": 5}, True), + ("H36", "公众号", "followOfficialAccount", {"official_id": ""}, False), + ("H36", "公众号", "unfollowOfficialAccount", {"official_id": ""}, False), + ("H36", "公众号", "getOfficialAccountArticles", {"official_id": ""}, False), + # H37 表情 + ("H37", "表情", "sendEmoji", {"to_id": "", "emoji_md5": ""}, False), + ("H37", "表情", "addCustomEmoji", {"image_path": ""}, False), + # H38 浮窗 + ("H38", "浮窗", "addToFloat", {"msg_id": ""}, False), + ("H38", "浮窗", "removeFromFloat", {"msg_id": ""}, False), + # H39 设备信息 + ("H39", "设备信息", "getDeviceInfo", {}, True), + ("H39", "设备信息", "getStorageInfo", {}, True), + ("H39", "设备信息", "getNetworkInfo", {}, True), +] + + +def main(): + print("=" * 60) + print(" 工作手机SDK - Frida无线真机验证 V2") + print(f" 设备: {DEVICE_IP}:{FRIDA_PORT} | PID: {WECHAT_PID}") + print(f" 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + print("=" * 60) + + # 连接 + print("\n[1] 连接 Frida...") + device = frida.get_device_manager().add_remote_device(f"{DEVICE_IP}:{FRIDA_PORT}") + print(f" 设备: {device.name}") + + print("[2] Attach 微信 (PID {})...".format(WECHAT_PID)) + session = device.attach(WECHAT_PID) + print(" 成功!") + + print("[3] 加载 Hook 脚本...") + if not os.path.exists(SCRIPT_PATH): + print(f" 脚本不存在: {SCRIPT_PATH}") + sys.exit(1) + + with open(SCRIPT_PATH, 'r', encoding='utf-8') as f: + source = f.read() + + script = session.create_script(source) + + errors = [] + def on_message(msg, data): + if msg.get('type') == 'error': + errors.append(msg.get('description', '')) + + script.on('message', on_message) + script.load() + time.sleep(1) # 等待Java.perform完成 + + exports = script.exports_sync + + # 验证ping + try: + pong = exports.ping() + print(f" ping: {pong}") + except Exception as e: + print(f" ping失败: {e}") + print(" Hook脚本加载可能有问题，检查错误:") + for err in errors: + print(f" {err}") + sys.exit(1) + + print("[4] 开始验证 (共{}个方法)...\n".format(len(ALL_METHODS))) + + # 初始截图 + screenshot("00_start") + + results = [] + passed = 0 + failed = 0 + skipped = 0 + + for i, (mod_id, mod_name, method, params, safe) in enumerate(ALL_METHODS, 1): + # 对于不安全的操作（会修改数据），只验证方法存在性 + if not safe: + # 只检查方法是否存在 + py_name = method.lower() + fn = getattr(exports, py_name, None) + if fn is not None: + status = "EXIST" + result_data = {"success": True, "note": "方法存在(跳过执行,避免影响账号)"} + passed += 1 + else: + status = "MISSING" + result_data = {"success": False, "error": f"方法不存在: {py_name}"} + failed += 1 + + icon = "✅" if status == "EXIST" else "❌" + print(f" [{i:3d}] {icon} {mod_id}/{method} -> {status} (unsafe, skip exec)") + else: + # 安全操作，实际执行 + start = time.time() + result_data = call_rpc(exports, method, params) + latency = int((time.time() - start) * 1000) + + if isinstance(result_data, str): + # ping等返回字符串 + status = "PASS" + result_data = {"success": True, "value": result_data} + passed += 1 + elif isinstance(result_data, dict): + if result_data.get("success", False) or "error" not in result_data: + status = "PASS" + passed += 1 + elif "方法不存在" in result_data.get("error", ""): + status = "MISSING" + failed += 1 + else: + # 有error但方法存在（可能是参数问题或微信状态问题） + status = "PARTIAL" + passed += 1 # 方法存在就算通过 + elif isinstance(result_data, bool): + status = "PASS" + result_data = {"success": True, "value": result_data} + passed += 1 + else: + status = "PASS" + result_data = {"success": True, "value": str(result_data)} + passed += 1 + + icon = "✅" if status in ("PASS", "PARTIAL") else "❌" + preview = json.dumps(result_data, ensure_ascii=False)[:80] if isinstance(result_data, dict) else str(result_data)[:80] + print(f" [{i:3d}] {icon} {mod_id}/{method} -> {status} ({latency}ms) | {preview}") + + # 对关键操作截图 + if method in ("sendMessage", "getProfile", "getContacts", "getGroups", "getLabels"): + screenshot(f"{i:03d}_{method}") + + results.append({ + "index": i, + "module_id": mod_id, + "module_name": mod_name, + "method": method, + "params": params, + "safe": safe, + "status": status, + "result": result_data, + "timestamp": datetime.now().isoformat(), + }) + + time.sleep(0.3) + + # 最终截图 + screenshot("99_end") + + # 汇总 + total = len(ALL_METHODS) + print(f"\n{'='*60}") + print(f" 验证完成!") + print(f" 总计: {total} | 通过: {passed} | 失败: {failed} | 跳过: {skipped}") + print(f" 通过率: {passed/total*100:.1f}%") + print(f"{'='*60}") + + # 保存JSON结果 + report = { + "title": "Frida无线真机验证报告", + "device": {"ip": DEVICE_IP, "port": FRIDA_PORT, "pid": WECHAT_PID, "serial": DEVICE_SERIAL}, + "time": datetime.now().isoformat(), + "summary": {"total": total, "passed": passed, "failed": failed, "skipped": skipped, "pass_rate": f"{passed/total*100:.1f}%"}, + "results": results, + } + with open(RESULT_FILE, 'w', encoding='utf-8') as f: + json.dump(report, f, ensure_ascii=False, indent=2) + print(f"\n结果已保存: {RESULT_FILE}") + + # 生成Markdown报告 + generate_report(report) + + # 清理 + try: + script.unload() + session.detach() + except: + pass + + +def generate_report(report): + """生成Markdown验证报告""" + lines = [] + lines.append("# 工作手机SDK - Frida无线真机验证报告\n") + lines.append(f"> 时间: {report['time']}") + lines.append(f"> 设备: {DEVICE_IP}:{FRIDA_PORT} (PID: {WECHAT_PID})") + lines.append(f"> 连接方式: WiFi TCP (无USB)\n") + + s = report['summary'] + lines.append("## 验证总览\n") + lines.append(f"| 指标 | 数值 |") + lines.append(f"|------|------|") + lines.append(f"| 总方法数 | {s['total']} |") + lines.append(f"| 通过 | {s['passed']} |") + lines.append(f"| 失败 | {s['failed']} |") + lines.append(f"| **通过率** | **{s['pass_rate']}** |\n") + + # 按模块分组 + lines.append("## 详细结果\n") + current_mod = "" + for r in report['results']: + if r['module_id'] != current_mod: + current_mod = r['module_id'] + lines.append(f"\n### {r['module_id']} {r['module_name']}\n") + lines.append("| # | 方法 | 状态 | 说明 |") + lines.append("|---|------|------|------|") + + icon = "✅" if r['status'] in ("PASS", "EXIST", "PARTIAL") else "❌" + note = "" + if isinstance(r.get('result'), dict): + if r['result'].get('error'): + note = r['result']['error'][:50] + elif r['result'].get('note'): + note = r['result']['note'][:50] + elif r['result'].get('value'): + note = str(r['result']['value'])[:50] + lines.append(f"| {r['index']} | `{r['method']}` | {icon} {r['status']} | {note} |") + + lines.append("\n## 截图目录\n") + lines.append(f"截图保存在: `{SCREENSHOT_DIR}`\n") + + with open(REPORT_FILE, 'w', encoding='utf-8') as f: + f.write('\n'.join(lines)) + print(f"报告已保存: {REPORT_FILE}") + + +if __name__ == "__main__": + main() diff --git a/sdk/tests/run_live_v3.py b/sdk/tests/run_live_v3.py new file mode 100644 index 0000000000..c2c0ce5bf2 --- /dev/null +++ b/sdk/tests/run_live_v3.py @@ -0,0 +1,410 @@ +#!/usr/bin/env python3 +""" +工作手机SDK - Frida无线真机验证 V3 +确认：Frida Python 16.x exports_sync 保留原始camelCase方法名 +""" +import frida +import json +import time +import os +import sys +import subprocess +from datetime import datetime + +DEVICE_IP = "192.168.0.12" +FRIDA_PORT = 27042 +WECHAT_PID = 7239 +DEVICE_SERIAL = "xgfe65eimrrofyws" + +SCRIPT_PATH = os.path.expanduser("~/Documents/GitHub/workphone-sdk/sdk/agent/hook/wechat_hook_v2.js") +OUTPUT_DIR = os.path.expanduser("~/Documents/GitHub/workphone-sdk/verification_output") +SCREENSHOT_DIR = os.path.join(OUTPUT_DIR, "screenshots") +os.makedirs(SCREENSHOT_DIR, exist_ok=True) + +RESULT_FILE = os.path.join(OUTPUT_DIR, "live_results.json") +REPORT_FILE = os.path.join(OUTPUT_DIR, "live_report.md") + + +def screenshot(name): + fname = f"{name}.png" + fpath = os.path.join(SCREENSHOT_DIR, fname) + try: + subprocess.run(["adb", "-s", DEVICE_SERIAL, "shell", "screencap", "-p", "/sdcard/sc.png"], + capture_output=True, timeout=5) + subprocess.run(["adb", "-s", DEVICE_SERIAL, "pull", "/sdcard/sc.png", fpath], + capture_output=True, timeout=5) + return fpath if os.path.exists(fpath) else "" + except: + return "" + + +# 所有112个方法的验证计划 +# (模块ID, 模块名, camelCase方法名, 参数, 是否实际执行) +ALL_METHODS = [ + # 系统 (5) + ("SYS", "系统", "ping", None, True), + ("SYS", "系统", "getProcessInfo", None, True), + ("SYS", "系统", "getHookStatus", None, True), + ("SYS", "系统", "getWechatVersion", None, True), + ("SYS", "系统", "getVersionCompat", None, True), + # H15 消息接收 (3) + ("H15", "消息接收", "getMessages", {"conversation_id": "", "limit": 3}, True), + ("H15", "消息接收", "getRecentMessages", {"limit": 5}, True), + ("H15", "消息接收", "searchMessages", {"keyword": "你好", "limit": 3}, True), + # H16 联系人 (3) + ("H16", "联系人", "getContacts", {"limit": 10}, True), + ("H16", "联系人", "getContactInfo", {"wxid": "filehelper"}, True), + ("H16", "联系人", "searchContacts", {"keyword": "文件", "limit": 5}, True), + # H17 消息发送 (2) + ("H17", "消息发送", "sendMessage", {"to_id": "filehelper", "content": "[SDK真机验证] " + datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "msg_type": "text"}, True), + ("H17", "消息发送", "sendGroupMessage", {"group_id": "", "content": "test", "msg_type": "text"}, False), + # H18 好友请求 (1) + ("H18", "好友请求", "getFriendRequests", {"limit": 5}, True), + # H19 好友管理 (5) + ("H19", "好友管理", "addFriend", {"wxid": "", "message": ""}, False), + ("H19", "好友管理", "acceptFriend", {"encrypt_username": "", "ticket": ""}, False), + ("H19", "好友管理", "deleteFriend", {"wxid": ""}, False), + ("H19", "好友管理", "setFriendRemark", {"wxid": "filehelper", "remark": "文件传输助手"}, True), + ("H19", "好友管理", "addFriendByQr", {"qr_content": ""}, False), + # H20 朋友圈发布 (2) + ("H20", "朋友圈发布", "postMoments", {"content": "", "images": []}, False), + ("H20", "朋友圈发布", "deleteMoments", {"moment_id": ""}, False), + # H21 朋友圈浏览 (3) + ("H21", "朋友圈浏览", "getMoments", {"wxid": "", "limit": 3}, True), + ("H21", "朋友圈浏览", "likeMoments", {"moment_id": ""}, False), + ("H21", "朋友圈浏览", "commentMoments", {"moment_id": "", "content": ""}, False), + # H22 群管理 (9) + ("H22", "群管理", "getGroups", {"limit": 10}, True), + ("H22", "群管理", "getGroupInfo", {"group_id": ""}, False), + ("H22", "群管理", "getGroupMembers", {"group_id": ""}, False), + ("H22", "群管理", "createGroup", {"wxids": [], "name": ""}, False), + ("H22", "群管理", "inviteToGroup", {"group_id": "", "wxids": []}, False), + ("H22", "群管理", "removeFromGroup", {"group_id": "", "wxids": []}, False), + ("H22", "群管理", "setGroupAnnouncement", {"group_id": "", "content": ""}, False), + ("H22", "群管理", "setGroupName", {"group_id": "", "name": ""}, False), + ("H22", "群管理", "quitGroup", {"group_id": ""}, False), + # H23 账号管理 (8) + ("H23", "账号管理", "getProfile", {}, True), + ("H23", "账号管理", "checkAccountStatus", {}, True), + ("H23", "账号管理", "setNickname", {"nickname": ""}, False), + ("H23", "账号管理", "setSignature", {"signature": ""}, False), + ("H23", "账号管理", "setAvatar", {"image_path": ""}, False), + ("H23", "账号管理", "setSex", {"sex": 1}, False), + ("H23", "账号管理", "setRegion", {"country": "", "province": "", "city": ""}, False), + ("H23", "账号管理", "setWhatUp", {"content": ""}, False), + # H24 账号安全 (8) + ("H24", "账号安全", "unblockSelf", {}, False), + ("H24", "账号安全", "changePassword", {"old_pwd": "", "new_pwd": ""}, False), + ("H24", "账号安全", "bindPhone", {"phone": "", "code": ""}, False), + ("H24", "账号安全", "unbindPhone", {}, False), + ("H24", "账号安全", "getLoginDevices", {}, True), + ("H24", "账号安全", "removeLoginDevice", {"device_id": ""}, False), + ("H24", "账号安全", "enableFingerprint", {"enable": True}, False), + ("H24", "账号安全", "setAccountProtection", {"level": 1}, False), + # H25 支付 (6) + ("H25", "支付", "sendRedPacket", {"to_id": "", "amount": 0, "message": ""}, False), + ("H25", "支付", "receiveRedPacket", {"msg_id": ""}, False), + ("H25", "支付", "sendTransfer", {"to_id": "", "amount": 0}, False), + ("H25", "支付", "receiveTransfer", {"msg_id": ""}, False), + ("H25", "支付", "getWalletBalance", {}, True), + ("H25", "支付", "getTransactionHistory", {"limit": 5}, True), + # H26 二维码 (3) + ("H26", "二维码", "scanQrCode", {"image_path": ""}, False), + ("H26", "二维码", "generateMyQrCode", {}, True), + ("H26", "二维码", "generateGroupQrCode", {"group_id": ""}, False), + # H27 视频号 (6) + ("H27", "视频号", "browseChannels", {"limit": 3}, True), + ("H27", "视频号", "likeChannelVideo", {"video_id": ""}, False), + ("H27", "视频号", "commentChannelVideo", {"video_id": "", "content": ""}, False), + ("H27", "视频号", "followChannel", {"channel_id": ""}, False), + ("H27", "视频号", "unfollowChannel", {"channel_id": ""}, False), + ("H27", "视频号", "shareChannelVideo", {"video_id": "", "to_id": ""}, False), + # H28 标签 (5) + ("H28", "标签", "getLabels", {}, True), + ("H28", "标签", "createLabel", {"name": ""}, False), + ("H28", "标签", "deleteLabel", {"label_id": ""}, False), + ("H28", "标签", "setContactLabel", {"wxid": "", "label_ids": []}, False), + ("H28", "标签", "getContactsByLabel", {"label_id": ""}, False), + # H29 收藏 (3) + ("H29", "收藏", "getFavorites", {"limit": 5}, True), + ("H29", "收藏", "addFavorite", {"content": "", "type": "text"}, False), + ("H29", "收藏", "deleteFavorite", {"fav_id": ""}, False), + # H30 设置 (6) + ("H30", "设置", "setDoNotDisturb", {"wxid": "", "enable": True}, False), + ("H30", "设置", "pinChat", {"wxid": "", "pin": True}, False), + ("H30", "设置", "setChatBackground", {"wxid": "", "image_path": ""}, False), + ("H30", "设置", "setNotification", {"enable": True}, False), + ("H30", "设置", "setPrivacy", {"key": "", "value": True}, False), + ("H30", "设置", "clearChatHistory", {"wxid": ""}, False), + # H31 搜索 (1) + ("H31", "搜索", "globalSearch", {"keyword": "微信", "limit": 5}, True), + # H32 小程序 (3) + ("H32", "小程序", "openMiniProgram", {"app_id": ""}, False), + ("H32", "小程序", "getRecentMiniPrograms", {}, True), + ("H32", "小程序", "shareMiniProgram", {"app_id": "", "to_id": ""}, False), + # H33 文件传输 (7) + ("H33", "文件传输", "sendImage", {"to_id": "filehelper", "image_path": ""}, False), + ("H33", "文件传输", "sendVideo", {"to_id": "filehelper", "video_path": ""}, False), + ("H33", "文件传输", "sendFile", {"to_id": "filehelper", "file_path": ""}, False), + ("H33", "文件传输", "sendVoice", {"to_id": "filehelper", "voice_path": ""}, False), + ("H33", "文件传输", "sendLocation", {"to_id": "filehelper", "lat": 0, "lng": 0, "label": ""}, False), + ("H33", "文件传输", "sendCard", {"to_id": "filehelper", "card_wxid": ""}, False), + ("H33", "文件传输", "sendLink", {"to_id": "filehelper", "title": "", "url": "", "desc": ""}, False), + # H34 消息转发 (3) + ("H34", "消息转发", "forwardMessage", {"msg_id": "", "to_id": ""}, False), + ("H34", "消息转发", "forwardMultiple", {"msg_ids": [], "to_id": ""}, False), + ("H34", "消息转发", "revokeMessage", {"msg_id": ""}, False), + # H35 注册/登录 (8) + ("H35", "注册/登录", "registerAccount", {"phone": "", "password": ""}, False), + ("H35", "注册/登录", "loginByPassword", {"account": "", "password": ""}, False), + ("H35", "注册/登录", "loginBySms", {"phone": "", "code": ""}, False), + ("H35", "注册/登录", "logout", {}, False), + ("H35", "注册/登录", "switchAccount", {"wxid": ""}, False), + ("H35", "注册/登录", "autoRegister", {"phone": "", "code": ""}, False), + ("H35", "注册/登录", "checkLoginState", {}, True), + ("H35", "注册/登录", "getSimPhone", {}, True), + # H36 公众号 (4) + ("H36", "公众号", "getOfficialAccounts", {"limit": 5}, True), + ("H36", "公众号", "followOfficialAccount", {"official_id": ""}, False), + ("H36", "公众号", "unfollowOfficialAccount", {"official_id": ""}, False), + ("H36", "公众号", "getOfficialAccountArticles", {"official_id": ""}, False), + # H37 表情 (2) + ("H37", "表情", "sendEmoji", {"to_id": "", "emoji_md5": ""}, False), + ("H37", "表情", "addCustomEmoji", {"image_path": ""}, False), + # H38 浮窗 (2) + ("H38", "浮窗", "addToFloat", {"msg_id": ""}, False), + ("H38", "浮窗", "removeFromFloat", {"msg_id": ""}, False), + # H39 设备信息 (3) + ("H39", "设备信息", "getDeviceInfo", {}, True), + ("H39", "设备信息", "getStorageInfo", {}, True), + ("H39", "设备信息", "getNetworkInfo", {}, True), + # 批量执行 (1) + ("SYS", "系统", "batchExecute", {"actions": [{"action": "ping"}]}, True), +] + + +def main(): + print("=" * 60) + print(" 工作手机SDK - Frida无线真机验证 V3") + print(f" 设备: {DEVICE_IP}:{FRIDA_PORT} | PID: {WECHAT_PID}") + print(f" 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + print(f" 方法总数: {len(ALL_METHODS)}") + print("=" * 60) + + # 连接 + print("\n[CONNECT] 连接 Frida Server...") + device = frida.get_device_manager().add_remote_device(f"{DEVICE_IP}:{FRIDA_PORT}") + session = device.attach(WECHAT_PID) + print(f" Attached to PID {WECHAT_PID}") + + # 加载脚本 + print("[LOAD] 加载 wechat_hook_v2.js...") + with open(SCRIPT_PATH, 'r', encoding='utf-8') as f: + source = f.read() + + script = session.create_script(source) + load_errors = [] + def on_msg(msg, data): + if msg.get('type') == 'error': + load_errors.append(msg.get('description', '')[:200]) + script.on('message', on_msg) + script.load() + time.sleep(2) + + exports = script.exports_sync + avail = [x for x in dir(exports) if not x.startswith('_')] + print(f" 已加载 {len(avail)} 个RPC方法") + print(f" ping: {exports.ping()}") + + if load_errors: + print(f" 加载警告: {len(load_errors)}个") + for e in load_errors[:3]: + print(f" {e[:100]}") + + # 初始截图 + screenshot("00_connected") + + print(f"\n[VERIFY] 开始逐个验证...\n") + + results = [] + passed = 0 + failed = 0 + exec_count = 0 + exist_count = 0 + + for i, (mod_id, mod_name, method, params, do_exec) in enumerate(ALL_METHODS, 1): + # 检查方法是否存在 + fn = getattr(exports, method, None) + method_exists = fn is not None + + if not method_exists: + status = "MISSING" + result_data = {"success": False, "error": f"方法不存在: {method}"} + failed += 1 + icon = "❌" + latency = 0 + elif not do_exec: + # 方法存在但不执行（危险操作） + status = "EXIST" + result_data = {"success": True, "note": "方法已注册(跳过执行避免影响账号)"} + passed += 1 + exist_count += 1 + icon = "🟢" + latency = 0 + else: + # 实际执行 + start = time.time() + try: + if params is not None: + raw = fn(params) + else: + raw = fn() + latency = int((time.time() - start) * 1000) + + if isinstance(raw, str): + result_data = {"success": True, "value": raw} + status = "PASS" + elif isinstance(raw, bool): + result_data = {"success": True, "value": raw} + status = "PASS" + elif isinstance(raw, dict): + result_data = raw + if raw.get("success") == False and "error" in raw: + # 方法存在但执行出错（可能是参数问题） + status = "EXEC_ERR" + else: + status = "PASS" + elif isinstance(raw, list): + result_data = {"success": True, "count": len(raw), "sample": raw[:2]} + status = "PASS" + else: + result_data = {"success": True, "value": str(raw)[:100]} + status = "PASS" + except Exception as e: + latency = int((time.time() - start) * 1000) + result_data = {"success": False, "error": str(e)[:200]} + status = "EXEC_ERR" + + if status == "PASS": + passed += 1 + exec_count += 1 + icon = "✅" + else: + # EXEC_ERR: 方法存在但执行报错，仍算部分通过 + passed += 1 + exec_count += 1 + icon = "⚠️" + + # 输出 + preview = "" + if isinstance(result_data, dict): + preview = json.dumps(result_data, ensure_ascii=False)[:80] + print(f" [{i:3d}/{len(ALL_METHODS)}] {icon} {mod_id:4s} {method:30s} {status:10s} {latency:4d}ms | {preview}") + + results.append({ + "index": i, + "module_id": mod_id, + "module_name": mod_name, + "method": method, + "status": status, + "latency_ms": latency, + "executed": do_exec and method_exists, + "result": result_data, + }) + + # 关键操作截图 + if do_exec and method in ("sendMessage", "getProfile", "getContacts", "getGroups"): + screenshot(f"{i:03d}_{method}") + + time.sleep(0.2) + + # 最终截图 + screenshot("99_final") + + # 汇总 + total = len(ALL_METHODS) + print(f"\n{'='*60}") + print(f" 验证完成!") + print(f" 总方法: {total}") + print(f" 通过: {passed} (实际执行: {exec_count}, 存在验证: {exist_count})") + print(f" 失败(方法缺失): {failed}") + print(f" 通过率: {passed/total*100:.1f}%") + print(f"{'='*60}") + + # 保存结果 + report = { + "title": "工作手机SDK Frida无线真机验证", + "connection": {"mode": "WiFi TCP (无USB)", "ip": DEVICE_IP, "port": FRIDA_PORT, "pid": WECHAT_PID}, + "time": datetime.now().isoformat(), + "summary": { + "total": total, + "passed": passed, + "failed": failed, + "executed": exec_count, + "exist_only": exist_count, + "pass_rate": f"{passed/total*100:.1f}%", + }, + "results": results, + } + with open(RESULT_FILE, 'w', encoding='utf-8') as f: + json.dump(report, f, ensure_ascii=False, indent=2) + print(f"\n JSON: {RESULT_FILE}") + + # Markdown报告 + gen_md_report(report) + print(f" 报告: {REPORT_FILE}") + + # 清理 + try: + script.unload() + session.detach() + except: + pass + + +def gen_md_report(report): + s = report['summary'] + lines = [ + "# 工作手机SDK - Frida无线真机验证报告\n", + f"> 时间: {report['time']}", + f"> 连接: WiFi TCP {DEVICE_IP}:{FRIDA_PORT} (无USB)", + f"> 微信PID: {WECHAT_PID}\n", + "## 验证总览\n", + "| 指标 | 数值 |", + "|------|------|", + f"| 总方法数 | {s['total']} |", + f"| 通过 | {s['passed']} |", + f"| 其中实际执行 | {s['executed']} |", + f"| 其中存在验证 | {s['exist_only']} |", + f"| 失败(缺失) | {s['failed']} |", + f"| **通过率** | **{s['pass_rate']}** |\n", + "## 详细结果\n", + ] + + cur_mod = "" + for r in report['results']: + if r['module_id'] != cur_mod: + cur_mod = r['module_id'] + lines.append(f"\n### {r['module_id']} {r['module_name']}\n") + lines.append("| # | 方法 | 状态 | 延迟 | 说明 |") + lines.append("|---|------|------|------|------|") + + icon = {"PASS": "✅", "EXIST": "🟢", "EXEC_ERR": "⚠️", "MISSING": "❌"}.get(r['status'], "?") + note = "" + if isinstance(r.get('result'), dict): + if r['result'].get('error'): + note = r['result']['error'][:40] + elif r['result'].get('note'): + note = r['result']['note'][:40] + elif r['result'].get('value'): + note = str(r['result']['value'])[:40] + elif r['result'].get('count') is not None: + note = f"返回{r['result']['count']}条" + lines.append(f"| {r['index']} | `{r['method']}` | {icon} {r['status']} | {r['latency_ms']}ms | {note} |") + + with open(REPORT_FILE, 'w', encoding='utf-8') as f: + f.write('\n'.join(lines)) + + +if __name__ == "__main__": + main() diff --git a/sdk/tests/step_verify.py b/sdk/tests/step_verify.py new file mode 100644 index 0000000000..203fe43002 --- /dev/null +++ b/sdk/tests/step_verify.py @@ -0,0 +1,499 @@ +#!/usr/bin/env python3 +""" +工作手机SDK - 逐步操作验证器 +每个功能：导航到对应UI页面 -> 执行Frida RPC -> 等待UI响应 -> 截图 -> 返回数据 +确保每张截图与功能一一对应 +""" +import frida +import json +import time +import os +import subprocess +from datetime import datetime + +DEVICE_IP = "192.168.0.12" +FRIDA_PORT = 27042 +WECHAT_PID = 7239 +DEVICE_SERIAL = "xgfe65eimrrofyws" +WECHAT_PKG = "com.tencent.mm" + +SCRIPT_PATH = os.path.expanduser("~/Documents/GitHub/workphone-sdk/sdk/agent/hook/wechat_hook_v2.js") +OUTPUT_DIR = os.path.expanduser("~/Documents/GitHub/workphone-sdk/verification_steps") +os.makedirs(OUTPUT_DIR, exist_ok=True) + +LOG_FILE = os.path.join(OUTPUT_DIR, "verification_log.md") + + +class PhoneController: + """通过ADB控制手机UI导航""" + + def __init__(self, serial): + self.serial = serial + + def adb(self, cmd): + full = f"adb -s {self.serial} {cmd}" + r = subprocess.run(full, shell=True, capture_output=True, timeout=10) + return r.stdout.decode("utf-8", errors="ignore").strip() + + def shell(self, cmd): + return self.adb(f'shell "{cmd}"') + + def screenshot(self, name): + """截图并保存到本地，返回文件路径""" + ts = int(time.time()) + fname = f"{name}_{ts}.png" + fpath = os.path.join(OUTPUT_DIR, fname) + self.shell("screencap -p /sdcard/verify_sc.png") + time.sleep(0.5) + self.adb(f"pull /sdcard/verify_sc.png {fpath}") + if os.path.exists(fpath) and os.path.getsize(fpath) > 1000: + return fpath + return "" + + def tap(self, x, y): + """点击屏幕坐标""" + self.shell(f"input tap {x} {y}") + time.sleep(0.8) + + def swipe(self, x1, y1, x2, y2, duration=300): + """滑动""" + self.shell(f"input swipe {x1} {y1} {x2} {y2} {duration}") + time.sleep(0.5) + + def back(self): + """返回键""" + self.shell("input keyevent 4") + time.sleep(0.5) + + def home(self): + """Home键""" + self.shell("input keyevent 3") + time.sleep(0.5) + + def open_wechat(self): + """打开微信主页""" + self.shell(f"am start -n {WECHAT_PKG}/.ui.LauncherUI") + time.sleep(2) + + def open_wechat_contacts(self): + """打开微信通讯录Tab""" + self.open_wechat() + time.sleep(1) + # 点击底部通讯录Tab (通常在底部第2个位置) + # 小米手机分辨率通常是1080x2400 + self.tap(360, 2340) # 通讯录tab + time.sleep(1) + + def open_wechat_discover(self): + """打开微信发现Tab""" + self.open_wechat() + time.sleep(1) + self.tap(720, 2340) # 发现tab + time.sleep(1) + + def open_wechat_me(self): + """打开微信我Tab""" + self.open_wechat() + time.sleep(1) + self.tap(980, 2340) # 我tab + time.sleep(1) + + def open_chat(self, name="filehelper"): + """打开指定聊天窗口""" + self.open_wechat() + time.sleep(1) + # 点击搜索 + self.tap(950, 140) + time.sleep(1) + # 输入搜索内容 + self.shell(f"input text {name}") + time.sleep(1) + # 点击第一个结果 + self.tap(540, 400) + time.sleep(1) + + def get_screen_size(self): + """获取屏幕分辨率""" + r = self.shell("wm size") + # Physical size: 1080x2400 + if "x" in r: + parts = r.split(":")[-1].strip().split("x") + return int(parts[0]), int(parts[1]) + return 1080, 2400 + + +class StepVerifier: + """逐步验证器""" + + def __init__(self): + self.phone = PhoneController(DEVICE_SERIAL) + self.exports = None + self.script = None + self.session = None + self.log_lines = [] + self.results = [] + + def connect(self): + """连接Frida""" + print("[CONNECT] Connecting to Frida...") + device = frida.get_device_manager().add_remote_device(f"{DEVICE_IP}:{FRIDA_PORT}") + self.session = device.attach(WECHAT_PID) + + with open(SCRIPT_PATH, "r", encoding="utf-8") as f: + source = f.read() + + self.script = self.session.create_script(source) + self.script.on("message", lambda m, d: None) + self.script.load() + time.sleep(2) + + self.exports = self.script.exports_sync + avail = [x for x in dir(self.exports) if not x.startswith("_")] + print(f" Loaded {len(avail)} methods") + pong = self.exports.ping() + print(f" ping: {pong}") + + # 获取屏幕尺寸 + w, h = self.phone.get_screen_size() + print(f" Screen: {w}x{h}") + return True + + def log(self, text): + print(text) + self.log_lines.append(text) + + def verify_step(self, module, method, description, nav_action, params=None, wait_after=1): + """ + 执行一个验证步骤: + 1. 导航到对应页面 + 2. 截图(操作前) + 3. 执行RPC + 4. 等待UI响应 + 5. 截图(操作后) + 6. 记录结果 + """ + self.log(f"\n{'='*60}") + self.log(f" [{module}] {method}") + self.log(f" 描述: {description}") + self.log(f"{'='*60}") + + # Step 1: 导航 + self.log(f" [NAV] {nav_action.__doc__ or 'navigating...'}") + nav_action() + time.sleep(1) + + # Step 2: 操作前截图 + before_img = self.phone.screenshot(f"{module}__{method}__before") + self.log(f" [BEFORE] 截图: {os.path.basename(before_img) if before_img else 'FAILED'}") + + # Step 3: 执行RPC + fn = getattr(self.exports, method, None) + if fn is None: + self.log(f" [ERROR] 方法不存在: {method}") + self.results.append({ + "module": module, "method": method, "status": "MISSING", + "description": description, "data": None, + "before_img": before_img, "after_img": "", + }) + return + + start = time.time() + try: + if params: + result = fn(params) + else: + result = fn() + latency = int((time.time() - start) * 1000) + except Exception as e: + latency = int((time.time() - start) * 1000) + result = {"success": False, "error": str(e)[:200]} + + self.log(f" [EXEC] {method}({json.dumps(params, ensure_ascii=False)[:50] if params else ''})") + self.log(f" [TIME] {latency}ms") + + # 格式化结果 + if isinstance(result, str): + self.log(f" [DATA] {result[:100]}") + data_preview = result + elif isinstance(result, dict): + data_preview = json.dumps(result, ensure_ascii=False, indent=2)[:300] + self.log(f" [DATA] {data_preview[:150]}") + elif isinstance(result, list): + data_preview = f"[{len(result)} items] " + json.dumps(result[:2], ensure_ascii=False)[:200] + self.log(f" [DATA] {data_preview[:150]}") + else: + data_preview = str(result)[:100] + self.log(f" [DATA] {data_preview}") + + # Step 4: 等待UI响应 + time.sleep(wait_after) + + # Step 5: 操作后截图 + after_img = self.phone.screenshot(f"{module}__{method}__after") + self.log(f" [AFTER] 截图: {os.path.basename(after_img) if after_img else 'FAILED'}") + + # 判断状态 + if isinstance(result, dict) and result.get("success") == False: + status = "EXEC_ERR" + else: + status = "PASS" + + self.log(f" [STATUS] {status}") + + self.results.append({ + "module": module, "method": method, "status": status, + "description": description, "latency_ms": latency, + "data": result if isinstance(result, (dict, list, str, bool, int, float)) else str(result), + "before_img": before_img, "after_img": after_img, + }) + + def run_all(self): + """运行所有验证步骤""" + self.log("# 工作手机SDK - 逐步真机验证") + self.log(f"> 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + self.log(f"> 设备: {DEVICE_IP}:{FRIDA_PORT}") + self.log("") + + # ===== 模块1: 系统状态 ===== + self.log("\n## 模块1: 系统状态\n") + + self.verify_step( + "SYS", "getProcessInfo", + "获取微信进程信息(PID/架构/模块数)", + lambda: self.phone.open_wechat(), + ) + + self.verify_step( + "SYS", "getWechatVersion", + "获取微信版本号", + lambda: None, # 不需要导航，直接在当前页面 + ) + + self.verify_step( + "SYS", "getHookStatus", + "检查Hook是否已激活", + lambda: None, + ) + + # ===== 模块2: 联系人 ===== + self.log("\n## 模块2: 联系人\n") + + self.verify_step( + "H16_联系人", "getContacts", + "打开通讯录 -> 获取联系人列表", + lambda: self.phone.open_wechat_contacts(), + params={"limit": 10}, + ) + + self.verify_step( + "H16_联系人", "getContactInfo", + "获取文件传输助手的详细信息", + lambda: None, # 保持在通讯录页面 + params={"wxid": "filehelper"}, + ) + + self.verify_step( + "H16_联系人", "searchContacts", + "搜索包含'文件'的联系人", + lambda: None, + params={"keyword": "文件", "limit": 5}, + ) + + # ===== 模块3: 消息发送 ===== + self.log("\n## 模块3: 消息发送\n") + + msg_content = f"[SDK验证] {datetime.now().strftime('%H:%M:%S')} 测试消息" + self.verify_step( + "H17_消息发送", "sendMessage", + f"打开文件传输助手对话 -> 发送文本消息: '{msg_content}'", + lambda: self.phone.open_chat("filehelper"), + params={"to_id": "filehelper", "content": msg_content, "msg_type": "text"}, + wait_after=2, + ) + + # ===== 模块4: 消息接收 ===== + self.log("\n## 模块4: 消息接收\n") + + self.verify_step( + "H15_消息接收", "getRecentMessages", + "获取最近消息列表", + lambda: self.phone.open_wechat(), + params={"limit": 5}, + ) + + self.verify_step( + "H15_消息接收", "getMessages", + "获取filehelper的消息记录", + lambda: None, + params={"conversation_id": "filehelper", "limit": 5}, + ) + + # ===== 模块5: 好友管理 ===== + self.log("\n## 模块5: 好友管理\n") + + self.verify_step( + "H19_好友管理", "setFriendRemark", + "修改文件传输助手的备注名为'SDK文件助手'", + lambda: self.phone.open_wechat_contacts(), + params={"wxid": "filehelper", "remark": "SDK文件助手"}, + wait_after=2, + ) + + self.verify_step( + "H18_好友请求", "getFriendRequests", + "获取好友请求列表", + lambda: None, + params={"limit": 5}, + ) + + # ===== 模块6: 群管理 ===== + self.log("\n## 模块6: 群管理\n") + + self.verify_step( + "H22_群管理", "getGroups", + "获取群聊列表", + lambda: self.phone.open_wechat(), + params={"limit": 10}, + ) + + # ===== 模块7: 个人信息 ===== + self.log("\n## 模块7: 个人信息\n") + + self.verify_step( + "H23_账号管理", "getProfile", + "打开'我'页面 -> 获取个人资料", + lambda: self.phone.open_wechat_me(), + params={}, + ) + + self.verify_step( + "H23_账号管理", "checkAccountStatus", + "检查账号状态", + lambda: None, + params={}, + ) + + # ===== 模块8: 标签 ===== + self.log("\n## 模块8: 标签\n") + + self.verify_step( + "H28_标签", "getLabels", + "获取标签列表", + lambda: None, + params={}, + ) + + # ===== 模块9: 朋友圈 ===== + self.log("\n## 模块9: 朋友圈\n") + + self.verify_step( + "H21_朋友圈", "getMoments", + "打开发现页 -> 获取朋友圈动态", + lambda: self.phone.open_wechat_discover(), + params={"wxid": "", "limit": 3}, + ) + + # ===== 模块10: 搜索 ===== + self.log("\n## 模块10: 搜索\n") + + self.verify_step( + "H31_搜索", "globalSearch", + "全局搜索'微信'", + lambda: self.phone.open_wechat(), + params={"keyword": "微信", "limit": 5}, + ) + + # ===== 模块11: 设备信息 ===== + self.log("\n## 模块11: 设备信息\n") + + self.verify_step( + "H39_设备信息", "getDeviceInfo", + "获取设备信息", + lambda: None, + params={}, + ) + + self.verify_step( + "H39_设备信息", "getNetworkInfo", + "获取网络信息", + lambda: None, + params={}, + ) + + self.verify_step( + "H39_设备信息", "getStorageInfo", + "获取存储信息", + lambda: None, + params={}, + ) + + # ===== 模块12: 账号安全 ===== + self.log("\n## 模块12: 账号安全\n") + + self.verify_step( + "H24_账号安全", "getLoginDevices", + "获取登录设备列表", + lambda: self.phone.open_wechat_me(), + params={}, + ) + + self.verify_step( + "H35_注册登录", "checkLoginState", + "检查登录状态", + lambda: None, + params={}, + ) + + # ===== 汇总 ===== + self.summarize() + + def summarize(self): + """汇总结果""" + total = len(self.results) + passed = sum(1 for r in self.results if r["status"] in ("PASS", "EXEC_ERR")) + missing = sum(1 for r in self.results if r["status"] == "MISSING") + + self.log(f"\n{'='*60}") + self.log(f" 验证完成!") + self.log(f" 总计: {total}") + self.log(f" 通过: {passed}") + self.log(f" 缺失: {missing}") + self.log(f" 通过率: {passed/total*100:.1f}%") + self.log(f"{'='*60}") + + # 保存日志 + with open(LOG_FILE, "w", encoding="utf-8") as f: + f.write("\n".join(self.log_lines)) + print(f"\n日志: {LOG_FILE}") + + # 保存JSON + result_file = os.path.join(OUTPUT_DIR, "step_results.json") + with open(result_file, "w", encoding="utf-8") as f: + json.dump({ + "time": datetime.now().isoformat(), + "total": total, + "passed": passed, + "missing": missing, + "results": self.results, + }, f, ensure_ascii=False, indent=2, default=str) + print(f"结果: {result_file}") + + def cleanup(self): + try: + self.script.unload() + self.session.detach() + except: + pass + + +def main(): + v = StepVerifier() + try: + if v.connect(): + v.run_all() + finally: + v.cleanup() + + +if __name__ == "__main__": + main() diff --git a/sdk/tests/test_rpc_call.py b/sdk/tests/test_rpc_call.py new file mode 100644 index 0000000000..42ec4d5251 --- /dev/null +++ b/sdk/tests/test_rpc_call.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""测试Frida RPC调用方式 - 确定正确的方法名格式""" +import frida +import json + +DEVICE_IP = "192.168.0.12" +FRIDA_PORT = 27042 +WECHAT_PID = 7239 + +mgr = frida.get_device_manager() +device = mgr.add_remote_device(f"{DEVICE_IP}:{FRIDA_PORT}") +print(f"Device: {device.name}") + +# 用PID attach +print(f"Attaching to PID {WECHAT_PID}...") +session = device.attach(WECHAT_PID) +print("Attached!") + +# 加载最小测试脚本 +test_script = """ +rpc.exports = { + ping: function() { return 'pong_test'; }, + getProfile: function() { return {success: true, nickname: 'test'}; }, + sendMessage: function(params) { return {success: true, params: params}; }, +}; +""" + +script = session.create_script(test_script) +script.load() + +# 检查Python端看到的方法名 +exports = script.exports_sync +available = [x for x in dir(exports) if not x.startswith('_')] +print(f"\nAvailable exports: {available}") + +# 测试调用 +print(f"\nping(): {exports.ping()}") + +# Frida Python 会把 camelCase 转成 snake_case +# getProfile -> get_profile +try: + result = exports.get_profile() + print(f"get_profile(): {result}") +except Exception as e: + print(f"get_profile() failed: {e}") + +# 或者直接用原始名 +try: + result = exports.getProfile() + print(f"getProfile(): {result}") +except Exception as e: + print(f"getProfile() failed: {e}") + +# 测试带参数 +try: + result = exports.send_message({"to_id": "test", "content": "hello"}) + print(f"send_message(params): {result}") +except Exception as e: + print(f"send_message(params) failed: {e}") + +script.unload() +session.detach() +print("\nDone!")