| 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.create(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, exe_ctx: lldb.SBExecutionContext, args : lldb.SBStructuredData): |
| super().__init__(exe_ctx, 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, error: lldb.SBError) -> 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 |
| |
