237 lines
9.8 KiB
Python
237 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Summarize a stalker_trace.js log: block histogram, skeleton, call edges.
|
|
|
|
WHY THIS EXISTS
|
|
---------------
|
|
scripts/stalker_trace.js records, for one target module, the basic blocks a
|
|
thread really executed. Raw, that log answers nothing -- the value comes from
|
|
three reductions, and each maps to a deobfuscation decision:
|
|
|
|
* execution histogram (BLK lines) -- in a control-flow-flattened function
|
|
the dispatcher block runs orders of magnitude more often than any real
|
|
block; the top of this histogram IS the dispatcher;
|
|
* first-visit order (BB lines) -- the deduplicated block set in first-visit
|
|
order approximates the real CFG skeleton with the flattening state machine
|
|
stripped out;
|
|
* call edges (CALL lines) -- which functions inside the module talk to each
|
|
other, independent of symbol names.
|
|
|
|
This tool is deliberately stdlib-only and parses text, so it works on logs
|
|
written by run_probe.py (each event line embedded in a `[host ts] [dev ts]
|
|
TRACE ...` wrapper) and on raw one-event-per-line logs alike.
|
|
|
|
Usage
|
|
-----
|
|
python stalker_report.py trace.log
|
|
python stalker_report.py trace.log --top 30 --skeleton 80
|
|
python stalker_report.py trace.log --json report.json
|
|
python stalker_report.py trace.log --quiet # summary only
|
|
|
|
Reading the output
|
|
------------------
|
|
* DONE ... truncated=1 -- the trace hit maxBlocks; ratios are still useful,
|
|
absolute counts are not.
|
|
* zero BLK but nonzero BB -- the thread stopped between translation and
|
|
execution (short follow, or the trigger returned immediately).
|
|
* zero BB and zero BLK with DONE present -- module never ran on the followed
|
|
thread; pick a real trigger instead of `main`.
|
|
* everything zero, no DONE -- the trace never started: check READY/TRIG-FAIL
|
|
lines in the log first.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from collections import Counter, OrderedDict
|
|
|
|
MOD_RE = re.compile(r'\bMOD (\S+) base=(0x[0-9a-fA-F]+) size=(\d+)(?:\s+path=(\S+))?')
|
|
BB_RE = re.compile(r'\bBB (\d+) (\S+)\+0x([0-9a-fA-F]+)')
|
|
BLK_RE = re.compile(r'\bBLK (\d+) (\S+)\+0x([0-9a-fA-F]+)(?:\s+size=(\d+))?')
|
|
CALL_RE = re.compile(r'\bCALL (\S+) (\S+) -> (\S+)')
|
|
DONE_RE = re.compile(r'\bDONE reason=(\S+) blocks=(\d+) blk=(\d+) calls=(\d+) truncated=(\d)')
|
|
FATAL_RE = re.compile(r'\b(FATAL|TRIG-FAIL) (.*)')
|
|
|
|
|
|
def parse_log(path):
|
|
stats = {
|
|
'mods': OrderedDict(), # name -> {base, size, path}
|
|
'bb': [], # (seq, mod, offset) first-visit order
|
|
'blk': [], # (seq, mod, offset, size|None) executions
|
|
'calls': [], # (depth, from, to)
|
|
'done': None,
|
|
'fatals': [],
|
|
}
|
|
with open(path, encoding='utf-8', errors='replace') as fh:
|
|
for raw in fh:
|
|
m = MOD_RE.search(raw)
|
|
if m:
|
|
name, base, size, path = m.groups()
|
|
if name not in stats['mods']:
|
|
stats['mods'][name] = {
|
|
'base': base, 'size': int(size), 'path': path or ''}
|
|
continue
|
|
m = BLK_RE.search(raw)
|
|
if m:
|
|
seq, mod, off, size = m.groups()
|
|
stats['blk'].append((int(seq), mod, int(off, 16),
|
|
int(size) if size else None))
|
|
continue
|
|
m = BB_RE.search(raw)
|
|
if m:
|
|
seq, mod, off = m.groups()
|
|
stats['bb'].append((int(seq), mod, int(off, 16)))
|
|
continue
|
|
m = CALL_RE.search(raw)
|
|
if m:
|
|
stats['calls'].append(m.groups())
|
|
continue
|
|
m = DONE_RE.search(raw)
|
|
if m and stats['done'] is None:
|
|
stats['done'] = m.groups()
|
|
continue
|
|
m = FATAL_RE.search(raw)
|
|
if m:
|
|
stats['fatals'].append(m.group(0).strip())
|
|
return stats
|
|
|
|
|
|
def blockkey(mod, off):
|
|
return '%s+0x%x' % (mod, off)
|
|
|
|
|
|
def collapse_consecutive(items):
|
|
"""[(mod, off), ...] -> [(key, repeats), ...] with runs collapsed."""
|
|
out = []
|
|
for mod, off in items:
|
|
key = blockkey(mod, off)
|
|
if out and out[-1][0] == key:
|
|
out[-1] = (key, out[-1][1] + 1)
|
|
else:
|
|
out.append((key, 1))
|
|
return out
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(
|
|
description='Summarize a stalker_trace.js log into a block histogram, '
|
|
'a first-visit CFG skeleton, and call-edge tables.')
|
|
ap.add_argument('log', help='trace log written by run_probe.py (wrapped '
|
|
'lines are fine) or a raw event-per-line log')
|
|
ap.add_argument('--top', type=int, default=20,
|
|
help='rows in the execution histogram (default 20)')
|
|
ap.add_argument('--skeleton', type=int, default=40,
|
|
help='rows of the first-visit order to print (default 40)')
|
|
ap.add_argument('--edges', type=int, default=15,
|
|
help='rows in the call-edge table (default 15)')
|
|
ap.add_argument('--seq', type=int, default=60,
|
|
help='entries of the collapsed execution sequence to print '
|
|
'(default 60)')
|
|
ap.add_argument('--json', metavar='PATH', default=None,
|
|
help='also write the full reduced data as JSON')
|
|
ap.add_argument('--quiet', action='store_true',
|
|
help='print the summary block only')
|
|
args = ap.parse_args()
|
|
|
|
st = parse_log(args.log)
|
|
|
|
print('== summary ==')
|
|
for name, info in st['mods'].items():
|
|
print(' module %-24s base=%s size=%d %s'
|
|
% (name, info['base'], info['size'],
|
|
('(target)' if st['bb'] and st['bb'][0][1] == name else '')))
|
|
print(' unique blocks first-seen (BB) : %d' % len(st['bb']))
|
|
print(' block executions (BLK) : %d' % len(st['blk']))
|
|
print(' call edges (CALL) : %d' % len(st['calls']))
|
|
if st['done']:
|
|
reason, nb, nk, nc, trunc = st['done']
|
|
print(' DONE reason=%s truncated=%s (device-side counts: bb=%s blk=%s call=%s)'
|
|
% (reason, trunc, nb, nk, nc))
|
|
else:
|
|
print(' DONE line missing -- trace ended without a clean stop '
|
|
'(detach or crash?)')
|
|
for f in st['fatals'][:5]:
|
|
print(' NOTE %s' % f)
|
|
|
|
if not st['bb'] and not st['blk']:
|
|
print('\n no target-module blocks at all. In order of likelihood:\n'
|
|
' 1. the module was never loaded on the followed thread '
|
|
'(check the MOD lines);\n'
|
|
' 2. the trigger never fired (TRIG-FAIL above, or the app '
|
|
'never called it);\n'
|
|
' 3. the follow window closed before any code ran '
|
|
'(raise followMs, or use a call trigger instead of main).')
|
|
return 0 if st['done'] else 1
|
|
|
|
first_seen = {}
|
|
for seq, mod, off in st['bb']:
|
|
first_seen.setdefault(blockkey(mod, off), seq)
|
|
|
|
if not args.quiet:
|
|
hist = Counter(blockkey(mod, off) for _, mod, off, _ in st['blk'])
|
|
total = sum(hist.values())
|
|
if hist:
|
|
print('\n== execution histogram: top %d (dispatcher candidates) ==' % args.top)
|
|
print(' %-6s %-8s %-7s %-28s %s'
|
|
% ('rank', 'count', 'pct', 'block', 'firstSeen'))
|
|
for rank, (key, cnt) in enumerate(hist.most_common(args.top), 1):
|
|
pct = 100.0 * cnt / total if total else 0.0
|
|
print(' %-6d %-8d %5.1f%% %-28s #%s'
|
|
% (rank, cnt, pct, key, first_seen.get(key, '-')))
|
|
print(' (a block at the top by a wide margin, with many distinct '
|
|
'successors below,\n is the classic control-flow-'
|
|
'flattening dispatcher; see\n references/'
|
|
'native-dbi-and-deobfuscation.md)')
|
|
|
|
if st['bb']:
|
|
print('\n== first-visit order (CFG skeleton, first %d of %d) =='
|
|
% (min(args.skeleton, len(st['bb'])), len(st['bb'])))
|
|
for seq, mod, off in st['bb'][:args.skeleton]:
|
|
print(' #%d %s' % (seq, blockkey(mod, off)))
|
|
|
|
if st['calls']:
|
|
edges = Counter('%s -> %s' % (frm, to) for _, frm, to in st['calls'])
|
|
print('\n== call edges (top %d of %d) ==' % (args.edges, len(edges)))
|
|
for edge, cnt in edges.most_common(args.edges):
|
|
print(' %-8d %s' % (cnt, edge))
|
|
|
|
collapsed = collapse_consecutive([(mod, off) for _, mod, off, _ in st['blk']])
|
|
if collapsed:
|
|
print('\n== collapsed execution sequence (first %d runs of %d) =='
|
|
% (min(args.seq, len(collapsed)), len(collapsed)))
|
|
line = ' '
|
|
for key, reps in collapsed[:args.seq]:
|
|
piece = key + ('*%d' % reps if reps > 1 else '')
|
|
if len(line) + len(piece) > 100:
|
|
print(line)
|
|
line = ' '
|
|
line += piece + ' -> '
|
|
if line.strip():
|
|
print(line[:-4])
|
|
|
|
if args.json:
|
|
payload = {
|
|
'modules': st['mods'],
|
|
'summary': {
|
|
'bb': len(st['bb']), 'blk': len(st['blk']),
|
|
'calls': len(st['calls']), 'done': st['done'],
|
|
},
|
|
'first_seen': first_seen,
|
|
'histogram': dict(Counter(
|
|
blockkey(mod, off) for _, mod, off, _ in st['blk'])),
|
|
'bb_sequence': [blockkey(mod, off) for _, mod, off in st['bb']],
|
|
'collapsed': collapsed if st['blk'] else [],
|
|
'call_edges': dict(Counter('%s -> %s' % (f, t)
|
|
for _, f, t in st['calls'])),
|
|
}
|
|
with open(args.json, 'w', encoding='utf-8') as fh:
|
|
json.dump(payload, fh, indent=1)
|
|
print('\njson written to %s' % args.json)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|