dragonx/qa/fd-guard/shim/symbolize.py

#!/usr/bin/env python3
"""
symbolize.py -- turn fdguard log frames into function + file:line.

Every stack frame in an fdguard report looks like
    #3 0x5b96b433c991 /path/to/module+0x1991 c_dbforeign+0x2d [bdb]
where "+0x1991" is the ELF virtual address inside that module (runtime address
minus the module's load bias, so PIE executables and shared libraries need no
further adjustment; for a non-PIE executable it equals the absolute address).
Frames are return addresses, so we ask addr2line about vaddr-1 (the call
instruction), with -f -C -i (function, demangle, inline chain).

backtrace_symbols_fd lines "module(+0x1991)[0x...]" are symbolized too.

usage:
  symbolize.py [options] LOG [LOG...]          annotated log on stdout
  symbolize.py --summary LOG                    one line per distinct report
options:
  --map OLD=NEW      read module OLD from path NEW instead (repeatable),
                     e.g. --map /path/to/deployed/dragonxd=./dragonxd.unstripped
  --addr2line PATH   addr2line binary (default: addr2line)
  --only-reports     drop T/S lines (trace + stats) from the output
"""
import os
import re
import subprocess
import sys
from collections import defaultdict, OrderedDict

FRAME_RE = re.compile(r'^(\s*#\d+ 0x[0-9a-f]+ )(\S+)\+0x([0-9a-f]+)(.*)$')
BSFD_RE = re.compile(r'^(\S+)\(\+0x([0-9a-f]+)\)\[0x[0-9a-f]+\]\s*$')
HDR_RE = re.compile(r'^=== FDGUARD #(\d+) (\S+)')


def parse_args(argv):
    opts = {'map': {}, 'addr2line': 'addr2line', 'summary': False, 'only': False, 'logs': []}
    it = iter(argv)
    for a in it:
        if a == '--map':
            old, new = next(it).split('=', 1)
            opts['map'][old] = new
        elif a.startswith('--map='):
            old, new = a[6:].split('=', 1)
            opts['map'][old] = new
        elif a == '--addr2line':
            opts['addr2line'] = next(it)
        elif a == '--summary':
            opts['summary'] = True
        elif a == '--only-reports':
            opts['only'] = True
        elif a in ('-h', '--help'):
            print(__doc__)
            sys.exit(0)
        else:
            opts['logs'].append(a)
    if not opts['logs']:
        opts['logs'] = ['-']
    return opts


def read_lines(paths):
    out = []
    for p in paths:
        f = sys.stdin if p == '-' else open(p, errors='replace')
        out.extend(f.read().splitlines())
        if f is not sys.stdin:
            f.close()
    return out


def resolve_module(mod, opts, known):
    if mod in opts['map']:
        return opts['map'][mod]
    if os.path.exists(mod):
        return mod
    base = os.path.basename(mod)
    for k in known:
        if os.path.basename(k) == base and os.path.exists(opts['map'].get(k, k)):
            return opts['map'].get(k, k)
    return None


def run_addr2line(binary, addrs, a2l):
    """addrs: list of int vaddrs (already -1'd). returns {addr: [(func, loc), ...]}"""
    res = {}
    if not addrs:
        return res
    uniq = sorted(set(addrs))
    try:
        p = subprocess.run([a2l, '-a', '-f', '-C', '-i', '-e', binary] + ['0x%x' % a for a in uniq],
                           stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, check=False)
    except FileNotFoundError:
        return res
    cur = None
    lines = p.stdout.splitlines()
    i = 0
    while i < len(lines):
        ln = lines[i]
        if re.match(r'^0x[0-9a-f]+$', ln):
            cur = int(ln, 16)
            res[cur] = []
            i += 1
            continue
        func = ln
        loc = lines[i + 1] if i + 1 < len(lines) else '??:0'
        if cur is not None:
            res[cur].append((func, loc))
        i += 2
    return res


def main():
    opts = parse_args(sys.argv[1:])
    lines = read_lines(opts['logs'])

    if opts['summary']:
        summary(lines)
        return

    # pass 1: collect modules + addresses
    want = defaultdict(set)
    known = OrderedDict()
    for ln in lines:
        m = FRAME_RE.match(ln)
        if m:
            known[m.group(2)] = 1
            want[m.group(2)].add(max(int(m.group(3), 16) - 1, 0))
    for ln in lines:
        m = BSFD_RE.match(ln)
        if m:
            want[m.group(1)].add(max(int(m.group(2), 16) - 1, 0))

    table = {}
    for mod, addrs in want.items():
        path = resolve_module(mod, opts, known)
        if not path:
            continue
        table[mod] = run_addr2line(path, list(addrs), opts['addr2line'])

    def fmt(mod, vaddr):
        r = table.get(mod, {}).get(vaddr)
        if not r:
            return []
        out = []
        for j, (func, loc) in enumerate(r):
            if func == '??' and loc.startswith('??'):
                continue
            # addr2line -i lists the innermost (inlined) frame first
            tag = 'inlined:' if j < len(r) - 1 else 'in'
            out.append('%s %s at %s' % (tag, func, loc))
        return out

    w = sys.stdout.write
    for ln in lines:
        if opts['only'] and (ln.startswith('T ') or ln.startswith('S ')):
            continue
        w(ln + '\n')
        m = FRAME_RE.match(ln)
        if m:
            v = max(int(m.group(3), 16) - 1, 0)
            for s in fmt(m.group(2), v):
                w('         -> ' + s + '\n')
            continue
        m = BSFD_RE.match(ln)
        if m:
            v = max(int(m.group(2), 16) - 1, 0)
            for s in fmt(m.group(1), v):
                w('         -> ' + s + '\n')


def summary(lines):
    """one line per (type, reason, class, first exe closer frame)"""
    groups = OrderedDict()
    cur = None
    in_closer = False
    for ln in lines:
        h = HDR_RE.match(ln)
        if h:
            cur = {'type': h.group(2), 'reason': '', 'cls': '', 'top': '', 'call': ''}
            in_closer = False
            continue
        if cur is None:
            continue
        s = ln.strip()
        if s.startswith('call:'):
            cur['call'] = s.split('(')[0][6:]
        elif s.startswith('target=') and ' class=' in s and not cur['cls']:
            cur['cls'] = s.rsplit('class=', 1)[-1]
        elif s.startswith('verdict='):
            cur['reason'] = s.split('reason=', 1)[-1].split(' ')[0]
        elif s == 'closer stack:':
            in_closer = True
        elif in_closer and FRAME_RE.match(ln):
            parts = s.split(' ')
            if len(parts) >= 4 and not cur['top'] and '/libc.so' not in parts[2] and '/libstdc++' not in parts[2]:
                cur['top'] = parts[3]
        elif in_closer and not s.startswith('#'):
            in_closer = False
        if s.startswith('=== END'):
            k = (cur['type'], cur['call'], cur['reason'], cur['cls'], cur['top'])
            groups[k] = groups.get(k, 0) + 1
            cur = None
    print('%6s  %-22s %-12s %-6s %-40s %s' % ('count', 'type', 'call', 'class', 'reason', 'first closer frame'))
    for (t, c, r, cl, top), n in groups.items():
        print('%6d  %-22s %-12s %-6s %-40s %s' % (n, t[:22], c, cl, r[:40], top))
    stats = [ln for ln in lines if ln.startswith('S ')]
    if stats:
        print('\nlast stats line:\n  ' + stats[-1])


if __name__ == '__main__':
    main()