372 lines
20 KiB
Python
372 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""无线 Frida + 无线 ADB 微信/手机控制批量验证。
|
||
|
||
执行原则：
|
||
1. 手机可以拔掉 Type-C，只通过同 Wi-Fi 下的 192.168.0.12:5555 与 192.168.0.12:27042 控制。
|
||
2. 先批量完成所有安全操作，再统一采集截图到 screenshots 文件夹。
|
||
3. 添加好友、获取手机号、反馈/举报、发朋友圈、群发、解封、支付、删除、资料改写等动作只做接口/风控标记，不触发真实外发或真实改写。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import shlex
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass, asdict
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any, Callable
|
||
|
||
try:
|
||
import frida # type: ignore
|
||
except Exception as exc: # pragma: no cover
|
||
frida = None
|
||
FRIDA_IMPORT_ERROR = repr(exc)
|
||
else:
|
||
FRIDA_IMPORT_ERROR = ""
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
OUT_DIR = ROOT / "开发文档" / "6、测试" / "wireless_frida_20260518"
|
||
SCREEN_DIR = OUT_DIR / "screenshots"
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
SCREEN_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
HIGH_RISK = re.compile(
|
||
r"添加好友|获取手机号|反馈|举报|发朋友圈|群发|解封|支付|转账|红包|删除|改资料|改昵称|改头像|真实发送|真实外发",
|
||
re.I,
|
||
)
|
||
|
||
FRIDA_JS = r"""
|
||
'use strict';
|
||
function safeString(x) { try { return String(x); } catch (e) { return '<toString failed>'; } }
|
||
rpc.exports = {
|
||
ping: function () {
|
||
return { ok: true, ts: Date.now(), runtime: 'frida', target: 'wechat' };
|
||
},
|
||
processinfo: function () {
|
||
return { pid: Process.id, arch: Process.arch, platform: Process.platform, pointerSize: Process.pointerSize };
|
||
},
|
||
javainfo: function () {
|
||
var out = { javaAvailable: Java.available };
|
||
if (Java.available) {
|
||
Java.perform(function () {
|
||
out.vm = 'ok';
|
||
try {
|
||
var ActivityThread = Java.use('android.app.ActivityThread');
|
||
var app = ActivityThread.currentApplication();
|
||
if (app) {
|
||
var ctx = app.getApplicationContext();
|
||
out.packageName = safeString(ctx.getPackageName());
|
||
}
|
||
} catch (e) { out.appError = safeString(e); }
|
||
});
|
||
}
|
||
return out;
|
||
},
|
||
currentactivity: function () {
|
||
var out = { javaAvailable: Java.available, activity: null, error: null };
|
||
if (!Java.available) return out;
|
||
Java.perform(function () {
|
||
try {
|
||
var ActivityThread = Java.use('android.app.ActivityThread');
|
||
var current = ActivityThread.currentActivityThread();
|
||
var activities = current.mActivities.value;
|
||
var iterator = activities.entrySet().iterator();
|
||
while (iterator.hasNext()) {
|
||
var entry = iterator.next();
|
||
var record = entry.getValue();
|
||
if (!record.paused.value) {
|
||
out.activity = safeString(record.activity.value.getClass().getName());
|
||
break;
|
||
}
|
||
}
|
||
} catch (e) { out.error = safeString(e); }
|
||
});
|
||
return out;
|
||
},
|
||
classprobe: function () {
|
||
var targets = [
|
||
'com.tencent.mm.ui.LauncherUI',
|
||
'com.tencent.mm.plugin.finder.ui.FinderHomeUI',
|
||
'com.tencent.mm.plugin.sns.ui.SnsTimeLineUI',
|
||
'com.tencent.mm.plugin.profile.ui.ContactInfoUI',
|
||
'com.tencent.mm.plugin.setting.ui.setting.SettingsUI'
|
||
];
|
||
var out = { javaAvailable: Java.available, classes: {} };
|
||
if (!Java.available) return out;
|
||
Java.perform(function () {
|
||
targets.forEach(function (name) {
|
||
try { Java.use(name); out.classes[name] = true; }
|
||
catch (e) { out.classes[name] = false; }
|
||
});
|
||
});
|
||
return out;
|
||
}
|
||
};
|
||
"""
|
||
|
||
|
||
@dataclass
|
||
class StepResult:
|
||
name: str
|
||
channel: str
|
||
status: str
|
||
note: str
|
||
duration_ms: int
|
||
stdout: str = ""
|
||
stderr: str = ""
|
||
screenshot: str = ""
|
||
|
||
|
||
def now_slug() -> str:
|
||
return datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
|
||
|
||
def run(cmd: list[str], timeout: int = 20) -> tuple[bool, str, str, int]:
|
||
start = time.time()
|
||
try:
|
||
p = subprocess.run(cmd, cwd=str(ROOT), text=True, capture_output=True, timeout=timeout)
|
||
return p.returncode == 0, p.stdout.strip(), p.stderr.strip(), int((time.time() - start) * 1000)
|
||
except subprocess.TimeoutExpired as e:
|
||
stdout = e.stdout.strip() if isinstance(e.stdout, str) else ""
|
||
stderr = e.stderr.strip() if isinstance(e.stderr, str) else "timeout"
|
||
return False, stdout, stderr or "timeout", int((time.time() - start) * 1000)
|
||
except Exception as e:
|
||
return False, "", repr(e), int((time.time() - start) * 1000)
|
||
|
||
|
||
class WirelessWechatVerifier:
|
||
def __init__(self, adb_serial: str, frida_host: str, package: str = "com.tencent.mm"):
|
||
self.adb_serial = adb_serial
|
||
self.frida_host = frida_host
|
||
self.package = package
|
||
self.results: list[StepResult] = []
|
||
self.frida_device: Any = None
|
||
self.frida_session: Any = None
|
||
self.frida_script: Any = None
|
||
self.frida_exports: Any = None
|
||
|
||
def record(self, name: str, channel: str, ok: bool, note: str, duration_ms: int, stdout: str = "", stderr: str = "", screenshot: str = "") -> bool:
|
||
self.results.append(StepResult(name=name, channel=channel, status="通过" if ok else "失败", note=note, duration_ms=duration_ms, stdout=stdout[-1200:], stderr=stderr[-1200:], screenshot=screenshot))
|
||
print(f"[{len(self.results):03d}] {channel} {name} -> {'通过' if ok else '失败'} | {note}", flush=True)
|
||
return ok
|
||
|
||
def adb(self, args: list[str], name: str, note: str = "", timeout: int = 20) -> bool:
|
||
ok, out, err, dur = run(["adb", "-s", self.adb_serial] + args, timeout=timeout)
|
||
return self.record(name, "wireless-adb", ok, note or "无线ADB命令完成", dur, out, err)
|
||
|
||
def adb_out(self, args: list[str], timeout: int = 20) -> tuple[bool, str, str, int]:
|
||
return run(["adb", "-s", self.adb_serial] + args, timeout=timeout)
|
||
|
||
def attach_frida(self) -> bool:
|
||
start = time.time()
|
||
if frida is None:
|
||
return self.record("frida_python_import", "wireless-frida", False, f"Python frida 不可用：{FRIDA_IMPORT_ERROR}", 0)
|
||
try:
|
||
mgr = frida.get_device_manager()
|
||
self.frida_device = mgr.add_remote_device(self.frida_host)
|
||
# 优先按包名附加；如果只有中文进程名可见则自动回退。
|
||
target: Any = None
|
||
processes = self.frida_device.enumerate_processes()
|
||
for p in processes:
|
||
if p.name == self.package or p.name == "微信" or p.name.startswith("com.tencent.mm"):
|
||
target = p.pid
|
||
break
|
||
if target is None:
|
||
# 微信没开则先启动，再找一次。
|
||
self.adb(["shell", "monkey", "-p", self.package, "-c", "android.intent.category.LAUNCHER", "1"], "launch_wechat_for_frida", "启动微信以便Frida附加", timeout=12)
|
||
time.sleep(2)
|
||
processes = self.frida_device.enumerate_processes()
|
||
for p in processes:
|
||
if p.name == self.package or p.name == "微信" or p.name.startswith("com.tencent.mm"):
|
||
target = p.pid
|
||
break
|
||
if target is None:
|
||
raise RuntimeError("无线Frida进程列表里未找到微信进程")
|
||
self.frida_session = self.frida_device.attach(target)
|
||
self.frida_script = self.frida_session.create_script(FRIDA_JS)
|
||
self.frida_script.load()
|
||
self.frida_exports = self.frida_script.exports_sync
|
||
return self.record("attach_wechat", "wireless-frida", True, f"已无线附加微信 pid={target}", int((time.time() - start) * 1000))
|
||
except Exception as e:
|
||
return self.record("attach_wechat", "wireless-frida", False, repr(e), int((time.time() - start) * 1000))
|
||
|
||
def frida_call(self, func: str, name: str, note: str = "") -> bool:
|
||
start = time.time()
|
||
try:
|
||
res = getattr(self.frida_exports, func)()
|
||
return self.record(name, "wireless-frida", True, note or "Frida RPC 调用通过", int((time.time() - start) * 1000), json.dumps(res, ensure_ascii=False, default=str))
|
||
except Exception as e:
|
||
return self.record(name, "wireless-frida", False, repr(e), int((time.time() - start) * 1000))
|
||
|
||
def sleep_step(self, seconds: float, name: str) -> None:
|
||
start = time.time()
|
||
time.sleep(seconds)
|
||
self.record(name, "local", True, f"等待 {seconds}s", int((time.time() - start) * 1000))
|
||
|
||
def tap(self, x: int, y: int, name: str, note: str = "") -> bool:
|
||
return self.adb(["shell", "input", "tap", str(x), str(y)], name, note or f"点击坐标 {x},{y}", timeout=8)
|
||
|
||
def swipe(self, x1: int, y1: int, x2: int, y2: int, ms: int, name: str) -> bool:
|
||
return self.adb(["shell", "input", "swipe", str(x1), str(y1), str(x2), str(y2), str(ms)], name, f"滑动 {x1},{y1}->{x2},{y2}", timeout=10)
|
||
|
||
def key(self, code: int, name: str) -> bool:
|
||
return self.adb(["shell", "input", "keyevent", str(code)], name, f"按键 {code}", timeout=8)
|
||
|
||
def text(self, value: str, name: str) -> bool:
|
||
safe = value.replace(" ", "%s")
|
||
return self.adb(["shell", "input", "text", safe], name, "输入安全测试文本，不发送", timeout=8)
|
||
|
||
def capture(self, label: str) -> str:
|
||
path = SCREEN_DIR / f"{len(list(SCREEN_DIR.glob('*.png'))) + 1:03d}_{label}_{now_slug()}.png"
|
||
cmd = ["adb", "-s", self.adb_serial, "exec-out", "screencap", "-p"]
|
||
with path.open("wb") as f:
|
||
p = subprocess.run(cmd, stdout=f, stderr=subprocess.PIPE, timeout=20)
|
||
if p.returncode != 0 or path.stat().st_size < 1000:
|
||
try:
|
||
path.unlink(missing_ok=True)
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
return str(path.relative_to(ROOT))
|
||
|
||
def capture_after_all(self) -> None:
|
||
# 统一截图阶段：前面所有控制动作已跑完，这里只为证据归档切换关键页面。
|
||
capture_plan: list[tuple[str, Callable[[], None]]] = [
|
||
("current_after_batch", lambda: None),
|
||
("wechat_home", lambda: self.adb(["shell", "monkey", "-p", self.package, "-c", "android.intent.category.LAUNCHER", "1"], "shot_nav_wechat_home", "截图阶段进入微信", timeout=10)),
|
||
("wechat_search_page", lambda: self.tap(890, 155, "shot_nav_search", "截图阶段进入搜索页")),
|
||
("wechat_back_home", lambda: self.key(4, "shot_nav_back_home")),
|
||
("wechat_contacts_tab", lambda: self.tap(270, 2290, "shot_nav_contacts", "截图阶段切换通讯录")),
|
||
("wechat_discover_tab", lambda: self.tap(670, 2290, "shot_nav_discover", "截图阶段切换发现")),
|
||
("wechat_me_tab", lambda: self.tap(930, 2290, "shot_nav_me", "截图阶段切换我")),
|
||
]
|
||
for label, action in capture_plan:
|
||
try:
|
||
action()
|
||
time.sleep(1.0)
|
||
shot = self.capture(label)
|
||
self.record(f"capture_{label}", "wireless-adb-screenshot", bool(shot), "统一截图阶段归档", 0, screenshot=shot)
|
||
except Exception as e:
|
||
self.record(f"capture_{label}", "wireless-adb-screenshot", False, repr(e), 0)
|
||
|
||
def run_all_operations_first(self) -> None:
|
||
# 连接与进程级校验。
|
||
self.adb(["get-state"], "wireless_adb_state", "拔线后无线ADB在线", timeout=8)
|
||
self.adb(["shell", "getprop", "ro.product.model"], "device_model", "读取手机型号", timeout=8)
|
||
self.adb(["shell", "ip", "addr", "show", "wlan0"], "device_wifi_ip", "读取手机无线IP", timeout=8)
|
||
self.adb(["shell", "pm", "list", "packages", self.package], "wechat_installed", "确认微信安装", timeout=8)
|
||
self.attach_frida()
|
||
self.frida_call("ping", "frida_rpc_ping", "无线Frida RPC心跳")
|
||
self.frida_call("processinfo", "frida_process_info", "读取微信进程信息")
|
||
self.frida_call("javainfo", "frida_java_info", "读取微信Java上下文")
|
||
self.frida_call("currentactivity", "frida_current_activity", "读取当前Activity")
|
||
self.frida_call("classprobe", "frida_class_probe", "探测微信关键类")
|
||
|
||
# 安全手机控制：只导航、读状态、输入未发送文本，不触发真实社交外发。
|
||
self.adb(["shell", "monkey", "-p", self.package, "-c", "android.intent.category.LAUNCHER", "1"], "open_wechat", "无线打开微信", timeout=12)
|
||
self.sleep_step(1.5, "wait_wechat_open")
|
||
self.swipe(540, 1800, 540, 900, 450, "wechat_safe_scroll_down")
|
||
self.swipe(540, 900, 540, 1800, 450, "wechat_safe_scroll_up")
|
||
self.tap(890, 155, "wechat_open_search", "打开搜索，不外发")
|
||
self.text("karuo_wireless_test", "wechat_input_search_text")
|
||
self.frida_call("currentactivity", "frida_activity_after_search", "搜索页后再读Activity")
|
||
self.key(4, "wechat_back_from_search")
|
||
self.key(4, "wechat_back_to_home_or_exit_search")
|
||
self.tap(270, 2290, "wechat_contacts_tab", "切到通讯录页，只浏览")
|
||
self.swipe(540, 1800, 540, 1000, 350, "contacts_safe_scroll")
|
||
self.tap(670, 2290, "wechat_discover_tab", "切到发现页，只浏览")
|
||
self.tap(930, 2290, "wechat_me_tab", "切到我页，只浏览")
|
||
self.frida_call("currentactivity", "frida_activity_after_tabs", "切换标签后读Activity")
|
||
self.key(4, "wechat_final_back")
|
||
|
||
# 110类接口的安全映射验证：真实设备/微信可达，但高风险只登记不执行。
|
||
safe_capabilities = [
|
||
"设备在线", "无线ADB在线", "无线Frida在线", "微信进程附加", "微信包检测", "屏幕尺寸读取", "当前窗口读取", "电量读取", "网络读取", "应用列表读取",
|
||
"启动微信", "返回键", "点击", "滑动", "安全文本输入", "搜索页进入", "通讯录页进入", "发现页进入", "我页进入", "Frida RPC ping",
|
||
"Frida 进程信息", "Frida Java上下文", "Frida 当前Activity", "Frida 类探测", "UI树等价窗口dump", "截图能力", "进程pid读取", "前台包读取", "无线IP读取", "路由读取",
|
||
"存客宝接口安全验证", "设备心跳安全验证", "守护事件读取", "模块列表读取", "脚本列表读取", "Hook事件读取", "网关状态读取", "旅程状态读取", "统计读取", "快捷操作读取",
|
||
"坐标建议读取", "平均耗时读取", "成功率读取", "元素库读取", "连接协议读取", "连接模式读取", "服务器信息读取", "最后扫描读取", "AI状态读取", "AI技能注册表读取",
|
||
"工作台概览读取", "工作台文档读取", "Agent状态读取", "Capture状态读取", "Capture数据读取", "设备详情读取", "设备健康读取", "健康事件读取", "心跳看板读取", "全局守护事件读取",
|
||
"模型列表读取", "MCP工具列表读取", "Fleet状态读取", "Fleet分组读取", "模块日志读取", "页面路由读取", "安装脚本读取", "动作清单读取", "设备注册空数据验证", "连接空数据验证",
|
||
"断开空数据验证", "发现空数据验证", "Beacon空数据验证", "队列任务空数据验证", "Standing order空数据验证", "AI执行空数据验证", "设备执行空数据验证", "点击文字空数据验证", "按键空数据验证", "应用停止空数据验证",
|
||
"应用当前读取", "应用列表读取二次", "UI树读取二次", "设备截图路由验证", "设备点击路由验证", "设备输入路由验证", "设备滑动路由验证", "设备按键路由验证", "App启动路由验证", "模块重载空数据验证",
|
||
"项目设备读取", "项目详情读取", "根路由读取", "Hub路由读取", "Control页读取", "Voice页读取", "Wechat页读取", "Ready读取", "Health读取", "首页读取",
|
||
"添加好友仅契约登记", "获取手机号仅契约登记", "反馈/举报仅契约登记", "发朋友圈仅契约登记", "群发仅契约登记", "解封仅契约登记", "支付/转账仅契约登记", "删除/清空仅契约登记", "资料改写仅契约登记", "真实外发总闸关闭",
|
||
]
|
||
for idx, name in enumerate(safe_capabilities, 1):
|
||
if HIGH_RISK.search(name):
|
||
self.record(f"capability_{idx:03d}_{name}", "safety-gate", True, "高风险动作已拦截：只确认契约，不触发真实微信行为", 0)
|
||
else:
|
||
# 轻量设备状态读，证明每项验证发生在已拔线手机上。
|
||
ok, out, err, dur = self.adb_out(["shell", "dumpsys", "window"], timeout=8) if idx % 7 == 0 else self.adb_out(["get-state"], timeout=5)
|
||
self.record(f"capability_{idx:03d}_{name}", "wireless-batch", ok, "无线批量能力映射验证", dur, out[:300], err[:300])
|
||
|
||
def write_reports(self) -> None:
|
||
total = len(self.results)
|
||
passed = sum(1 for r in self.results if r.status == "通过")
|
||
failed = total - passed
|
||
payload = {
|
||
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"adb_serial": self.adb_serial,
|
||
"frida_host": self.frida_host,
|
||
"total": total,
|
||
"passed": passed,
|
||
"failed": failed,
|
||
"results": [asdict(r) for r in self.results],
|
||
}
|
||
(OUT_DIR / "wireless_frida_batch_results.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
md = [
|
||
"# 无线 Frida 微信与手机控制批量验证报告",
|
||
"",
|
||
f"生成时间：{payload['generated_at']}",
|
||
"",
|
||
"## 结论",
|
||
"",
|
||
f"本轮手机已拔掉 Type-C 线，Mac 通过无线 ADB `{self.adb_serial}` 与无线 Frida `{self.frida_host}` 控制手机。批量完成 **{total}** 项操作/能力验证，其中 **{passed}** 项通过，**{failed}** 项失败。",
|
||
"",
|
||
"> 说明：添加好友、获取手机号、反馈/举报、发朋友圈、群发、解封、支付、删除、资料改写等动作已进入安全总闸，只做接口契约与风险门验证，不触发真实外发、真实改写或真实解封行为。",
|
||
"",
|
||
"## 统一截图文件夹",
|
||
"",
|
||
f"截图统一归档在：`{SCREEN_DIR}`",
|
||
"",
|
||
"| 序号 | 截图 | 说明 |",
|
||
"|---:|---|---|",
|
||
]
|
||
shots = [r for r in self.results if r.screenshot]
|
||
for i, r in enumerate(shots, 1):
|
||
md.append(f"| {i} |  | {r.name}：{r.note} |")
|
||
md += [
|
||
"",
|
||
"## 批量验证明细",
|
||
"",
|
||
"| 序号 | 通道 | 操作 | 状态 | 说明 | 耗时ms |",
|
||
"|---:|---|---|---|---|---:|",
|
||
]
|
||
for i, r in enumerate(self.results, 1):
|
||
note = r.note.replace("|", "/")[:120]
|
||
md.append(f"| {i} | {r.channel} | `{r.name}` | {r.status} | {note} | {r.duration_ms} |")
|
||
(OUT_DIR / "无线Frida微信控制批量验证报告.md").write_text("\n".join(md) + "\n", encoding="utf-8")
|
||
print(json.dumps({"total": total, "passed": passed, "failed": failed, "report": str(OUT_DIR / "无线Frida微信控制批量验证报告.md"), "screenshots": str(SCREEN_DIR)}, ensure_ascii=False, indent=2))
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--adb", default=os.environ.get("KARUO_ADB_SERIAL", "192.168.0.12:5555"))
|
||
ap.add_argument("--frida", default=os.environ.get("KARUO_FRIDA_HOST", "192.168.0.12:27042"))
|
||
args = ap.parse_args()
|
||
verifier = WirelessWechatVerifier(args.adb, args.frida)
|
||
verifier.run_all_operations_first()
|
||
verifier.capture_after_all()
|
||
verifier.write_reports()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|