| #!/usr/bin/env python |
| #===- lib/asan/scripts/asan_symbolize.py -----------------------------------===# |
| # |
| # 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 |
| # |
| #===------------------------------------------------------------------------===# |
| """ |
| Example of use: |
| asan_symbolize.py -c "$HOME/opt/cross/bin/arm-linux-gnueabi-" -s "$HOME/SymbolFiles" < asan.log |
| |
| PLUGINS |
| |
| This script provides a way for external plug-ins to hook into the behaviour of |
| various parts of this script (see `--plugins`). This is useful for situations |
| where it is necessary to handle site-specific quirks (e.g. binaries with debug |
| symbols only accessible via a remote service) without having to modify the |
| script itself. |
| |
| """ |
| import argparse |
| import bisect |
| import errno |
| import getopt |
| import logging |
| import os |
| import re |
| import subprocess |
| import sys |
| from distutils.spawn import find_executable |
| |
| symbolizers = {} |
| demangle = False |
| binutils_prefix = None |
| fix_filename_patterns = None |
| logfile = sys.stdin |
| allow_system_symbolizer = True |
| force_system_symbolizer = False |
| |
| # FIXME: merge the code that calls fix_filename(). |
| def fix_filename(file_name): |
| if fix_filename_patterns: |
| for path_to_cut in fix_filename_patterns: |
| file_name = re.sub('.*' + path_to_cut, '', file_name) |
| file_name = re.sub('.*asan_[a-z_]*.(cc|cpp):[0-9]*', '_asan_rtl_', file_name) |
| file_name = re.sub('.*crtstuff.c:0', '???:0', file_name) |
| return file_name |
| |
| def is_valid_arch(s): |
| return s in ["i386", "x86_64", "x86_64h", "arm", "armv6", "armv7", "armv7s", |
| "armv7k", "arm64", "powerpc64", "powerpc64le", "s390x", "s390", |
| "riscv64"] |
| |
| def guess_arch(addr): |
| # Guess which arch we're running. 10 = len('0x') + 8 hex digits. |
| if len(addr) > 10: |
| return 'x86_64' |
| else: |
| return 'i386' |
| |
| class Symbolizer(object): |
| def __init__(self): |
| pass |
| |
| def symbolize(self, addr, binary, offset): |
| """Symbolize the given address (pair of binary and offset). |
| |
| Overriden in subclasses. |
| Args: |
| addr: virtual address of an instruction. |
| binary: path to executable/shared object containing this instruction. |
| offset: instruction offset in the @binary. |
| Returns: |
| list of strings (one string for each inlined frame) describing |
| the code locations for this instruction (that is, function name, file |
| name, line and column numbers). |
| """ |
| return None |
| |
| |
| class LLVMSymbolizer(Symbolizer): |
| def __init__(self, symbolizer_path, default_arch, system, dsym_hints=[]): |
| super(LLVMSymbolizer, self).__init__() |
| self.symbolizer_path = symbolizer_path |
| self.default_arch = default_arch |
| self.system = system |
| self.dsym_hints = dsym_hints |
| self.pipe = self.open_llvm_symbolizer() |
| |
| def open_llvm_symbolizer(self): |
| cmd = [self.symbolizer_path, |
| ('--demangle' if demangle else '--no-demangle'), |
| '--functions=linkage', |
| '--inlines', |
| '--default-arch=%s' % self.default_arch] |
| if self.system == 'Darwin': |
| for hint in self.dsym_hints: |
| cmd.append('--dsym-hint=%s' % hint) |
| logging.debug(' '.join(cmd)) |
| try: |
| result = subprocess.Popen(cmd, stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| bufsize=0, |
| universal_newlines=True) |
| except OSError: |
| result = None |
| return result |
| |
| def symbolize(self, addr, binary, offset): |
| """Overrides Symbolizer.symbolize.""" |
| if not self.pipe: |
| return None |
| result = [] |
| try: |
| symbolizer_input = '"%s" %s' % (binary, offset) |
| logging.debug(symbolizer_input) |
| self.pipe.stdin.write("%s\n" % symbolizer_input) |
| while True: |
| function_name = self.pipe.stdout.readline().rstrip() |
| if not function_name: |
| break |
| file_name = self.pipe.stdout.readline().rstrip() |
| file_name = fix_filename(file_name) |
| if (not function_name.startswith('??') or |
| not file_name.startswith('??')): |
| # Append only non-trivial frames. |
| result.append('%s in %s %s' % (addr, function_name, |
| file_name)) |
| except Exception: |
| result = [] |
| if not result: |
| result = None |
| return result |
| |
| |
| def LLVMSymbolizerFactory(system, default_arch, dsym_hints=[]): |
| symbolizer_path = os.getenv('LLVM_SYMBOLIZER_PATH') |
| if not symbolizer_path: |
| symbolizer_path = os.getenv('ASAN_SYMBOLIZER_PATH') |
| if not symbolizer_path: |
| # Assume llvm-symbolizer is in PATH. |
| symbolizer_path = 'llvm-symbolizer' |
| return LLVMSymbolizer(symbolizer_path, default_arch, system, dsym_hints) |
| |
| |
| class Addr2LineSymbolizer(Symbolizer): |
| def __init__(self, binary): |
| super(Addr2LineSymbolizer, self).__init__() |
| self.binary = binary |
| self.pipe = self.open_addr2line() |
| self.output_terminator = -1 |
| |
| def open_addr2line(self): |
| addr2line_tool = 'addr2line' |
| if binutils_prefix: |
| addr2line_tool = binutils_prefix + addr2line_tool |
| logging.debug('addr2line binary is %s' % find_executable(addr2line_tool)) |
| cmd = [addr2line_tool, '-fi'] |
| if demangle: |
| cmd += ['--demangle'] |
| cmd += ['-e', self.binary] |
| logging.debug(' '.join(cmd)) |
| return subprocess.Popen(cmd, |
| stdin=subprocess.PIPE, stdout=subprocess.PIPE, |
| bufsize=0, |
| universal_newlines=True) |
| |
| def symbolize(self, addr, binary, offset): |
| """Overrides Symbolizer.symbolize.""" |
| if self.binary != binary: |
| return None |
| lines = [] |
| try: |
| self.pipe.stdin.write("%s\n" % offset) |
| self.pipe.stdin.write("%s\n" % self.output_terminator) |
| is_first_frame = True |
| while True: |
| function_name = self.pipe.stdout.readline().rstrip() |
| logging.debug("read function_name='%s' from addr2line" % function_name) |
| # If llvm-symbolizer is installed as addr2line, older versions of |
| # llvm-symbolizer will print -1 when presented with -1 and not print |
| # a second line. In that case we will block for ever trying to read the |
| # file name. This also happens for non-existent files, in which case GNU |
| # addr2line exits immediate, but llvm-symbolizer does not (see |
| # https://llvm.org/PR42754). |
| if function_name == '-1': |
| logging.debug("got function '-1' -> no more input") |
| break |
| file_name = self.pipe.stdout.readline().rstrip() |
| logging.debug("read file_name='%s' from addr2line" % file_name) |
| if is_first_frame: |
| is_first_frame = False |
| elif function_name == '??': |
| assert file_name == '??:0', file_name |
| logging.debug("got function '??' -> no more input") |
| break |
| elif not function_name: |
| assert not file_name, file_name |
| logging.debug("got empty function name -> no more input") |
| break |
| if not function_name and not file_name: |
| logging.debug("got empty function and file name -> unknown function") |
| function_name = '??' |
| file_name = '??:0' |
| lines.append((function_name, file_name)) |
| except IOError as e: |
| # EPIPE happens if addr2line exits early (which some implementations do |
| # if an invalid file is passed). |
| if e.errno == errno.EPIPE: |
| logging.debug(f"addr2line exited early (broken pipe) returncode={self.pipe.poll()}") |
| else: |
| logging.debug("unexpected I/O exception communicating with addr2line", exc_info=e) |
| lines.append(('??', '??:0')) |
| except Exception as e: |
| logging.debug("got unknown exception communicating with addr2line", exc_info=e) |
| lines.append(('??', '??:0')) |
| return ['%s in %s %s' % (addr, function, fix_filename(file)) for (function, file) in lines] |
| |
| class UnbufferedLineConverter(object): |
| """ |
| Wrap a child process that responds to each line of input with one line of |
| output. Uses pty to trick the child into providing unbuffered output. |
| """ |
| def __init__(self, args, close_stderr=False): |
| # Local imports so that the script can start on Windows. |
| import pty |
| import termios |
| pid, fd = pty.fork() |
| if pid == 0: |
| # We're the child. Transfer control to command. |
| if close_stderr: |
| dev_null = os.open('/dev/null', 0) |
| os.dup2(dev_null, 2) |
| os.execvp(args[0], args) |
| else: |
| # Disable echoing. |
| attr = termios.tcgetattr(fd) |
| attr[3] = attr[3] & ~termios.ECHO |
| termios.tcsetattr(fd, termios.TCSANOW, attr) |
| # Set up a file()-like interface to the child process |
| self.r = os.fdopen(fd, "r", 1) |
| self.w = os.fdopen(os.dup(fd), "w", 1) |
| |
| def convert(self, line): |
| self.w.write(line + "\n") |
| return self.readline() |
| |
| def readline(self): |
| return self.r.readline().rstrip() |
| |
| |
| class DarwinSymbolizer(Symbolizer): |
| def __init__(self, addr, binary, arch): |
| super(DarwinSymbolizer, self).__init__() |
| self.binary = binary |
| self.arch = arch |
| self.open_atos() |
| |
| def open_atos(self): |
| logging.debug('atos -o %s -arch %s', self.binary, self.arch) |
| cmdline = ['atos', '-o', self.binary, '-arch', self.arch] |
| self.atos = UnbufferedLineConverter(cmdline, close_stderr=True) |
| |
| def symbolize(self, addr, binary, offset): |
| """Overrides Symbolizer.symbolize.""" |
| if self.binary != binary: |
| return None |
| if not os.path.exists(binary): |
| # If the binary doesn't exist atos will exit which will lead to IOError |
| # exceptions being raised later on so just don't try to symbolize. |
| return ['{} ({}:{}+{})'.format(addr, binary, self.arch, offset)] |
| atos_line = self.atos.convert('0x%x' % int(offset, 16)) |
| while "got symbolicator for" in atos_line: |
| atos_line = self.atos.readline() |
| # A well-formed atos response looks like this: |
| # foo(type1, type2) (in object.name) (filename.cc:80) |
| # NOTE: |
| # * For C functions atos omits parentheses and argument types. |
| # * For C++ functions the function name (i.e., `foo` above) may contain |
| # templates which may contain parentheses. |
| match = re.match('^(.*) \(in (.*)\) \((.*:\d*)\)$', atos_line) |
| logging.debug('atos_line: %s', atos_line) |
| if match: |
| function_name = match.group(1) |
| file_name = fix_filename(match.group(3)) |
| return ['%s in %s %s' % (addr, function_name, file_name)] |
| else: |
| return ['%s in %s' % (addr, atos_line)] |
| |
| |
| # Chain several symbolizers so that if one symbolizer fails, we fall back |
| # to the next symbolizer in chain. |
| class ChainSymbolizer(Symbolizer): |
| def __init__(self, symbolizer_list): |
| super(ChainSymbolizer, self).__init__() |
| self.symbolizer_list = symbolizer_list |
| |
| def symbolize(self, addr, binary, offset): |
| """Overrides Symbolizer.symbolize.""" |
| for symbolizer in self.symbolizer_list: |
| if symbolizer: |
| result = symbolizer.symbolize(addr, binary, offset) |
| if result: |
| return result |
| return None |
| |
| def append_symbolizer(self, symbolizer): |
| self.symbolizer_list.append(symbolizer) |
| |
| |
| def BreakpadSymbolizerFactory(binary): |
| suffix = os.getenv('BREAKPAD_SUFFIX') |
| if suffix: |
| filename = binary + suffix |
| if os.access(filename, os.F_OK): |
| return BreakpadSymbolizer(filename) |
| return None |
| |
| |
| def SystemSymbolizerFactory(system, addr, binary, arch): |
| if system == 'Darwin': |
| return DarwinSymbolizer(addr, binary, arch) |
| elif system in ['Linux', 'FreeBSD', 'NetBSD', 'SunOS']: |
| return Addr2LineSymbolizer(binary) |
| |
| |
| class BreakpadSymbolizer(Symbolizer): |
| def __init__(self, filename): |
| super(BreakpadSymbolizer, self).__init__() |
| self.filename = filename |
| lines = file(filename).readlines() |
| self.files = [] |
| self.symbols = {} |
| self.address_list = [] |
| self.addresses = {} |
| # MODULE mac x86_64 A7001116478B33F18FF9BEDE9F615F190 t |
| fragments = lines[0].rstrip().split() |
| self.arch = fragments[2] |
| self.debug_id = fragments[3] |
| self.binary = ' '.join(fragments[4:]) |
| self.parse_lines(lines[1:]) |
| |
| def parse_lines(self, lines): |
| cur_function_addr = '' |
| for line in lines: |
| fragments = line.split() |
| if fragments[0] == 'FILE': |
| assert int(fragments[1]) == len(self.files) |
| self.files.append(' '.join(fragments[2:])) |
| elif fragments[0] == 'PUBLIC': |
| self.symbols[int(fragments[1], 16)] = ' '.join(fragments[3:]) |
| elif fragments[0] in ['CFI', 'STACK']: |
| pass |
| elif fragments[0] == 'FUNC': |
| cur_function_addr = int(fragments[1], 16) |
| if not cur_function_addr in self.symbols.keys(): |
| self.symbols[cur_function_addr] = ' '.join(fragments[4:]) |
| else: |
| # Line starting with an address. |
| addr = int(fragments[0], 16) |
| self.address_list.append(addr) |
| # Tuple of symbol address, size, line, file number. |
| self.addresses[addr] = (cur_function_addr, |
| int(fragments[1], 16), |
| int(fragments[2]), |
| int(fragments[3])) |
| self.address_list.sort() |
| |
| def get_sym_file_line(self, addr): |
| key = None |
| if addr in self.addresses.keys(): |
| key = addr |
| else: |
| index = bisect.bisect_left(self.address_list, addr) |
| if index == 0: |
| return None |
| else: |
| key = self.address_list[index - 1] |
| sym_id, size, line_no, file_no = self.addresses[key] |
| symbol = self.symbols[sym_id] |
| filename = self.files[file_no] |
| if addr < key + size: |
| return symbol, filename, line_no |
| else: |
| return None |
| |
| def symbolize(self, addr, binary, offset): |
| if self.binary != binary: |
| return None |
| res = self.get_sym_file_line(int(offset, 16)) |
| if res: |
| function_name, file_name, line_no = res |
| result = ['%s in %s %s:%d' % ( |
| addr, function_name, file_name, line_no)] |
| print(result) |
| return result |
| else: |
| return None |
| |
| |
| class SymbolizationLoop(object): |
| def __init__(self, plugin_proxy=None, dsym_hint_producer=None): |
| self.plugin_proxy = plugin_proxy |
| if sys.platform == 'win32': |
| # ASan on Windows uses dbghelp.dll to symbolize in-process, which works |
| # even in sandboxed processes. Nothing needs to be done here. |
| self.process_line = self.process_line_echo |
| else: |
| # Used by clients who may want to supply a different binary name. |
| # E.g. in Chrome several binaries may share a single .dSYM. |
| self.dsym_hint_producer = dsym_hint_producer |
| self.system = os.uname()[0] |
| if self.system not in ['Linux', 'Darwin', 'FreeBSD', 'NetBSD','SunOS']: |
| raise Exception('Unknown system') |
| self.llvm_symbolizers = {} |
| self.last_llvm_symbolizer = None |
| self.dsym_hints = set([]) |
| self.frame_no = 0 |
| self.process_line = self.process_line_posix |
| self.using_module_map = plugin_proxy.has_plugin(ModuleMapPlugIn.get_name()) |
| |
| def symbolize_address(self, addr, binary, offset, arch): |
| # On non-Darwin (i.e. on platforms without .dSYM debug info) always use |
| # a single symbolizer binary. |
| # On Darwin, if the dsym hint producer is present: |
| # 1. check whether we've seen this binary already; if so, |
| # use |llvm_symbolizers[binary]|, which has already loaded the debug |
| # info for this binary (might not be the case for |
| # |last_llvm_symbolizer|); |
| # 2. otherwise check if we've seen all the hints for this binary already; |
| # if so, reuse |last_llvm_symbolizer| which has the full set of hints; |
| # 3. otherwise create a new symbolizer and pass all currently known |
| # .dSYM hints to it. |
| result = None |
| if not force_system_symbolizer: |
| if not binary in self.llvm_symbolizers: |
| use_new_symbolizer = True |
| if self.system == 'Darwin' and self.dsym_hint_producer: |
| dsym_hints_for_binary = set(self.dsym_hint_producer(binary)) |
| use_new_symbolizer = bool(dsym_hints_for_binary - self.dsym_hints) |
| self.dsym_hints |= dsym_hints_for_binary |
| if self.last_llvm_symbolizer and not use_new_symbolizer: |
| self.llvm_symbolizers[binary] = self.last_llvm_symbolizer |
| else: |
| self.last_llvm_symbolizer = LLVMSymbolizerFactory( |
| self.system, arch, self.dsym_hints) |
| self.llvm_symbolizers[binary] = self.last_llvm_symbolizer |
| # Use the chain of symbolizers: |
| # Breakpad symbolizer -> LLVM symbolizer -> addr2line/atos |
| # (fall back to next symbolizer if the previous one fails). |
| if not binary in symbolizers: |
| symbolizers[binary] = ChainSymbolizer( |
| [BreakpadSymbolizerFactory(binary), self.llvm_symbolizers[binary]]) |
| result = symbolizers[binary].symbolize(addr, binary, offset) |
| else: |
| symbolizers[binary] = ChainSymbolizer([]) |
| if result is None: |
| if not allow_system_symbolizer: |
| raise Exception('Failed to launch or use llvm-symbolizer.') |
| # Initialize system symbolizer only if other symbolizers failed. |
| symbolizers[binary].append_symbolizer( |
| SystemSymbolizerFactory(self.system, addr, binary, arch)) |
| result = symbolizers[binary].symbolize(addr, binary, offset) |
| # The system symbolizer must produce some result. |
| assert result |
| return result |
| |
| def get_symbolized_lines(self, symbolized_lines, inc_frame_counter=True): |
| if not symbolized_lines: |
| if inc_frame_counter: |
| self.frame_no += 1 |
| return [self.current_line] |
| else: |
| assert inc_frame_counter |
| result = [] |
| for symbolized_frame in symbolized_lines: |
| result.append(' #%s %s' % (str(self.frame_no), symbolized_frame.rstrip())) |
| self.frame_no += 1 |
| return result |
| |
| def process_logfile(self): |
| self.frame_no = 0 |
| for line in logfile: |
| processed = self.process_line(line) |
| print('\n'.join(processed)) |
| |
| def process_line_echo(self, line): |
| return [line.rstrip()] |
| |
| def process_line_posix(self, line): |
| self.current_line = line.rstrip() |
| # Unsymbolicated: |
| # #0 0x7f6e35cf2e45 (/blah/foo.so+0x11fe45) |
| # Partially symbolicated: |
| # #0 0x7f6e35cf2e45 in foo (foo.so+0x11fe45) |
| # NOTE: We have to very liberal with symbol |
| # names in the regex because it could be an |
| # Objective-C or C++ demangled name. |
| stack_trace_line_format = ( |
| '^( *#([0-9]+) *)(0x[0-9a-f]+) *(?:in *.+)? *\((.*)\+(0x[0-9a-f]+)\)') |
| match = re.match(stack_trace_line_format, line) |
| if not match: |
| logging.debug('Line "{}" does not match regex'.format(line)) |
| # Not a frame line so don't increment the frame counter. |
| return self.get_symbolized_lines(None, inc_frame_counter=False) |
| logging.debug(line) |
| _, frameno_str, addr, binary, offset = match.groups() |
| |
| if not self.using_module_map and not os.path.isabs(binary): |
| # Do not try to symbolicate if the binary is just the module file name |
| # and a module map is unavailable. |
| # FIXME(dliew): This is currently necessary for reports on Darwin that are |
| # partially symbolicated by `atos`. |
| return self.get_symbolized_lines(None) |
| arch = "" |
| # Arch can be embedded in the filename, e.g.: "libabc.dylib:x86_64h" |
| colon_pos = binary.rfind(":") |
| if colon_pos != -1: |
| maybe_arch = binary[colon_pos+1:] |
| if is_valid_arch(maybe_arch): |
| arch = maybe_arch |
| binary = binary[0:colon_pos] |
| if arch == "": |
| arch = guess_arch(addr) |
| if frameno_str == '0': |
| # Assume that frame #0 is the first frame of new stack trace. |
| self.frame_no = 0 |
| original_binary = binary |
| binary = self.plugin_proxy.filter_binary_path(binary) |
| if binary is None: |
| # The binary filter has told us this binary can't be symbolized. |
| logging.debug('Skipping symbolication of binary "%s"', original_binary) |
| return self.get_symbolized_lines(None) |
| symbolized_line = self.symbolize_address(addr, binary, offset, arch) |
| if not symbolized_line: |
| if original_binary != binary: |
| symbolized_line = self.symbolize_address(addr, original_binary, offset, arch) |
| return self.get_symbolized_lines(symbolized_line) |
| |
| class AsanSymbolizerPlugInProxy(object): |
| """ |
| Serves several purposes: |
| - Manages the lifetime of plugins (must be used a `with` statement). |
| - Provides interface for calling into plugins from within this script. |
| """ |
| def __init__(self): |
| self._plugins = [ ] |
| self._plugin_names = set() |
| |
| def _load_plugin_from_file_impl_py_gt_2(self, file_path, globals_space): |
| with open(file_path, 'r') as f: |
| exec(f.read(), globals_space, None) |
| |
| def load_plugin_from_file(self, file_path): |
| logging.info('Loading plugins from "{}"'.format(file_path)) |
| globals_space = dict(globals()) |
| # Provide function to register plugins |
| def register_plugin(plugin): |
| logging.info('Registering plugin %s', plugin.get_name()) |
| self.add_plugin(plugin) |
| globals_space['register_plugin'] = register_plugin |
| if sys.version_info.major < 3: |
| execfile(file_path, globals_space, None) |
| else: |
| # Indirection here is to avoid a bug in older Python 2 versions: |
| # `SyntaxError: unqualified exec is not allowed in function ...` |
| self._load_plugin_from_file_impl_py_gt_2(file_path, globals_space) |
| |
| def add_plugin(self, plugin): |
| assert isinstance(plugin, AsanSymbolizerPlugIn) |
| self._plugins.append(plugin) |
| self._plugin_names.add(plugin.get_name()) |
| plugin._receive_proxy(self) |
| |
| def remove_plugin(self, plugin): |
| assert isinstance(plugin, AsanSymbolizerPlugIn) |
| self._plugins.remove(plugin) |
| self._plugin_names.remove(plugin.get_name()) |
| logging.debug('Removing plugin %s', plugin.get_name()) |
| plugin.destroy() |
| |
| def has_plugin(self, name): |
| """ |
| Returns true iff the plugin name is currently |
| being managed by AsanSymbolizerPlugInProxy. |
| """ |
| return name in self._plugin_names |
| |
| def register_cmdline_args(self, parser): |
| plugins = list(self._plugins) |
| for plugin in plugins: |
| plugin.register_cmdline_args(parser) |
| |
| def process_cmdline_args(self, pargs): |
| # Use copy so we can remove items as we iterate. |
| plugins = list(self._plugins) |
| for plugin in plugins: |
| keep = plugin.process_cmdline_args(pargs) |
| assert isinstance(keep, bool) |
| if not keep: |
| self.remove_plugin(plugin) |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, exc_type, exc_val, exc_tb): |
| for plugin in self._plugins: |
| plugin.destroy() |
| # Don't suppress raised exceptions |
| return False |
| |
| def _filter_single_value(self, function_name, input_value): |
| """ |
| Helper for filter style plugin functions. |
| """ |
| new_value = input_value |
| for plugin in self._plugins: |
| result = getattr(plugin, function_name)(new_value) |
| if result is None: |
| return None |
| new_value = result |
| return new_value |
| |
| def filter_binary_path(self, binary_path): |
| """ |
| Consult available plugins to filter the path to a binary |
| to make it suitable for symbolication. |
| |
| Returns `None` if symbolication should not be attempted for this |
| binary. |
| """ |
| return self._filter_single_value('filter_binary_path', binary_path) |
| |
| def filter_module_desc(self, module_desc): |
| """ |
| Consult available plugins to determine the module |
| description suitable for symbolication. |
| |
| Returns `None` if symbolication should not be attempted for this module. |
| """ |
| assert isinstance(module_desc, ModuleDesc) |
| return self._filter_single_value('filter_module_desc', module_desc) |
| |
| class AsanSymbolizerPlugIn(object): |
| """ |
| This is the interface the `asan_symbolize.py` code uses to talk |
| to plugins. |
| """ |
| @classmethod |
| def get_name(cls): |
| """ |
| Returns the name of the plugin. |
| """ |
| return cls.__name__ |
| |
| def _receive_proxy(self, proxy): |
| assert isinstance(proxy, AsanSymbolizerPlugInProxy) |
| self.proxy = proxy |
| |
| def register_cmdline_args(self, parser): |
| """ |
| Hook for registering command line arguments to be |
| consumed in `process_cmdline_args()`. |
| |
| `parser` - Instance of `argparse.ArgumentParser`. |
| """ |
| pass |
| |
| def process_cmdline_args(self, pargs): |
| """ |
| Hook for handling parsed arguments. Implementations |
| should not modify `pargs`. |
| |
| `pargs` - Instance of `argparse.Namespace` containing |
| parsed command line arguments. |
| |
| Return `True` if plug-in should be used, otherwise |
| return `False`. |
| """ |
| return True |
| |
| def destroy(self): |
| """ |
| Hook called when a plugin is about to be destroyed. |
| Implementations should free any allocated resources here. |
| """ |
| pass |
| |
| # Symbolization hooks |
| def filter_binary_path(self, binary_path): |
| """ |
| Given a binary path return a binary path suitable for symbolication. |
| |
| Implementations should return `None` if symbolication of this binary |
| should be skipped. |
| """ |
| return binary_path |
| |
| def filter_module_desc(self, module_desc): |
| """ |
| Given a ModuleDesc object (`module_desc`) return |
| a ModuleDesc suitable for symbolication. |
| |
| Implementations should return `None` if symbolication of this binary |
| should be skipped. |
| """ |
| return module_desc |
| |
| class ModuleDesc(object): |
| def __init__(self, name, arch, start_addr, end_addr, module_path, uuid): |
| self.name = name |
| self.arch = arch |
| self.start_addr = start_addr |
| self.end_addr = end_addr |
| # Module path from an ASan report. |
| self.module_path = module_path |
| # Module for performing symbolization, by default same as above. |
| self.module_path_for_symbolization = module_path |
| self.uuid = uuid |
| assert self.is_valid() |
| |
| def __str__(self): |
| assert self.is_valid() |
| return "{name} {arch} {start_addr:#016x}-{end_addr:#016x} {module_path} {uuid}".format( |
| name=self.name, |
| arch=self.arch, |
| start_addr=self.start_addr, |
| end_addr=self.end_addr, |
| module_path=self.module_path if self.module_path == self.module_path_for_symbolization else '{} ({})'.format(self.module_path_for_symbolization, self.module_path), |
| uuid=self.uuid |
| ) |
| |
| def is_valid(self): |
| if not isinstance(self.name, str): |
| return False |
| if not isinstance(self.arch, str): |
| return False |
| if not isinstance(self.start_addr, int): |
| return False |
| if self.start_addr < 0: |
| return False |
| if not isinstance(self.end_addr, int): |
| return False |
| if self.end_addr <= self.start_addr: |
| return False |
| if not isinstance(self.module_path, str): |
| return False |
| if not os.path.isabs(self.module_path): |
| return False |
| if not isinstance(self.module_path_for_symbolization, str): |
| return False |
| if not os.path.isabs(self.module_path_for_symbolization): |
| return False |
| if not isinstance(self.uuid, str): |
| return False |
| return True |
| |
| class GetUUIDFromBinaryException(Exception): |
| def __init__(self, msg): |
| super(GetUUIDFromBinaryException, self).__init__(msg) |
| |
| _get_uuid_from_binary_cache = dict() |
| |
| def get_uuid_from_binary(path_to_binary, arch=None): |
| cache_key = (path_to_binary, arch) |
| cached_value = _get_uuid_from_binary_cache.get(cache_key) |
| if cached_value: |
| return cached_value |
| if not os.path.exists(path_to_binary): |
| raise GetUUIDFromBinaryException('Binary "{}" does not exist'.format(path_to_binary)) |
| cmd = [ '/usr/bin/otool', '-l'] |
| if arch: |
| cmd.extend(['-arch', arch]) |
| cmd.append(path_to_binary) |
| output = subprocess.check_output(cmd, stderr=subprocess.STDOUT) |
| # Look for this output: |
| # cmd LC_UUID |
| # cmdsize 24 |
| # uuid 4CA778FE-5BF9-3C45-AE59-7DF01B2BE83F |
| if isinstance(output, str): |
| output_str = output |
| else: |
| assert isinstance(output, bytes) |
| output_str = output.decode() |
| assert isinstance(output_str, str) |
| lines = output_str.split('\n') |
| uuid = None |
| for index, line in enumerate(lines): |
| stripped_line = line.strip() |
| if not stripped_line.startswith('cmd LC_UUID'): |
| continue |
| uuid_line = lines[index+2].strip() |
| if not uuid_line.startswith('uuid'): |
| raise GetUUIDFromBinaryException('Malformed output: "{}"'.format(uuid_line)) |
| split_uuid_line = uuid_line.split() |
| uuid = split_uuid_line[1] |
| break |
| if uuid is None: |
| logging.error('Failed to retrieve UUID from binary {}'.format(path_to_binary)) |
| logging.error('otool output was:\n{}'.format(output_str)) |
| raise GetUUIDFromBinaryException('Failed to retrieve UUID from binary "{}"'.format(path_to_binary)) |
| else: |
| # Update cache |
| _get_uuid_from_binary_cache[cache_key] = uuid |
| return uuid |
| |
| class ModuleMap(object): |
| def __init__(self): |
| self._module_name_to_description_map = dict() |
| |
| def add_module(self, desc): |
| assert isinstance(desc, ModuleDesc) |
| assert desc.name not in self._module_name_to_description_map |
| self._module_name_to_description_map[desc.name] = desc |
| |
| def find_module_by_name(self, name): |
| return self._module_name_to_description_map.get(name, None) |
| |
| def __str__(self): |
| s = '{} modules:\n'.format(self.num_modules) |
| for module_desc in sorted(self._module_name_to_description_map.values(), key=lambda v: v.start_addr): |
| s += str(module_desc) + '\n' |
| return s |
| |
| @property |
| def num_modules(self): |
| return len(self._module_name_to_description_map) |
| |
| @property |
| def modules(self): |
| return set(self._module_name_to_description_map.values()) |
| |
| def get_module_path_for_symbolication(self, module_name, proxy, validate_uuid): |
| module_desc = self.find_module_by_name(module_name) |
| if module_desc is None: |
| return None |
| # Allow a plug-in to change the module description to make it |
| # suitable for symbolication or avoid symbolication altogether. |
| module_desc = proxy.filter_module_desc(module_desc) |
| if module_desc is None: |
| return None |
| if validate_uuid: |
| logging.debug('Validating UUID of {}'.format(module_desc.module_path_for_symbolization)) |
| try: |
| uuid = get_uuid_from_binary(module_desc.module_path_for_symbolization, arch = module_desc.arch) |
| if uuid != module_desc.uuid: |
| logging.warning("Detected UUID mismatch {} != {}".format(uuid, module_desc.uuid)) |
| # UUIDs don't match. Tell client to not symbolize this. |
| return None |
| except GetUUIDFromBinaryException as e: |
| logging.error('Failed to get binary from UUID: %s', str(e)) |
| return None |
| else: |
| logging.warning('Skipping validation of UUID of {}'.format(module_desc.module_path_for_symbolization)) |
| return module_desc.module_path_for_symbolization |
| |
| @staticmethod |
| def parse_from_file(module_map_path): |
| if not os.path.exists(module_map_path): |
| raise Exception('module map "{}" does not exist'.format(module_map_path)) |
| with open(module_map_path, 'r') as f: |
| mm = None |
| # E.g. |
| # 0x2db4000-0x102ddc000 /path/to (arm64) <0D6BBDE0-FF90-3680-899D-8E6F9528E04C> |
| hex_regex = lambda name: r'0x(?P<' + name + r'>[0-9a-f]+)' |
| module_path_regex = r'(?P<path>.+)' |
| arch_regex = r'\((?P<arch>.+)\)' |
| uuid_regex = r'<(?P<uuid>[0-9A-Z-]+)>' |
| line_regex = r'^{}-{}\s+{}\s+{}\s+{}'.format( |
| hex_regex('start_addr'), |
| hex_regex('end_addr'), |
| module_path_regex, |
| arch_regex, |
| uuid_regex |
| ) |
| matcher = re.compile(line_regex) |
| line_num = 0 |
| line = 'dummy' |
| while line != '': |
| line = f.readline() |
| line_num += 1 |
| if mm is None: |
| if line.startswith('Process module map:'): |
| mm = ModuleMap() |
| continue |
| if line.startswith('End of module map'): |
| break |
| m_obj = matcher.match(line) |
| if not m_obj: |
| raise Exception('Failed to parse line {} "{}"'.format(line_num, line)) |
| arch = m_obj.group('arch') |
| start_addr = int(m_obj.group('start_addr'), base=16) |
| end_addr = int(m_obj.group('end_addr'), base=16) |
| module_path = m_obj.group('path') |
| uuid = m_obj.group('uuid') |
| module_desc = ModuleDesc( |
| name=os.path.basename(module_path), |
| arch=arch, |
| start_addr=start_addr, |
| end_addr=end_addr, |
| module_path=module_path, |
| uuid=uuid |
| ) |
| mm.add_module(module_desc) |
| if mm is not None: |
| logging.debug('Loaded Module map from "{}":\n{}'.format( |
| f.name, |
| str(mm)) |
| ) |
| return mm |
| |
| class SysRootFilterPlugIn(AsanSymbolizerPlugIn): |
| """ |
| Simple plug-in to add sys root prefix to all binary paths |
| used for symbolication. |
| """ |
| def __init__(self): |
| self.sysroot_path = "" |
| |
| def register_cmdline_args(self, parser): |
| parser.add_argument('-s', dest='sys_root', metavar='SYSROOT', |
| help='set path to sysroot for sanitized binaries') |
| |
| def process_cmdline_args(self, pargs): |
| if pargs.sys_root is None: |
| # Not being used so remove ourselves. |
| return False |
| self.sysroot_path = pargs.sys_root |
| return True |
| |
| def filter_binary_path(self, path): |
| return self.sysroot_path + path |
| |
| class ModuleMapPlugIn(AsanSymbolizerPlugIn): |
| def __init__(self): |
| self._module_map = None |
| self._uuid_validation = True |
| def register_cmdline_args(self, parser): |
| parser.add_argument('--module-map', |
| help='Path to text file containing module map' |
| 'output. See print_module_map ASan option.') |
| parser.add_argument('--skip-uuid-validation', |
| default=False, |
| action='store_true', |
| help='Skips validating UUID of modules using otool.') |
| |
| def process_cmdline_args(self, pargs): |
| if not pargs.module_map: |
| return False |
| self._module_map = ModuleMap.parse_from_file(args.module_map) |
| if self._module_map is None: |
| msg = 'Failed to find module map' |
| logging.error(msg) |
| raise Exception(msg) |
| self._uuid_validation = not pargs.skip_uuid_validation |
| return True |
| |
| def filter_binary_path(self, binary_path): |
| if os.path.isabs(binary_path): |
| # This is a binary path so transform into |
| # a module name |
| module_name = os.path.basename(binary_path) |
| else: |
| module_name = binary_path |
| return self._module_map.get_module_path_for_symbolication( |
| module_name, |
| self.proxy, |
| self._uuid_validation |
| ) |
| |
| def add_logging_args(parser): |
| parser.add_argument('--log-dest', |
| default=None, |
| help='Destination path for script logging (default stderr).', |
| ) |
| parser.add_argument('--log-level', |
| choices=['debug', 'info', 'warning', 'error', 'critical'], |
| default='info', |
| help='Log level for script (default: %(default)s).' |
| ) |
| |
| def setup_logging(): |
| # Set up a parser just for parsing the logging arguments. |
| # This is necessary because logging should be configured before we |
| # perform the main argument parsing. |
| parser = argparse.ArgumentParser(add_help=False) |
| add_logging_args(parser) |
| pargs, unparsed_args = parser.parse_known_args() |
| |
| log_level = getattr(logging, pargs.log_level.upper()) |
| if log_level == logging.DEBUG: |
| log_format = '%(levelname)s: [%(funcName)s() %(filename)s:%(lineno)d] %(message)s' |
| else: |
| log_format = '%(levelname)s: %(message)s' |
| basic_config = { |
| 'level': log_level, |
| 'format': log_format |
| } |
| log_dest = pargs.log_dest |
| if log_dest: |
| basic_config['filename'] = log_dest |
| logging.basicConfig(**basic_config) |
| logging.debug('Logging level set to "{}" and directing output to "{}"'.format( |
| pargs.log_level, |
| 'stderr' if log_dest is None else log_dest) |
| ) |
| return unparsed_args |
| |
| def add_load_plugin_args(parser): |
| parser.add_argument('-p', '--plugins', |
| help='Load plug-in', nargs='+', default=[]) |
| |
| def setup_plugins(plugin_proxy, args): |
| parser = argparse.ArgumentParser(add_help=False) |
| add_load_plugin_args(parser) |
| pargs , unparsed_args = parser.parse_known_args() |
| for plugin_path in pargs.plugins: |
| plugin_proxy.load_plugin_from_file(plugin_path) |
| # Add built-in plugins. |
| plugin_proxy.add_plugin(ModuleMapPlugIn()) |
| plugin_proxy.add_plugin(SysRootFilterPlugIn()) |
| return unparsed_args |
| |
| if __name__ == '__main__': |
| remaining_args = setup_logging() |
| with AsanSymbolizerPlugInProxy() as plugin_proxy: |
| remaining_args = setup_plugins(plugin_proxy, remaining_args) |
| parser = argparse.ArgumentParser( |
| formatter_class=argparse.RawDescriptionHelpFormatter, |
| description='ASan symbolization script', |
| epilog=__doc__) |
| parser.add_argument('path_to_cut', nargs='*', |
| help='pattern to be cut from the result file path ') |
| parser.add_argument('-d','--demangle', action='store_true', |
| help='demangle function names') |
| parser.add_argument('-c', metavar='CROSS_COMPILE', |
| help='set prefix for binutils') |
| parser.add_argument('-l','--logfile', default=sys.stdin, |
| type=argparse.FileType('r'), |
| help='set log file name to parse, default is stdin') |
| parser.add_argument('--force-system-symbolizer', action='store_true', |
| help='don\'t use llvm-symbolizer') |
| # Add logging arguments so that `--help` shows them. |
| add_logging_args(parser) |
| # Add load plugin arguments so that `--help` shows them. |
| add_load_plugin_args(parser) |
| plugin_proxy.register_cmdline_args(parser) |
| args = parser.parse_args(remaining_args) |
| plugin_proxy.process_cmdline_args(args) |
| if args.path_to_cut: |
| fix_filename_patterns = args.path_to_cut |
| if args.demangle: |
| demangle = True |
| if args.c: |
| binutils_prefix = args.c |
| if args.logfile: |
| logfile = args.logfile |
| else: |
| logfile = sys.stdin |
| if args.force_system_symbolizer: |
| force_system_symbolizer = True |
| if force_system_symbolizer: |
| assert(allow_system_symbolizer) |
| loop = SymbolizationLoop(plugin_proxy) |
| loop.process_logfile() |
