1# DExTer : Debugging Experience Tester
2# ~~~~~~   ~         ~~         ~   ~~
3#
4# Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
5# See https://llvm.org/LICENSE.txt for license information.
6# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
7"""Interface for communicating with the LLDB debugger via its python interface.
8"""
9
10import imp
11import os
12from subprocess import CalledProcessError, check_output, STDOUT
13import sys
14
15from dex.debugger.DebuggerBase import DebuggerBase, watch_is_active
16from dex.dextIR import FrameIR, LocIR, StepIR, StopReason, ValueIR
17from dex.dextIR import StackFrame, SourceLocation, ProgramState
18from dex.utils.Exceptions import DebuggerException, LoadDebuggerException
19from dex.utils.ReturnCode import ReturnCode
20
21
22class LLDB(DebuggerBase):
23    def __init__(self, context, *args):
24        self.lldb_executable = context.options.lldb_executable
25        self._debugger = None
26        self._target = None
27        self._process = None
28        self._thread = None
29        # Map {id (int): condition (str)} for breakpoints which have a
30        # condition. See get_triggered_breakpoint_ids usage for more info.
31        self._breakpoint_conditions = {}
32        super(LLDB, self).__init__(context, *args)
33
34    def _custom_init(self):
35        self._debugger = self._interface.SBDebugger.Create()
36        self._debugger.SetAsync(False)
37        self._target = self._debugger.CreateTargetWithFileAndArch(
38            self.context.options.executable, self.context.options.arch)
39        if not self._target:
40            raise LoadDebuggerException(
41                'could not create target for executable "{}" with arch:{}'.
42                format(self.context.options.executable,
43                       self.context.options.arch))
44
45    def _custom_exit(self):
46        if getattr(self, '_process', None):
47            self._process.Kill()
48        if getattr(self, '_debugger', None) and getattr(self, '_target', None):
49            self._debugger.DeleteTarget(self._target)
50
51    def _translate_stop_reason(self, reason):
52        if reason == self._interface.eStopReasonNone:
53            return None
54        if reason == self._interface.eStopReasonBreakpoint:
55            return StopReason.BREAKPOINT
56        if reason == self._interface.eStopReasonPlanComplete:
57            return StopReason.STEP
58        if reason == self._interface.eStopReasonThreadExiting:
59            return StopReason.PROGRAM_EXIT
60        if reason == self._interface.eStopReasonException:
61            return StopReason.ERROR
62        return StopReason.OTHER
63
64    def _load_interface(self):
65        try:
66            args = [self.lldb_executable, '-P']
67            pythonpath = check_output(
68                args, stderr=STDOUT).rstrip().decode('utf-8')
69        except CalledProcessError as e:
70            raise LoadDebuggerException(str(e), sys.exc_info())
71        except OSError as e:
72            raise LoadDebuggerException(
73                '{} ["{}"]'.format(e.strerror, self.lldb_executable),
74                sys.exc_info())
75
76        if not os.path.isdir(pythonpath):
77            raise LoadDebuggerException(
78                'path "{}" does not exist [result of {}]'.format(
79                    pythonpath, args), sys.exc_info())
80
81        try:
82            module_info = imp.find_module('lldb', [pythonpath])
83            return imp.load_module('lldb', *module_info)
84        except ImportError as e:
85            msg = str(e)
86            if msg.endswith('not a valid Win32 application.'):
87                msg = '{} [Are you mixing 32-bit and 64-bit binaries?]'.format(
88                    msg)
89            raise LoadDebuggerException(msg, sys.exc_info())
90
91    @classmethod
92    def get_name(cls):
93        return 'lldb'
94
95    @classmethod
96    def get_option_name(cls):
97        return 'lldb'
98
99    @property
100    def version(self):
101        try:
102            return self._interface.SBDebugger_GetVersionString()
103        except AttributeError:
104            return None
105
106    def clear_breakpoints(self):
107        self._target.DeleteAllBreakpoints()
108
109    def _add_breakpoint(self, file_, line):
110        return self._add_conditional_breakpoint(file_, line, None)
111
112    def _add_conditional_breakpoint(self, file_, line, condition):
113        bp = self._target.BreakpointCreateByLocation(file_, line)
114        if not bp:
115            raise DebuggerException(
116                  'could not add breakpoint [{}:{}]'.format(file_, line))
117        id = bp.GetID()
118        if condition:
119            bp.SetCondition(condition)
120            assert id not in self._breakpoint_conditions
121            self._breakpoint_conditions[id] = condition
122        return id
123
124    def _evaulate_breakpoint_condition(self, id):
125        """Evaluate the breakpoint condition and return the result.
126
127        Returns True if a conditional breakpoint with the specified id cannot
128        be found (i.e. assume it is an unconditional breakpoint).
129        """
130        try:
131            condition = self._breakpoint_conditions[id]
132        except KeyError:
133            # This must be an unconditional breakpoint.
134            return True
135        valueIR = self.evaluate_expression(condition)
136        return valueIR.type_name == 'bool' and valueIR.value == 'true'
137
138    def get_triggered_breakpoint_ids(self):
139        # Breakpoints can only have been triggered if we've hit one.
140        stop_reason = self._translate_stop_reason(self._thread.GetStopReason())
141        if stop_reason != StopReason.BREAKPOINT:
142            return []
143        breakpoint_ids = set()
144        # When the stop reason is eStopReasonBreakpoint, GetStopReasonDataCount
145        # counts all breakpoints associated with the location that lldb has
146        # stopped at, regardless of their condition. I.e. Even if we have two
147        # breakpoints at the same source location that have mutually exclusive
148        # conditions, both will be counted by GetStopReasonDataCount when
149        # either condition is true. Check each breakpoint condition manually to
150        # filter the list down to breakpoints that have caused this stop.
151        #
152        # Breakpoints have two data parts: Breakpoint ID, Location ID. We're
153        # only interested in the Breakpoint ID so we skip every other item.
154        for i in range(0, self._thread.GetStopReasonDataCount(), 2):
155            id = self._thread.GetStopReasonDataAtIndex(i)
156            if self._evaulate_breakpoint_condition(id):
157                breakpoint_ids.add(id)
158        return breakpoint_ids
159
160    def delete_breakpoint(self, id):
161        bp = self._target.FindBreakpointByID(id)
162        if not bp:
163            # The ID is not valid.
164            raise KeyError
165        try:
166            del self._breakpoint_conditions[id]
167        except KeyError:
168            # This must be an unconditional breakpoint.
169            pass
170        self._target.BreakpointDelete(id)
171
172    def launch(self):
173        self._process = self._target.LaunchSimple(None, None, os.getcwd())
174        if not self._process or self._process.GetNumThreads() == 0:
175            raise DebuggerException('could not launch process')
176        if self._process.GetNumThreads() != 1:
177            raise DebuggerException('multiple threads not supported')
178        self._thread = self._process.GetThreadAtIndex(0)
179        assert self._thread, (self._process, self._thread)
180
181    def step(self):
182        self._thread.StepInto()
183
184    def go(self) -> ReturnCode:
185        self._process.Continue()
186        return ReturnCode.OK
187
188    def _get_step_info(self, watches, step_index):
189        frames = []
190        state_frames = []
191
192        for i in range(0, self._thread.GetNumFrames()):
193            sb_frame = self._thread.GetFrameAtIndex(i)
194            sb_line = sb_frame.GetLineEntry()
195            sb_filespec = sb_line.GetFileSpec()
196
197            try:
198                path = os.path.join(sb_filespec.GetDirectory(),
199                                    sb_filespec.GetFilename())
200            except (AttributeError, TypeError):
201                path = None
202
203            function = self._sanitize_function_name(sb_frame.GetFunctionName())
204
205            loc_dict = {
206                'path': path,
207                'lineno': sb_line.GetLine(),
208                'column': sb_line.GetColumn()
209            }
210            loc = LocIR(**loc_dict)
211            valid_loc_for_watch = loc.path and os.path.exists(loc.path)
212
213            frame = FrameIR(
214                function=function, is_inlined=sb_frame.IsInlined(), loc=loc)
215
216            if any(
217                    name in (frame.function or '')  # pylint: disable=no-member
218                    for name in self.frames_below_main):
219                break
220
221            frames.append(frame)
222
223            state_frame = StackFrame(function=frame.function,
224                                     is_inlined=frame.is_inlined,
225                                     location=SourceLocation(**loc_dict),
226                                     watches={})
227            if valid_loc_for_watch:
228                for expr in map(
229                    # Filter out watches that are not active in the current frame,
230                    # and then evaluate all the active watches.
231                    lambda watch_info, idx=i:
232                        self.evaluate_expression(watch_info.expression, idx),
233                    filter(
234                        lambda watch_info, idx=i, line_no=loc.lineno, loc_path=loc.path:
235                            watch_is_active(watch_info, loc_path, idx, line_no),
236                        watches)):
237                    state_frame.watches[expr.expression] = expr
238            state_frames.append(state_frame)
239
240        if len(frames) == 1 and frames[0].function is None:
241            frames = []
242            state_frames = []
243
244        reason = self._translate_stop_reason(self._thread.GetStopReason())
245
246        return StepIR(
247            step_index=step_index, frames=frames, stop_reason=reason,
248            program_state=ProgramState(state_frames))
249
250    @property
251    def is_running(self):
252        # We're not running in async mode so this is always False.
253        return False
254
255    @property
256    def is_finished(self):
257        return not self._thread.GetFrameAtIndex(0)
258
259    @property
260    def frames_below_main(self):
261        return ['__scrt_common_main_seh', '__libc_start_main']
262
263    def evaluate_expression(self, expression, frame_idx=0) -> ValueIR:
264        result = self._thread.GetFrameAtIndex(frame_idx
265            ).EvaluateExpression(expression)
266        error_string = str(result.error)
267
268        value = result.value
269        could_evaluate = not any(s in error_string for s in [
270            "Can't run the expression locally",
271            "use of undeclared identifier",
272            "no member named",
273            "Couldn't lookup symbols",
274            "reference to local variable",
275            "invalid use of 'this' outside of a non-static member function",
276        ])
277
278        is_optimized_away = any(s in error_string for s in [
279            'value may have been optimized out',
280        ])
281
282        is_irretrievable = any(s in error_string for s in [
283            "couldn't get the value of variable",
284            "couldn't read its memory",
285            "couldn't read from memory",
286            "Cannot access memory at address",
287            "invalid address (fault address:",
288        ])
289
290        if could_evaluate and not is_irretrievable and not is_optimized_away:
291            assert error_string == 'success', (error_string, expression, value)
292            # assert result.value is not None, (result.value, expression)
293
294        if error_string == 'success':
295            error_string = None
296
297        # attempt to find expression as a variable, if found, take the variable
298        # obj's type information as it's 'usually' more accurate.
299        var_result = self._thread.GetFrameAtIndex(frame_idx).FindVariable(expression)
300        if str(var_result.error) == 'success':
301            type_name = var_result.type.GetDisplayTypeName()
302        else:
303            type_name = result.type.GetDisplayTypeName()
304
305        return ValueIR(
306            expression=expression,
307            value=value,
308            type_name=type_name,
309            error_string=error_string,
310            could_evaluate=could_evaluate,
311            is_optimized_away=is_optimized_away,
312            is_irretrievable=is_irretrievable,
313        )
314