Convert KSCrash log to Apple format

ks2apple.py

#!/usr/bin/env python
#-*- coding: utf-8 -*-

import json
import os
from datetime import datetime
from dateutil import parser, tz

def dump_json(obj):
    return json.dumps(obj, indent=4,
            sort_keys=True, separators=(',',': '))

def get_system_info(report):
    '''Get system information from report'''
    return report.get('system', None)

def get_report_info(report):
    '''Get report information from report'''
    return report.get('report', None)

def get_crash_info(report):
    '''Get crash information from report'''
    return report.get('crash', None)

def get_cpu_type(arch):
    return {'arm64': 'ARM-64',
            'arm': 'ARM',
            'x86': 'X86',
            'x86_64': 'X86_64'}.get(arch, 'Unknow')

def get_cpu_arch(report):
    system = get_system_info(report)
    arch = system.get('cpu_arch', '')
    return get_cpu_type(arch)

CPU_TYPE_ARM = 12
CPU_ARCH_ABI64 = 0x01000000
CPU_TYPE_ARM64 = CPU_TYPE_ARM | CPU_ARCH_ABI64
CPU_TYPE_X86 = 7
CPU_TYPE_X86_64 = CPU_TYPE_X86 | CPU_ARCH_ABI64

CPU_SUBTYPE_ARM_V6 = 6
CPU_SUBTYPE_ARM_V7 = 9
CPU_SUBTYPE_ARM_V7F = 10
CPU_SUBTYPE_ARM_V7S = 11
CPU_SUBTYPE_ARM_V7K = 12
CPU_SUBTYPE_ARM_V6M = 14
CPU_SUBTYPE_ARM_V7M = 15
CPU_SUBTYPE_ARM_V7EM = 16
CPU_SUBTYPE_ARM_V8 = 13

CPU_ARM_TYPES = {
    CPU_SUBTYPE_ARM_V6: 'armv6',
    CPU_SUBTYPE_ARM_V7: 'armv7',
    CPU_SUBTYPE_ARM_V7F: 'armv7f',
    CPU_SUBTYPE_ARM_V7S: 'armv7s',
    CPU_SUBTYPE_ARM_V7K: 'armv7k',
    CPU_SUBTYPE_ARM_V6M: 'armv6m',
    CPU_SUBTYPE_ARM_V7M: 'armv7m',
    CPU_SUBTYPE_ARM_V7EM: 'armv7em',
    CPU_SUBTYPE_ARM_V8: 'armv8',
}

def get_detail_cpu_arch(major, minor):
    if major == CPU_TYPE_ARM:
        cpu_type = CPU_ARM_TYPES.get(minor, 'arm')
    elif major == CPU_TYPE_ARM64:
        return 'arm64'
    elif major == CPU_TYPE_X86:
        return 'i386'
    elif major == CPU_TYPE_X86_64:
        return 'x86_64'

    return 'unknown({},{})'.format(major, minor)

def get_last_exception(report):
    report_info = get_report_info(report)
    process = report_info.get('process', None)
    if not process:
        return None
    return process.get('last_dealloced_nsexception', None)

def get_time(report):
    '''Get crash time from report
    @return time like "2015-12-28 17:48:03 +0800"
    '''

    info = get_report_info(report)
    timestamp = info.get('timestamp', None)
    if not timestamp:
        return ''

    date = parser.parse(timestamp)
    return date.astimezone(tz.tzlocal())\
            .strftime('%Y-%m-%d %H:%M:%S.000 %z')

def get_crash_thread(report):
    crash = get_crash_info(report)
    threads = crash['threads']
    for thread in threads:
        crashed = thread.get('crashed', False)
        if crashed:
            return thread

    return crash.get('crashed_thread', None)

def parse_system_info(report):
    '''Parse system information from report.
    @return header lines
    '''

    sys = get_system_info(report)
    info = get_report_info(report)

    _s = lambda x: sys.get(x, '')
    _i = lambda x: info.get(x, '')

    headers = []
    headers.append('Incident Identifier: {}'.format(_i('id')))
    headers.append('CrashReporter Key:   {}'.format(_s('device_app_hash')))
    headers.append('Hardware Model:      {}'.format(_s('machine')))
    headers.append('Process:         {} [{}]'.format(_s('process_name'), _s('process_id')))
    headers.append('Path:            {}'.format(_s('CFBundleExecutablePath')))
    headers.append('Identifier:      {}'.format(_s('CFBundleIdentifier')))
    headers.append('Version:         {} ({})'.format(_s('CFBundleShortVersionString'), _s('CFBundleVersion')))
    headers.append('Code Type:       {}'.format(get_cpu_arch(report)))
    headers.append('Parent Process:  {} [{}]'.format(_s('parent_process_name'), _s('parent_process_id')))
    headers.append('')
    headers.append('Date/Time:       {}'.format(get_time(report)))
    headers.append('OS Version:      {} {} ({})'.format(_s('system_name'), _s('system_version'), _s('os_version')))
    headers.append('Report Version:  104')

    return headers

def zombie_exception(report):
    crash = get_crash_info(report)
    error = crash['error']
    mach = error['mach']
    exc_name = mach.get('exception_name', '0')
    code_name = mach.get('code_name', '0x00000000')

    if exc_name != 'EXC_BAD_ACCESS' or code_name != 'KERN_INVALID_ADDRESS':
        return False

    last_exception = get_last_exception(report)
    if not last_exception:
        return False

    last_addr = last_exception['address']
    thread = get_crash_thread(report)
    registers = thread['registers']['basic']
    for reg, addr in registers:
        if addr == last_addr:
            return True

    return False


def parse_error_info(report):

    crash = get_crash_info(report)
    error = crash['error']
    mach = error['mach']
    signal = error['signal']

    exc_name = mach.get('exception_name', '0')
    sig_name = signal.get('name', None)
    if not sig_name:
        sig_name = signal.get('signal', '')

    code_name = mach.get('code_name', '0x00000000')
    addr = error.get('address', '0')

    crash_thread = 0
    thread = get_crash_thread(report)
    if thread:
        crash_thread = thread['index']

    result = ['']
    result.append('Exception Type:  {} ({})'.format(exc_name, sig_name))
    result.append('Exception Codes: {} at 0x{:016x}'.format(code_name, int(addr)))
    result.append('Crashed Thread:  {}'.format(crash_thread))

    return result

def parse_crash_reason(report):
    result = ['']
    reason_fmt = ("Application Specific Information:\n"
                  "*** Terminating app due to uncaught exception '{}', "
                  "reason: '{}'")
    fmt = lambda x, y: reason_fmt.format(x, y)

    crash = get_crash_info(report)
    error = crash['error']
    crash_type = error['type']

    user_exception = error.get('user_reported', None)
    ns_exception = error.get('nsexception', None)
    if ns_exception:
        result.append(fmt(ns_exception['name'], error['reason']))
    elif zombie_exception(report):
        last_exception = get_last_exception(report)
        if (last_exception):
            result.append(fmt(last_exception['name'], last_exception['reason']))
    elif user_exception:
        result.append(fmt(user_exception['name'], error['reason']))
        line = user_exception.get('line_of_code', None)
        backtrace = user_exception.get('backtrace', [])
        if line or backtrace:
            result.append('Custom Backtrace:')
        if line:
            result.append('Line: {}'.format(line))
        for entry in backtrace:
            result.append(entry)

    elif crash_type == 'cpp_exception':
        cpp_exception = error['cppexception']
        result.append(fmt(cpp_exception['name'], error['reason']))
    elif crash_type == 'deadlock':
        result.append('Application main thread deadlocked')
    return result

def parse_backtrace(backtrace, app):
    num = 0
    result = []
    for trace in backtrace.get('contents', []):
        pc = trace['instruction_addr']
        obj_addr = trace['object_addr']
        obj_name = trace['object_name']
        symbol_name = trace.get('symbol_name', None)
        symbol_addr = trace.get('symbol_addr', 0)

        preamble = '{:<4}{:31} 0x{:016x}'.format(num, obj_name, pc)
        unsymbolicated = '0x{:04x} + {}'.format(obj_addr, pc-obj_addr)
        symbolicated = '(null)'
        is_unsymbolicated = False
        if symbol_name:
            symbolicated = '{} + {}'.format(symbol_name, pc-symbol_addr)
        else:
            is_unsymbolicated = True
        if symbol_name == '<redacted>':
            is_unsymbolicated = True

        if is_unsymbolicated:
            result.append('{} {}'.format(preamble, unsymbolicated))
        else:
            result.append('{} {} ({})'.format(preamble, unsymbolicated, symbolicated))

        num += 1

    return result

def parse_thread_info(thread):
    result = []
    crashed = thread.get('crashed', False)
    index = thread['index']
    name = thread.get('name', None)
    queue = thread.get('dispatch_queue', None)

    if name:
        result.append('Thread {} name:  {}'.format(index, name))
    elif queue:
        result.append('Thread {} name:  Dispatch queue: {}'\
                .format(index, queue))
    if crashed:
        result.append('Thread {} Crashed:'.format(index))
    else:
        result.append('Thread {}:'.format(index))

    info = get_report_info(report)
    app = info['process_name']
    backtrace = parse_backtrace(thread['backtrace'], app)
    result += backtrace

    return result

def parse_thread_list(report):
    crash = get_crash_info(report)
    threads = crash['threads']

    result = []
    for thread in threads:
        result.append('')
        result += parse_thread_info(thread)

    return result

def get_register_order(cpu):
    cpu = cpu.lower()
    arm = ['r0', 'r1', 'r2', 'r3', 'r4', 'r5', 'r6', 'r7',
           'r8', 'r9', 'r10', 'r11', 'ip','sp', 'lr', 'pc',
           'cpsr']
    x86 = ['eax', 'ebx', 'ecx', 'edx', 'edi', 'esi', 'ebp',
           'esp', 'ss', 'eflags', 'eip', 'cs', 'ds', 'es',
           'fs', 'gs']
    x86_64 = ['rax', 'rbx', 'rcx', 'rdx', 'rdi', 'rsi', 'rbp',
              'rsp', 'r8', 'r9', 'r10', 'r11', 'r12', 'r13',
              'r14', 'r15', 'rip', 'rflags', 'cs', 'fs', 'gs']
    if cpu.startswith('arm'):
        return arm
    elif cpu in ('x86', 'i386', 'i486', 'i686'):
        return x86
    elif cpu == 'x86_64':
        return x86_64

def parse_cpu_state(report):
    result = ['']
    crashed = get_crash_thread(report)
    if not crashed:
        return result
    index = crashed['index']

    system = get_system_info(report)
    cpu = get_detail_cpu_arch(system['binary_cpu_type'],
            system['binary_cpu_subtype'])
    cpu = get_cpu_type(cpu)
    result.append('Thread {} crashed with {} Thread State:'.format(index, cpu))

    registers = crashed.get('registers', {}).get('basic', {})
    order = get_register_order(cpu)

    line = ''
    for i, reg in enumerate(order):
        if (i != 0) and (i % 4 == 0):
            result.append(line[:-1])
            line = ''
        line += '{:>6}: 0x{:016x} '.format(reg, registers.get(reg, 0))

    if line:
        result.append(line[:-1])

    return result

def parse_binary_images(report):
    result = ['\nBinary Images:']
    system = get_system_info(report)
    exe_path = system['CFBundleExecutablePath']

    images = report.get('binary_images', [])
    #images = sorted(images, key=lambda k: k['image_addr'])

    for image in images:
        cpu = image['cpu_type']
        cpu_sub = image['cpu_subtype']
        addr = image['image_addr']
        size = image['image_size']
        path = image['name']
        name = os.path.basename(path)
        uuid = image['uuid'].lower().replace('-', '')
        is_base = '+' if path==exe_path else ' '
        result.append('{:>#18x} - {:>#18x} {}{} {}  <{}> {}'\
                .format(addr, addr+size-1, is_base, name,
                        get_detail_cpu_arch(cpu, cpu_sub),
                        uuid, path))

    return result

def parse_extra_info(report):
    result = ['', 'Extra Information:']
    system = get_system_info(report)
    crash = get_crash_info(report)
    error = crash['error']
    ns_exception = error.get('nsexception', None)
    if ns_exception:
        ref_obj = ns_exception.get('referenced_object', None)
        if ref_obj:
            result.append('Object referenced by NSException:')
            result.append(dump_json(ref_obj))

    crashed = get_crash_thread(report)
    if crashed:
        stack = crashed.get('stack', None)
        if stack:
            result.append('Stack Dump (0x{:08x}-0x{:08x}):'\
                    .format(stack['dump_start'], stack['dump_end']))
            result.append('')
            result.append(stack.get('contents', ''))
        notable_addr = crashed.get('notable_addresses', None)
        if notable_addr:
            result.append('Notable Addresses:\n{}'\
                    .format(dump_json(notable_addr)))
    last_exception = get_last_exception(report)
    if last_exception:
        addr = last_exception['address']
        name = last_exception['name']
        reason = last_exception['reason']

        result.append('\nLast deallocated NSException (0x{:016x}): {}: {}'\
                .format(addr, name, reason))

        ref_obj = last_exception.get('referenced_object', None)
        if ref_obj:
            result.append('Referenced object:\n{}'\
                    .format(dump_json(ref_obj)))
        info = get_report_info(report)
        app = info['process_name']
        backtrace = parse_backtrace(crashed['backtrace'], app)
        result += backtrace

    app_stats = system.get('application_stats', None)
    if app_stats:
        result.append('\nApplication Stats:\n{}'\
            .format(dump_json(app_stats)))

    crash = get_crash_info(report)
    diagnosis = crash.get('diagnosis', None)
    if diagnosis:
        result.append('\nCrashDoctor Diagnosis: {}'.format(diagnosis))

    return result


if __name__ == '__main__':
    report = json.load(open('demo.json'))
    headers = parse_system_info(report)
    for line in headers:
        print(line)

    errors = parse_error_info(report)
    for line in errors:
        print(line)

    reason = parse_crash_reason(report)
    for line in reason:
        print(line)

    threads = parse_thread_list(report)
    for line in threads:
        print(line)

    cpu_state = parse_cpu_state(report)
    for line in cpu_state:
        print(line)

    images = parse_binary_images(report)
    for line in images:
        print(line)

    extra = parse_extra_info(report)
    for line in extra:
        print(line)