live-hub-py/scripts/capture_propedge.py

#!/usr/bin/env python3
"""Runtime edge capture on live libhydeviceid generation:
1) hook __system_property_get -> what system inputs drive triple/device derivation
2) hook SSL_write -> the dfpReport wire (to correlate)
Only needs the ~3-6s spawn window (dfpReport fires ~1s after start).
"""
from pathlib import Path
import frida, time, subprocess, json, sys

REMOTE="127.0.0.1:31878"; PACKAGE="com.duowan.kiwi"
RE=Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
OUT=Path("/Users/yml/codes/douyu_login_py/evidence/propedge_" + time.strftime("%H%M%S") + ".json")

JS = r"""
'use strict';
send({type:'armed'});
function rcs(p,n){try{return p.readCString(n)||'';}catch(e){return '';}}
function hexb(p,n){try{return Array.from(new Uint8Array(p.readByteArray(n))).map(b=>('0'+b.toString(16)).slice(-2)).join('');}catch(e){return '';}}
var START = Date.now(); var t0 = Math.floor(Date.now()/1000);

// 1) system property reads
try{
  var pg = Module.findExportByName('libc.so','__system_property_get');
  if(pg){
    Interceptor.attach(pg,{onEnter:function(a){this.name=rcs(a[0],256);this.valbuf=a[1];},
      onLeave:function(ret){
        try{
          var val=rcs(this.valbuf,512);
          // 只关心设备/身份派生相关属性, 避免把框架进程的读取混进来
          var isDevice = this.name.indexOf('ro.product')===0 || this.name.indexOf('ro.serialno')===0
            || this.name.indexOf('ro.boot')===0 || this.name.indexOf('ro.hardware')===0
            || this.name.indexOf('ro.build')===0 || this.name=== 'ro.secure'
            || this.name==='ro.debuggable' || this.name.indexOf('ro.kernel')===0
            || this.name.indexOf('gsm.')===0 || this.name.indexOf('persist.sys')===0
            || this.name.indexOf('qemu')===0;
          if(isDevice){ send({type:'prop',dt:Date.now()-START,name:this.name,val:val.slice(0,200)}); }
        }catch(e){}
      }});
  }
  send({type:'prop_hooked'});
}catch(e){send({type:'err',e:String(e)});}

// 2) SSL_write dfpReport wire
try{
  var r=new ApiResolver('module');
  r.enumerateMatchesSync('exports:*!SSL_write').forEach(function(m){
    Interceptor.attach(m.address,{onEnter:function(a){
      var len=a[2].toInt32();
      if(len<50||len>50000)return;
      var h=rcs(a[1],Math.min(len,1500));
      if(h.indexOf('dfpReport')>=0){
        send({type:'dfp',dt:Date.now()-START,len:len,hex:hexb(a[1],len)});
      }
    }});
  });
  send({type:'dfp_hooked'});
}catch(e){send({type:'dfperr',e:String(e)});}

// 3) also capture plaintext JSON candidate: hook memcpy/strlen leaving std::string? skip, noisy.
"""

def one_run(d, a, result, dump_props):
    subprocess.run(["adb","-s","127.0.0.1:5555","shell","am","force-stop",PACKAGE],capture_output=True)
    time.sleep(1.2)
    pid=d.spawn([PACKAGE])
    print(f"[att{a}] spawn pid={pid}", flush=True)
    s=d.attach(pid)
    try:
        b=s.create_script((RE/"evidence/scripts/bypass_msaoaid_maps_skip_cleanup.js").read_text()); b.load()
    except Exception as e:
        print(f"[att{a}] bypass err {e}", flush=True)
    d.resume(pid)
    print(f"[att{a}] resumed", flush=True)
    got=False; dfp_hook=False
    def on(m,dd):
        nonlocal got, dfp_hook
        if m.get('type')=='error': print(f"[att{a}] JSErr {str(m)[:120]}", flush=True); return
        p=m.get('payload') or {}
        t=p.get('type')
        if t=='prop_hooked': print(f"[att{a}] prop hooked", flush=True)
        elif t=='dfp_hooked': dfp_hook=True; print(f"[att{a}] SSL_write hooked", flush=True)
        elif t=='prop':
            result.setdefault('props',[]).append({'dt':p.get('dt'),'name':p['name'],'val':p['val']})
        elif t=='dfp':
            got=True; print(f"[att{a}] dfp len={p['len']}", flush=True)
            result['dfp_wire']=p['hex']; result['dfp_dt']=p.get('dt'); result['pid']=pid
    sc=s.create_script(JS); sc.on('message',on); sc.load()
    t0=time.time()
    while time.time()-t0<20:
        time.sleep(2)
        if got: time.sleep(3); break
    print(f"[att{a}] end got_dfp={got} (props={len(result.get('props',[]))})", flush=True)
    # unique props by name
    seen={}
    for pr in result.get('props',[]):
        seen.setdefault(pr['name'],[]).append(pr['val'])
    if dump_props and seen:
        print("--- unique props read ---")
        for k,v in sorted(seen.items()):
            print(f"  {k} = {v[0][:120]}  (x{len(v)})", flush=True)
    try: d.kill(pid)
    except: pass
    return got

def main():
    d=frida.get_device_manager().add_remote_device(REMOTE)
    result={}
    ok=False
    for a in range(1,6):
        try:
            if one_run(d,a,result,dump_props=(a>=1)):
                print(f"[*] SUCCESS att{a}", flush=True)
                ok=True; break
        except Exception as e:
            print(f"[att{a}] ERR {repr(e)}", flush=True); time.sleep(2)
    if ok or result.get('props'):
        json.dump(result, open(OUT,'w'), indent=2)
        print(f"[*] saved {OUT}", flush=True)
    else:
        print("[*] no capture", flush=True)

if __name__=="__main__": main()