| import os, json, struct, signal, uuid, tempfile | |
| 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): | |
| if self.crashlog.process_id: | |
| if type(self.crashlog.process_id) is int: | |
| self.pid = self.crashlog.process_id | |
| elif type(self.crashlog.process_id) is str: | |
| self.pid = int(self.crashlog.process_id, 0) | |
| else: | |
| self.pid = super().get_process_id() | |
| self.addr_mask = self.crashlog.addr_mask | |
| self.crashed_thread_idx = self.crashlog.crashed_thread_idx | |
| self.loaded_images = [] | |
| self.exception = self.crashlog.exception | |
| self.app_specific_thread = None | |
| if hasattr(self.crashlog, "asi"): | |
| self.metadata["asi"] = self.crashlog.asi | |
| if hasattr(self.crashlog, "asb"): | |
| self.extended_thread_info = self.crashlog.asb | |
| for thread in self.crashlog.threads: | |
| 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 | |
| ) | |
| class CrashLogOptions: | |
| load_all_images = False | |
| crashed_only = True | |
| no_parallel_image_loading = False | |
| 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 | |
| self.options = self.CrashLogOptions() | |
| 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.options.load_all_images = load_all_images.GetBooleanValue() | |
| crashed_only = args.GetValueForKey("crashed_only") | |
| if crashed_only and crashed_only.IsValid(): | |
| if crashed_only.GetType() == lldb.eStructuredDataTypeBoolean: | |
| self.options.crashed_only = crashed_only.GetBooleanValue() | |
| no_parallel_image_loading = args.GetValueForKey("no_parallel_image_loading") | |
| if no_parallel_image_loading and no_parallel_image_loading.IsValid(): | |
| if no_parallel_image_loading.GetType() == lldb.eStructuredDataTypeBoolean: | |
| self.options.no_parallel_image_loading = ( | |
| no_parallel_image_loading.GetBooleanValue() | |
| ) | |
| self.crashlog = None | |
| crashlog = args.GetValueForKey("crashlog") | |
| if crashlog and crashlog.IsValid(): | |
| if crashlog.GetType() == lldb.eStructuredDataTypeGeneric: | |
| self.crashlog = crashlog.GetGenericValue() | |
| if not self.crashlog: | |
| # Return error | |
| return | |
| self.pid = super().get_process_id() | |
| self.crashed_thread_idx = 0 | |
| self.exception = None | |
| self.extended_thread_info = None | |
| self.parse_crashlog() | |
| 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): | |
| if len(self.loaded_images) == 0: | |
| self.crashlog.load_images(self.options, self.loaded_images) | |
| return self.loaded_images | |
| 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.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"] | |
| reg_alt_name = None | |
| if "alt-name" in reg: | |
| reg_alt_name = reg["alt-name"] | |
| if reg_name in self.backing_thread.registers: | |
| self.register_ctx[reg_name] = self.backing_thread.registers[reg_name] | |
| elif reg_alt_name and reg_alt_name in self.backing_thread.registers: | |
| self.register_ctx[reg_name] = self.backing_thread.registers[ | |
| reg_alt_name | |
| ] | |
| else: | |
| self.register_ctx[reg_name] = 0 | |
| return self.register_ctx | |
| def resolve_stackframes(thread, addr_mask, target): | |
| frames = [] | |
| for i, frame in enumerate(thread.frames): | |
| frame_pc = frame.pc & addr_mask | |
| # Skip inlined frames so LLDB can reconstruct them from | |
| # debug info when it processes the concrete frames we provide. | |
| # Use the inlined attribute when available (symbolicated reports), | |
| # otherwise fall back to comparing PCs (non-symbolicated reports). | |
| if frame.inlined: | |
| continue | |
| next_frame = thread.frames[i + 1] if i + 1 < len(thread.frames) else None | |
| if next_frame and frame_pc == (next_frame.pc & addr_mask): | |
| continue | |
| # Don't subtract 1 from the first concrete frame (it's the actual | |
| # PC, not a return address) or from null PCs. | |
| if len(frames) == 0 or frame_pc == 0: | |
| pc = frame_pc | |
| else: | |
| pc = frame_pc - 1 | |
| sym_addr = lldb.SBAddress() | |
| sym_addr.SetLoadAddress(pc, target) | |
| if not sym_addr.IsValid(): | |
| continue | |
| frames.append({"idx": len(frames), "pc": pc}) | |
| return frames | |
| def create_stackframes(self): | |
| if not (self.originating_process.options.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.originating_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 | |
| self.name = self.backing_thread.name | |
| self.queue = self.backing_thread.queue | |
| self.has_crashed = self.originating_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.originating_process.exception: | |
| stop_reason["data"]["mach_exception"] = self.originating_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.originating_process.extended_thread_info | |
| return self.extended_info |