live-hub-py/scripts/diag_phone_lifecycle.py

#!/usr/bin/env python3
"""真机版 存活/闪退四组对照诊断器 (M2102J2SC 5dd8c93f, fs152 15.2.2).

背景: 模拟器 GC3VE 上结论 = attach 主进程必死(msaoaidsec 静默 _exit) /
spawn+脚本必 EGL 崩。换真机后两项都要重测 —— 真机 GPU 正常、无模拟器特征,
msaoaidsec 在真机上的 frida 检测行为是未知数。

A. baseline      无 frida 正常启动                       (应该活)
B. spawn-zero    spawn 立即 resume, 不加载任何脚本       (模拟器上稳定活)
C. spawn-bypass  spawn 挂起中加载 art_callsite 补丁后 resume (模拟器上 EGL 崩)
D. attach        正常启动 7s 后 attach 主进程, 裸 attach  (模拟器上静默 _exit)

指标(与模拟器版一致, 可交叉对比):
  proc_main  /proc/<pid>/cmdline 精确等于 com.duowan.kiwi
  ui_focus  dumpsys window mCurrentFocus 是否含包名
  crash     logcat -b crash 的 Fatal/signal/Abort 计数
  detached  frida session detached reason/detail

用法:
  /Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0/.venv/bin/python \
      scripts/diag_phone_lifecycle.py [a|b|c|d] [serial] [remote]
"""
from __future__ import annotations

import json
import re
import subprocess
import sys
import time
from pathlib import Path

import frida

SERIAL = sys.argv[2] if len(sys.argv) > 2 else "5dd8c93f"
REMOTE = sys.argv[3] if len(sys.argv) > 3 else "127.0.0.1:31878"
PACKAGE = "com.duowan.kiwi"
REPO = Path("/Users/yml/codes/douyu_login_py")
OUT_DIR = REPO / "evidence" / "diag_phone"
RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
ART_CALLSITE = RE / "evidence" / "scripts" / "bypass_msaoaid_maps_art_callsite.js"

DURATION = 22.0  # 单轮观察时长(s)


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


def adb(*a):
    return sh("adb", "-s", SERIAL, *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_state():
    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_tail():
    r = adb("shell", "logcat", "-d", "-b", "crash", "-t", "60")
    n = r.stdout.count("Fatal") + r.stdout.count("Abort message") + r.stdout.count("signal ")
    return n, r.stdout[-1200:]


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


DETACHED = []


def detach_cb(reason, detail):
    DETACHED.append({"reason": reason, "detail": str(detail)})
    print(f"    [detached] reason={reason} detail={detail}", flush=True)


def run_observe(d, pid, label, extra=None):
    """观察进程到超时或主进程死亡, 每 0.5s 记一次存活. 返回 timeline."""
    t0 = time.time()
    timeline = []
    last_pid = pid
    while time.time() - t0 < DURATION:
        ok, foc = focus_state()
        cur = main_pid()
        timeline.append({"t": round(time.time() - t0, 1), "proc": cur, "ui_ok": ok, "focus": foc})
        if cur is None or (last_pid is not None and cur != last_pid):
            break
        if extra is not None and extra():
            break
        time.sleep(0.5)
    return timeline


def case_a():
    adb("shell", "logcat", "-c")
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.5)
    adb("shell", "monkey", "-p", PACKAGE, "-c", "android.intent.category.LAUNCHER", "1")
    t0 = time.time()
    time.sleep(3)
    pid = main_pid()
    tl = run_observe(None, pid, "A")
    return {"group": "A", "t_spawn": round(time.time() - t0, 1), "pid": pid, "timeline": tl}


def case_b():
    d = frida.get_device_manager().add_remote_device(REMOTE)
    adb("shell", "logcat", "-c")
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.5)
    t0 = time.time()
    pid = d.spawn([PACKAGE])
    d.resume(pid)
    tl = run_observe(d, pid, "B", extra=lambda: False)
    return {"group": "B", "pid": pid, "t_spawn": round(time.time() - t0, 1),
            "timeline": tl, "detached": DETACHED}


def case_c():
    d = frida.get_device_manager().add_remote_device(REMOTE)
    adb("shell", "logcat", "-c")
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.5)
    t0 = time.time()
    pid = d.spawn([PACKAGE])
    suspend_start = time.time()
    session = d.attach(pid)
    session.on("detached", detach_cb)
    sc = session.create_script(ART_CALLSITE.read_text())
    sc.load()
    suspend_ms = (time.time() - suspend_start) * 1000
    d.resume(pid)
    print(f"    [C] spawn pid={pid} 挂起{suspend_ms:.0f}ms(art_callsite) resumed", flush=True)
    tl = run_observe(d, pid, "C", extra=lambda: False)
    try:
        session.detach()
    except Exception:
        pass
    return {"group": "C", "pid": pid, "suspend_ms": round(suspend_ms, 1),
            "timeline": tl, "detached": DETACHED}


def case_d():
    d = frida.get_device_manager().add_remote_device(REMOTE)
    adb("shell", "logcat", "-c")
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.5)
    adb("shell", "monkey", "-p", PACKAGE, "-c", "android.intent.category.LAUNCHER", "1")
    time.sleep(7)
    pid = main_pid()
    t_att = time.time()
    try:
        session = d.attach(pid)
        session.on("detached", detach_cb)
        att_ms = (time.time() - t_att) * 1000
        print(f"    [D] attach pid={pid} 耗时{att_ms:.0f}ms", flush=True)
    except Exception as exc:
        return {"group": "D", "pid": pid, "error": f"attach失败: {exc}",
                "detached": DETACHED}
    tl = run_observe(d, pid, "D", extra=lambda: False)
    try:
        session.detach()
    except Exception:
        pass
    return {"group": "D", "pid": pid, "attach_ms": round(att_ms, 1),
            "timeline": tl, "detached": DETACHED}


def main():
    which = (sys.argv[1] if len(sys.argv) > 1 else "a").lower()
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    cases = {"a": case_a, "b": case_b, "c": case_c, "d": case_d}
    result = cases[which]()
    n, tail = crash_tail()
    result["crash_count"] = n
    result["crash_tail"] = tail
    out = OUT_DIR / f"phone_{which}.json"
    out.write_text(json.dumps(result, indent=1, ensure_ascii=False), encoding="utf-8")
    print(f"[{which.upper()}] -> {out}")
    print("  主进程最终:", result["timeline"][-1] if result["timeline"] else None)
    print("  crash:", n)
    if DETACHED:
        print("  detached:", DETACHED)


if __name__ == "__main__":
    main()