notiMessage/reverse/frida/run_tng_compare.py

# -*- coding: utf-8 -*-
"""Compare: Xposed-only vs Frida-stealth-spawn."""
import subprocess
import time

import frida

ADB = r"C:\Users\Administrator\AppData\Local\Android\Sdk\platform-tools\adb.exe"
PKG = "my.com.tngdigital.ewallet"

STEALTH = r"""
"use strict";
function log(m){ send("[stealth] "+m); }
function findExport(mod, name) {
  try {
    var m = Process.getModuleByName(mod);
    var a = m.findExportByName(name);
    if (a) return a;
  } catch (e) {}
  try { return Module.getGlobalExportByName(name); } catch (e2) { return null; }
}

// Rename frida threads
try {
  var pthread_setname = findExport("libc.so", "pthread_setname_np");
  // also patch existing: best-effort via Java later
} catch (e) {}

// Hide maps
var markers = ["frida","gadget","linjector","gum-js","gmain","pool-frida","hluda"];
var tracked = {};
function hideLine(line) {
  var l = (line||"").toLowerCase();
  for (var i=0;i<markers.length;i++) if (l.indexOf(markers[i])>=0) return true;
  return false;
}
function filterBuf(buf, len) {
  try {
    var t = buf.readUtf8String(len);
    if (!t) return len;
    var out = t.split("\n").filter(function(x){return !hideLine(x);}).join("\n");
    var b = Memory.allocUtf8String(out);
    var n = Math.min(len, out.length);
    Memory.copy(buf, b, n);
    return n;
  } catch (e) { return len; }
}
var openat = findExport("libc.so","openat");
var readFn = findExport("libc.so","read");
if (openat) {
  Interceptor.attach(openat, {
    onEnter: function(args){ this.path = args[1].isNull()?null:args[1].readCString(); },
    onLeave: function(retval){
      var fd=retval.toInt32();
      if (fd>=0 && this.path && (this.path.indexOf("maps")>=0 || this.path.indexOf("status")>=0 || this.path.indexOf("task")>=0))
        tracked[fd]=this.path;
    }
  });
}
if (readFn) {
  Interceptor.attach(readFn, {
    onEnter: function(args){ this.fd=args[0].toInt32(); this.buf=args[1]; },
    onLeave: function(retval){
      var n=retval.toInt32();
      if (n>0 && tracked[this.fd]) retval.replace(ptr(filterBuf(this.buf, n)));
    }
  });
}

// Patch libc exit_group SVC
var libc = Process.getModuleByName("libc.so");
libc.enumerateRanges("r-x").forEach(function(r){
  for (var off=0; off+8<r.size; off+=4) {
    var p=r.base.add(off);
    var w;
    try { w=p.readU32(); } catch(e){ return; }
    if (w!==0xd2800bc8 && w!==0xd2800ba8 && w!==0x52800bc8 && w!==0x52800ba8) continue;
    for (var j=4;j<=24;j+=4) {
      var s=p.add(j);
      try {
        if (s.readU32()===0xd4000001) {
          Memory.protect(s,4,"rwx");
          s.writeU32(0xd65f03c0);
          log("SVC->RET "+s);
        }
      } catch(e2){}
    }
  }
});

// Replace exit APIs
["_exit","exit","abort"].forEach(function(n){
  var a=findExport("libc.so",n);
  if(!a) return;
  try {
    Interceptor.replace(a, new NativeCallback(function(c){ log("block "+n+"("+c+")"); }, "void", ["int"]));
  } catch(e){}
});

log("stealth ready");
"""


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


def on_msg(m, _d):
    print(m.get("payload", m), flush=True)


def wait_alive(sec, label):
    for i in range(sec):
        time.sleep(1)
        p = adb("shell", "pidof", PKG).stdout.strip()
        act = ""
        dump = adb("shell", "dumpsys", "activity", "activities").stdout
        for line in dump.splitlines():
            if "tngdigital" in line.lower() and any(
                x in line for x in ("UserLogin", "SecurityError", "Splash", "topResumedActivity")
            ):
                act = line.strip()[:120]
                break
        print("%s t=%ds pid=%s act=%s" % (label, i + 1, p or "DEAD", act), flush=True)
        if not p:
            return i + 1
    return sec


def test_xposed_only():
    print("=== Xposed-only (monkey) ===", flush=True)
    adb("shell", "am", "force-stop", PKG)
    adb("logcat", "-c")
    time.sleep(0.3)
    adb(
        "shell",
        "monkey",
        "-p",
        PKG,
        "-c",
        "android.intent.category.LAUNCHER",
        "1",
    )
    alive = wait_alive(15, "XPOSED")
    print("xposed_alive_sec", alive, flush=True)
    for line in adb("logcat", "-d").stdout.splitlines():
        if "TngRoot" in line and any(
            x in line for x in ("install", "UserLogin", "short-circuit", "finishing", "blocked intent")
        ):
            print(line[line.find("TngRoot") :][:180], flush=True)
        if "Displayed" in line and "tngdigital" in line:
            print(line.strip()[:200], flush=True)
        if "exited cleanly" in line:
            print(line.strip()[:160], flush=True)


def test_frida_stealth():
    print("=== Frida stealth spawn ===", flush=True)
    adb("shell", "am", "force-stop", PKG)
    time.sleep(0.3)
    d = frida.get_usb_device(10)
    pid = d.spawn([PKG])
    s = d.attach(pid)
    sc = s.create_script(STEALTH)
    sc.on("message", on_msg)
    sc.load()
    d.resume(pid)
    alive = wait_alive(15, "FRIDA")
    print("frida_alive_sec", alive, flush=True)


def main():
    test_xposed_only()
    test_frida_stealth()


if __name__ == "__main__":
    main()