feat(frida): 自愈式多轮运行器 + 两阶段注入稳定性定案 (R13b)
- run_capture.py: spawn-only-bypass两阶段 + 进程死亡自愈重试 + 多轮事件汇总 - 配方: bypass_msaoaid_maps_art_callsite.js(唯一) + 延迟注入钩子; 记录三脚本冲突为反模式 - 实测存活>120s (旧配方<60s) - docs §11.43-10: 稳定性方法论
This commit is contained in:
yml2213
@@ -983,3 +983,10 @@ hypasswordLogin/MsgLoginReq/LogLoginReq → mid 各异, tail 恒定 8b38f1 → a
|
||||
- 后续路线: (a) libhydeviceid 0x64670 区静态攻坚 (datadiv 已解, OLLVM 状态机)
|
||||
(b) frida 真机 (stable bypass) 抓 setDeviceInfo 入参 → 32hex 直读
|
||||
(c) 接受"hdid=设备级证书不可纯代码铸造"结论, 维持金样本 hdid 共用方案 (已跑通多账号)
|
||||
|
||||
### 10) R13b: frida 稳定性优化 (2026-08-29)
|
||||
- 配方定案: **spawn 挂起 + 只注入 bypass_msaoaid_maps_art_callsite.js + resume**, 后延迟+6s 注入业务钩子
|
||||
(同时注入 frida_bypass.js/javaexit = 模拟器向配方, 会与 msaoaid 补丁冲突 → 真机启动卡死, 勿用)
|
||||
- 自愈运行器: tools/frida/run_capture.py (多轮自动重 spawn, 事件汇总单 JSONL, 进程死亡检测)
|
||||
- 实测: 两阶段配方下 app 存活 >120s, crypto-otp/getHdid 全触发 (vs 旧三脚本配方 <60s 被杀)
|
||||
- 已知残余: msaoaid solist 快照面 (G2-0019) 未掩; 登录流程仍不推荐带 frida (标记上报)
|
||||
|
||||
+80
-56
@@ -1,67 +1,67 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""虎牙 OTP 定向捕获运行器。
|
||||
"""虎牙 OTP / setDeviceInfo 定向捕获运行器 (自愈多轮版)。
|
||||
|
||||
稳定性设计 (2026-08-29 第二轮):
|
||||
- 两阶段注入: spawn 挂起只注入 bypass_msaoaid_maps_art_callsite.js (真机唯一稳定配方)
|
||||
-> resume -> 等 app 启动后 (+6s) 再注入 OTP/设备信息钩子 (避开 msaoaid 启动扫描竞争)
|
||||
- 自愈: 每轮 app 被杀/窗口结束自动 force-stop 重 spawn, 最多 N 轮, 事件汇总到同一 JSONL
|
||||
- 单轮窗口 150s (真机稳定绿区 ~90s+, 加上两阶段补偿)
|
||||
|
||||
流程:
|
||||
1) adb 启动手机端 frida-server (re.frida.server/fs152, 监听 127.0.0.1:31878)
|
||||
2) 远程连接 127.0.0.1:31878, spawn com.duowan.kiwi
|
||||
3) 依次注入: frida_bypass.js (maps 伪装) -> frida_java_exit.js (Java kill 拦截)
|
||||
-> hook_otp_capture.js (OTP 六元组 + setSafeDeviceId 捕获)
|
||||
4) resume, 收集 send() 事件 -> JSONL 落盘 + 实时打印
|
||||
用法:
|
||||
python3 run_capture.py # 默认输出 ./capture_<ts>.jsonl
|
||||
CAP_OUT=/tmp/otp.jsonl python3 run_capture.py
|
||||
python3 run_capture.py
|
||||
CAP_MAX_ATTEMPTS=12 CAP_WINDOW=160 python3 run_capture.py
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import threading
|
||||
from datetime import datetime
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
OUT_DIR = os.environ.get("CAP_OUT_DIR", HERE)
|
||||
PKG = "com.duowan.kiwi"
|
||||
SERVER_PATH = "/data/local/tmp/re.frida.server/fs152"
|
||||
SERVER_PATH = "/data/local/tmp/fs152"
|
||||
SERVER_PORT = "31878"
|
||||
BYpass_RE = "/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0/evidence/scripts/bypass_msaoaid_maps_art_callsite.js"
|
||||
JAVA_EXIT_RE = "" # 真机通道: 单一 bypass 脚本最稳, 不加载 java-exit
|
||||
HOOK_LOCAL = os.path.join(HERE, "hook_otp_capture.js")
|
||||
|
||||
MAX_ATTEMPTS = int(os.environ.get("CAP_MAX_ATTEMPTS", "8"))
|
||||
WINDOW_SEC = int(os.environ.get("CAP_WINDOW", "150"))
|
||||
HOOK_DELAY = float(os.environ.get("CAP_HOOK_DELAY", "6"))
|
||||
|
||||
_count = {"events": 0}
|
||||
|
||||
|
||||
def adb(*args):
|
||||
return subprocess.run(["adb"] + list(args), capture_output=True, text=True)
|
||||
|
||||
|
||||
def main():
|
||||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
out_path = os.path.join(OUT_DIR, f"capture_{ts}.jsonl")
|
||||
fout = open(out_path, "w", encoding="utf-8")
|
||||
print(f"[runner] output -> {out_path}", flush=True)
|
||||
|
||||
# 1) 启动手机端 frida-server (root, 后台)
|
||||
print("[runner] starting frida-server on device ...", flush=True)
|
||||
adb("shell", "su", "-c",
|
||||
f"nohup {SERVER_PATH} -l 127.0.0.1:{SERVER_PORT} >/data/local/tmp/re.frida.server/fs152.log 2>&1 &")
|
||||
time.sleep(2.0)
|
||||
|
||||
def run_once(fout, out_path):
|
||||
"""单次 spawn 捕获一轮; 返回该轮事件数 (0 = 该轮无捕获)."""
|
||||
import frida
|
||||
|
||||
# 1) frida-server 存活
|
||||
adb("shell", "su", "-c",
|
||||
f"if ! pgrep -f {SERVER_PATH}; then nohup {SERVER_PATH} -l 127.0.0.1:{SERVER_PORT} >/data/local/tmp/fs152.log 2>&1 & fi")
|
||||
time.sleep(1.5)
|
||||
|
||||
dev = None
|
||||
for i in range(6):
|
||||
try:
|
||||
dev = frida.get_device_manager().add_remote_device(f"127.0.0.1:{SERVER_PORT}")
|
||||
apps = dev.enumerate_processes()
|
||||
print(f"[runner] frida-server OK ({len(apps)} procs)", flush=True)
|
||||
n = len(dev.enumerate_processes())
|
||||
print(f"[runner] frida-server OK ({n} procs)", flush=True)
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"[runner] wait frida-server [{i}] {str(e)[:80]}", flush=True)
|
||||
time.sleep(2.0)
|
||||
if dev is None:
|
||||
print("[runner] FATAL: frida-server unreachable", flush=True)
|
||||
sys.exit(1)
|
||||
return 0
|
||||
|
||||
# 2) 确保旧进程关闭, 然后 spawn
|
||||
adb("shell", "am", "force-stop", PKG)
|
||||
time.sleep(1.0)
|
||||
try:
|
||||
@@ -69,7 +69,7 @@ def main():
|
||||
print(f"[runner] spawned pid={pid}", flush=True)
|
||||
except Exception as e:
|
||||
print(f"[runner] spawn failed: {e}", flush=True)
|
||||
sys.exit(1)
|
||||
return 0
|
||||
|
||||
session = dev.attach(pid)
|
||||
|
||||
@@ -80,57 +80,81 @@ def main():
|
||||
line = json.dumps(payload, ensure_ascii=False)
|
||||
fout.write(line + "\n")
|
||||
fout.flush()
|
||||
_count["events"] += 1
|
||||
ev = payload.get("event", "")
|
||||
if ev in ("crypto-otp", "setdi", "gethdid", "hook-missing", "hook-installed"):
|
||||
print(f"[cap] {line[:400]}", flush=True)
|
||||
elif msg.get("type") == "error":
|
||||
print(f"[cap-err] {msg.get('stack', msg)}", flush=True)
|
||||
print(f"[cap-err] {msg.get('stack', msg)[:300]}", flush=True)
|
||||
elif msg.get("type") == "device":
|
||||
print(f"[cap-dev] {msg.get('payload')}", flush=True)
|
||||
|
||||
scripts = []
|
||||
# 阶段1: 只注入 bypass (STATUS.md 验证的 90s 稳定配方, 避免启动期注入竞争)
|
||||
with open(BYpass_RE, "r", encoding="utf-8") as f:
|
||||
sc = session.create_script(f.read())
|
||||
sc.on("message", on_message)
|
||||
sc.load()
|
||||
scripts.append(sc)
|
||||
print("[runner] loaded bypass (phase-1)", flush=True)
|
||||
# 阶段1: 仅注入 bypass
|
||||
try:
|
||||
with open(BYpass_RE, "r", encoding="utf-8") as f:
|
||||
sc = session.create_script(f.read())
|
||||
sc.on("message", on_message)
|
||||
sc.load()
|
||||
print("[runner] loaded bypass (phase-1)", flush=True)
|
||||
except Exception as e:
|
||||
print(f"[runner] bypass load failed: {e}", flush=True)
|
||||
session.detach()
|
||||
return 0
|
||||
|
||||
dev.resume(pid)
|
||||
print("[runner] RESUMED (phase-1). 等待 app 启动 ...", flush=True)
|
||||
print("[runner] RESUMED (phase-1). 等待 app 启动...", flush=True)
|
||||
|
||||
# 阶段2: app 启动后延迟注入 otp 钩子 (避开 msaoaid 启动扫描 + EGL 竞争)
|
||||
# 阶段2: 延迟注入 otp 钩子
|
||||
def inject_phase2():
|
||||
time.sleep(8)
|
||||
time.sleep(HOOK_DELAY)
|
||||
try:
|
||||
with open(HOOK_LOCAL, "r", encoding="utf-8") as f:
|
||||
src = f.read()
|
||||
sc2 = session.create_script(src)
|
||||
sc2 = session.create_script(f.read())
|
||||
sc2.on("message", on_message)
|
||||
sc2.load()
|
||||
scripts.append(sc2)
|
||||
print("[runner] loaded otp-hook (phase-2)", flush=True)
|
||||
except Exception as e:
|
||||
print(f"[runner] phase-2 inject failed: {e}", flush=True)
|
||||
import threading
|
||||
print(f"[runner] phase-2 inject failed: {e[:200]}", flush=True)
|
||||
threading.Thread(target=inject_phase2, daemon=True).start()
|
||||
|
||||
# 稳定窗口 ~120s 后自动收工
|
||||
def auto_stop():
|
||||
time.sleep(120)
|
||||
print("[runner] auto-stop", flush=True)
|
||||
os._exit(0)
|
||||
threading.Thread(target=auto_stop, daemon=True).start()
|
||||
|
||||
# 窗口等待 + 进程死亡检测 (自愈: 死了立刻重试)
|
||||
start = time.time()
|
||||
while time.time() - start < WINDOW_SEC:
|
||||
time.sleep(1.0)
|
||||
out = adb("shell", "su", "-c", f"kill -0 {pid} 2>/dev/null && echo alive || echo dead")
|
||||
if "dead" in out.stdout:
|
||||
print(f"[runner] process {pid} died at {int(time.time()-start)}s", flush=True)
|
||||
break
|
||||
try:
|
||||
while True:
|
||||
time.sleep(1.0)
|
||||
except KeyboardInterrupt:
|
||||
session.detach()
|
||||
except Exception:
|
||||
pass
|
||||
adb("shell", "am", "force-stop", PKG)
|
||||
print(f"[runner] round done: {out_path}", flush=True)
|
||||
return _count["events"]
|
||||
|
||||
|
||||
def main():
|
||||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
out_path = os.path.join(OUT_DIR, f"capture_{ts}.jsonl")
|
||||
fout = open(out_path, "w", encoding="utf-8")
|
||||
print(f"[runner] output -> {out_path} attempts<={MAX_ATTEMPTS} window={WINDOW_SEC}s", flush=True)
|
||||
|
||||
attempts = 0
|
||||
while attempts < MAX_ATTEMPTS:
|
||||
attempts += 1
|
||||
before = _count["events"]
|
||||
try:
|
||||
run_once(fout, out_path)
|
||||
except Exception as e:
|
||||
print(f"[runner] round {attempts} exception: {str(e)[:150]}", flush=True)
|
||||
if _count["events"] > before:
|
||||
print(f"[runner] attempts={attempts} events={_count['events']} — stopping (capture got data)", flush=True)
|
||||
break
|
||||
time.sleep(3)
|
||||
|
||||
fout.close()
|
||||
print(f"[runner] done -> {out_path}", flush=True)
|
||||
print(f"[runner] FINAL attempts={attempts} events={_count['events']} -> {out_path}", flush=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user