212 lines
7.8 KiB
Python
212 lines
7.8 KiB
Python
import os,json,struct,signal,uuid
|
|
|
|
from typing import Any, Dict
|
|
|
|
import lldb
|
|
from lldb.plugins.scripted_process import ScriptedProcess
|
|
from lldb.plugins.scripted_process import ScriptedThread
|
|
|
|
from lldb.macosx.crashlog import CrashLog,CrashLogParser
|
|
|
|
class CrashLogScriptedProcess(ScriptedProcess):
|
|
def parse_crashlog(self):
|
|
crashlog_parser = CrashLogParser(self.dbg, self.crashlog_path, False)
|
|
crash_log = crashlog_parser.parse()
|
|
|
|
self.pid = crash_log.process_id
|
|
self.addr_mask = crash_log.addr_mask
|
|
self.crashed_thread_idx = crash_log.crashed_thread_idx
|
|
self.loaded_images = []
|
|
self.exception = crash_log.exception
|
|
self.app_specific_thread = None
|
|
if hasattr(crash_log, 'asi'):
|
|
self.metadata['asi'] = crash_log.asi
|
|
if hasattr(crash_log, 'asb'):
|
|
self.extended_thread_info = crash_log.asb
|
|
|
|
def load_images(self, images):
|
|
#TODO: Add to self.loaded_images and load images in lldb
|
|
if images:
|
|
for image in images:
|
|
if image not in self.loaded_images:
|
|
if image.uuid == uuid.UUID(int=0):
|
|
continue
|
|
err = image.add_module(self.target)
|
|
if err:
|
|
# Append to SBCommandReturnObject
|
|
print(err)
|
|
else:
|
|
self.loaded_images.append(image)
|
|
|
|
for thread in crash_log.threads:
|
|
if self.load_all_images:
|
|
load_images(self, crash_log.images)
|
|
elif thread.did_crash():
|
|
for ident in thread.idents:
|
|
load_images(self, crash_log.find_images_with_identifier(ident))
|
|
|
|
if hasattr(thread, 'app_specific_backtrace') and thread.app_specific_backtrace:
|
|
# We don't want to include the Application Specific Backtrace
|
|
# Thread into the Scripted Process' Thread list.
|
|
# Instead, we will try to extract the stackframe pcs from the
|
|
# backtrace and inject that as the extended thread info.
|
|
self.app_specific_thread = thread
|
|
continue
|
|
|
|
self.threads[thread.index] = CrashLogScriptedThread(self, None, thread)
|
|
|
|
|
|
if self.app_specific_thread:
|
|
self.extended_thread_info = \
|
|
CrashLogScriptedThread.resolve_stackframes(self.app_specific_thread,
|
|
self.addr_mask,
|
|
self.target)
|
|
|
|
def __init__(self, target: lldb.SBTarget, args : lldb.SBStructuredData):
|
|
super().__init__(target, args)
|
|
|
|
if not self.target or not self.target.IsValid():
|
|
# Return error
|
|
return
|
|
|
|
self.crashlog_path = None
|
|
|
|
crashlog_path = args.GetValueForKey("file_path")
|
|
if crashlog_path and crashlog_path.IsValid():
|
|
if crashlog_path.GetType() == lldb.eStructuredDataTypeString:
|
|
self.crashlog_path = crashlog_path.GetStringValue(4096)
|
|
|
|
if not self.crashlog_path:
|
|
# Return error
|
|
return
|
|
|
|
load_all_images = args.GetValueForKey("load_all_images")
|
|
if load_all_images and load_all_images.IsValid():
|
|
if load_all_images.GetType() == lldb.eStructuredDataTypeBoolean:
|
|
self.load_all_images = load_all_images.GetBooleanValue()
|
|
|
|
if not self.load_all_images:
|
|
self.load_all_images = False
|
|
|
|
self.pid = super().get_process_id()
|
|
self.crashed_thread_idx = 0
|
|
self.exception = None
|
|
self.extended_thread_info = None
|
|
self.parse_crashlog()
|
|
|
|
def get_memory_region_containing_address(self, addr: int) -> lldb.SBMemoryRegionInfo:
|
|
return None
|
|
|
|
def get_thread_with_id(self, tid: int):
|
|
return {}
|
|
|
|
def get_registers_for_thread(self, tid: int):
|
|
return {}
|
|
|
|
def read_memory_at_address(self, addr: int, size: int) -> lldb.SBData:
|
|
# NOTE: CrashLogs don't contain any memory.
|
|
return lldb.SBData()
|
|
|
|
def get_loaded_images(self):
|
|
# TODO: Iterate over corefile_target modules and build a data structure
|
|
# from it.
|
|
return self.loaded_images
|
|
|
|
def get_process_id(self) -> int:
|
|
return self.pid
|
|
|
|
def should_stop(self) -> bool:
|
|
return True
|
|
|
|
def is_alive(self) -> bool:
|
|
return True
|
|
|
|
def get_scripted_thread_plugin(self):
|
|
return CrashLogScriptedThread.__module__ + "." + CrashLogScriptedThread.__name__
|
|
|
|
def get_process_metadata(self):
|
|
return self.metadata
|
|
|
|
class CrashLogScriptedThread(ScriptedThread):
|
|
def create_register_ctx(self):
|
|
if not self.has_crashed:
|
|
return dict.fromkeys([*map(lambda reg: reg['name'], self.register_info['registers'])] , 0)
|
|
|
|
if not self.backing_thread or not len(self.backing_thread.registers):
|
|
return dict.fromkeys([*map(lambda reg: reg['name'], self.register_info['registers'])] , 0)
|
|
|
|
for reg in self.register_info['registers']:
|
|
reg_name = reg['name']
|
|
if reg_name in self.backing_thread.registers:
|
|
self.register_ctx[reg_name] = self.backing_thread.registers[reg_name]
|
|
else:
|
|
self.register_ctx[reg_name] = 0
|
|
|
|
return self.register_ctx
|
|
|
|
def resolve_stackframes(thread, addr_mask, target):
|
|
frames = []
|
|
for frame in thread.frames:
|
|
frame_pc = frame.pc & addr_mask
|
|
pc = frame_pc if frame.index == 0 or frame_pc == 0 else frame_pc - 1
|
|
sym_addr = lldb.SBAddress()
|
|
sym_addr.SetLoadAddress(pc, target)
|
|
if not sym_addr.IsValid():
|
|
continue
|
|
frames.append({"idx": frame.index, "pc": pc})
|
|
return frames
|
|
|
|
|
|
def create_stackframes(self):
|
|
if not (self.scripted_process.load_all_images or self.has_crashed):
|
|
return None
|
|
|
|
if not self.backing_thread or not len(self.backing_thread.frames):
|
|
return None
|
|
|
|
self.frames = CrashLogScriptedThread.resolve_stackframes(self.backing_thread,
|
|
self.scripted_process.addr_mask,
|
|
self.target)
|
|
|
|
return self.frames
|
|
|
|
def __init__(self, process, args, crashlog_thread):
|
|
super().__init__(process, args)
|
|
|
|
self.backing_thread = crashlog_thread
|
|
self.idx = self.backing_thread.index
|
|
self.tid = self.backing_thread.id
|
|
if self.backing_thread.app_specific_backtrace:
|
|
self.name = "Application Specific Backtrace - " + str(self.idx)
|
|
else:
|
|
self.name = self.backing_thread.name
|
|
self.queue = self.backing_thread.queue
|
|
self.has_crashed = (self.scripted_process.crashed_thread_idx == self.idx)
|
|
self.create_stackframes()
|
|
|
|
def get_state(self):
|
|
if not self.has_crashed:
|
|
return lldb.eStateStopped
|
|
return lldb.eStateCrashed
|
|
|
|
def get_stop_reason(self) -> Dict[str, Any]:
|
|
if not self.has_crashed:
|
|
return { "type": lldb.eStopReasonNone }
|
|
# TODO: Investigate what stop reason should be reported when crashed
|
|
stop_reason = { "type": lldb.eStopReasonException, "data": { }}
|
|
if self.scripted_process.exception:
|
|
stop_reason['data']['mach_exception'] = self.scripted_process.exception
|
|
return stop_reason
|
|
|
|
def get_register_context(self) -> str:
|
|
if not self.register_ctx:
|
|
self.register_ctx = self.create_register_ctx()
|
|
|
|
return struct.pack("{}Q".format(len(self.register_ctx)), *self.register_ctx.values())
|
|
|
|
def get_extended_info(self):
|
|
if (self.has_crashed):
|
|
self.extended_info = self.scripted_process.extended_thread_info
|
|
return self.extended_info
|
|
|