| # DExTer : Debugging Experience Tester |
| # ~~~~~~ ~ ~~ ~ ~~ |
| # |
| # Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions. |
| # See https://llvm.org/LICENSE.txt for license information. |
| # SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception |
| """Interface for communicating with the LLDB debugger via its python interface. |
| """ |
| |
| import imp |
| import os |
| from subprocess import CalledProcessError, check_output, STDOUT |
| import sys |
| |
| from dex.debugger.DebuggerBase import DebuggerBase, watch_is_active |
| from dex.dextIR import FrameIR, LocIR, StepIR, StopReason, ValueIR |
| from dex.dextIR import StackFrame, SourceLocation, ProgramState |
| from dex.utils.Exceptions import DebuggerException, LoadDebuggerException |
| from dex.utils.ReturnCode import ReturnCode |
| |
| |
| class LLDB(DebuggerBase): |
| def __init__(self, context, *args): |
| self.lldb_executable = context.options.lldb_executable |
| self._debugger = None |
| self._target = None |
| self._process = None |
| self._thread = None |
| # Map {id (int): condition (str)} for breakpoints which have a |
| # condition. See get_triggered_breakpoint_ids usage for more info. |
| self._breakpoint_conditions = {} |
| super(LLDB, self).__init__(context, *args) |
| |
| def _custom_init(self): |
| self._debugger = self._interface.SBDebugger.Create() |
| self._debugger.SetAsync(False) |
| self._target = self._debugger.CreateTargetWithFileAndArch( |
| self.context.options.executable, self.context.options.arch) |
| if not self._target: |
| raise LoadDebuggerException( |
| 'could not create target for executable "{}" with arch:{}'. |
| format(self.context.options.executable, |
| self.context.options.arch)) |
| |
| def _custom_exit(self): |
| if getattr(self, '_process', None): |
| self._process.Kill() |
| if getattr(self, '_debugger', None) and getattr(self, '_target', None): |
| self._debugger.DeleteTarget(self._target) |
| |
| def _translate_stop_reason(self, reason): |
| if reason == self._interface.eStopReasonNone: |
| return None |
| if reason == self._interface.eStopReasonBreakpoint: |
| return StopReason.BREAKPOINT |
| if reason == self._interface.eStopReasonPlanComplete: |
| return StopReason.STEP |
| if reason == self._interface.eStopReasonThreadExiting: |
| return StopReason.PROGRAM_EXIT |
| if reason == self._interface.eStopReasonException: |
| return StopReason.ERROR |
| return StopReason.OTHER |
| |
| def _load_interface(self): |
| try: |
| args = [self.lldb_executable, '-P'] |
| pythonpath = check_output( |
| args, stderr=STDOUT).rstrip().decode('utf-8') |
| except CalledProcessError as e: |
| raise LoadDebuggerException(str(e), sys.exc_info()) |
| except OSError as e: |
| raise LoadDebuggerException( |
| '{} ["{}"]'.format(e.strerror, self.lldb_executable), |
| sys.exc_info()) |
| |
| if not os.path.isdir(pythonpath): |
| raise LoadDebuggerException( |
| 'path "{}" does not exist [result of {}]'.format( |
| pythonpath, args), sys.exc_info()) |
| |
| try: |
| module_info = imp.find_module('lldb', [pythonpath]) |
| return imp.load_module('lldb', *module_info) |
| except ImportError as e: |
| msg = str(e) |
| if msg.endswith('not a valid Win32 application.'): |
| msg = '{} [Are you mixing 32-bit and 64-bit binaries?]'.format( |
| msg) |
| raise LoadDebuggerException(msg, sys.exc_info()) |
| |
| @classmethod |
| def get_name(cls): |
| return 'lldb' |
| |
| @classmethod |
| def get_option_name(cls): |
| return 'lldb' |
| |
| @property |
| def version(self): |
| try: |
| return self._interface.SBDebugger_GetVersionString() |
| except AttributeError: |
| return None |
| |
| def clear_breakpoints(self): |
| self._target.DeleteAllBreakpoints() |
| |
| def _add_breakpoint(self, file_, line): |
| return self._add_conditional_breakpoint(file_, line, None) |
| |
| def _add_conditional_breakpoint(self, file_, line, condition): |
| bp = self._target.BreakpointCreateByLocation(file_, line) |
| if not bp: |
| raise DebuggerException( |
| 'could not add breakpoint [{}:{}]'.format(file_, line)) |
| id = bp.GetID() |
| if condition: |
| bp.SetCondition(condition) |
| assert id not in self._breakpoint_conditions |
| self._breakpoint_conditions[id] = condition |
| return id |
| |
| def _evaulate_breakpoint_condition(self, id): |
| """Evaluate the breakpoint condition and return the result. |
| |
| Returns True if a conditional breakpoint with the specified id cannot |
| be found (i.e. assume it is an unconditional breakpoint). |
| """ |
| try: |
| condition = self._breakpoint_conditions[id] |
| except KeyError: |
| # This must be an unconditional breakpoint. |
| return True |
| valueIR = self.evaluate_expression(condition) |
| return valueIR.type_name == 'bool' and valueIR.value == 'true' |
| |
| def get_triggered_breakpoint_ids(self): |
| # Breakpoints can only have been triggered if we've hit one. |
| stop_reason = self._translate_stop_reason(self._thread.GetStopReason()) |
| if stop_reason != StopReason.BREAKPOINT: |
| return [] |
| breakpoint_ids = set() |
| # When the stop reason is eStopReasonBreakpoint, GetStopReasonDataCount |
| # counts all breakpoints associated with the location that lldb has |
| # stopped at, regardless of their condition. I.e. Even if we have two |
| # breakpoints at the same source location that have mutually exclusive |
| # conditions, both will be counted by GetStopReasonDataCount when |
| # either condition is true. Check each breakpoint condition manually to |
| # filter the list down to breakpoints that have caused this stop. |
| # |
| # Breakpoints have two data parts: Breakpoint ID, Location ID. We're |
| # only interested in the Breakpoint ID so we skip every other item. |
| for i in range(0, self._thread.GetStopReasonDataCount(), 2): |
| id = self._thread.GetStopReasonDataAtIndex(i) |
| if self._evaulate_breakpoint_condition(id): |
| breakpoint_ids.add(id) |
| return breakpoint_ids |
| |
| def delete_breakpoint(self, id): |
| bp = self._target.FindBreakpointByID(id) |
| if not bp: |
| # The ID is not valid. |
| raise KeyError |
| try: |
| del self._breakpoint_conditions[id] |
| except KeyError: |
| # This must be an unconditional breakpoint. |
| pass |
| self._target.BreakpointDelete(id) |
| |
| def launch(self): |
| self._process = self._target.LaunchSimple(None, None, os.getcwd()) |
| if not self._process or self._process.GetNumThreads() == 0: |
| raise DebuggerException('could not launch process') |
| if self._process.GetNumThreads() != 1: |
| raise DebuggerException('multiple threads not supported') |
| self._thread = self._process.GetThreadAtIndex(0) |
| assert self._thread, (self._process, self._thread) |
| |
| def step(self): |
| self._thread.StepInto() |
| |
| def go(self) -> ReturnCode: |
| self._process.Continue() |
| return ReturnCode.OK |
| |
| def _get_step_info(self, watches, step_index): |
| frames = [] |
| state_frames = [] |
| |
| for i in range(0, self._thread.GetNumFrames()): |
| sb_frame = self._thread.GetFrameAtIndex(i) |
| sb_line = sb_frame.GetLineEntry() |
| sb_filespec = sb_line.GetFileSpec() |
| |
| try: |
| path = os.path.join(sb_filespec.GetDirectory(), |
| sb_filespec.GetFilename()) |
| except (AttributeError, TypeError): |
| path = None |
| |
| function = self._sanitize_function_name(sb_frame.GetFunctionName()) |
| |
| loc_dict = { |
| 'path': path, |
| 'lineno': sb_line.GetLine(), |
| 'column': sb_line.GetColumn() |
| } |
| loc = LocIR(**loc_dict) |
| valid_loc_for_watch = loc.path and os.path.exists(loc.path) |
| |
| frame = FrameIR( |
| function=function, is_inlined=sb_frame.IsInlined(), loc=loc) |
| |
| if any( |
| name in (frame.function or '') # pylint: disable=no-member |
| for name in self.frames_below_main): |
| break |
| |
| frames.append(frame) |
| |
| state_frame = StackFrame(function=frame.function, |
| is_inlined=frame.is_inlined, |
| location=SourceLocation(**loc_dict), |
| watches={}) |
| if valid_loc_for_watch: |
| for expr in map( |
| # Filter out watches that are not active in the current frame, |
| # and then evaluate all the active watches. |
| lambda watch_info, idx=i: |
| self.evaluate_expression(watch_info.expression, idx), |
| filter( |
| lambda watch_info, idx=i, line_no=loc.lineno, loc_path=loc.path: |
| watch_is_active(watch_info, loc_path, idx, line_no), |
| watches)): |
| state_frame.watches[expr.expression] = expr |
| state_frames.append(state_frame) |
| |
| if len(frames) == 1 and frames[0].function is None: |
| frames = [] |
| state_frames = [] |
| |
| reason = self._translate_stop_reason(self._thread.GetStopReason()) |
| |
| return StepIR( |
| step_index=step_index, frames=frames, stop_reason=reason, |
| program_state=ProgramState(state_frames)) |
| |
| @property |
| def is_running(self): |
| # We're not running in async mode so this is always False. |
| return False |
| |
| @property |
| def is_finished(self): |
| return not self._thread.GetFrameAtIndex(0) |
| |
| @property |
| def frames_below_main(self): |
| return ['__scrt_common_main_seh', '__libc_start_main'] |
| |
| def evaluate_expression(self, expression, frame_idx=0) -> ValueIR: |
| result = self._thread.GetFrameAtIndex(frame_idx |
| ).EvaluateExpression(expression) |
| error_string = str(result.error) |
| |
| value = result.value |
| could_evaluate = not any(s in error_string for s in [ |
| "Can't run the expression locally", |
| "use of undeclared identifier", |
| "no member named", |
| "Couldn't lookup symbols", |
| "reference to local variable", |
| "invalid use of 'this' outside of a non-static member function", |
| ]) |
| |
| is_optimized_away = any(s in error_string for s in [ |
| 'value may have been optimized out', |
| ]) |
| |
| is_irretrievable = any(s in error_string for s in [ |
| "couldn't get the value of variable", |
| "couldn't read its memory", |
| "couldn't read from memory", |
| "Cannot access memory at address", |
| "invalid address (fault address:", |
| ]) |
| |
| if could_evaluate and not is_irretrievable and not is_optimized_away: |
| assert error_string == 'success', (error_string, expression, value) |
| # assert result.value is not None, (result.value, expression) |
| |
| if error_string == 'success': |
| error_string = None |
| |
| # attempt to find expression as a variable, if found, take the variable |
| # obj's type information as it's 'usually' more accurate. |
| var_result = self._thread.GetFrameAtIndex(frame_idx).FindVariable(expression) |
| if str(var_result.error) == 'success': |
| type_name = var_result.type.GetDisplayTypeName() |
| else: |
| type_name = result.type.GetDisplayTypeName() |
| |
| return ValueIR( |
| expression=expression, |
| value=value, |
| type_name=type_name, |
| error_string=error_string, |
| could_evaluate=could_evaluate, |
| is_optimized_away=is_optimized_away, |
| is_irretrievable=is_irretrievable, |
| ) |
