live-hub-py/scripts/t1_traces.py

#!/usr/bin/env python3
"""T1: frida attach 痕迹快照 (自己的实验).

spawn -> attach(挂起) -> 加载【痕迹采集脚本】(只读, 不改任何东西) -> resume
采集(每~350ms):
  - 进程线程名列表 (/proc/self/task/*/comm)
  - /proc/self/maps 里含 frida/agent/re.linker/gum 的行
  - /proc/self/fd 的 symlink (linjector/pipe/socket)
  - 已加载模块含 frida/agent 的名字
目标: attach(不任何hook) 时进程内哪些痕迹可被 msaoaidsec 看到。
"""
from pathlib import Path
import frida, time, json, subprocess, sys

REMOTE = "127.0.0.1:31878"
ADB = ["adb", "-s", "127.0.0.1:5555"]
PACKAGE = "com.duowan.kiwi"
OUT = Path("/tmp/t1_traces.json")

TRACE_JS = r"""
'use strict';
function snap() {
  const s = { t: Date.now(), threads: [], maps_marks: [], fds: [], modules: [] };
  try {
    for (const t of Process.enumerateThreads()) {
      s.threads.push({ id: t.id, name: t.state == null ? '?' : t.name });
    }
  } catch (_) {}
  try {
    const maps = Process.enumerateMappings ? null : null;
  } catch (_) {}
  try {
    for (const m of Process.enumerateModules()) {
      if (/frida|agent|gum|re\.linker|linjector/i.test(m.name + m.path)) {
        s.modules.push({ name: m.name, path: m.path, base: m.base.toString() });
      }
    }
  } catch (_) {}
  try {
    const MEM = Process.findModuleByName('libc.so');
  } catch (_) {}
  // fd 列表: 用 openat 枚举 /proc/self/fd
  try {
    const dir = new File('/proc/self/fd', 'r');
    const names = dir.read().toString().split(/\s+/);
    for (const n of names) {
      if (!/^\d+$/.test(n)) continue;
      const link = new File('/proc/self/fd/' + n, 'r');
      // can't readlink via File; use readlink via NativeFunction
    }
  } catch (_) {}
  send(s);
}
try {
  const readlink = new NativeFunction(Module.findExportByName('libc.so', 'readlink'), 'long', ['pointer', 'pointer', 'ulong']);
  setInterval(() => {
    const s = { t: Date.now(), threads: [], fds: [], modules: [], links: [] };
    try {
      for (const t of Process.enumerateThreads()) s.threads.push({ id: t.id, name: t.name });
    } catch (_) {}
    try {
      for (const m of Process.enumerateModules()) {
        if (/frida|agent|gum|re\.linker|linjector/i.test(m.name + m.path)) {
          s.modules.push({ name: m.name, path: m.path });
        }
      }
    } catch (_) {}
    // fd symlinks
    try {
      const fp = new File('/proc/self/fd', 'r');
      const buf = fp.read().toString();
      for (const n of buf.split(/\s+/)) {
        if (!/^\d+$/.test(n)) continue;
        try {
          const p = Memory.alloc(512);
          const r = readlink('/proc/self/fd/' + n, p, 512);
          if (r > 0) {
            const link = p.readCString();
            if (/frida|agent|linjector|pipe|socket|memfd/i.test(link)) s.links.push(n + ' -> ' + link);
          }
        } catch (_) {}
      }
    } catch (_) {}
    send(s);
  }, 350);
} catch (e) { send({ err: String(e) }); }
send({ boot: true });
"""

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 main():
    d = frida.get_device_manager().add_remote_device(REMOTE)
    for rnd in (1, 2):
        adb("shell", "am", "force-stop", PACKAGE)
        time.sleep(1.5)
        pid = d.spawn([PACKAGE])
        session = d.attach(pid)
        t0 = time.time()
        snaps = []
        def on(m, dta):
            if m.get('type') != 'send': return
            p = m.get('payload') or {}
            p['elapsed'] = round(time.time() - t0, 2)
            if 'threads' in p or 'err' in p or 'boot' in p:
                snaps.append(p)
        sc = session.create_script(TRACE_JS)
        sc.on('message', on)
        sc.load()
        d.resume(pid)
        print(f"[T1.r{rnd}] spawn pid={pid} attach+采集脚本 resume +{(time.time()-t0)*1000:.0f}ms", flush=True)
        # 观察直到死(最多15s)
        died = None
        while time.time() - t0 < 15:
            time.sleep(1)
            mp = main_pid()
            if mp is None or mp != pid:
                died = time.time() - t0
                break
        print(f"[T1.r{rnd}] 死@+{died if died else 15}s 快照数={len(snaps)}", flush=True)
        if dies := (died or 15):
            print(f"  最后2个快照: {json.dumps(snaps[-2:], ensure_ascii=False)[:900]}", flush=True)
        try: session.detach()
        except Exception: pass
        try: d.kill(pid)
        except Exception: pass
        time.sleep(1)
        OUT.write_text(json.dumps(snaps, ensure_ascii=False, indent=1))
    print(f"[*] -> {OUT}", flush=True)

if __name__ == "__main__":
    main()