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
class StackCoreScriptedProcess(ScriptedProcess):
def get_module_with_name(self, target, name):
for module in target.modules:
if name in module.GetFileSpec().GetFilename():
return module
return None
def __init__(self, exe_ctx: lldb.SBExecutionContext, args: lldb.SBStructuredData):
super().__init__(exe_ctx, args)
self.corefile_target = None
self.corefile_process = None
self.backing_target_idx = args.GetValueForKey("backing_target_idx")
if self.backing_target_idx and self.backing_target_idx.IsValid():
if self.backing_target_idx.GetType() == lldb.eStructuredDataTypeInteger:
idx = self.backing_target_idx.GetIntegerValue(42)
if self.backing_target_idx.GetType() == lldb.eStructuredDataTypeString:
idx = int(self.backing_target_idx.GetStringValue(100))
self.corefile_target = self.target.GetDebugger().GetTargetAtIndex(idx)
self.corefile_process = self.corefile_target.GetProcess()
for corefile_thread in self.corefile_process:
structured_data = lldb.SBStructuredData()
structured_data.SetFromJSON(
json.dumps(
{
"backing_target_idx": idx,
"thread_idx": corefile_thread.GetIndexID(),
}
)
)
self.threads[corefile_thread.GetThreadID()] = StackCoreScriptedThread(
self, structured_data
)
if len(self.threads) == 2:
self.threads[len(self.threads) - 1].is_stopped = True
corefile_module = self.get_module_with_name(
self.corefile_target, "libbaz.dylib"
)
if not corefile_module or not corefile_module.IsValid():
return
module_path = os.path.join(
corefile_module.GetFileSpec().GetDirectory(),
corefile_module.GetFileSpec().GetFilename(),
)
if not os.path.exists(module_path):
return
module_load_addr = corefile_module.GetObjectFileHeaderAddress().GetLoadAddress(
self.corefile_target
)
self.loaded_images.append({"path": module_path, "load_addr": module_load_addr})
def get_memory_region_containing_address(
self, addr: int
) -> lldb.SBMemoryRegionInfo:
mem_region = lldb.SBMemoryRegionInfo()
error = self.corefile_process.GetMemoryRegionInfo(addr, mem_region)
if error.Fail():
return None
return mem_region
def read_memory_at_address(
self, addr: int, size: int, error: lldb.SBError
) -> lldb.SBData:
data = lldb.SBData()
bytes_read = self.corefile_process.ReadMemory(addr, size, error)
if error.Fail():
return data
data.SetDataWithOwnership(
error,
bytes_read,
self.corefile_target.GetByteOrder(),
self.corefile_target.GetAddressByteSize(),
)
return data
def get_loaded_images(self):
return self.loaded_images
def get_process_id(self) -> int:
return 42
def should_stop(self) -> bool:
return True
def is_alive(self) -> bool:
return True
def get_scripted_thread_plugin(self):
return (
StackCoreScriptedThread.__module__ + "." + StackCoreScriptedThread.__name__
)
class StackCoreScriptedThread(ScriptedThread):
def __init__(self, process, args):
super().__init__(process, args)
backing_target_idx = args.GetValueForKey("backing_target_idx")
thread_idx = args.GetValueForKey("thread_idx")
self.is_stopped = False
def extract_value_from_structured_data(data, default_val):
if data and data.IsValid():
if data.GetType() == lldb.eStructuredDataTypeInteger:
return data.GetIntegerValue(default_val)
if data.GetType() == lldb.eStructuredDataTypeString:
return int(data.GetStringValue(100))
return None
# TODO: Change to Walrus operator (:=) with oneline if assignment
# Requires python 3.8
val = extract_value_from_structured_data(thread_idx, 0)
if val is not None:
self.idx = val
self.corefile_target = None
self.corefile_process = None
self.corefile_thread = None
# TODO: Change to Walrus operator (:=) with oneline if assignment
# Requires python 3.8
val = extract_value_from_structured_data(backing_target_idx, 42)
if val is not None:
self.corefile_target = self.target.GetDebugger().GetTargetAtIndex(val)
self.corefile_process = self.corefile_target.GetProcess()
self.corefile_thread = self.corefile_process.GetThreadByIndexID(self.idx)
if self.corefile_thread:
self.id = self.corefile_thread.GetThreadID()
def get_thread_id(self) -> int:
return self.id
def get_name(self) -> str:
return StackCoreScriptedThread.__name__ + ".thread-" + str(self.id)
def get_stop_reason(self) -> Dict[str, Any]:
stop_reason = {"type": lldb.eStopReasonInvalid, "data": {}}
if (
self.corefile_thread
and self.corefile_thread.IsValid()
and self.get_thread_id() == self.corefile_thread.GetThreadID()
):
stop_reason["type"] = lldb.eStopReasonNone
if self.is_stopped:
if "arm64" in self.scripted_process.arch:
stop_reason["type"] = lldb.eStopReasonException
stop_reason["data"][
"desc"
] = self.corefile_thread.GetStopDescription(100)
elif self.scripted_process.arch == "x86_64":
stop_reason["type"] = lldb.eStopReasonSignal
stop_reason["data"]["signal"] = signal.SIGTRAP
else:
stop_reason["type"] = self.corefile_thread.GetStopReason()
return stop_reason
def get_register_context(self) -> str:
if not self.corefile_thread or self.corefile_thread.GetNumFrames() == 0:
return None
frame = self.corefile_thread.GetFrameAtIndex(0)
GPRs = None
registerSet = frame.registers # Returns an SBValueList.
for regs in registerSet:
if "general purpose" in regs.name.lower():
GPRs = regs
break
if not GPRs:
return None
for reg in GPRs:
self.register_ctx[reg.name] = int(reg.value, base=16)
return struct.pack(
"{}Q".format(len(self.register_ctx)), *self.register_ctx.values()
)
def __lldb_init_module(debugger, dict):
if not "SKIP_SCRIPTED_PROCESS_LAUNCH" in os.environ:
debugger.HandleCommand(
"process launch -C %s.%s" % (__name__, StackCoreScriptedProcess.__name__)
)
else:
print(
"Name of the class that will manage the scripted process: '%s.%s'"
% (__name__, StackCoreScriptedProcess.__name__)
)
Codebase Browser
llvm