XFEEDToolBox

"""Read crash metadata locally; never upload dump contents or scan private strings."""
import argparse
import hashlib
import json
import struct
from pathlib import Path


def inspect(path):
    data = path.read_bytes()
    def unpack(fmt, at):
        return struct.unpack_from('<' + fmt, data, at)
    if data[:4] != b'MDMP':
        raise ValueError('Not a Windows minidump')
    count, directory = unpack('II', 8)
    streams = {unpack('III', directory + i * 12)[0]: unpack('III', directory + i * 12)[1:]
               for i in range(count)}
    def text(at):
        size, = unpack('I', at)
        return data[at + 4:at + 4 + size].decode('utf-16-le', errors='replace')
    modules = []
    if 4 in streams:
        _, at = streams[4]
        n, = unpack('I', at)
        for i in range(n):
            p = at + 4 + i * 108
            base, size, checksum, stamp, name = unpack('QIIII', p)
            modules.append(dict(base=base, size=size, checksum=checksum, timestamp=stamp, name=text(name)))
    def location(address):
        for m in modules:
            if m['base'] <= address < m['base'] + m['size']:
                return {'module': m['name'], 'rva': hex(address - m['base'])}
        return None
    memory = []
    if 5 in streams:
        _, at = streams[5]
        n, = unpack('I', at)
        memory += [unpack('QII', at + 4 + i * 16) for i in range(n)]
    if 9 in streams:
        _, at = streams[9]
        n, file_rva = unpack('QQ', at)
        for i in range(n):
            address, size = unpack('QQ', at + 16 + i * 16)
            memory.append((address, size, file_rva))
            file_rva += size
    def read(address, size):
        for base, length, file_rva in memory:
            if base <= address and address + size <= base + length:
                return data[file_rva + address - base:file_rva + address - base + size]
        return b''
    result = {'file': str(path.resolve()), 'sha256': hashlib.sha256(data).hexdigest(),
              'streams': {str(k): v[0] for k, v in streams.items()},
              'modules': [{**m, 'base': hex(m['base']), 'size': hex(m['size']), 'timestamp': hex(m['timestamp'])} for m in modules]}
    if 6 not in streams:
        return result
    _, ex = streams[6]
    thread, = unpack('I', ex)
    code, flags, record, address, params, _ = unpack('IIQQII', ex + 8)
    infos = unpack('15Q', ex + 40)[:min(params, 15)]
    context_size, context_rva = unpack('II', ex + 160)
    regs = {name: unpack('Q', context_rva + 0x78 + i * 8)[0] for i, name in enumerate(
        ['rax', 'rcx', 'rdx', 'rbx', 'rsp', 'rbp', 'rsi', 'rdi', 'r8', 'r9', 'r10', 'r11', 'r12', 'r13', 'r14', 'r15', 'rip'])}
    exception = {'thread': thread, 'code': hex(code), 'flags': flags, 'address': hex(address),
                 'location': location(address), 'parameters': [hex(x) for x in infos],
                 'registers': {k: hex(v) for k, v in regs.items()}, 'contextSize': context_size}
    code_bytes = read(regs['rip'], 48)
    exception['instructionBytes'] = code_bytes.hex()
    try:
        import capstone
        dis = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64)
        exception['instructions'] = [{'address': hex(i.address), 'instruction': i.mnemonic + ' ' + i.op_str}
                                     for i in dis.disasm(code_bytes, regs['rip'])]
    except ImportError:
        pass
    # These are candidates only, not unwound frames. A debugger must validate the calling stack.
    candidates = []
    for offset in range(0, 2048, 8):
        value = read(regs['rsp'] + offset, 8)
        if value:
            pointer, = struct.unpack('<Q', value)
            loc = location(pointer)
            if loc:
                candidates.append({'stackOffset': hex(offset), 'address': hex(pointer), **loc})
    exception['stackPointerCandidates'] = candidates
    result['exception'] = exception
    if 7 in streams:
        result['architecture'] = unpack('H', streams[7][1])[0]
    if 15 in streams:
        misc = streams[15][1]
        size, flags, pid = unpack('III', misc)
        if size >= 12 and flags & 1:
            result['processId'] = pid
    return result


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('dumps', type=Path, nargs='+')
    parser.add_argument('--output', type=Path, required=True)
    args = parser.parse_args()
    results = [inspect(p) for p in args.dumps]
    args.output.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding='utf-8')
    for r in results:
        ex = r.get('exception', {})
        print(json.dumps({'dump': Path(r['file']).name, 'processId': r.get('processId'),
                          'exception': {k: v for k, v in ex.items() if k not in ('stackPointerCandidates',)},
                          'plugins': [m for m in r['modules'] if any(s in m['name'].lower() for s in ('dinput8', 'dbghelp', 'xfeed'))]}, ensure_ascii=False))