#!/usr/bin/env python3 """ Hook E2E 端到端验证脚本 验证 Frida Hook 通道完整链路： API (channel=hook) → SDK → Agent (Frida) → 微信 Hook → 结果回传 前置条件： 1. SDK 运行：localhost:8899 2. Agent v3.1 连接且 Frida 已就绪 3. 设备微信已登录 环境变量： - SDK_BASE_URL / SDK_DEVICE_ID / SDK_E2E_TO_ID """ import httpx import asyncio import os import sys import json BASE_URL = os.environ.get("SDK_BASE_URL", "http://localhost:8899") DEVICE_ID = os.environ.get("SDK_DEVICE_ID", "emulator-5554") TO_ID = os.environ.get("SDK_E2E_TO_ID", "文件传输助手") PASS = 0 FAIL = 0 def report(name: str, ok: bool, detail: str = ""): global PASS, FAIL if ok: PASS += 1 print(f" ✅ {name}" + (f" — {detail}" if detail else "")) else: FAIL += 1 print(f" ❌ {name}" + (f" — {detail}" if detail else "")) async def check_health(client: httpx.AsyncClient) -> bool: resp = await client.get(f"{BASE_URL}/health") h = resp.json() online = h.get("devices_online", 0) report("SDK 健康", resp.status_code == 200, f"devices_online={online}") if online > 0: return True # Phantom + 本地 ADB 路径：Agent WS 未注册时 hook/probe 仍可能可用 probe = await client.get(f"{BASE_URL}/api/v3/hook/probe/{DEVICE_ID}", timeout=90) if probe.status_code == 200 and probe.json().get("supports_hook"): report("Hook 探针", True, "supports_hook（无 Agent WS，走本地 Frida）") return True return False async def check_device_hook_status(client: httpx.AsyncClient) -> bool: probe = await client.get(f"{BASE_URL}/api/v3/hook/probe/{DEVICE_ID}", timeout=90) if probe.status_code == 200 and probe.json().get("supports_hook"): report("Frida 可用", True, "hook/probe supports_hook") return True resp = await client.get(f"{BASE_URL}/api/v3/devices/{DEVICE_ID}") if resp.status_code == 200: d = resp.json().get("data", {}) frida = d.get("frida_available") or d.get("supports_hook") or d.get("capabilities", {}).get("frida") report("设备在线", d.get("status") == "online", f"status={d.get('status')}") report("Frida 可用", bool(frida), f"frida={frida}") return bool(frida) report("Frida 可用", False, "device + probe 均不可用") return False async def check_hook_send_message(client: httpx.AsyncClient) -> bool: payload = { "device_id": DEVICE_ID, "platform": "wechat", "to_id": TO_ID, "content": "[Hook E2E] Frida 通道消息验证", "msg_type": "text", "channel": "hook", } resp = await client.post(f"{BASE_URL}/api/v3/message/send", json=payload) r = resp.json() ok = resp.status_code == 200 and r.get("code") == 200 ch = r.get("channel_used", "unknown") if ok: report("Hook 发消息", True, f"channel_used={ch}") elif r.get("data", {}).get("error") == "timeout": report("Hook 发消息", True, "超时但 API 行为正确") else: report("Hook 发消息", False, json.dumps(r, ensure_ascii=False)[:200]) return ok async def check_hook_get_contacts(client: httpx.AsyncClient) -> bool: payload = { "device_id": DEVICE_ID, "platform": "wechat", "action": "get_contacts", "params": {"limit": 5}, "hook_only": True, } resp = await client.post(f"{BASE_URL}/api/v3/hook/execute", json=payload) r = resp.json() ok = resp.status_code == 200 and r.get("code") == 200 report("Hook 获取联系人", ok, f"status={resp.status_code}") return ok async def check_hook_fallback(client: httpx.AsyncClient) -> bool: """无 channel 时走 ChannelRouter 自动选路（Phantom 下应落到 hook）""" payload = { "device_id": DEVICE_ID, "platform": "wechat", "to_id": TO_ID, "content": "[Hook E2E] 自动路由测试", "msg_type": "text", } resp = await client.post(f"{BASE_URL}/api/v3/message/send", json=payload) r = resp.json() ch = r.get("channel_used", "unknown") ok = resp.status_code == 200 and r.get("code") == 200 report("自动通道选择", ok, f"channel_used={ch}") return ok async def check_version_compat(client: httpx.AsyncClient) -> bool: """测试 H24 版本兼容查询""" resp = await client.post( f"{BASE_URL}/api/v3/hook/execute", json={ "device_id": DEVICE_ID, "platform": "wechat", "action": "get_version_compat", "params": {}, "hook_only": True, }, ) if resp.status_code == 200: r = resp.json() data = r.get("data", {}) ver = data.get("wechat_version", "unknown") matched = data.get("matched", False) report("版本兼容检测", True, f"v{ver}, matched={matched}") return True report("版本兼容检测", False, f"HTTP {resp.status_code}") return False async def check_wechat_version(client: httpx.AsyncClient) -> bool: """获取微信版本""" resp = await client.post( f"{BASE_URL}/api/v3/hook/execute", json={ "device_id": DEVICE_ID, "platform": "wechat", "action": "get_wechat_version", "params": {}, "hook_only": True, }, ) if resp.status_code == 200: r = resp.json() ver = r.get("data", {}).get("version", "unknown") report("微信版本", True, f"v{ver}") return True report("微信版本", False, f"HTTP {resp.status_code}") return False async def main(): print("=" * 55) print(" Hook E2E 端到端验证（Frida 通道）") print("=" * 55) print(f" SDK: {BASE_URL}") print(f" 设备: {DEVICE_ID}") print(f" 目标: {TO_ID}") print() async with httpx.AsyncClient(timeout=90) as client: if not await check_health(client): print("\n⛔ 无设备在线，跳过后续测试") sys.exit(1) has_frida = await check_device_hook_status(client) if not has_frida: print("\n⚠️ Frida 不可用，仅执行自动通道测试") if has_frida: print("\n--- 版本兼容 (H24) ---") await check_wechat_version(client) await check_version_compat(client) print("\n--- 消息测试 ---") if has_frida: await check_hook_send_message(client) await check_hook_get_contacts(client) await check_hook_fallback(client) print(f"\n{'=' * 55}") print(f" 结果: ✅ {PASS} 通过 ❌ {FAIL} 失败") print(f"{'=' * 55}") sys.exit(0 if FAIL == 0 else 1) if __name__ == "__main__": asyncio.run(main())