1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
import os,json,struct,signal
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):
try:
crash_log = CrashLogParser().parse(self.dbg, self.crashlog_path, False)
except Exception as e:
return
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 = []
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:
err = image.add_module(self.target)
if err:
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))
self.threads[thread.index] = CrashLogScriptedThread(self, None, thread)
def __init__(self, target: lldb.SBTarget, args : lldb.SBStructuredData):
super().__init__(target, args)
if not self.target or not self.target.IsValid():
return
self.crashlog_path = None
crashlog_path = args.GetValueForKey("crashlog_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
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.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__
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 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
for frame in self.backing_thread.frames:
frame_pc = frame.pc & self.scripted_process.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, self.target)
if not sym_addr.IsValid():
continue
self.frames.append({"idx": frame.index, "pc": pc})
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.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, "data": { }}
# TODO: Investigate what stop reason should be reported when crashed
return { "type": lldb.eStopReasonException, "data": { "desc": "EXC_BAD_ACCESS" }}
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())
|
