live-hub-py/scripts/diag_lifecycle.py

#!/usr/bin/env python3
"""存活/闪退四组对照诊断器 (统一度量):
  指标1 proc_main: /proc/<pid>/cmdline == 包名 (主进程存活)
  指标2 ui_focus:  dumpsys window mCurrentFocus 是否 com.duowan.kiwi/* (UI 未闪退)
  指标3 crash:     logcat -b crash 是否出现 Fatal (native/java 崩溃)

  A. 无frida  正常启动 (baseline)
  B. spawn    挂起~0s 立即resume, 不加载任何脚本
  C. spawn    挂起 加载2个bypass脚本后再resume (模拟v8/v9/stale)
  D. attach   正常启动7s后 attach 主进程 (裸attach, 无脚本)

用法: python3 diag_lifecycle.py [a|b|c|d]
"""
from pathlib import Path
import frida, time, subprocess, sys, json, re

REMOTE = "127.0.0.1:31878"
ADB = ["adb", "-s", "127.0.0.1:5555"]
PACKAGE = "com.duowan.kiwi"
RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
OUT_DIR = Path("/Users/yml/codes/douyu_login_py/evidence/diag_lifecycle")

def sh(*a): return subprocess.run(a, capture_output=True, text=True)
def adb(*a): return sh(*ADB, *a)

def main_pid():
    r = adb("shell", "pidof", PACKAGE)
    for p in r.stdout.strip().split():
        rr = adb("shell", "cat", f"/proc/{p}/cmdline")
        if rr.stdout.strip("\x00").strip() == PACKAGE:
            return int(p)
    return None

def focus_ok():
    r = adb("shell", "dumpsys", "window")
    m = re.search(r"mCurrentFocus=Window\{([^}]+)\}", r.stdout)
    if m:
        f = m.group(1)
        return PACKAGE in f, f
    return False, "?"

def crash_count():
    r = adb("shell", "logcat", "-d", "-b", "crash", "-t", "50")
    return r.stdout.count("Fatal"), r.stdout[-2000:]

def snapshot(label):
    p = main_pid()
    ok, foc = focus_ok()
    return {"label": label, "proc": p, "ui_ok": ok, "focus": foc}

def main():
    which = (sys.argv[1] if len(sys.argv) > 1 else "a").lower()
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    adb("shell", "logcat", "-c")
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.5)
    d = frida.get_device_manager().add_remote_device(REMOTE)
    pid = None
    t_start = time.time()
    timeline = []
    label = which.upper()

    if which in ("a",):
        print("[A] 无frida 正常启动", flush=True)
        adb("shell", "monkey", "-p", PACKAGE, "-c", "android.intent.category.LAUNCHER", "1")

    elif which in ("b",):
        print("[B] spawn 挂起~0 立即 resume", flush=True)
        pid = d.spawn([PACKAGE])
        dt = time.time() - t_start
        t_start = time.time()
        d.resume(pid)
        print(f"    spawn->resume 挂起 {dt*1000:.0f}ms", flush=True)

    elif which in ("c",):
        print("[C] spawn 挂起+加载2个bypass再resume", flush=True)
        pid = d.spawn([PACKAGE])
        s = d.attach(pid)
        for name in ("bypass_msaoaid_maps_art_callsite.js", "mask_frida_maps_only.js"):
            t0 = time.time()
            sc = s.create_script((RE / "evidence/scripts" / name).read_text())
            sc.load()
            print(f"    loaded {name} +{time.time()-t0:.2f}s", flush=True)
        dt = time.time() - t_start
        t_start = time.time()
        d.resume(pid)
        print(f"    spawn->resume 挂起 {dt*1000:.0f}ms", flush=True)

    elif which in ("d",):
        print("[D] 正常启动7s后 attach 主进程(裸)", flush=True)
        adb("shell", "monkey", "-p", PACKAGE, "-c", "android.intent.category.LAUNCHER", "1")
        time.sleep(7)
        pid = main_pid()
        print(f"    pid={pid}, 开始attach", flush=True)
        try:
            s = d.attach(pid)
            sc = s.create_script("console.log('bare attach ok');")
            sc.load()
            print("    bare attach loaded", flush=True)
        except Exception as e:
            print(f"    attach err {e}", flush=True)
        t_start = time.time()

    else:
        print(f"未知: {which}"); return

    # 观察 60s, 每 2s 快照
    t0 = time.time()
    while time.time() - t0 < 25:
        time.sleep(2)
        s = snapshot(label)
        s["elapsed"] = round(time.time() - t0, 1)
        timeline.append(s)
        if s["ui_ok"]:
            print(f"  +{s['elapsed']:5.1f}s proc={s['proc']} UI={s['focus']}", flush=True)
        else:
            print(f"  +{s['elapsed']:5.1f}s proc={s['proc']} UI_FAIL focus={s['focus']}", flush=True)
        if not s["proc"] and s["elapsed"] > 3:
            break
    crashes, tail = crash_count()
    final = snapshot(label)
    result = {"case": label, "timeline": timeline, "final": final,
              "crashes_in_buffer": crashes, "crash_tail": tail[:1500]}
    out = OUT_DIR / f"diag_{label.lower()}.json"
    out.write_text(json.dumps(result, ensure_ascii=False, indent=1))
    print(f"[*] 结果 -> {out} | 最终: proc={final['proc']} ui_ok={final['ui_ok']} crashes={crashes}", flush=True)

if __name__ == "__main__":
    main()