# Copyright (c) 2012-2021, The Linux Foundation. All rights reserved. # Copyright (c) 2022-2024, Qualcomm Innovation Center, Inc. All rights reserved. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 and # only version 2 as published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. from __future__ import print_function import sys import re import os import struct import gzip import functools import string import random import platform import stat import subprocess import enum import copy import threading from boards import get_supported_boards, get_supported_ids from tempfile import NamedTemporaryFile import gdbmi from print_out import print_out_str, print_out_exception from mmu import Armv7MMU, Armv7LPAEMMU, Armv8MMU import parser_util import minidump_util import ramreduction_util as elfutil from importlib import import_module import module_table from mm import mm_init from register import Register from collections import namedtuple import shlex SP = 13 LR = 14 PC = 15 # The smem code is very stable and unlikely to go away or be changed. # Rather than go through the hassel of parsing the id through gdb, # just hard code it SMEM_HW_SW_BUILD_ID = 0x89 SMEM_PRIVATE_CANARY = 0xa5a5 PARTITION_MAGIC = 0x54525024 BUILD_ID_LENGTH = 32 primary_types = ["int", "unsigned", "unsigned int", "signed", "signed int", "char", "signed char", "unsigned char", "long unsigned int", "long signed int", "short", "short int", "unsigned short", "unsigned short int", "signed short", "signed short int", "long", "long int", "unsigned long", "signed long", "unsigned long int", "signed long int", "long long", "long long int", "signed long long", "signed long long int", "unsigned long long", "long long unsigned int", "unsigned long long int", "float", "double", "long double", "u8", "u16", "u32", "u64", "bool", "uint32_t", "s8", "s16", "s32", "s64", "uint8_t", "uint16_t", "uint64_t", "__u32", "size_t", "_Bool", "boolean" ] storage_classes = ["static", "volatile", "extern", "register", "auto", "const"] class InvalidDatatype(Exception): """ This exception will be raised when a vaiable of invalid datatype is passed as an argument to a function. """ pass class InvalidInput(Exception): """ This exception will be raised when a vaiable of invalid value is passed as an argument to a function. """ pass class SymbolNotFound(Exception): """ This exception will be raised when an invalid symbol is passed as an argument to a function. """ pass def is_ramdump_file(val, minidump): if not minidump: ddr = re.compile(r'(DDR|EBI|VMDDR)[0-9_CS]+[.]BIN', re.IGNORECASE) imem = re.compile(r'.*IMEM.BIN', re.IGNORECASE) if ddr.match(val) or imem.match(val) and not ("md_" in val): return True else: ddr = re.compile(r'(md_)[0-9_A-Z]+[.]BIN', re.IGNORECASE) if ddr.match(val): return True return False def is_reduceddump_file(val, is_vm): hlos = re.compile(r'MR_HLOS.*[.]ELF', re.IGNORECASE) smem = re.compile(r'MR_smem.*[.]bin', re.IGNORECASE) imem = re.compile(r'.*IMEM.BIN', re.IGNORECASE) memdump = re.compile(r'MR_(mem_dump|linux,cma).*[.]bin', re.IGNORECASE) hyp = re.compile(r'MR_(hyp|hyp_region).*[.]bin', re.IGNORECASE) if hyp.match(val) or smem.match(val) or hlos.match(val) or imem.match(val) or memdump.match(val): return True if is_vm: vm = re.compile(r'.*MR_.*(trustedvm).*[.]bin', re.IGNORECASE) hyp_carve = hyp = re.compile(r'MR_hyp_(smmu|trace)_carveout.*[.]bin', re.IGNORECASE) if vm.match(val) or hyp_carve.match(val): return True return False class AutoDumpInfo(object): priority = 0 def __init__(self, autodumpdir, minidump, reduceddump, svm=None): self.autodumpdir = autodumpdir self.minidump = minidump self.reduceddump = reduceddump self.svm = svm self.ebi_files = [] self.elf_files = [] def parse(self): for (filename, base_addr) in self._parse(): fullpath = os.path.join(self.autodumpdir, filename) if not os.path.exists(fullpath): continue if self.reduceddump and filename.lower().endswith(".elf"): self.elf_files.append(fullpath) else: end = base_addr + os.path.getsize(fullpath) - 1 self.ebi_files.append((open(fullpath, 'rb'), base_addr, end, fullpath)) # sort by addr, DDR files first. The goal is for # self.ebi_files[0] to be the DDR file with the lowest address. self.ebi_files.sort(key=lambda x: (x[1])) def _parse(self): # Implementations should return an interable of (filename, base_addr) raise NotImplementedError class AutoDumpInfoCMM(AutoDumpInfo): # Parses CMM scripts (like load.cmm) def _parse(self): filename = 'load.cmm' if not os.path.exists(os.path.join(self.autodumpdir, filename)): print_out_str('!!! AutoParse could not find load.cmm!') return with open(os.path.join(self.autodumpdir, filename)) as f: for line in f.readlines(): words = line.split() if len(words) == 4 and is_ramdump_file(words[1], self.minidump): fname = words[1] start = int(words[2], 16) yield fname, start class AutoDumpInfoDumpInfoTXT(AutoDumpInfo): # Parses dump_info.txt priority = 1 def _parse(self): filename = 'dump_info.txt' if not os.path.exists(os.path.join(self.autodumpdir, filename)): print_out_str('!!! AutoParse could not find dump_info.txt!') return with open(os.path.join(self.autodumpdir, filename)) as f: try: for line in f.readlines(): words = line.split() if not words or not is_ramdump_file(words[-1], self.minidump): continue fname = words[-1] start = int(words[1], 16) size = int(words[2]) filesize = os.path.getsize( os.path.join(self.autodumpdir, fname)) if size != filesize: print_out_str( ("!!! Size of %s on disk (%d) doesn't match size " + "from dump_info.txt (%d). Skipping...") % (fname, filesize, size)) continue yield fname, start except: print_out_str('!!! Cannot parse dump_info.txt due to improper format!') return class AutoDumpInfoReducedDump(AutoDumpInfo): # Parses binoffsets.txt, dump_info.txt # Finds HLOS elf file priority = 1 def _parse(self): filename = 'binoffsets.txt' added = [] if not os.path.exists(os.path.join(self.autodumpdir, filename)): print_out_str('!!! AutoParse could not find {}!'.format(filename)) return # Find smem, hyp, memdump file with open(os.path.join(self.autodumpdir, filename)) as f: for line in f.readlines(): words = line.split() if not words or not is_reduceddump_file(words[0], self.svm): continue fname = words[0].strip() start = int(words[1], 16) size = int(words[2], 16) added.append(fname) yield fname, start # Find HLOS elf's hlos = re.compile(r'.*MR_HLOS.*[.]ELF', re.IGNORECASE) for file in os.listdir(self.autodumpdir): if hlos.match(file): yield file, None # Find *IMEM bins filename = 'load.cmm' if not os.path.exists(os.path.join(self.autodumpdir, filename)): print_out_str('!!! AutoParse could not find load.cmm!') return with open(os.path.join(self.autodumpdir, filename)) as f: for line in f.readlines(): words = line.split() if len(words) != 4 or not is_reduceddump_file(words[1], self.svm): continue fname = words[1] if fname in added: continue start = int(words[2], 16) yield fname, start class AutoDumpInfodram_cs(AutoDumpInfo): # Parses dump_info.txt priority = 2 def _parse(self): if not os.path.exists(self.autodumpdir): print_out_str('!!! AutoParse could not find path {}!'.format(self.autodumpdir)) return filename_lst = os.listdir(self.autodumpdir) regex = re.compile(r'^(dram_cs|ocimem)\S*(0x[A-Fa-f0-9]+)\S+(0x[A-Fa-f0-9]+)') for filename in filename_lst: m = re.search(regex, filename) if m: start = int(m.group(2), 16) end = int(m.group(3), 16) filesize = os.path.getsize( os.path.join(self.autodumpdir, filename)) if end - start + 1 != filesize: print_out_str( ("!!! Size of %s on disk (%d) doesn't match size " + "from dump_info.txt (%d). Skipping...") % (filename, filesize, end - start + 1)) continue yield filename, start class RamDump(): """The main interface to the RAM dump""" class Unwinder (): class Stackframe (): def __init__(self, fp, sp, lr, pc): self.fp = fp self.sp = sp self.lr = lr self.pc = pc class UnwindCtrlBlock (): def __init__(self): self.vrs = 16 * [0] self.insn = 0 self.entries = -1 self.byte = -1 self.index = 0 def __init__(self, ramdump): start = ramdump.address_of('__start_unwind_idx') end = ramdump.address_of('__stop_unwind_idx') self.ramdump = ramdump if (start is None) or (end is None): if ramdump.arm64: self.unwind_frame = self.unwind_frame_generic64 else: self.unwind_frame = self.unwind_frame_generic return None # addresses self.unwind_frame = self.unwind_frame_tables self.start_idx = start self.stop_idx = end self.unwind_table = [] i = 0 for addr in range(start, end, 8): r = ramdump.read_string(addr, '> 1) if (self.unwind_table[mid][0] >= 0x40000000): start = mid + 1 else: stop = mid return stop def unwind_frame_generic64(self, frame, cpu_work_state=''): fp = frame.fp try: frame.sp = fp + 0x10 frame.fp = self.ramdump.read_word(fp) frame.pc = self.ramdump.read_word(fp + 8) if ((frame.fp == 0 and frame.pc == 0) or frame.pc is None or frame.lr is None): return -1 except: return -1 return 0 def unwind_frame_generic(self, frame, cpu_work_state=''): high = 0 fp = frame.fp low = frame.sp mask = (self.ramdump.thread_size) - 1 high = (low + mask) & (~mask) # ALIGN(low, THREAD_SIZE) # /* check current frame pointer is within bounds */ if (fp < (low + 12) or fp + 4 >= high): return -1 fp_is_at = self.ramdump.read_word(frame.fp - 12) sp_is_at = self.ramdump.read_word(frame.fp - 8) pc_is_at = self.ramdump.read_word(frame.fp - 4) frame.fp = fp_is_at frame.sp = sp_is_at frame.pc = pc_is_at return 0 def walk_stackframe_generic(self, frame): while True: symname = self.ramdump.addr_to_symbol(frame.pc) print_out_str(symname) ret = self.unwind_frame_generic(frame) if ret < 0: break def unwind_backtrace_generic(self, sp, fp, pc): frame = self.Stackframe() frame.fp = fp frame.pc = pc frame.sp = sp walk_stackframe_generic(frame) def search_idx_3_4(self, addr): start = 0 stop = len(self.unwind_table) orig = addr if (addr < self.start_idx): stop = self.origin else: start = self.origin if (start >= stop): return None addr = (addr - self.unwind_table[start][2]) & 0x7fffffff while (start < (stop - 1)): mid = start + ((stop - start) >> 1) dif = (self.unwind_table[mid][2] - self.unwind_table[start][2]) if ((addr - dif) < self.unwind_table[mid][0]): stop = mid else: addr = addr - dif start = mid if self.unwind_table[start][0] <= addr: return self.unwind_table[start] else: return None def search_idx_3_0(self, addr): first = 0 last = len(self.unwind_table) while (first < last - 1): mid = first + ((last - first + 1) >> 1) if (addr < self.unwind_table[mid][0]): last = mid else: first = mid return self.unwind_table[first] def unwind_get_byte(self, ctrl): if (ctrl.entries <= 0): print_out_str('unwind: Corrupt unwind table') return 0 val = self.ramdump.read_word(ctrl.insn) ret = (val >> (ctrl.byte * 8)) & 0xff if (ctrl.byte == 0): ctrl.insn += 4 ctrl.entries -= 1 ctrl.byte = 3 else: ctrl.byte -= 1 return ret def unwind_exec_insn(self, ctrl): insn = self.unwind_get_byte(ctrl) if ((insn & 0xc0) == 0x00): ctrl.vrs[SP] += ((insn & 0x3f) << 2) + 4 elif ((insn & 0xc0) == 0x40): ctrl.vrs[SP] -= ((insn & 0x3f) << 2) + 4 elif ((insn & 0xf0) == 0x80): vsp = ctrl.vrs[SP] reg = 4 insn = (insn << 8) | self.unwind_get_byte(ctrl) mask = insn & 0x0fff if (mask == 0): print_out_str("unwind: 'Refuse to unwind' instruction") return -1 # pop R4-R15 according to mask */ load_sp = mask & (1 << (13 - 4)) while (mask): if (mask & 1): ctrl.vrs[reg] = self.ramdump.read_word(vsp) if ctrl.vrs[reg] is None: return -1 vsp += 4 mask >>= 1 reg += 1 if not load_sp: ctrl.vrs[SP] = vsp elif ((insn & 0xf0) == 0x90 and (insn & 0x0d) != 0x0d): ctrl.vrs[SP] = ctrl.vrs[insn & 0x0f] elif ((insn & 0xf0) == 0xa0): vsp = ctrl.vrs[SP] a = list(range(4, 4 + (insn & 7))) a.append(4 + (insn & 7)) # pop R4-R[4+bbb] */ for reg in (a): ctrl.vrs[reg] = self.ramdump.read_word(vsp) if ctrl.vrs[reg] is None: return -1 vsp += 4 if (insn & 0x80): ctrl.vrs[14] = self.ramdump.read_word(vsp) if ctrl.vrs[14] is None: return -1 vsp += 4 ctrl.vrs[SP] = vsp elif (insn == 0xb0): if (ctrl.vrs[PC] == 0): ctrl.vrs[PC] = ctrl.vrs[LR] ctrl.entries = 0 elif (insn == 0xb1): mask = self.unwind_get_byte(ctrl) vsp = ctrl.vrs[SP] reg = 0 if (mask == 0 or mask & 0xf0): print_out_str('unwind: Spare encoding') return -1 # pop R0-R3 according to mask while mask: if (mask & 1): ctrl.vrs[reg] = self.ramdump.read_word(vsp) if ctrl.vrs[reg] is None: return -1 vsp += 4 mask >>= 1 reg += 1 ctrl.vrs[SP] = vsp elif (insn == 0xb2): uleb128 = self.unwind_get_byte(ctrl) ctrl.vrs[SP] += 0x204 + (uleb128 << 2) else: print_out_str('unwind: Unhandled instruction') return -1 return 0 def prel31_to_addr(self, addr): value = self.ramdump.read_word(addr) # offset = (value << 1) >> 1 # C wants this sign extended. Python doesn't do that. # Sign extend manually. if (value & 0x40000000): offset = value | 0x80000000 else: offset = value # This addition relies on integer overflow # Emulate this behavior temp = addr + offset return (temp & 0xffffffff) + ((temp >> 32) & 0xffffffff) def unwind_frame_tables(self, frame, cpu_work_state): low = frame.sp high = ((low + (self.ramdump.thread_size - 1)) & \ ~(self.ramdump.thread_size - 1)) + self.ramdump.thread_size idx = self.search_idx(frame.pc) if (idx is None): return -1 if cpu_work_state == "thumb": FP = 7 else: FP = 11 ctrl = self.UnwindCtrlBlock() ctrl.vrs[FP] = frame.fp ctrl.vrs[SP] = frame.sp ctrl.vrs[LR] = frame.lr ctrl.vrs[PC] = 0 if (idx[1] == 1): return -1 elif ((idx[1] & 0x80000000) == 0): ctrl.insn = self.prel31_to_addr(idx[2] + 4) elif (idx[1] & 0xff000000) == 0x80000000: ctrl.insn = idx[2] + 4 else: print_out_str('not supported') return -1 val = self.ramdump.read_word(ctrl.insn) if ((val & 0xff000000) == 0x80000000): ctrl.byte = 2 ctrl.entries = 1 elif ((val & 0xff000000) == 0x81000000): ctrl.byte = 1 ctrl.entries = 1 + ((val & 0x00ff0000) >> 16) else: return -1 while (ctrl.entries > 0): urc = self.unwind_exec_insn(ctrl) if (urc < 0): return urc if (ctrl.vrs[SP] < low or ctrl.vrs[SP] >= high): return -1 if (ctrl.vrs[PC] == 0): ctrl.vrs[PC] = ctrl.vrs[LR] # check for infinite loop */ if (frame.pc == ctrl.vrs[PC]): return -1 frame.fp = ctrl.vrs[FP] frame.sp = ctrl.vrs[SP] frame.lr = ctrl.vrs[LR] frame.pc = ctrl.vrs[PC] return 0 def pac_frame_update(self,frame): frame.fp = self.ramdump.pac_ignore(frame.fp) frame.sp = self.ramdump.pac_ignore(frame.sp) frame.lr = self.ramdump.pac_ignore(frame.lr) frame.pc = self.ramdump.pac_ignore(frame.pc) def unwind_backtrace(self, sp, fp, pc, lr, extra_str='', out_file=None): offset = 0 max_frames = 128 frame_count = 0 frame = self.Stackframe(fp, sp, lr, pc) frame.fp = fp frame.sp = sp frame.lr = lr frame.pc = pc self.pac_frame_update(frame) backtrace = '\n' cpu_work_state = '' if (pc & 0x1) or (lr & 0x1): cpu_work_state = 'thumb' while True: where = frame.pc offset = 0 if frame.pc is None: break r = self.ramdump.unwind_lookup(frame.pc) if r is None: symname = 'UNKNOWN' offset = 0x0 else: symname, offset = r pstring = ( extra_str + '[<{0:x}>] {1}+0x{2:x}'.format(frame.pc, symname, offset)) if out_file: out_file.write(pstring + '\n') else: print_out_str(pstring) backtrace += pstring + '\n' urc = self.unwind_frame(frame, cpu_work_state) self.pac_frame_update(frame) if urc < 0: break frame_count = frame_count + 1 if frame_count >= max_frames: if out_file != None: out_file.write("Max stack depth reached") break return backtrace def createMask(self,a, b): r = 0 i = a while(i <= b): r |= 1 << i i = i +1 return r; def pac_ignore(self,data): pac_check = self.createMask(self.va_bits, 63) top_bit_ignore = 0xff00000000000000 if data is None or not self.arm64: return data if (data & pac_check) == pac_check or (data & pac_check) == 0: return data # When address tagging is used # The PAC field is Xn[54:bottom_PAC_bit]. # In the PAC field definitions, bottom_PAC_bit == 64-TCR_ELx.TnSZ, # TCR_ELx.TnSZ is set to 25. so 64-25=39 pac_mack = self.createMask(self.va_bits, 54) result = pac_mack | data result = result | top_bit_ignore return result def load_phys_range(self, path): phys_base, phys_end = 0xffffffff, 0 with open(path, 'r') as _fd: phys_base, phys_end = _fd.read().strip().split("--") return int(phys_base), int(phys_end) def get_kimage_vaddr(self, need_aslr=True): kimage_vaddr = None try: kimage_vaddr = self.address_of('_text') if need_aslr: kimage_vaddr -= self.kaslr_offset except: if self.get_kernel_version() > (4, 20, 0): if self.get_kernel_version() >= (6, 4, 0): modules_vsize = 0x80000000 else: modules_vsize = 0x08000000 bpf_jit_vsize = 0x08000000 self.page_end = (0xffffffffffffffff << ( self.va_bits - 1)) & 0xffffffffffffffff if self.address_of("kasan_init") is None: self.kasan_shadow_size = 0 else: if self.is_config_defined("CONFIG_KASAN_SW_TAGS"): self.kasan_shadow_size = 1 << (self.va_bits - 4) else: self.kasan_shadow_size = 1 << (self.va_bits - 3) kimage_vaddr = self.page_end + modules_vsize if self.get_kernel_version() < (5, 10, 0): kimage_vaddr += bpf_jit_vsize # new since v5.11: https://lore.kernel.org/all/20201008153602.9467-3-ardb@kernel.org/ # The KASAN shadow region is reconfigured so that it ends at the start of # the vmalloc region, and grows downwards. That way, the arrangement of # the vmalloc space (which contains kernel mappings, modules, BPF region, # the vmemmap array etc) is identical between non-KASAN and KASAN builds, # which aids debugging. if self.get_kernel_version() < (5, 11, 0): kimage_vaddr = kimage_vaddr + self.kasan_shadow_size else: modules_vsize = 0x08000000 self.va_start = (0xffffffffffffffff << self.va_bits) & 0xffffffffffffffff if self.address_of("kasan_init") is None: self.kasan_shadow_size = 0 else: self.kasan_shadow_size = 1 << (self.va_bits - 3) kimage_vaddr = self.va_start + self.kasan_shadow_size + \ modules_vsize return kimage_vaddr def __init__(self, options, nm_path, gdb_path, objdump_path,gdb_ndk_path): self.ebi_files = [] ## used for read_physical in multi-thread mode self.use_multithread = False self.thread_maxcount = 0x8 self.ebi_files_mappings = {} self.thread_name_prefix = "ThreadPoolExecutor-0" ## self.ebi_files_minidump = [] self.ebi_pa_name_map = {} self.md_dict = {} self.phys_offset = None self.kaslr_offset = options.kaslr_offset self.tz_start = 0 self.ebi_start = 0 self.hyp_diag_addr = None self.rm_debug_addr = None self.cpu_type = None self.tbi_mask = None self.svm_kaslr_offset = None self.iommu_pg_table_format = options.iommu_pg_table_format self.hw_id = options.force_hardware or None self.hw_version = options.force_hardware_version or None self.offset_table = [] self.vmlinux = options.vmlinux self.nm_path = nm_path self.gdb_path = gdb_path self.gdb_ndk_path = gdb_ndk_path self.objdump_path = objdump_path self.outdir = options.outdir self.ftrace_args = options.ftrace_args self.ftrace_max_size = options.ftrace_max_size self.imem_fname = None self.gdbmi = None self.gdbmi_hyp = None self.arm64 = options.arm64 self.logcat_limit_time = options.logcat_limit_time self.ftrace_limit_time = options.ftrace_limit_time self.ndk_compatible = False self.lookup_table = [] self.ko_file_names = [] self.datatype_dict = {} self.enum_data = {} self.available_cores = [] self.skip_TLB_Cache_parse = options.skip_TLB_Cache_parse self.module_layout_dict = {} if gdb_ndk_path: self.gdbmi = gdbmi.GdbMI(self.gdb_ndk_path, self.vmlinux, 0) self.gdbmi.open() sanity_data = self.address_of("kimage_voffset") self.kernel_version = (0, 0, 0) if self.arm64: self.gdbmi.setup_aarch('aarch64') if (sanity_data and (sanity_data & 0xFF000000000000) == 0): print_out_str('RELR tags not compatible with NDK GDB') elif sanity_data is not None and self.get_kernel_version() >= (5, 10): print_out_str('vmlinux is ndk-compatible') self.ndk_compatible = True if not self.ndk_compatible: print_out_str("vmlinux not ndk compatible\n") self.gdbmi.close() if not self.ndk_compatible: self.gdbmi = gdbmi.GdbMI(self.gdb_path, self.vmlinux, 0) self.gdbmi.open() if self.arm64: self.gdbmi.setup_aarch('aarch64') self.gdbmi.set_gdbmi_aslr_offset() self.page_offset = 0xc0000000 self.thread_size = 8192 self.qtf_path = options.qtf_path self.ftrace_format = options.ftrace_format self.skip_qdss_bin = options.skip_qdss_bin self.debug = options.debug self.dcc = False self.sysreg = False self.t32_host_system = options.t32_host_system or None self.use_stdout = options.stdout self.kernel_version = (0, 0, 0) self.linux_banner = None self.minidump = options.minidump self.reduceddump = options.reduceddump self.svm = options.svm self.elffile = None self.ram_elf_file = None self.ram_addr = options.ram_addr self.autodump = options.autodump self.elf_addr = None self.module_table = module_table.module_table_class() self.hyp = options.hyp # Save all paths given from --mod_path option. These will be searched for .ko.unstripped files if options.mod_path_list: for path in options.mod_path_list: self.module_table.add_sym_path(path) self.dump_module_symbol_table = options.dump_module_symbol_table self.dump_kernel_symbol_table = options.dump_kernel_symbol_table self.dump_module_kallsyms = options.dump_module_kallsyms self.dump_global_symbol_table = options.dump_global_symbol_table self.currentEL = options.currentEL or None self.hyp_dump = None self.ttbr = None self.vttbr = None self.hlos_tcr_el1 = None self.hlos_sctlr_el1 = None if self.hyp: self.gdbmi_hyp = gdbmi.GdbMI(self.gdb_path, self.hyp, 0) self.gdbmi_hyp.open() if self.minidump or self.reduceddump: try: mod = import_module('elftools.elf.elffile') ELFFile = mod.ELFFile StringTableSection = mod.StringTableSection mod = import_module('elftools.common.py3compat') bytes2str = mod.bytes2str except ImportError: print("Oops, missing required library for minidump. Check README") sys.exit(1) if options.ram_addr is not None: # TODO sanity check to make sure the memory regions don't overlap for file_path, start, end in options.ram_addr: fd = open(file_path, 'rb') if not fd: print_out_str( 'Could not open {0}. Will not be part of dump'.format(file_path)) continue self.ebi_files.append((fd, start, end, file_path)) elif not options.reduceddump: if not self.auto_parse(options.autodump, options.minidump, options.svm): print("Oops, auto-parse option failed. Please specify vmlinux & DDR files manually.") sys.exit(1) elif options.reduceddump: if not self.auto_parse(options.autodump, options.minidump, options.svm): print("Oops, auto-parse option failed. Please specify vmlinux & HLOS elf files manually.") sys.exit(1) if self.elf_addr is not None: # Setup the needed vector and hash table for binary search in read_physical vector, htable, filemap = elfutil.setup_elfmappings(self.elf_addr) self.elf_vector, self.elf_htable, self.elf_filemap = vector, htable, filemap if not self.has_debug_info(self.vmlinux): print('!!! Your vmlinux does not have debug info.') print('!!! Exiting now') sys.exit(1) if options.minidump: if not options.autodump: file_path = options.ram_elf_addr else: file_path = os.path.join(options.outdir, 'ap_minidump.elf') self.ram_elf_file = file_path if not os.path.exists(file_path): print_out_str("ELF file not exists, try to generate") if minidump_util.generate_elf(options.autodump, options.outdir, self.svm, self.get_kernel_version()): print_out_str("!!! ELF file generate failed") sys.exit(1) fd = open(file_path, 'rb') self.elffile = ELFFile(fd) for idx, s in enumerate(self.elffile.iter_segments()): pa = int(s['p_paddr']) va = int(s['p_vaddr']) size = int(s['p_filesz']) end_addr = pa + size - 1 for section in self.elffile.iter_sections(): if (not section.is_null() and s.section_in_segment(section)): self.ebi_pa_name_map[pa] = section.name if section.name == "KVA_DUMP": kva_dump_addr = pa self.ebi_files_minidump.append((idx, pa, end_addr, va,size)) if options.autodump and os.path.exists(os.path.join(options.autodump, "md_KVA_DUMP.BIN")): file_path = os.path.join(options.autodump, "md_KVA_DUMP.BIN") fd = open(file_path, 'rb') kva_elf = ELFFile(fd) for s in kva_elf.iter_sections(): start = int(s.header['sh_addr']) size = int(s.header['sh_size']) offset = int(s.header['sh_offset']) pa = kva_dump_addr + offset end_addr = pa + size - 1 if start == 0x0: continue self.ebi_files_minidump.append((idx, pa, end_addr, start, size)) if s.name not in self.md_dict.keys(): self.md_dict[s.name] = [[start,size]] else: self.md_dict[s.name].append([start,size]) if options.minidump: if self.ebi_start == 0: self.ebi_start = self.ebi_files_minidump[0][1] elif options.reduceddump: if self.ebi_start == 0: # options.elf_addr needs to be sorted for filename self.ebi_start = self.elf_filemap[options.elf_addr[0]][0] else: if self.ebi_start == 0: self.ebi_start = self.ebi_files[0][1] if self.phys_offset is None: self.get_hw_id() if options.phys_offset is not None: print_out_str( '[!!!] Phys offset was set to {0:x}'.format(\ options.phys_offset)) self.phys_offset = options.phys_offset self.s2_walk = False if self.svm and not self.minidump: from extensions.hyp_trace import HypDump hyp_dump = HypDump(self) hyp_dump.determine_kaslr() self.gdbmi_hyp.kaslr_offset = hyp_dump.hyp_kaslr_addr_offset hyp_dump.get_trace_phy() if hyp_dump.ttbr1 is None: print_out_str('!!! Could not find {}'.format(self.svm)) print_out_str('!!! Exiting now') sys.exit(1) self.ttbr = hyp_dump.ttbr1 self.vttbr = hyp_dump.vttbr self.TTBR0_EL1 = hyp_dump.TTBR0_EL1 self.SCTLR_EL1 = hyp_dump.SCTLR_EL1 self.TCR_EL1 = hyp_dump.TCR_EL1 self.VTCR_EL2 = hyp_dump.VTCR_EL2 self.HCR_EL2 = hyp_dump.HCR_EL2 self.ttbr_data = hyp_dump.ttbr1_data_info self.vttbr_data = hyp_dump.vttbr_el2_data self.s2_walk = True self.config = [] self.config_dict = {} if not self.get_config(): print_out_str('!!! Could not get saved configuration') print_out_str( '!!! This is really bad and probably indicates RAM corruption') print_out_str('!!! Some features may be disabled!') # extract kernel's configuration to kconfig.txt saved_config = self.open_file('kconfig.txt') for l in self.config: saved_config.write(l + '\n') saved_config.close() try: self.va_bits = int(self.get_config_val("CONFIG_ARM64_VA_BITS")) except: if self.arm64: self.va_bits = 39 else: # for arm32 self.va_bits = 32 try: self.page_shift = int(self.get_config_val("CONFIG_ARM64_PAGE_SHIFT")) except: self.page_shift = 12 try: self.pgtable_levels = int(self.get_config_val("CONFIG_PGTABLE_LEVELS")) except: self.pgtable_levels = 3 self.pfn_range = None self.vmemmap = None ''' determine kaslr_offset, phys_offset and kimage_voffset @start ''' self.thread_maxcount = len(self.ebi_files) # value is None in ARM32 self.__kimage_vaddr_var_va = self.address_of('kimage_vaddr') # Virtual address of the variable 'kimage_voffset' # value is None in ARM32 before kernel version 5.4 self.__kimage_voffset_var_va = self.address_of('kimage_voffset') # virtual address of start of kernel image self.__kimage_vaddr_va = self.get_kimage_vaddr(need_aslr=False) # address of linux_banner variable self.__linux_banner_va = self.address_of('linux_banner') self.kimage_voffset = None print_out_str(f"kimage_vaddr is: 0x{self.__kimage_vaddr_va:x}") # determine kaslr_offset and kimage_voffset here self.determine_kaslr_offset() self.gdbmi.kaslr_offset = self.kaslr_offset ''' determine kaslr_offset, phys_offset and kimage_voffset @end ''' self.wlan = options.wlan if self.arm64: if self.get_kernel_version() >= (5, 4): self.page_offset = -(1 << self.va_bits) % (1 << 64) if self.address_of('__start_init_task') is not None: self.thread_size = self.address_of('__end_init_task') - self.address_of('__start_init_task') else: self.thread_size = self.address_of('__end_init_stack') - self.address_of('__start_init_stack') else: self.page_offset = 0xffffffc000000000 self.thread_size = 16384 if options.page_offset is not None: print_out_str( '[!!!] Page offset was set to {0:x}'.format(options.page_offset)) self.page_offset = options.page_offset self.setup_symbol_tables() print_out_str("Kernel version vmlinux: {0}".format(self.kernel_version)) self.field_offset("struct trace_entry", "preempt_count") self.kimage_vaddr = self.__kimage_vaddr_va + self.get_kaslr_offset() print_out_str(f"kimage_vaddr with kaslr is: 0x{self.kimage_vaddr:x}") # The address of swapper_pg_dir can be used to determine # whether or not we're running with LPAE enabled since an # extra 4k is needed for LPAE. If it's 0x5000 below # PAGE_OFFSET + TEXT_OFFSET then we know we're using LPAE. For # non-LPAE it should be 0x4000 below PAGE_OFFSET + TEXT_OFFSET self.swapper_pg_dir_addr = self.address_of('swapper_pg_dir') if self.swapper_pg_dir_addr is None: print_out_str('!!! Could not get the swapper page directory!') if not self.minidump: print_out_str( '!!! Your vmlinux is probably wrong for these dumps') print_out_str('!!! Exiting now') sys.exit(1) if self.get_kernel_version() > (5, 7, 0) and self.arm64: stext = self.address_of('primary_entry') else: stext = self.address_of('stext') if self.kimage_voffset is None or not self.arm64: self.kernel_text_offset = stext - self.page_offset else: self.kernel_text_offset = stext - self.kimage_vaddr pg_dir_size = self.kernel_text_offset + self.page_offset \ - self.swapper_pg_dir_addr if self.arm64: print_out_str('Using 64bit MMU') self.mmu = Armv8MMU(self) elif pg_dir_size == 0x4000: print_out_str('Using non-LPAE MMU') if self.minidump: self.mmu = None else: self.mmu = Armv7MMU(self) elif pg_dir_size == 0x5000: print_out_str('Using LPAE MMU') text_offset = 0x8000 pg_dir_size = 0x5000 # 0x4000 for non-LPAE swapper_pg_dir_addr = self.phys_offset + text_offset - pg_dir_size # We deduce ttbr1 and ttbcr.t1sz based on the value of # PAGE_OFFSET. This is based on v7_ttb_setup in # arch/arm/mm/proc-v7-3level.S: # * TTBR0/TTBR1 split (PAGE_OFFSET): # * 0x40000000: T0SZ = 2, T1SZ = 0 (not used) # * 0x80000000: T0SZ = 0, T1SZ = 1 # * 0xc0000000: T0SZ = 0, T1SZ = 2 if self.page_offset == 0x40000000: t1sz = 0 elif self.page_offset == 0x80000000: t1sz = 1 elif self.page_offset == 0xc0000000: t1sz = 2 # need to fixup ttbr1 since we'll be skipping the # first-level lookup (see v7_ttb_setup): # /* PAGE_OFFSET == 0xc0000000, T1SZ == 2 */ # add \ttbr1, \ttbr1, #4096 * (1 + 3) @ only L2 used, skip # pgd+3*pmd swapper_pg_dir_addr += (4096 * (1 + 3)) else: raise Exception( 'Invalid phys_offset for page_table_walk: 0x%x' % self.page_offset) self.mmu = Armv7LPAEMMU(self, swapper_pg_dir_addr, t1sz) else: print_out_str( "!!! Couldn't determine whether or not we're using LPAE!") print_out_str( '!!! This is a BUG in the parser and should be reported.') sys.exit(1) if not self.match_version(): print_out_str('!!! Could not get the Linux version!') print_out_str( '!!! Your vmlinux is probably wrong for these dumps') print_out_str('!!! Exiting now') sys.exit(1) self.unwind = self.Unwinder(self) if self.module_table.sym_paths_exist(): self.setup_module_symbols() self.gdbmi.setup_module_table(self.module_table) if self.dump_global_symbol_table: self.dump_global_symbol_lookup_table() if not self.minidump: self.setup_module_layout() mm_init(self) self.set_available_cores() self.arm_smmu_v12 = self.is_arm_smmu_v12() def get_section_address(self,section): """ Function to return address and size corresponding to the section name in elf files. :param section: name of the section. :type addr: str :return: A list of list of addresses and sizes corresponding to the section name. """ res_dict = self.md_dict if (section in res_dict.keys()): return res_dict[section] else: raise InvalidInput def __del__(self): self.gdbmi.close() if self.hyp: self.gdbmi_hyp.close() def open_file(self, file_name, mode='wt'): """Open a file in the out directory. Example: >>> with self.ramdump.open_file('pizza.txt') as p: p.write('Pizza is the best\\n') """ file_path = os.path.join(self.outdir, file_name) f = None try: dir_path = os.path.dirname(file_path) if not os.path.exists(dir_path) and ('w' in mode or 'a' in mode): os.makedirs(dir_path) f = open(file_path, mode) except: print_out_str('Could not open path {0}'.format(file_path)) print_out_str('Do you have write/read permissions on the path?') sys.exit(1) return f def chmod(self, file_name, mode): file_path = os.path.join(self.outdir, file_name) return os.chmod(file_path, mode) def remove_file(self, file_name): file_path = os.path.join(self.outdir, file_name) try: if (os.path.exists(file_path)): os.remove(file_path) except: print_out_str('Could not remove file {0}'.format(file_path)) print_out_str('Do you have write/read permissions on the path?') sys.exit(1) def get_srcdir(self): """ Returns absolute path of directory containing ramdump.py """ return os.path.dirname(os.path.abspath(__file__)) def get_config(self): kconfig_addr = self.address_of('kernel_config_data') if kconfig_addr is None: return if self.get_kernel_version() > (5, 0, 0): kconfig_addr_end = self.address_of('kernel_config_data_end') if kconfig_addr_end is None: return kconfig_size = kconfig_addr_end - kconfig_addr # magic is 8 bytes before kconfig_addr and data # starts at kconfig_addr for kernel > 5.0.0 kconfig_addr = kconfig_addr - 8 else: kconfig_size = self.sizeof('kernel_config_data') # size includes magic, offset from it kconfig_size = kconfig_size - 16 - 1 # kconfig data starts with magic 8 byte string, go past that zconfig = os.path.join(self.outdir, "elf_temp.txt") temp_file = open(zconfig, 'wb+') size = kconfig_addr + 8 s = self.read_elf_memory(kconfig_addr, size, temp_file) temp_file.close() if s != 'IKCFG_ST': return temp_file = open(zconfig, 'wb+') kconfig_addr = kconfig_addr + 8 val = self.read_elf_memory(kconfig_addr, kconfig_size + kconfig_addr, temp_file) temp_file.close() zconfig_in = gzip.open(temp_file.name, 'rt') try: t = zconfig_in.readlines() except: return False zconfig_in.close() os.remove(zconfig) for l in t: self.config.append(l.rstrip()) if not l.startswith('#') and l.strip() != '': eql = l.find('=') cfg = l[:eql] val = l[eql+1:] self.config_dict[cfg] = val.strip() return True def get_config_val(self, config): """Gets the value of a kernel config option. Example: >>> va_bits = int(dump.get_config_val("CONFIG_ARM64_VA_BITS")) 39 """ return self.config_dict.get(config) def is_config_defined(self, config): return config in self.config_dict def get_kernel_version(self): if self.kernel_version == (0, 0, 0): vm_v = self.gdbmi.get_value_of_string('linux_banner') if vm_v is None: print_out_str('!!! Could not read linux_banner from vmlinux!') sys.exit(1) v = re.search('Linux version (\d{0,2}\.\d{0,2}\.\d{0,3})', vm_v) if v is None: print_out_str('!!! Could not extract version info!') sys.exit(1) self.version = v.group(1) match = re.search('(\d+)\.(\d+)\.(\d+)', self.version) if match is not None: self.version = tuple(map(int, match.groups())) self.kernel_version = self.version self.linux_banner = vm_v else: print_out_str('!!! Could not extract version info! {0}'.format(self.version)) sys.exit(1) return self.kernel_version def kernel_virt_to_phys(self, addr): if self.minidump: return minidump_util.minidump_virt_to_phys(self.ebi_files_minidump,addr) else: if self.kimage_voffset is None: return addr - self.page_offset + self.phys_offset else: if self.kernel_version > (4, 20, 0): if not (addr & (1 << (self.va_bits - 1))): return addr - self.page_offset + self.phys_offset else: return addr - (self.kimage_voffset) else: if addr & (1 << (self.va_bits - 1)): return addr - self.page_offset + self.phys_offset else: return addr - (self.kimage_voffset) def match_version(self): banner_addr = self.address_of('linux_banner') if banner_addr is not None: banner_addr = self.kernel_virt_to_phys(banner_addr) banner_len = len(self.linux_banner) b = self.read_cstring(banner_addr, banner_len, False) if b is None: print_out_str('!!! Banner not found in dumps!') return False print_out_str('Linux Banner: ' + b.rstrip()) if str(self.linux_banner) in str(b): print_out_str("Linux banner from vmlinux = %s" % self.linux_banner) print_out_str("Linux banner from dump = %s" % b) return True else: print_out_str("Expected Linux banner = %s" % self.linux_banner) print_out_str("Linux banner in Dumps = %s" % b) return False else: print_out_str('!!! linux_banner sym not found in vmlinux') return False def print_command_line(self): command_addr = self.address_of('saved_command_line') if command_addr is not None: command_addr = self.read_word(command_addr) b = self.read_cstring(command_addr, 2048) if b is None: print_out_str('!!! could not read saved command line address') static_command_addr = self.address_of('static_command_line') if static_command_addr is not None: static_command_addr = self.read_word(static_command_addr) b = self.read_cstring(static_command_addr, 2048) if b is None: print_out_str('!!! could not read static command line address') return False else: print_out_str('Satic Command Line: ' + b) return True return False else: print_out_str('Saved Command Line: ' + b) return True else: print_out_str('!!! Could not lookup saved command line address') return False def print_socinfo_minidump(self): content_socinfo = None boards = get_supported_boards() for board in boards: if self.hw_id == board.board_num: content_socinfo = board.ram_start + board.smem_addr_buildinfo break sernum_offset = self.field_offset('struct socinfo_v10', 'serial_number') if sernum_offset is None: sernum_offset = self.field_offset('struct socinfo_v0_10', 'serial_number') if sernum_offset is None: print_out_str("No serial number information available") return False if content_socinfo: addr_of_sernum = content_socinfo + sernum_offset serial_number = self.read_u32(addr_of_sernum, False) if serial_number is not None: print_out_str('Serial number %s' % hex(serial_number)) return True return False return False def print_socinfo(self): if self.kernel_version < (5, 4, 0): try: if self.read_pointer('socinfo') is None: return None content_socinfo = hex(self.read_pointer('socinfo')) content_socinfo = content_socinfo.strip('L') sernum_offset = self.field_offset('struct socinfo_v10', 'serial_number') if sernum_offset is None: sernum_offset = self.field_offset('struct socinfo_v0_10', 'serial_number') if sernum_offset is None: print_out_str("No serial number information available") return False addr_of_sernum = hex(int(content_socinfo, 16) + sernum_offset) addr_of_sernum = addr_of_sernum.strip('L') serial_number = self.read_u32(int(addr_of_sernum, 16)) if serial_number is not None: print_out_str('Serial number %s' % hex(serial_number)) return True except: pass return False else: socinfo = self.address_of('socinfo') if socinfo is None: return None socinfo = self.read_pointer(socinfo) if socinfo is None: return None ver = int(self.read_structure_field(socinfo, 'struct socinfo', 'ver') or 0) chip_ver_major = (ver & 0xFFFF0000) >> 16 chip_ver_minor = (ver & 0x0000FFFF) print_out_str("Chip Version: v{0}.{1}".format(chip_ver_major, chip_ver_minor)) serial_num = int(self.read_structure_field(socinfo, 'struct socinfo', 'serial_num') or 0) print_out_str("Chip Serial Number 0x{0:x}".format(serial_num)) return True def auto_parse(self, file_path, minidump, svm): if self.reduceddump: elfflag, ebiflag = False, False cls = None info = AutoDumpInfoReducedDump(file_path, minidump, self.reduceddump, svm) info.parse() if info is not None: if len(info.ebi_files) > 0: self.ebi_files = info.ebi_files self.thread_maxcount = len(self.ebi_files) self.phys_offset = self.ebi_files[0][1] if self.get_hw_id(): for (f, start, end, filename) in self.ebi_files: print_out_str('Adding {0} {1:x}--{2:x}'.format( filename, start, end)) ebiflag = True else: return False if len(info.elf_files) > 0: self.elf_addr = info.elf_files for filename in self.elf_addr: print_out_str('Adding {0}'.format(filename)) elfflag = True return elfflag and ebiflag else: for cls in sorted(AutoDumpInfo.__subclasses__(), key=lambda x: x.priority, reverse=True): info = cls(file_path, minidump, self.reduceddump, svm) info.parse() if info is not None and len(info.ebi_files) > 0: self.ebi_files = info.ebi_files self.phys_offset = self.ebi_files[0][1] if self.get_hw_id(): for (f, start, end, filename) in self.ebi_files: print_out_str('Adding {0} {1:x}--{2:x}'.format( filename, start, end)) return True self.ebi_files = None return False def create_t32_launcher(self): out_path = os.path.abspath(self.outdir) t32_host_system = self.t32_host_system or platform.system() launch_config = self.open_file('t32_config.t32') launch_config.write('OS=\n') launch_config.write('ID=T32_1000002\n') if t32_host_system != 'Linux': launch_config.write('TMP=C:\\TEMP\n') launch_config.write('SYS=C:\\T32\n') launch_config.write('HELP=C:\\T32\\pdf\n') else: launch_config.write('TMP=/tmp\n') launch_config.write('SYS=/opt/t32\n') launch_config.write('HELP=/opt/t32/pdf\n') launch_config.write('\n') launch_config.write('PBI=SIM\n') launch_config.write('\n') launch_config.write('SCREEN=\n') launch_config.write('FONT=LARGE\n') launch_config.write('HEADER=Trace32-ScorpionSimulator\n') launch_config.write('\n') if t32_host_system != 'Linux': launch_config.write('PRINTER=WINDOWS\n') launch_config.write('\n') launch_config.write('RCL=NETASSIST\n') launch_config.write('PACKLEN=1024\n') launch_config.write('PORT=%d\n' % random.randint(20000, 30000)) launch_config.write('\n') launch_config.close() startup_script = self.open_file('t32_startup_script.cmm') startup_script.write('title \"' + out_path + '\"\n') is_cortex_a53 = self.hw_id in ["8916", "8939", "8936", "bengal", "scuba"] if self.arm64 and is_cortex_a53: startup_script.write('sys.cpu CORTEXA53\n') else: if self.cpu_type == "ARMv8.2-A": startup_script.write("sys.cpu CORTEXA55\n") else: startup_script.write('sys.cpu {0}\n'.format(self.cpu_type)) if self.minidump: startup_script.write('SYStem.Option MMUSPACES OFF\n') else: startup_script.write('SYStem.Option MMUSPACES ON\n') startup_script.write('SYStem.Option ZONESPACES OFF\n') startup_script.write('sys.up\n') if is_cortex_a53 and not self.arm64: startup_script.write('r.s m 0x13\n') for ram in self.ebi_files: ebi_path = os.path.abspath(ram[3]) startup_script.write('data.load.binary {0} 0x{1:x}\n'.format( ebi_path, ram[1])) if self.minidump: dload_ram_elf = 'data.load.elf {} /LOGLOAD /nosymbol\n'.format(os.path.abspath(self.ram_elf_file)) startup_script.write(dload_ram_elf) # Check to include Reduced dump elf's if self.elf_addr: for file in self.elf_addr: startup_script.write('data.load.elf {0} /noclear\n'.format(file)) if not self.minidump: if self.arm64: if self.svm: startup_script.write('Data.Set SPR:0x30201 %Quad 0x{0:x}\n'.format( self.ttbr_data)) startup_script.write('Data.Set SPR:0x34210 %Quad 0x{0:x}\n'.format( self.vttbr_data)) startup_script.write('Data.Set SPR:0x30100 %Quad 0x{0:x}\n'.format( self.SCTLR_EL1)) startup_script.write('Data.Set SPR:0x30200 %Quad 0x{0:x}\n'.format( self.TTBR0_EL1)) startup_script.write('Data.Set SPR:0x30202 %Quad 0x{0:x}\n'.format( self.TCR_EL1)) startup_script.write('Data.Set SPR:0x34110 %Quad 0x{0:x}\n'.format( self.HCR_EL2)) startup_script.write('Data.Set SPR:0x34212 %Quad 0x{0:x}\n'.format( self.VTCR_EL2)) else: startup_script.write('Data.Set SPR:0x30201 %Quad 0x{0:x}\n'.format( self.kernel_virt_to_phys(self.swapper_pg_dir_addr))) tcr_el1 = self.get_tcr_el1(is_cortexa=is_cortex_a53) if is_cortex_a53: startup_script.write('Data.Set SPR:0x30202 %Quad 0x{0:016X}\n'.format(tcr_el1)) startup_script.write('Data.Set SPR:0x30A20 %Quad 0x000000FF440C0400\n') startup_script.write('Data.Set SPR:0x30A30 %Quad 0x0000000000000000\n') startup_script.write('Data.Set SPR:0x30100 %Quad 0x0000000034D5D91D\n') elif self.cpu_type == 'ARMV9-A': if self.hlos_tcr_el1: startup_script.write('Data.Set SPR:0x30202 %Quad 0x{0:x}\n'.format( self.hlos_tcr_el1)) else: startup_script.write('Data.Set SPR:0x30202 %Quad 0x{0:016X}\n'.format(tcr_el1)) startup_script.write('Data.Set SPR:0x30A20 %Quad 0x000000FF440C0400\n') startup_script.write('Data.Set SPR:0x30A30 %Quad 0x0000000000000000\n') if self.hlos_sctlr_el1: startup_script.write('Data.Set SPR:0x30100 %Quad 0x{0:x}\n'.format( self.hlos_sctlr_el1)) else: startup_script.write('Data.Set SPR:0x30100 %Quad 0x0000000084C5D93D\n') corevcpu_path = os.path.join(self.outdir,'corevcpu0_regs.cmm') if os.path.exists(corevcpu_path): startup_script.write('do ' + corevcpu_path + '\n') else: startup_script.write('Data.Set SPR:0x30202 %Quad 0x{0:016X}\n'.format(tcr_el1)) startup_script.write('Data.Set SPR:0x30A20 %Quad 0x000000FF440C0400\n') startup_script.write('Data.Set SPR:0x30A30 %Quad 0x0000000000000000\n') startup_script.write('Data.Set SPR:0x30100 %Quad 0x0000000004C5D93D\n') startup_script.write('Register.Set NS 1\n') startup_script.write('Register.Set CPSR 0x1C5\n') else: # ARM-32: MMU is enabled by default on most platforms. mmu_enabled = 1 if self.mmu is None: mmu_enabled = 0 startup_script.write( 'PER.S.simple C15:0x1 %L 0x{0:x}\n'.format(mmu_enabled)) startup_script.write( 'PER.S.simple C15:0x2 %L 0x{0:x}\n'.format(self.mmu.ttbr)) if isinstance(self.mmu, Armv7LPAEMMU): # TTBR1. This gets setup once and never change again even if TTBR0 # changes startup_script.write('PER.S.F C15:0x102 %L 0x{0:x}\n'.format( self.mmu.ttbr + 0x4000)) # TTBCR with EAE and T1SZ set approprately startup_script.write( 'PER.S.F C15:0x202 %L 0x80030000\n') startup_script.write('mmu.on\n') startup_script.write('mmu.scan\n') where = os.path.abspath(self.vmlinux) kaslr_offset = self.get_kaslr_offset() if kaslr_offset != 0: where += ' 0x{0:x}'.format(kaslr_offset) dloadelf = 'data.load.elf {} /nocode\n'.format(where) startup_script.write(dloadelf) if self.arm64 and not self.minidump: startup_script.write('TRANSlation.COMMON NS:0xF000000000000000--0xffffffffffffffff\n') startup_script.write('trans.tablewalk on\n') startup_script.write('trans.on\n') if not self.svm and self.cpu_type != 'ARMV9-A': startup_script.write('MMU.Delete\n') startup_script.write('MMU.SCAN PT 0xFFFFFF8000000000--0xFFFFFFFFFFFFFFFF\n') startup_script.write('mmu.on\n') startup_script.write('mmu.pt.list 0xffffff8000000000\n') if self.minidump: startup_script.write('y.pointer x29\n') startup_script.write('frame.config.eabi on\n') if self.arm64: startup_script.write('Register.Set CPSR 0x1C5\n') if t32_host_system != 'Linux': if self.arm64: startup_script.write('IF OS.DIR("C:\\T32\\demo\\arm64")\n') startup_script.write('(\n') startup_script.write( 'task.config C:\\T32\\demo\\arm64\\kernel\\linux\\awareness\\linux.t32 /ACCESS NS:\n') startup_script.write( 'menu.reprogram C:\\T32\\demo\\arm64\\kernel\\linux\\awareness\\linux.men\n') startup_script.write(')\n') startup_script.write('ELSE\n') startup_script.write('(\n') startup_script.write( 'task.config C:\\T32\\demo\\arm\\kernel\\linux\\awareness\\linux.t32 /ACCESS NS:\n') startup_script.write( 'menu.reprogram C:\\T32\\demo\\arm\\kernel\\linux\\awareness\\linux.men\n') startup_script.write(')\n') else: if self.kernel_version > (3, 0, 0): startup_script.write( 'task.config c:\\t32\\demo\\arm\\kernel\\linux\\linux-3.x\\linux3.t32\n') startup_script.write( 'menu.reprogram c:\\t32\\demo\\arm\\kernel\\linux\\linux-3.x\\linux.men\n') else: startup_script.write( 'task.config c:\\t32\\demo\\arm\\kernel\\linux\\linux.t32\n') startup_script.write( 'menu.reprogram c:\\t32\\demo\\arm\\kernel\\linux\\linux.men\n') else: if self.arm64: startup_script.write('IF OS.DIR("/opt/t32/demo/arm64")\n') startup_script.write('(\n') startup_script.write( 'task.config /opt/t32/demo/arm64/kernel/linux/linux-3.x/linux3.t32\n') startup_script.write( 'menu.reprogram /opt/t32/demo/arm64/kernel/linux/linux-3.x/linux.men\n') startup_script.write(')\n') startup_script.write('ELSE\n') startup_script.write('(\n') startup_script.write( 'task.config /opt/t32/demo/arm/kernel/linux/linux-3.x/linux3.t32\n') startup_script.write( 'menu.reprogram /opt/t32/demo/arm/kernel/linux/linux-3.x/linux.men\n') startup_script.write(')\n') else: if self.kernel_version > (3, 0, 0): startup_script.write( 'task.config /opt/t32/demo/arm/kernel/linux/linux-3.x/linux3.t32\n') startup_script.write( 'menu.reprogram /opt/t32/demo/arm/kernel/linux/linux-3.x/linux.men\n') else: startup_script.write( 'task.config /opt/t32/demo/arm/kernel/linux/linux.t32\n') startup_script.write( 'menu.reprogram /opt/t32/demo/arm/kernel/linux/linux.men\n') if self.get_kernel_version() >= (5, 10) and not self.minidump: mod_dir = os.path.dirname(self.vmlinux) mod_dir = os.path.abspath(mod_dir) if t32_host_system != 'Linux': startup_script.write('IF OS.DIR("C:\\T32\\demo\\arm64")\n') startup_script.write('(\n') startup_script.write('sYmbol.AUTOLOAD.CHECKCOMMAND ' + '"do C:\\T32\\demo\\arm64\\kernel\\linux\\awareness\\autoload.cmm"' + '\n') startup_script.write(')\n') startup_script.write('ELSE\n') startup_script.write('(\n') startup_script.write('sYmbol.AUTOLOAD.CHECKCOMMAND ' + '"do C:\\T32\\demo\\arm\\kernel\\linux\\etc\\gdb\\gdb_autoload.cmm"' + '\n') startup_script.write(')\n') else: startup_script.write('IF OS.DIR("/opt/t32/demo/arm64")\n') startup_script.write('(\n') startup_script.write('sYmbol.AUTOLOAD.CHECKCOMMAND ' + '"do /opt/t32/demo/arm64/kernel/linux/awareness/autoload.cmm"' + '\n') startup_script.write(')\n') startup_script.write('ELSE\n') startup_script.write('(\n') startup_script.write('sYmbol.AUTOLOAD.CHECKCOMMAND ' + '"do /opt/t32/demo/arm/kernel/linux/etc/gdb/gdb_autoload.cmm"' + '\n') startup_script.write(')\n') if self.module_table.sym_path_list: startup_script.write("y.spath = " +'"{0}"'.format(self.module_table.sym_path_list[0])+ '\n') if len(self.module_table.sym_path_list) > 1 : for path in self.module_table.sym_path_list[1:]: startup_script.write("y.spath += " +'"{0}"'.format(path)+ '\n') else: startup_script.write('sYmbol.SourcePATH.Set ' + '"' + mod_dir + '"' + "\n") startup_script.write('TASK.sYmbol.Option AutoLoad Module\n') startup_script.write('TASK.sYmbol.Option AutoLoad noprocess\n') startup_script.write('sYmbol.AutoLOAD.List\n') startup_script.write('sYmbol.AutoLOAD.CHECK\n') else: for mod_tbl_ent in self.module_table.module_table: mod_sym_path = mod_tbl_ent.get_sym_path() if mod_sym_path != '': ld_mod_sym = '' where = os.path.abspath(mod_sym_path) if self.minidump: if mod_tbl_ent.section_offsets: ld_mod_sym = "Data.LOAD.Elf " + where + " /NoClear /CODESEC /RELOC .text at " + str(hex(mod_tbl_ent.module_offset)) if ".data" in mod_tbl_ent.section_offsets.keys(): ld_mod_sym += " /RELOC .data at " + str(hex(mod_tbl_ent.section_offsets['.data'])) if ".bss" in mod_tbl_ent.section_offsets.keys() : ld_mod_sym += " /RELOC .bss at " + str(hex(mod_tbl_ent.section_offsets['.bss'])) ld_mod_sym += "\n" elif 'wlan' in mod_tbl_ent.name: ld_mod_sym = "Data.LOAD.Elf " + where + " " + str(hex(mod_tbl_ent.module_offset)) + " /NoCODE /NoClear /NAME " + mod_tbl_ent.name + " /reloctype 0x3" + "\n" else: ld_mod_sym = "Data.LOAD.Elf " + where + " /NoCODE /NoClear /NAME " + mod_tbl_ent.name + " /reloctype 0x3" + "\n" startup_script.write(ld_mod_sym) if not self.minidump: startup_script.write('task.dtask\n') startup_script.write( 'v.v %ASCII %STRING linux_banner\n') if os.path.exists(os.path.join(out_path, 'regs_panic.cmm')): startup_script.write( 'do {0}\n'.format(out_path + '/regs_panic.cmm')) elif os.path.exists(os.path.join(out_path, '/core0_regs.cmm')): startup_script.write( 'do {0}\n'.format(out_path + '/core0_regs.cmm')) startup_script.close() if t32_host_system != 'Linux': launch_file = 'launch_t32.bat' t32_bat = self.open_file(launch_file) if self.arm64: t32_binary = 'C:\\T32\\bin\\windows64\\t32MARM64.exe' elif is_cortex_a53: t32_binary = 'C:\\T32\\bin\\windows64\\t32MARM64.exe' else: t32_binary = 'c:\\T32\\bin\\windows64\\t32MARM.exe' t32_bat.write(('start '+ t32_binary + ' -c ' + out_path + '/t32_config.t32, ' + out_path + '/t32_startup_script.cmm')) t32_bat.close() else: launch_file = 'launch_t32.sh' t32_sh = self.open_file(launch_file) if self.arm64: t32_binary = '/opt/t32/bin/pc_linux64/t32marm64-qt' elif is_cortex_a53: t32_binary = '/opt/t32/bin/pc_linux64/t32marm-qt' else: t32_binary = '/opt/t32/bin/pc_linux64/t32marm-qt' t32_sh.write('#!/bin/sh\n\n') t32_sh.write('{0} -c {1}/t32_config.t32, {1}/t32_startup_script.cmm &\n'.format(t32_binary, out_path)) t32_sh.close() self.chmod(launch_file, stat.S_IRWXU) print_out_str( '--- Created a T32 Simulator launcher (run {})'.format(launch_file)) def get_tcr_el1(self, is_cortexa=False): if not is_cortexa: tcr_el1 = Register( 0x00000032B5193519, TG1=(31, 30), T1SZ=(21, 16), TG0=(15, 14), T0SZ=(5, 0) ) tg1_granule_size = { 4096 : 0b10, 16384 : 0b01, 65536 : 0b11 } tg0_granule_size = { 4096 : 0b00, 16384 : 0b10, 65536 : 0b01 } tcr_el1.TG1 = tg1_granule_size.get(self.get_page_size(), 0b10) tcr_el1.TG0 = tg0_granule_size.get(self.get_page_size(), 0b00) tcr_el1.T1SZ = tcr_el1.T0SZ = 64 - self.va_bits else: tcr_el1 = Register( 0x00000012B5193519, RES0=(63, 39), RES1=(38, 0) ) return tcr_el1.value def read_tz_offset(self): if self.tz_addr == 0: print_out_str( 'No TZ address was given, cannot read the magic value!') return None else: return self.read_word(self.tz_addr, False) def get_kaslr_offset(self): return self.kaslr_offset @parser_util.time_cost def determine_kaslr_offset(self): if self.svm and self.svm_kaslr_offset: self.kaslr_offset = self.svm_kaslr_offset self.kaslr_addr = None self.kimage_voffset = self.__kimage_vaddr_va + self.kaslr_offset - self.phys_offset return elif self.svm and not self.svm_kaslr_offset: self.kaslr_offset = 0 self.kaslr_addr = None self.kimage_voffset = self.__kimage_vaddr_va - self.phys_offset return else: __kaslr_offset = None if self.kaslr_offset is not None: __kaslr_offset = self.kaslr_offset elif not self.is_config_defined("CONFIG_RANDOMIZE_BASE"): __kaslr_offset = 0x0 print_out_str('!!!! Kaslr feature is not enabled.') else: if self.minidump: for a in self.ebi_files: if "md_SHRDIMEM".lower() in a[3].lower(): self.kaslr_addr = a[1] + 0x6d0 break if self.kaslr_addr: kaslr_magic = self.read_u32(self.kaslr_addr, False) if kaslr_magic == 0xdead4ead: __kaslr_offset = self.read_u64(self.kaslr_addr + 4, False) if __kaslr_offset: print_out_str(f"&kaslr_offset=0x{self.kaslr_addr:x} kaslr_offset=0x{__kaslr_offset:x}") try: self.kaslr_offset, self.kimage_voffset = self.validate_phys_offset(self.phys_offset, __kaslr_offset) except: print_out_str("Traverse DDR to find out correct kaslr_offset and phys_offset, it may take a little time to do!!") hasFound, kaslr_offset, kimage_voffset, phys_offset = self.determine_phys_offset(__kaslr_offset) if hasFound: self.kaslr_offset = kaslr_offset self.kimage_voffset = kimage_voffset self.phys_offset = phys_offset else: self.kaslr_offset = __kaslr_offset if __kaslr_offset else 0 self.kimage_voffset = self.__kimage_vaddr_va + self.kaslr_offset - self.phys_offset print_out_str("!!! Determine kaslr_offset failed") def determine_phys_offset(self, __kaslr_offset): fdtuple = namedtuple("FDTuple", ["base", "end", "path"]) bfiles = [] if self.reduceddump: # Load the phys range from phys_range.txt path = os.path.join(os.path.dirname(self.elf_addr[0]), 'phys_range.txt') if os.path.exists(path): phys_base, phys_end = self.load_phys_range(path) bfiles.append(fdtuple(phys_base, phys_end, "")) else: print_out_str("Unable to locate the phys_range.txt file !!! Linux banner will not be found !!!") return False, 0, 0, 0 else: for a in self.ebi_files: _, start, end, path = a bfiles.append(fdtuple(start, end, path)) if len(bfiles) == 0: print_out_str("No ddr file found!! check if there is DDRCSxxx.bin in your dumps") return False, 0, 0, 0 isFound = False kaslr_offset = 0 kimage_voffset = 0 phys_offset = 0 from concurrent import futures max_workers = max(len(bfiles), 8) self.executor = futures.ThreadPoolExecutor(max_workers) self.enable_multithread(max_workers, self.executor._thread_name_prefix) lock = threading.Lock() threads = [self.executor.submit(self.traverse_file_thread, __kaslr_offset, _bfile, lock) for _bfile in bfiles] for future in futures.as_completed(threads): isFound, kaslr_offset, kimage_voffset, phys_offset = future.result() if isFound: break self.executor.shutdown() self.disable_multithread() return isFound, kaslr_offset, kimage_voffset, phys_offset ''' Sometimes this function is runing in multi-thread so it's better to use self.read_word(kimage_vaddr_var_phy, virtual = False) instead of self.read_word(kimage_vaddr_var_phy, virtual = True) read data from physical address is allowed. ''' def traverse_file_thread(self, __kaslr_offset, _bfile, thread_lock): ''' traverse DDR file with min_image_align to find out correct kaslr_offset and kimage_voffset _bfile: FDTuple type argument ''' phys_base = _bfile.base & 0xfffff0000 phys_end = _bfile.end for min_image_align in [0x00200000, 0x00080000, 0x00008000]: kimage_load_addr = phys_base while (kimage_load_addr < phys_end): try: if self.__kaslr_found: # kaslr was found by other threads, cancel this thread return False, 0, 0, 0 except: pass try: kaslr_offset, kimage_voffset = self.validate_phys_offset(kimage_load_addr, __kaslr_offset) with thread_lock: self.__kaslr_found = True return True, kaslr_offset, kimage_voffset, kimage_load_addr except: kimage_load_addr += min_image_align continue return False, 0, 0, 0 ''' Sometimes this function is runing in multi-thread so it's better to use self.read_word(kimage_vaddr_var_phy, virtual = False) instead of self.read_word(kimage_vaddr_var_phy, virtual = True) read data from physical address is allowed. ''' def validate_phys_offset(self, phys_offset, kaslr_offset=None): ''' validate whether given phys_offset was the right value phys_offset: phys_offset neet to be validated how to validate: First step: calculate kaslr_offset and kimage_voffset according to phys_offset Virtual Address Physical Address -------- -------- -------- -------- phys_offset kimage_vaddr -------- -------- -------- -------- kimage_vaddr var -------- -------- -------- -------- kimage_vaddr with kaslr -------- -------- phyiscal address of kimage_vaddr var = kimage_vaddr var - kimage_vaddr + phys_offset kimage_vaddr with kaslr = read content from phyiscal address of kimage_vaddr var kaslr_offset = kimage_vaddr with kaslr - kimage_vaddr kimage_voffset = kimage_vaddr with kaslr - phys_offset Second step: check if kimage_voffset value(calculated) == kimage_voffset read from vmlinux Third step: check if linux_banner read from DDR == linux_banner from vmlinux ''' ## First step, calculate kaslr_offset and kimage_voffset if self.arm64: kimage_vaddr_var_phy = phys_offset + self.__kimage_vaddr_var_va - self.__kimage_vaddr_va if kaslr_offset != None: kimage_voffset = self.__kimage_vaddr_var_va + kaslr_offset - kimage_vaddr_var_phy else: kimage_vaddr_va_kaslr = self.read_word(kimage_vaddr_var_phy, False) if kimage_vaddr_va_kaslr and kimage_vaddr_va_kaslr >= self.__kimage_vaddr_va: kaslr_offset = kimage_vaddr_va_kaslr - self.__kimage_vaddr_va kimage_voffset = kimage_vaddr_va_kaslr - phys_offset else: raise Exception("!!! Determine kaslr_voffset failed") else: kimage_voffset = self.page_offset - phys_offset if not self.__kimage_voffset_var_va: #print_out_str("!!!! Skip validate phys_offset for ARM32 with older kernel version") return kaslr_offset, kimage_voffset '''print_out_str(f" kaslr_offset {kaslr_offset:x}, kimage_voffset {kimage_voffset:x} phys_offset {phys_offset:x} self.__kimage_vaddr_va {self.__kimage_vaddr_va:x}")''' ## Second step, check if kimage_voffset value == "&kimage_voffset" kimage_voffset_var_phys = self.__kimage_voffset_var_va + kaslr_offset - kimage_voffset kimage_voffset_var_val = self.read_word(kimage_voffset_var_phys, False) if kimage_voffset_var_val == kimage_voffset: ## check if string from &linux_banner on DDR == string from &linux_banner on vmlinux linux_banner_phys = self.__linux_banner_va + kaslr_offset - kimage_voffset banner_string = self.read_cstring(linux_banner_phys, len(self.linux_banner), False) if banner_string and (banner_string == self.linux_banner): print_out_str(f"<-= Determined kaslr_offset: 0x{kaslr_offset:x} phys_offset: 0x{phys_offset:x} kimage_voffset: 0x{kimage_voffset:x} =->") return kaslr_offset, kimage_voffset else: raise Exception("!!! Validate linux_banner failed") else: raise Exception("!!! Validate kimage_voffset failed") def get_page_size(self): return 1 << self.page_shift def is_arm_smmu_v12(self): boards = get_supported_boards() chosen_board = None for board in boards: if self.hw_id == board.board_num: chosen_board = board break if hasattr(chosen_board, 'arm_smmu_v12') and chosen_board.arm_smmu_v12: return True else: return False def get_hw_id(self, add_offset=True): socinfo_format = -1 socinfo_id = -1 socinfo_version = 0 socinfo_build_id = 'DUMMY' chosen_board = None use_predefined = False boards = get_supported_boards() if (self.hw_id is None): if not self.minidump: heap_toc_offset = self.field_offset('struct smem_shared', 'heap_toc') global_partition_offset_offset = self.field_offset('struct smem_header', 'free_offset') if not heap_toc_offset is None: smem_heap_entry_size = self.sizeof('struct smem_heap_entry') offset_offset = self.field_offset('struct smem_heap_entry', 'offset') elif not global_partition_offset_offset is None: smem_partition_header_size = self.sizeof('struct smem_partition_header') uncached_offset_offset = self.field_offset('struct smem_partition_header', 'offset_free_uncached') smem_private_entry_size = self.sizeof('struct smem_private_entry') entry_item_offset = self.field_offset('struct smem_private_entry', 'item') item_size_offset = self.field_offset('struct smem_private_entry', 'size') else: print_out_str('!!!! Could not get a necessary offset for auto detection!') print_out_str('!!!! Try to use predefined offset!') use_predefined = True for board in boards: if self.minidump or use_predefined: if hasattr(board, 'smem_addr_buildinfo'): socinfo_start = board.smem_addr_buildinfo if add_offset: socinfo_start += board.ram_start else: continue elif not heap_toc_offset is None: socinfo_start_addr = board.smem_addr + heap_toc_offset + smem_heap_entry_size * SMEM_HW_SW_BUILD_ID + offset_offset if add_offset: socinfo_start_addr += board.ram_start soc_start = self.read_int(socinfo_start_addr, False) if soc_start is None: continue socinfo_start = board.smem_addr + soc_start if add_offset: socinfo_start += board.ram_start else: found = False global_partition_offset_addr = board.smem_addr + global_partition_offset_offset uncached_offset_addr = board.smem_addr + uncached_offset_offset if add_offset: global_partition_offset_addr += board.ram_start uncached_offset_addr += board.ram_start global_partition_offset = self.read_int(global_partition_offset_addr, False) if global_partition_offset is None: continue uncached_offset_addr += global_partition_offset uncached_offset = self.read_int(uncached_offset_addr, False) partition_magic_addr = board.smem_addr + global_partition_offset if add_offset: partition_magic_addr += board.ram_start if self.read_int(partition_magic_addr, False) != PARTITION_MAGIC: continue entry_addr = board.smem_addr + global_partition_offset + smem_partition_header_size uncached_end_addr = board.smem_addr + global_partition_offset + uncached_offset if add_offset: entry_addr += board.ram_start uncached_end_addr += board.ram_start while entry_addr < uncached_end_addr: if self.read_u16(entry_addr, False) != SMEM_PRIVATE_CANARY: break if self.read_u16(entry_addr + entry_item_offset, False) == SMEM_HW_SW_BUILD_ID: found = True socinfo_start = entry_addr + smem_private_entry_size break entry_addr += self.read_int(entry_addr + item_size_offset, False) entry_addr += smem_private_entry_size if not found: continue socinfo_id = self.read_int(socinfo_start + 4, False) if socinfo_id is None: break if (socinfo_id & 0xFFFF) != board.socid: continue socinfo_format = self.read_int(socinfo_start, False) socinfo_version = self.read_int(socinfo_start + 8, False) socinfo_build_id = self.read_cstring( socinfo_start + 12, BUILD_ID_LENGTH, virtual=False) chosen_board = board break if chosen_board is None: print_out_str('!!!! Could not find hardware') print_out_str("!!!! The SMEM didn't match anything") print_out_str( '!!!! You can use --force-hardware to use a specific set of values') sys.exit(1) else: for board in boards: if self.hw_id == board.board_num: print_out_str( '!!! Hardware id found! The socinfo values given are bogus') print_out_str('!!! Proceed with caution!') chosen_board = board break if chosen_board is None: print_out_str( '!!! A bogus hardware id was specified: {0}'.format(self.hw_id)) print_out_str('!!! Supported ids:') ids = get_supported_ids() if not len(ids): print_out_str('!!! No registered Boards found - check extensions/board_def.py') for b in ids: print_out_str(' {0}'.format(b)) sys.exit(1) print_out_str('\nHardware match: {0}'.format(board.board_num)) print_out_str('Socinfo id = {0}, version {1:x}.{2:x}'.format( socinfo_id, socinfo_version >> 16, socinfo_version & 0xFFFF)) print_out_str('Socinfo build = {0}'.format(socinfo_build_id)) print_out_str( 'Now setting phys_offset to {0:x}'.format(board.phys_offset)) if board.wdog_addr is not None: print_out_str( 'TZ address: {0:x}'.format(board.wdog_addr)) if board.phys_offset is not None: self.phys_offset = board.phys_offset self.tz_addr = board.wdog_addr self.ebi_start = board.ram_start self.tz_start = board.imem_start self.hw_id = board.board_num self.cpu_type = board.cpu self.imem_fname = board.imem_file_name if hasattr(board, 'hyp_diag_addr'): self.hyp_diag_addr = board.hyp_diag_addr else: self.hyp_diag_addr = None if hasattr(board, 'rm_debug_addr'): self.rm_debug_addr = board.rm_debug_addr else: self.rm_debug_addr = None if hasattr(board, 'tbi_mask'): self.tbi_mask = board.tbi_mask if hasattr(board, 'kaslr_addr'): self.kaslr_addr = board.kaslr_addr else: self.kaslr_addr = None if hasattr(board, 'svm_kaslr_offset'): self.svm_kaslr_offset = board.svm_kaslr_offset self.board = board return True def resolve_virt(self, virt_or_name): """Takes a virtual address or variable name, returns a virtual address """ if not isinstance(virt_or_name, str): return virt_or_name return self.address_of(virt_or_name) def virt_to_phys(self, virt_or_name): """Does a virtual-to-physical address lookup of the virtual address or variable name.""" if self.minidump: return minidump_util.minidump_virt_to_phys(self.ebi_files_minidump,self.resolve_virt(virt_or_name)) else: return self.mmu.virt_to_phys(self.resolve_virt(virt_or_name)) def setup_symbol_tables(self): args = [self.nm_path, '-n', self.vmlinux] p = subprocess.run(args, stdout=subprocess.PIPE) symbols = p.stdout.decode().splitlines() kaslr = self.get_kaslr_offset() # The beginning and ending of kernel image, from vmlinux.lds.S _text = self.address_of('_text') if _text is None: _text = 0 _end = self.address_of('_end') if _end is None: _end = 0xFFFFFFFFFFFFFFFF for line in symbols: s = line.split(' ') if len(s) != 3: continue if (("$x" in s[2].rstrip()) or ("$d" in s[2].rstrip())): continue entry = (int(s[0], 16) + kaslr, s[2].rstrip(),"") # The symbol file contains many artificial symbols which we don't care about. if entry[0] < _text or entry[0] >= _end: continue self.lookup_table.append(entry) if not len(self.lookup_table): print_out_str('!!! Unable to retrieve symbols... Exiting') sys.exit(1) if self.dump_kernel_symbol_table: self.dump_mod_sym_table('vmlinux', self.lookup_table) def retrieve_modules(self): mod_list = self.address_of('modules') next_offset = self.field_offset('struct list_head', 'next') list_offset = self.field_offset('struct module', 'list') name_offset = self.field_offset('struct module', 'name') if self.is_config_defined('CONFIG_SMP'): percpu_offset = self.field_offset('struct module', 'percpu') percpu_size_offset = self.field_offset('struct module', 'percpu_size') if self.kernel_version >= (6, 4, 0): module_core_offset = self.field_offset('struct module', 'mem[0].base') elif self.kernel_version > (4, 9, 0): module_core_offset = self.field_offset('struct module', 'core_layout.base') else: module_core_offset = self.field_offset('struct module', 'module_core') if self.field_offset('struct module_sect_attr', 'battr') is not None: sect_name_offset = self.field_offset('struct module_sect_attr', 'battr') + self.field_offset('struct bin_attribute', 'attr') + self.field_offset('struct attribute', 'name') else: sect_name_offset = self.field_offset('struct module_sect_attr', 'name') kallsyms_offset = self.field_offset('struct module', 'kallsyms') sect_addr_offset = self.field_offset('struct module_sect_attr', 'address') nsections_offset = self.field_offset('struct module_sect_attrs', 'nsections') section_attrs_offset = self.field_offset('struct module_sect_attrs', 'attrs') section_attr_size = self.sizeof('struct module_sect_attr') mod_sect_attrs_offset = self.field_offset('struct module', 'sect_attrs') mod_state_offset = self.field_offset('struct module', 'state') mod_attr_grp_name_offest = self.field_offset('struct module_sect_attrs', 'grp') + self.field_offset('struct attribute_group', 'name') module_states = self.gdbmi.get_enum_lookup_table('module_state', 5) next_list_ent = self.read_pointer(mod_list + next_offset) while next_list_ent and next_list_ent != mod_list: module = next_list_ent - list_offset mod_tbl_ent = module_table.module_table_entry() mod_tbl_ent.name = self.read_cstring(module + name_offset) state = self.read_u32(module + mod_state_offset) if mod_tbl_ent.name is None or state is None or state > len(module_states) or module_states[state] not in ['MODULE_STATE_LIVE']: msg = 'module state @{:x}'.format(module) if mod_tbl_ent.name: msg += ' [{}]'.format(mod_tbl_ent.name) msg += ' is {}'.format(state) if state is not None and state < len(module_states): msg += '({})'.format(module_states[state]) print_out_str(msg) next_list_ent = self.read_pointer(next_list_ent + next_offset) continue mod_tbl_ent.module_offset = self.read_pointer(module + module_core_offset) if mod_tbl_ent.module_offset is None: mod_tbl_ent.module_offset = 0 mod_tbl_ent.kallsyms_addr = self.read_pointer(module + kallsyms_offset) # Loop through sect_attrs mod_tbl_ent.section_offsets = {} mod_sect_attrs = self.read_pointer(module + mod_sect_attrs_offset) # module.sect_attrs if self.read_cstring(self.read_pointer(mod_sect_attrs + mod_attr_grp_name_offest)) != 'sections': # Observed some ramdumps did not have proper attribute set up yet when module is being loaded. # "LIVE" state check not good enough, so add one more sanity check print_out_str('Unexpected variation in module section group name, skipping loading sections for {}'.format(mod_tbl_ent.name)) next_list_ent = self.read_pointer(next_list_ent + next_offset) continue for i in range(0, self.read_u32(mod_sect_attrs + nsections_offset)): # attr_ptr = module.sect_attrs.attrs[i] attr_ptr = mod_sect_attrs + section_attrs_offset + (i * section_attr_size) # sect_name = attr_ptr.battr.attr.name (for 5.4+) sect_name = self.read_cstring(self.read_pointer(attr_ptr + sect_name_offset)) # sect_addr = attr_ptr.address sect_addr = self.read_word(attr_ptr + sect_addr_offset) # https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/tree/scripts/gdb/linux/symbols.py?h=v5.14#n102 if sect_name not in ['.data', '.data..read_mostly', '.rodata', '.bss', '.text', '.text.bss', '.text.hot', '.text.unlikely']: continue mod_tbl_ent.section_offsets[sect_name] = sect_addr if self.is_config_defined('CONFIG_SMP'): percpu_size = self.read_u32(module + percpu_size_offset) if percpu_size != 0: percpu_pointer = self.read_pointer(module + percpu_offset) mod_tbl_ent.section_offsets['.data..percpu'] = percpu_pointer self.module_table.add_entry(mod_tbl_ent) next_list_ent = self.read_pointer(next_list_ent + next_offset) def retrieve_minidump_modules(self): kmodules_seg = next((s for s in self.elffile.iter_sections() if s.name == 'KMODULES'), None) if kmodules_seg is None: return kmodules_lines = self.read_cstring(kmodules_seg['sh_addr'], max_length=kmodules_seg['sh_size']) for line in kmodules_lines.splitlines(): m = re.fullmatch(r"^name: (.+), base: (?:0x)?([0-9a-fA-F]+)\b.*", line) if m is not None: mod_tbl_ent = module_table.module_table_entry() mod_tbl_ent.name = m.group(1) mod_tbl_ent.module_offset = int(m.group(2), base=16) n = re.search(r"\.bss: (?:0x)?([0-9a-fA-F]+).*", line) if n is not None: mod_tbl_ent.section_offsets['.bss'] = int(n.group(1), base=16) n = re.search(r"\.data: (?:0x)?([0-9a-fA-F]+).*", line) if n is not None: mod_tbl_ent.section_offsets['.data'] = int(n.group(1), base=16) self.module_table.add_entry(mod_tbl_ent) def parse_symbols_of_one_module(self, mod_tbl_ent, ko_file_list): name_index = [s for s in ko_file_list.keys() if mod_tbl_ent.name in s] if len(name_index) == 0: print_out_str('!! Object not found for {}'.format(mod_tbl_ent.name)) return if mod_tbl_ent.name not in ko_file_list and name_index[0] in ko_file_list: temp_data = ko_file_list[name_index[0]] del ko_file_list[name_index[0]] ko_file_list[mod_tbl_ent.name] = temp_data if not mod_tbl_ent.set_sym_path(ko_file_list[mod_tbl_ent.name]): return if self.is_config_defined("CONFIG_KALLSYMS") and not self.minidump: symtab_offset = self.field_offset('struct mod_kallsyms', 'symtab') num_symtab_offset = self.field_offset('struct mod_kallsyms', 'num_symtab') strtab_offset = self.field_offset('struct mod_kallsyms', 'strtab') if self.arm64: sym_struct_name = 'struct elf64_sym' sym_struct_size = self.sizeof(sym_struct_name) else: sym_struct_name = 'struct elf32_sym' sym_struct_size = self.sizeof(sym_struct_name) st_info_offset = self.field_offset(sym_struct_name, 'st_info') symtab = self.read_pointer(mod_tbl_ent.kallsyms_addr + symtab_offset) num_symtab = self.read_pointer(mod_tbl_ent.kallsyms_addr + num_symtab_offset) strtab = self.read_pointer(mod_tbl_ent.kallsyms_addr + strtab_offset) if symtab is None or num_symtab is None or strtab is None: return KSYM_NAME_LEN = 128 for i in range(0, num_symtab): elf_sym = symtab + sym_struct_size * i st_value = self.read_structure_field(elf_sym, sym_struct_name, 'st_value') st_info = self.read_byte(elf_sym + st_info_offset) sym_type = chr(st_info) st_name = self.read_structure_field(elf_sym, sym_struct_name, 'st_name') sym_addr = st_value sym_name = self.read_cstring(strtab + st_name, KSYM_NAME_LEN) st_shndx = self.read_structure_field(elf_sym, sym_struct_name, 'st_shndx') st_size = self.read_structure_field(elf_sym, sym_struct_name, 'st_size') ### # FORMAT of record: # sym_addr, syn_name[mod_name], sym_type, idx_elf_sym, st_name, st_shndx, st_size ### if (sym_name is None or mod_tbl_ent.name is None): continue if sym_addr: # when sym_addr is 0, it means the symbol is undefined # will not add undefined symbols here to avoid address 0x0 # being treated as belonging to a particular kernel module mod_tbl_ent.kallsyms_table.append( (sym_addr, sym_name + '[' + mod_tbl_ent.name + ']', sym_type, i, st_name, st_shndx, st_size,sym_name)) mod_tbl_ent.kallsyms_table.sort() if self.dump_module_kallsyms: self.dump_mod_kallsyms_sym_table(mod_tbl_ent.name, mod_tbl_ent.kallsyms_table) else: args = [self.nm_path, '-n', mod_tbl_ent.get_sym_path()] p = subprocess.run(args, stdout=subprocess.PIPE) symbols = p.stdout.decode().splitlines() for line in symbols: s = line.split(' ') if len(s) == 3: mod_tbl_ent.sym_lookup_table.append( (int(s[0], 16) + mod_tbl_ent.module_offset, s[2].rstrip() + '[' + mod_tbl_ent.name + ']')) mod_tbl_ent.sym_lookup_table.sort() if self.dump_module_symbol_table: self.dump_mod_sym_table(mod_tbl_ent.name, mod_tbl_ent.sym_lookup_table) def walk_depth(self, path, on_file, depth=10): if depth <= 0: return try: for basename in os.listdir(path): file = os.path.join(path, basename) if os.path.isdir(file) and not os.path.islink(file): self.walk_depth(file, on_file, depth=depth-1) elif os.path.isfile(file): on_file(file) except Exception as e: print_out_str(str(e)) def has_debug_info(self, file): cmd = self.objdump_path + ' -h ' + file if platform.system() != "Linux": objdump = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True, ) else: objdump = subprocess.Popen(shlex.split(cmd), shell=False, stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True, ) out, err = objdump.communicate() if '.debug_info' in out: return True else: return False def parse_module_symbols(self): # Recursively search all files under mod_path ending in '.ko.unstripped' and store in a list ko_file_list = {} for path in self.module_table.sym_path_list: def on_file(file): if file.endswith('.ko.unstripped'): name = file[:-len('.ko.unstripped')] elif file.endswith('.ko'): name = file[:-len('.ko')] else: return name = os.path.basename(name) name = name.replace("-","_") # Prefer .ko.unstripped if ko_file_list.get(name, '').endswith('.ko.unstripped') and file.endswith('.ko'): return # Prefer ko with debug info if name in ko_file_list and self.has_debug_info(ko_file_list.get(name)): return ko_file_list[name] = file self.ko_file_names.append(name) self.walk_depth(path, on_file) for mod_tbl_ent in self.module_table.module_table: if mod_tbl_ent.name is None: print_out_str('!! Object name not extracted properly..checking next!!') continue self.parse_symbols_of_one_module(mod_tbl_ent, ko_file_list) def add_symbols_to_global_lookup_table(self): if self.is_config_defined("CONFIG_KALLSYMS") and not self.minidump: for mod_tbl_ent in self.module_table.module_table: for sym in mod_tbl_ent.kallsyms_table: if sym[1].startswith('$x') or sym[1].startswith('$d'): continue self.lookup_table.append((sym[0], sym[1],sym[7])) else: for mod_tbl_ent in self.module_table.module_table: for sym in mod_tbl_ent.sym_lookup_table: self.lookup_table.append(sym) self.lookup_table.sort() def setup_module_symbols(self): if self.minidump: self.retrieve_minidump_modules() else: self.retrieve_modules() self.parse_module_symbols(); self.add_symbols_to_global_lookup_table() def dump_mod_sym_table(self, mod_name, sym_lookup_tbl): sym_dump_file = self.open_file('sym_tbl_'+mod_name+'.txt') for line in sym_lookup_tbl: sym_dump_file.write('0x{0:x} {1}\n'.format(line[0], line[1])) sym_dump_file.close() def dump_mod_kallsyms_sym_table(self, mod_name, mod_kallsyms_table): kallsyms_header_format = '{0: >18} {1} {2: >64} {3} {4} {5} {6}\n' kallsyms_record_format = '0x{0:0>16x} {1: >8} {2: >64} {3: >11} {4: >7} {5: >8} {6: >7}\n' kallsyms_file = self.open_file('sym_tbl_kallsyms_'+mod_name+'.txt') kallsyms_file.write('KALLSYMS symbol lookup table['+mod_name+']\n') kallsyms_file.write( kallsyms_header_format.format( 'sym_addr', 'sym_type', 'syn_name[mod_name]', 'idx_elf_sym', 'st_name', 'st_shndx', 'st_size')) for mod_sym_line in mod_kallsyms_table: kallsyms_file.write( kallsyms_record_format.format( mod_sym_line[0], mod_sym_line[2], mod_sym_line[1], mod_sym_line[3], hex(mod_sym_line[4]), mod_sym_line[5], mod_sym_line[6])) kallsyms_file.close() def dump_global_symbol_lookup_table(self): sym_dump_file = self.open_file('sym_table.txt') for line in self.lookup_table: sym_dump_file.write('0x{0:x} {1}\n'.format(line[0], line[1])) sym_dump_file.close() def address_of(self, symbol): """Returns the address of a symbol. :param symbol: name of the symbol. :type symbol: str :return: address value Example: >>> hex(dump.address_of('linux_banner')) '0xffffffc000c7a0a8L' """ try: return self.gdbmi.address_of(symbol) except gdbmi.GdbMIException: if self.hyp: try: return self.gdbmi_hyp.address_of(symbol) except gdbmi.GdbMIException: pass def symbol_at(self, addr): """ Function to return symbol using gdbmi. :param addr: address value. :type addr: int :return: symbol value """ try: return self.gdbmi.symbol_at(addr) except gdbmi.GdbMIException: if self.hyp: try: return self.gdbmi_hyp.symbol_at(addr) except gdbmi.GdbMIException: pass def sizeof(self, the_type): try: return self.gdbmi.sizeof(the_type) except gdbmi.GdbMIException: if self.hyp: try: return self.gdbmi_hyp.sizeof(the_type) except gdbmi.GdbMIException: pass def array_index(self, addr, the_type, index): """Index into the array of type ``the_type`` located at ``addr``. I.e., given:: int my_arr[3]; my_arr[2] = 42; You could do the following: >>> addr = dump.address_of("my_arr") >>> dump.read_word(dump.array_index(addr, "int", 2)) 42 """ offset = self.gdbmi.sizeof(the_type) * index return addr + offset def frame_field_offset(self, frame_name, the_type, field): try: return self.gdbmi.frame_field_offset(frame_name, the_type, field) except gdbmi.GdbMIException: if self.hyp: try: return self.gdbmi_hyp.frame_field_offset(frame_name, the_type, field) except gdbmi.GdbMIException: pass def get_symbol_info1(self,addr): kaslr = self.get_kaslr_offset() if kaslr: addr1 = addr - kaslr else: addr1 = addr #print "hex of address in get_symbol_info1 {0}".format(hex(addr1)) addr1, desc = self.step_through_jump_table(addr1) symbol_obj = self.gdbmi.get_symbol_info(addr1) module = symbol_obj.section.split('\\\\')[-1] if self.minidump: if module == 'vmlinux': return symbol_obj.symbol + desc + " " + str(symbol_obj.offset) else: return symbol_obj.symbol + desc + " " + str(symbol_obj.offset) + " [" + module + "]" return symbol_obj.symbol + desc def type_of(self, symbol): """ this will be helpful to get the type of symbol. eg : >>>dump.type_of("kgsl_driver") struct kgsl_driver """ try: return self.gdbmi.type_of(symbol) except gdbmi.GdbMIException: pass def field_offset(self, the_type, field): """Gets the offset of a field from the base of its containing struct. This can be useful when reading struct fields, although you should consider using :func:`~read_structure_field` if you're reading a word-sized value. Example: >>> dump.field_offset('struct device', 'bus') 168 """ try: return self.gdbmi.field_offset(the_type, field) except gdbmi.GdbMIException: if self.hyp: try: return self.gdbmi_hyp.field_offset(the_type, field) except gdbmi.GdbMIException: pass def container_of(self, ptr, the_type, member): """Like ``container_of`` in the kernel.""" try: return self.gdbmi.container_of(ptr, the_type, member) except gdbmi.GdbMIException: if self.hyp: try: return self.gdbmi_hyp.container_of(ptr, the_type, member) except gdbmi.GdbMIException: pass def sibling_field_addr(self, ptr, parent_type, member, sibling): """Gets the address of a sibling structure field. Given the address of some field within a structure, returns the address of the requested sibling field. """ try: return self.gdbmi.sibling_field_addr(ptr, parent_type, member, sibling) except gdbmi.GdbMIException: if self.hyp: try: return self.gdbmi_hyp.sibling_field_addr(ptr, parent_type, member, sibling) except gdbmi.GdbMIException: pass def step_through_jump_table(self, addr): """ Steps through a jump table, if the address points to a unconditional branch """ if addr is None: return addr, '' fn_addr = addr if self.is_config_defined('CONFIG_ARM64_BTI_KERNEL'): # Skip past BTI instruction to the real branch instr fn_addr += 4 if self.cpu_type in ['ARMV9-A', 'CORTEXA53']: instr = self.read_u32(fn_addr) if instr is None or (instr & 0xFC000000) != 0x14000000: return addr, '' imm26_mask = 0x3FFFFFF offset = instr & imm26_mask if (offset & imm26_mask) >> 25: offset -= (imm26_mask + 1) fn_addr += 4 * offset return fn_addr, '[jt]' return addr, '' def setup_module_layout(self): mod_list = self.address_of('modules') next_offset = self.field_offset('struct list_head', 'next') list_offset = self.field_offset('struct module', 'list') name_offset = self.field_offset('struct module', 'name') next_list_ent = self.read_pointer(mod_list + next_offset) while next_list_ent and next_list_ent != mod_list: mod = next_list_ent - list_offset mod_name = self.read_cstring(mod + name_offset) ent_array = [] # setup module init layout if self.kernel_version >= (6, 4, 0): mem_offset = self.field_offset('struct module', 'mem') module_memory_size = self.sizeof('struct module_memory') # enum mod_mem_type for init layout: 4 - 6 for i in range(4, 7): base = self.read_structure_field(mod + mem_offset + i * module_memory_size, 'struct module_memory', 'base') size = self.read_structure_field(mod + mem_offset + i * module_memory_size, 'struct module_memory', 'size') ent_array.append((base, size)) elif self.kernel_version > (4, 9, 0): init_layout_offset = self.field_offset('struct module', 'init_layout') base = self.read_structure_field(mod + init_layout_offset, 'struct module_layout', 'base') size = self.read_structure_field(mod + init_layout_offset, 'struct module_layout', 'size') ent_array.append((base, size)) else: base = self.read_structure_field(mod, 'struct module', 'module_init') size = self.read_structure_field(mod, 'struct module', 'init_size') ent_array.append((base, size)) # setup module core layout if self.kernel_version >= (6, 4, 0): mem_offset = self.field_offset('struct module', 'mem') module_memory_size = self.sizeof('struct module_memory') # enum mod_mem_type for core layout: 0 - 3 for i in range(0, 4): base = self.read_structure_field(mod + mem_offset + i * module_memory_size, 'struct module_memory', 'base') size = self.read_structure_field(mod + mem_offset + i * module_memory_size, 'struct module_memory', 'size') ent_array.append((base, size)) elif self.kernel_version > (4, 9, 0): core_layout_offset = self.field_offset('struct module', 'core_layout') base = self.read_structure_field(mod + core_layout_offset, 'struct module_layout', 'base') size = self.read_structure_field(mod + core_layout_offset, 'struct module_layout', 'size') ent_array.append((base, size)) else: base = self.read_structure_field(mod, 'struct module', 'module_core') size = self.read_structure_field(mod, 'struct module', 'core_size') ent_array.append((base, size)) self.module_layout_dict[mod_name] = ent_array next_list_ent = self.read_pointer(next_list_ent + next_offset) def validate_module_sym_addr(self, sym_addr, mod_name): """ Validate that if sym_addr is in specified module layaout or not """ value = self.module_layout_dict.get(mod_name) if value is None: return False for base, size in value: if sym_addr >= base and sym_addr < base + size: return True return False def match_name_for_module_sym_addr(self, sym_addr): """ When matched ko is not provided to lrdp or this ko is not live, lrdp can't find a matched symbol name in lookup table. So search sym_addr in module_layout_dict to find its module name to show like UNKNOWN_SYMBOL[si_core_module]. With this, we can know where is this symbol in easily. """ for key in self.module_layout_dict.keys(): value = self.module_layout_dict[key] for base, size in value: if sym_addr >= base and sym_addr < base + size: return ('UNKNOWN_SYMBOL[{}]'.format(key), 0) return None def __unwind_lookup(self, addr, symbol_size=0): """ Returns closest symbols <= addr and either the relative offset or the symbol size. The loop constant is: table[low] <= addr <= table[high] """ addr = self.pac_ignore(addr) table = self.lookup_table low = 0 high = len(self.lookup_table) - 1 addr, desc = self.step_through_jump_table(addr) if self.minidump: symbol_str = self.get_symbol_info1(addr) words = symbol_str.split(" ") symbol = words[0] offset = words[1] if len(words) == 3: module = words[2] return (symbol + ' ' + module, int(offset)) return (symbol, int(offset)) if addr is None or addr < table[low][0] or addr > table[high][0]: return None while(True): # Python now complains about division producing floats mid = (high + low) >> 1 if mid == low or mid == high: break if addr <= table[mid][0]: high = mid elif addr >= table[mid][0]: low = mid if addr == table[low][0]: high = low elif addr == table[high][0]: low = high offset = addr - table[low][0] #how to calculate size for the last symbol? if low == len(self.lookup_table) - 1: size = 0 else: size = table[low + 1][0] - table[low][0] _text = self.address_of('_text') if _text is None: _text = 0 _end = self.address_of('_end') if _end is None: _end = 0xFFFFFFFFFFFFFFFF # do checking for symbol which is not in vmlinux if not (addr > _text and addr < _end): is_matched = re.match(r'.*\[(.+)\]', table[low][1]) if is_matched: mod_name = is_matched.group(1) if not self.validate_module_sym_addr(addr, mod_name): return None if symbol_size == 0: return (table[low][1] + desc, offset) else: return (table[low][1] + desc, size) def unwind_lookup(self, addr, symbol_size=0): r = self.__unwind_lookup(addr, symbol_size) if r is not None: return r # when fail to lookup the symbol name, show the module name as much as possible. return self.match_name_for_module_sym_addr(addr) def read_elf_memory(self, addr, length, temp_file): s = self.gdbmi.read_elf_memory(addr, length, temp_file) if s is not None: a = s.decode('ascii', 'ignore') return a.split('\0')[0] else: return s def read_physical(self, addr, length): if not isinstance(addr, int) or not isinstance(length, int): return None if self.minidump: addr_data = minidump_util.read_physical_minidump( self.ebi_files_minidump, self.ebi_files,self.elffile, addr, length) return addr_data elif self.reduceddump: data = elfutil.read_physical(self.elf_vector, self.elf_htable, self.elf_filemap, self.ebi_files, addr, length) #if addr == 0x83cc09268: # print("addr read yielded none : {:x}".format(addr)) return data else: if self.use_multithread and threading.current_thread().name.startswith(self.thread_name_prefix): '''multi-thread enabled''' ebi_files = self.ebi_files_mappings[threading.current_thread().name] return self.__read_physical(addr, length, ebi_files) else: return self.__read_physical(addr, length, self.ebi_files) def __read_physical(self, addr, length, ebi_files): ebi = (-1, -1, -1) for a in ebi_files: fd, start, end, path = a if addr >= start and addr <= end: ebi = a break if ebi[0] == -1: return None offset = addr - ebi[1] ebi[0].seek(offset) a = ebi[0].read(length) return a def enable_multithread(self, thread_max_count, thread_name_prefix): if self.use_multithread: print_out_str("Ramparser is already running in multi-thread mode!!!") return if thread_max_count > self.thread_maxcount or thread_max_count <= 1: thread_max_count = self.thread_maxcount self.use_multithread = True self.thread_name_prefix = thread_name_prefix self.ebi_files_mappings = {} for idx in range(thread_max_count): tmp_ebi = [] for file in self.ebi_files: _, start, end, path = file tmp_ebi.append([open(path, 'rb'), start, end, path]) self.ebi_files_mappings[f"{thread_name_prefix}_{idx}"] = tmp_ebi def disable_multithread(self): if not self.use_multithread: return self.use_multithread = False for eib_files in self.ebi_files_mappings.values(): for file in eib_files: fd, start, end, path = file fd.close() self.ebi_files_mappings = {} def read_dword(self, addr_or_name, virtual=True, cpu=None, allow_elf=False): s = self.read_string(addr_or_name, '{1}".format(struct_name, field)) addr = self.resolve_virt(addr_or_name) if addr is None or size is None: return None addr += self.field_offset(struct_name, field) if size == 2: return self.read_u16(addr, virtual) if size == 4: return self.read_u32(addr, virtual) if size == 8: return self.read_u64(addr, virtual) return None def read(self, cmd): """Reads the value of a C-style expression provided it doesn't have a pointer. Supported syntax for cmd: dump.read('device_3d0.dev.open_count') """ size = self.sizeof("{0}".format(cmd)) addr = self.address_of(cmd) if addr is None or size is None: return None if size == 2: return self.read_u16(addr) if size == 4: return self.read_u32(addr) if size == 8: return self.read_u64(addr) if size != 0 and size != 4 and size != 8: return addr return None def read_structure_cstring(self, addr_or_name, struct_name, field, max_length=100): """reads a C string from a structure field. The C string field will be dereferenced before reading, so it should be a ``char *``, not a ``char []``. """ virt = self.resolve_virt(addr_or_name) cstring_addr = virt + self.field_offset(struct_name, field) return self.read_cstring(self.read_pointer(cstring_addr), max_length) def read_cstring(self, addr_or_name, max_length=100, virtual=True, cpu=None, allow_elf=False): """Reads a C string.""" addr = addr_or_name s = None if virtual: if cpu is not None: pcpu_offset = self.per_cpu_offset(cpu) addr_or_name = self.resolve_virt(addr_or_name) addr_or_name += pcpu_offset + self.per_cpu_offset(cpu) addr = self.virt_to_phys(addr_or_name) if allow_elf and addr is None: s = self.gdbmi.read_memory(addr_or_name, '{}+{}'.format(addr_or_name, max_length)) if not s: s = self.read_physical(addr, max_length) if s is not None: a = s.decode('ascii', 'ignore') return a.split('\0')[0] else: return s def read_binarystring(self, addr_or_name, length, virtual=True, cpu=None, allow_elf=False): """Reads binary data of specified length from addr_or_name.""" addr = addr_or_name s = None if virtual: if cpu is not None: pcpu_offset = self.per_cpu_offset(cpu) addr_or_name = self.resolve_virt(addr_or_name) addr_or_name += pcpu_offset addr = self.virt_to_phys(addr_or_name) if allow_elf and addr is None: s = self.gdbmi.read_memory(addr_or_name, '{}+{}'.format(addr_or_name, length)) if not s: s = self.read_physical(addr, length) return s def read_string(self, addr_or_name, format_string, virtual=True, cpu=None, allow_elf=False): """Reads data using a format string. Reads data from addr_or_name using format_string (which should be a struct.unpack format). Returns the tuple returned by struct.unpack. """ addr = addr_or_name per_cpu_string = '' s = None if virtual: if cpu is not None: pcpu_offset = self.per_cpu_offset(cpu) addr_or_name = self.resolve_virt(addr_or_name) addr_or_name += pcpu_offset per_cpu_string = ' with per-cpu offset of ' + hex(pcpu_offset) addr = self.virt_to_phys(addr_or_name) if allow_elf and addr is None: s = self.gdbmi.read_memory(addr_or_name, '{}+{}'.format(addr_or_name, struct.calcsize(format_string))) if not s: s = self.read_physical(addr, struct.calcsize(format_string)) if (s is None) or (s == ''): return None return struct.unpack(format_string, s) def hexdump(self, addr_or_name, length, virtual=True, file_object=None, little_endian=True): """Returns a string with a hexdump (in the format of ``xxd``). ``length`` is in bytes. Example (intentionally not in doctest format since it would require a specific dump to be loaded to pass as a doctest): >>> print(dump.hexdump('linux_banner', 0x80)) ffffffc000c610a8: 4c69 6e75 7820 7665 7273 696f 6e20 332e Linux version 3. ffffffc000c610b8: 3138 2e32 302d 6761 3762 3238 6539 2d31 18.20-ga7b28e9-1 ffffffc000c610c8: 3333 3830 2d67 3036 3032 6531 3020 286c 3380-g0602e10 (l ffffffc000c610d8: 6e78 6275 696c 6440 6162 6169 7431 3532 nxbuild@abait152 ffffffc000c610e8: 2d73 642d 6c6e 7829 2028 6763 6320 7665 -sd-lnx) (gcc ve ffffffc000c610f8: 7273 696f 6e20 342e 392e 782d 676f 6f67 rsion 4.9.x-goog ffffffc000c61108: 6c65 2032 3031 3430 3832 3720 2870 7265 le 20140827 (pre ffffffc000c61118: 7265 6c65 6173 6529 2028 4743 4329 2029 release) (GCC) ) If little_endian = False, each 4 byte chunk will be printed in big endian form, like how Trace32 displays memory. The string below looks jumbled but this format is useful when decoding USB TRB rings, for instance. Ex: >>> print(dump.hexdump('linux_banner', 0x80, little_endian=False)) ffffffe03b562920: 756e 694c 6576 2078 6f69 7372 2e35 206e uniLev xoisr.5 n ffffffe03b562930: 372e 3531 6b71 2d38 6f63 2d69 6c6f 736e 7.51kq-8oc-ilosn ffffffe03b562940: 7461 6469 6e61 2d65 696f 7264 2d33 3164 tadina-eiord-31d ffffffe03b562950: 6667 2d38 3934 3035 6632 3139 2034 3731 fg-89405f219 471 ffffffe03b562960: 6975 6228 752d 646c 4072 6573 6c69 7562 iub(u-dl@resliub ffffffe03b562970: 6f68 2d64 2029 7473 646e 4128 6469 6f72 oh-d )tsdnA(dior ffffffe03b562980: 3538 2820 3036 3830 6220 2c38 6465 7361 58( 0680b ,8desa ffffffe03b562990: 206e 6f20 3035 3472 6534 3837 6c63 2029 no 054re487lc ) """ from io import StringIO sio = StringIO() address = self.resolve_virt(addr_or_name) if little_endian: parser_util.xxd( address, [self.read_byte(address + i, virtual=virtual) or 0 for i in range(length)], file_object=sio) else: places = [] for i in range(int(length / 4)): places.extend([(i + 1) * 4 - 1, (i + 1) * 4 - 2, (i + 1) * 4 - 3, i * 4]) parser_util.xxd( address, [self.read_byte(address + i, virtual=virtual) or 0 for i in places], file_object=sio) ret = sio.getvalue() sio.close() return ret def get_read_physical_offset(self, addr): if not self.reduceddump: offset = None input = None for file in self.ebi_files: fd, start, end, path = file if addr >= start and addr <= end: input = path offset = addr - start break return offset, input return elfutil.get_read_physical_offset_helper(self.elf_vector, self.elf_htable, self.elf_filemap, self.ebi_files, addr) def per_cpu_offset(self, cpu): """ __per_cpu_offset has been observed to be a negative number on kernel 5.4, even though it is stored in a unsigned long. Since python supports numbers larger than 64 bits, the behavior on overflow differs with that of C. Therefore, read it as a signed value instead. """ per_cpu_offset_addr = self.address_of('__per_cpu_offset') if per_cpu_offset_addr is None: return 0 per_cpu_offset_addr_indexed = self.array_index( per_cpu_offset_addr, 'unsigned long', cpu) return self.read_slong(per_cpu_offset_addr_indexed) def set_available_cores(self): """set available core numbers in the system.""" major, minor, patch = self.kernel_version cpu_present_bits_addr = self.address_of('cpu_present_bits') cpu_present_bits = self.read_word(cpu_present_bits_addr) ind = 0 if (major, minor) >= (4, 5): cpu_present_bits_addr = self.address_of('__cpu_present_mask') bits_offset = self.field_offset('struct cpumask', 'bits') cpu_present_bits = self.read_word(cpu_present_bits_addr + bits_offset) self.available_cores.clear() while cpu_present_bits: if cpu_present_bits & 1: self.available_cores.append(ind) cpu_present_bits = cpu_present_bits >> 1 ind += 1 def get_num_cpus(self): """Gets the number of CPUs in the system.""" if not(len(self.available_cores)): self.set_available_cores() return len(self.available_cores) def iter_cpus(self): """Returns an iterator over all CPUs in the system. Example: >>> list(dump.iter_cpus()) [0, 1, 2, 3] """ return (self.available_cores) def is_thread_info_in_task(self): return self.is_config_defined('CONFIG_THREAD_INFO_IN_TASK') def get_thread_info_addr(self, task_addr): if self.is_thread_info_in_task(): thread_info_address = task_addr + self.field_offset('struct task_struct', 'thread_info') else: thread_info_ptr = task_addr + self.field_offset('struct task_struct', 'stack') thread_info_address = self.read_word(thread_info_ptr, True) return thread_info_address def get_task_cpu(self, task_struct_addr, thread_info_struct_addr): if self.is_thread_info_in_task() and self.get_kernel_version() < (5, 19, 0): offset_cpu = self.field_offset('struct task_struct', 'cpu') cpu = self.read_int(task_struct_addr + offset_cpu) else: offset_cpu = self.field_offset('struct thread_info', 'cpu') cpu = self.read_int(thread_info_struct_addr + offset_cpu) return cpu def thread_saved_field_common_32(self, task, reg_offset): thread_info = self.get_thread_info_addr(task) cpu_context_offset = self.field_offset('struct thread_info', 'cpu_context') val = self.read_word(thread_info + cpu_context_offset + reg_offset) return val def thread_saved_field_common_64(self, task, reg_offset): thread_offset = self.field_offset('struct task_struct', 'thread') cpu_context_offset = self.field_offset('struct thread_struct', 'cpu_context') val = self.read_word(task + thread_offset + cpu_context_offset + reg_offset) return val def thread_saved_pc(self, task): if self.arm64: return self.thread_saved_field_common_64(task, self.field_offset('struct cpu_context', 'pc')) else: return self.thread_saved_field_common_32(task, self.field_offset('struct cpu_context_save', 'pc')) def thread_saved_sp(self, task): if self.arm64: return self.thread_saved_field_common_64(task, self.field_offset('struct cpu_context', 'sp')) else: return self.thread_saved_field_common_32(task, self.field_offset('struct cpu_context_save', 'sp')) def thread_saved_fp(self, task): if self.arm64: return self.thread_saved_field_common_64(task, self.field_offset('struct cpu_context', 'fp')) else: pc = self.thread_saved_pc(task) # PC value last bit is 1 for Thumb and 0 for ARM if (pc & 0x1): return self.thread_saved_field_common_32(task, self.field_offset('struct cpu_context_save', 'r7')) return self.thread_saved_field_common_32(task, self.field_offset('struct cpu_context_save', 'fp')) def for_each_process(self): """ create a generator for traversing through each valid process""" init_task = self.address_of('init_task') tasks_offset = self.field_offset('struct task_struct', 'tasks') prev_offset = self.field_offset('struct list_head', 'prev') next = init_task seen_tasks = [] try: while (1): task_pointer = self.read_word(next + tasks_offset, True) if not task_pointer: break task_struct = task_pointer - tasks_offset if ((self.validate_task_struct(task_struct) == -1) or ( self.validate_sched_class(task_struct) == -1)): next = init_task while (1): task_pointer = self.read_word(next + tasks_offset + prev_offset, True) if not task_pointer: break task_struct = task_pointer - tasks_offset if (self.validate_task_struct(task_struct) == -1): break if (self.validate_sched_class(task_struct) == -1): break if task_struct in seen_tasks: break yield task_struct seen_tasks.append(task_struct) next = task_struct if (next == init_task): break break if task_struct in seen_tasks: break yield task_struct seen_tasks.append(task_struct) next = task_struct if (next == init_task): break except: print_out_exception() ''' task_struct->signal->thread_head struct list_head { struct list_head *next; -->task_struct->thread_node struct list_head *prev; -->task_struct->thread_node } thread_head; ''' def for_each_thread(self, task_addr): offset_thread_node =self.field_offset( 'struct task_struct', 'thread_node') offset_signal = self.field_offset( 'struct task_struct', 'signal') offset_thread_head = self.field_offset( 'struct signal_struct', 'thread_head') try: signal_addr = self.read_word(task_addr + offset_signal) thread_head_addr = self.read_word(signal_addr + offset_thread_head) next_thread_head = thread_head_addr seen_threads = [] while True: task_addr = next_thread_head - offset_thread_node if (self.validate_task_struct(task_addr) == -1) or ( self.validate_sched_class(task_addr) == -1): break yield task_addr next_thr = self.read_word(next_thread_head) if (next_thr == next_thread_head) and (next_thr != thread_head_addr): print_out_str('!!!! Cycle in thread group! The list is corrupt!\n') break if (next_thr in seen_threads): break seen_threads.append(next_thr) next_thread_head = next_thr if next_thread_head == thread_head_addr: break except: print_out_exception() def validate_task_struct(self, task): thread_info_address = self.get_thread_info_addr(task) if self.is_thread_info_in_task(): task_struct = task else: task_address = thread_info_address + self.field_offset( 'struct thread_info', 'task') task_struct = self.read_word(task_address, True) cpu_number = self.get_task_cpu(task_struct, thread_info_address) if cpu_number is None: return -1 if ((task != task_struct) or (thread_info_address == 0x0)): return -1 if ((cpu_number < 0) or (cpu_number > self.get_num_cpus())): return -1 def validate_sched_class(self, task): sc_top = self.address_of('stop_sched_class') sc_rt = self.address_of('rt_sched_class') sc_idle = self.address_of('idle_sched_class') sc_fair = self.address_of('fair_sched_class') sched_class = self.read_structure_field( task, 'struct task_struct', 'sched_class') if not ((sched_class == sc_top) or (sched_class == sc_rt) or ( sched_class == sc_idle) or (sched_class == sc_fair)): return -1 def __ignore_storage_class(self, line): line_split = line.split() result_words = [word for word in line_split if word.lower() not in storage_classes] return ' '.join(result_words) def __ignore_expanded_pointer(self, text, d_type): if '} *' in text[-1].lstrip(): name_re = re.search(r"type = ([a-zA-Z0-9_ ]+){", text[0]) if name_re: name = name_re.group(1) + "*" return name return d_type def __is_primary_type(self, d_type): d_type = d_type.rstrip() if d_type[-1] == "]": re_obj = re.search("(.*)\[\d+\]",d_type) d_type = re_obj.group(1) if "*" in d_type or "enum " in d_type or d_type == "enum": return True if d_type.lstrip().rstrip() in primary_types: return True else: return False def __create_object(self, text, base_offset, curr_index): """ Function to create a python object from the gdb text output with meta data like size and offset of all the members, needed to populate the values from the binary dump files. :param text: text gdb output for a particular symbol/type. :type the_type: str :param base_offset: base offset value. :type field: int :param curr_index: current line index in 'text'. :type field: int :return: py object created based on 'text', array check flag, current index """ if curr_index == 0: d_type = text[0].split("{")[0] else: d_type = text[curr_index-1].split("{")[0] d_type = d_type.split("[")[0] d_type = d_type.strip() d_type = d_type.split()[-1] newclass = type(d_type,(), {}) curr_obj = newclass() curr_offset = base_offset total_size = len(text) size = 0 while total_size > curr_index: line = text[curr_index] curr_index = curr_index + 1 if line is None: break if "/* offset | size */" in line or line.lstrip().rstrip() == "": continue re1 = re2 = 0 for i in range(1): # using a one iteration loop to implement break # sample match : "/* 0 | 40 */ struct thread_info {" re1 = re.search('\s+(\d+)\s+[|]\s+(\d+) \*\/\s+(struct|union) .*{', line) #sample match:"/* 0 | 40 */ struct thread_info {" if re1: curr_offset = int(re1.group(1)) size = int(re1.group(2)) break # sample match : "/* 8 */ struct {" re2 = re.search('\/\*\s+(\d+) \*\/\s+(struct|union) .*{', line) if re2: size = int(re2.group(1)) if re1 or re2: obj, attr_name, curr_index = self.__create_object(text, curr_offset, curr_index) if attr_name is not None: setattr(curr_obj, attr_name, [obj, curr_offset - base_offset, size]) else: # adding anonimous union members to parent for attr, value in vars(obj).items(): temp_offset = curr_offset - base_offset if isinstance(value[0], int) or isinstance(value[0], float): value[0] += temp_offset else: value[1] += temp_offset setattr(curr_obj, attr, value) continue else: re1 = re2 = re3 = re4 = 0 for i in range(1): # using a one iteration loop to implement break # sample match : "/* 20 | 4 */ u32 need_resched;" re1 = re.search('/\*\s+(\d+)\s+[|]\s+(\d+)\s\*/\s+([^:]+) (\S+);', line) if re1 is not None: curr_offset = int(re1.group(1)) size = int(re1.group(2)) datatype = re1.group(3) attr_name = (re1.group(4)) break # sample match : "/* 4 */ uint32_t v;" re2 = re.search('/\*\s+(\d+)\s\*/\s+([^:]+) (\S+);', line) if re2 is not None: size = int(re2.group(1)) datatype = re2.group(2) attr_name = (re2.group(3)) break # sample match : "/* 868: 3 | 4 */ unsigned int dl_overrun : 1;" re3 = re.search('/\*\s+(\d+)[:]\s*(\d+)\s+[|]\s+(\d+)\s\*/\s+([^:]+) (\S+) [:] (\d+);', line) if re3 is not None: curr_offset = int(re3.group(1)) + (int(re3.group(2))/100) size = int(re3.group(3)) + (int(re3.group(6))/100) datatype = re3.group(4) attr_name = (re3.group(5)) break # sample match : "/* 4 */ unsigned int x : 1;" re4 = re.search('/\*\s+(\d+)\s\*/\s+([^:]+) (\S+) [:] (\d+);', line) if re4 is not None: size = int(re4.group(1)) + (int(re4.group(4))/100) datatype = re4.group(2) attr_name = (re4.group(3)) if re1 or re2 or re3 or re4: if ")(" in datatype: attr_name = datatype.split(")(")[0].split("(")[1] if attr_name.lstrip()[0] == '*': datatype = datatype + " *" attr_name = attr_name.lstrip('*') if not self.__is_primary_type(datatype): temp_obj = self.__get_type_info(datatype) if isinstance(temp_obj[0], str): setattr(curr_obj, attr_name, [curr_offset - base_offset, size, temp_obj[0]]) else: setattr(curr_obj, attr_name, [temp_obj[0], curr_offset - base_offset, size]) else: setattr(curr_obj, attr_name, [curr_offset - base_offset, size, datatype]) continue re_obj = re.search('\s*} (\S+);', line) if re_obj is not None: return curr_obj, re_obj.group(1), curr_index re_obj = re.search('\s*};', line) if re_obj: return curr_obj, None, curr_index re_obj = re.search('\s*}\s*(\[\d+\])', line) if re_obj: return curr_obj, re_obj.group(1), curr_index # None means unnamed union or struct return curr_obj, None, curr_index def __get_datatype_from_ptr(self, ptr_addr_or_name): if isinstance(ptr_addr_or_name, str): var_type, vsize, temp_name = self.__get_type_info(ptr_addr_or_name) if var_type[-1] != "*": raise InvalidDatatype else: return var_type[:-1] else: raise InvalidDatatype def __unpack_format(self, size, ty): if ty == "bool" or ty == "_Bool": return " length): bin_data += self.read_physical(self.virt_to_phys(addr), length) addr +=length size -= length length = PAGE_SIZE if(size > 0): addr = self.virt_to_phys(addr) bin_data += self.read_physical(addr, size) return bin_data def enum_lookup(self, enum, val): """ Function to return string corresponding to the value for an enum. :param enum: enum type / gdb output string for a typedef enum. :type enum: str :param val: enum value. :type val: int :return: string corresponding to the enum value. """ if "{" in enum and "}" in enum: temp = enum.split("{")[1].split("}")[0] temp = temp.split(",") res_dict = {} count = 0 for i in temp: if "=" in i: v = i.split("=")[0].strip() k = int(i.split("=")[1].strip()) res_dict[k] = v count = k+1 else: v = i.strip() res_dict[count] = v count += 1 if val in res_dict.keys(): return res_dict[val] else: return None else: if "enum " in enum: enum = enum.split()[1] if val >= 0: return self.gdbmi.get_enum_lookup_table(enum, val+1)[val] else: return None def __populate_primary(self, struct_bin_data, t): length = len(struct_bin_data) st_format = self.__unpack_format(length, t) if st_format is None: return None else: return struct.unpack_from(st_format, struct_bin_data)[0] def __populate_bitfield(self, struct_bin_data, t, bit_offset, bit_length): length = len(struct_bin_data) st_format = self.__unpack_format(length, t) if st_format is None: return None else: temp_data = struct_bin_data temp_bin_data = bin(int.from_bytes(temp_data, byteorder="little")) temp_bin_data = temp_bin_data[2:] #remove 0b temp_bin_data = temp_bin_data.rjust(length*8, '0') temp_bin_data = temp_bin_data[::-1] #reverse string temp_bin_data = temp_bin_data[bit_offset:bit_offset+bit_length] temp_bin_data = temp_bin_data[::-1] #reverse string temp_bin_data = int(temp_bin_data, 2) temp_bin_data = temp_bin_data.to_bytes(length, byteorder="little") return struct.unpack_from(st_format, temp_bin_data)[0] def __populate_enum(self, struct_bin_data, var_type): val = self.__populate_primary(struct_bin_data, 'unsigned int') enum_var,enum_ty = self.__get_enum(var_type) enum_vals = [member.value for member in enum_var] if val not in enum_vals: val = self.__populate_primary(struct_bin_data, 'int') if val not in enum_vals: temp_key = "UNKNOWN_" + str(val) enum_dict = {i.name:i.value for i in enum_var} enum_dict.update({temp_key:val}) enum_var = enum.Enum(enum_ty,enum_dict) res = enum_var(val) return res def __populate_array(self, struct_bin_data, var_type): t = var_type.split("[")[0].strip() temp_s = var_type.split("[")[1].split("]")[0] if temp_s == '' or temp_s == '0': return None else: s = int(temp_s) length = len(struct_bin_data) arr_len = s l = length // arr_len arr = [] st_format = self.__unpack_format(l, t) if st_format is None: return None for i in range(arr_len): start = (i * l) end = ((i + 1) * l) if "enum" in var_type: res = self.__populate_enum(struct_bin_data[start:end], var_type) arr.append(res) else: arr.append(struct.unpack_from(st_format, struct_bin_data[start:end])[0]) if t == 'char' or t == 'unsigned char' or t == 'signed char': temp = ''.join(chr(x) for x in arr) temp = temp.split('\0')[0] return temp return arr def get_enum_data(self, enum): if "{" not in enum: enum_data = self.gdbmi.getStructureData(enum) if enum_data: enum = enum_data[0].split("type = ")[1] res_dict = {} if "{" in enum and "}" in enum: temp = filter(None, enum.split("{")[1].split("}")[0].split(",")) count = 0 for i in temp: if "=" in i: k = i.split("=")[0].strip() v = int(i.split("=")[1].strip()) res_dict[k] = v count = v+1 else: k = i.strip() res_dict[k] = count count += 1 return res_dict def __get_enum(self, var_type): if var_type not in self.enum_data.keys(): temp_enum = self.get_enum_data(var_type) enum_ty = var_type.split()[1].split("[")[0] if "{" in enum_ty: enum_ty = "enum" enum_var = enum.Enum(enum_ty,temp_enum) self.enum_data[var_type] = (enum_var,enum_ty) return self.enum_data[var_type] def __object_value(self, var_type, struct_bin_data, bin_offset, temp_name, addr, attr_dict=None): """ Function to return structure value for the given type and type offset. :param var_type: name of the structure field type. :type var_type: str :param struct_bin_data: bin data structure. :type struct_bin_data: str :param bin_offset: offset value. :type bin_offset: int :param temp_name: temporary name of the structure field type. :type temp_name: str :param attr_dict: dictionary of attributes to be read(optional) :type attr_list: dictionary :return: The data read from the dumps. """ if temp_name is None: temp_name = "" if isinstance(var_type, str): t = var_type if (t not in primary_types) and ('*' not in t) and ('enum' not in t) and ("[" not in t): return None elif "enum" in var_type and "*" not in var_type and "[" not in var_type: return self.__populate_enum(struct_bin_data, var_type) elif "[" not in var_type: return self.__populate_primary(struct_bin_data, t) else: return self.__populate_array(struct_bin_data, var_type) else: if "[" in temp_name: temp_s = temp_name.split("[")[1].split("]")[0] if temp_s == '' or temp_s == '0': return None else: ar_len = int(temp_s) length = len(struct_bin_data) length = length // ar_len ar = [] for i in range(ar_len): ar.append(self.__object_value(var_type, struct_bin_data, i * length, None, addr, attr_dict)) return ar else: newclass = type(var_type) temp_structure = newclass() for key, value in var_type.__dict__.items(): if (attr_dict == None) or (key.split("[")[0] in list(attr_dict.keys())): if isinstance(value[0], int) or isinstance(value[0], float): offset = int(value[0]) length = int(value[1]) bit_offset = int(round((value[0] - offset),2)*100) bit_length = int(round((value[1] - length),2)*100) ty = value[2] if (ty not in primary_types) and ('*' not in ty) and ('enum' not in ty): setattr(temp_structure, key, None) continue if ("enum" in ty) and ('*' not in ty) and ("[" not in key): res = self.__populate_enum(struct_bin_data[bin_offset + offset:bin_offset + offset + length], ty) setattr(temp_structure, key, res) continue if "[" not in key: # member in neither an array nor another struct/union if (bit_length == 0) and (bit_offset == 0): res = self.__populate_primary(struct_bin_data[bin_offset + offset:bin_offset + offset + length], ty) setattr(temp_structure, key, res) else: res = self.__populate_bitfield(struct_bin_data[bin_offset + offset:bin_offset + offset + length], ty, bit_offset, bit_length) setattr(temp_structure, key, res) elif "[]" in key and length == 0: #flex array res = addr + offset setattr(temp_structure, key.split("[")[0], res) else: # member is an array but not of struct/union temp_ty = key.split("[")[1] temp_ty = ty + "[" + temp_ty res = self.__populate_array(struct_bin_data[bin_offset + offset:bin_offset + offset + length], temp_ty) setattr(temp_structure, key.split("[")[0], res) else: if "[" not in key: # member is another struct/union/obj but not an array if attr_dict != None: setattr(temp_structure, key, self.__object_value(value[0], struct_bin_data, value[1] + bin_offset, None, addr, attr_dict[key])) else: setattr(temp_structure, key, self.__object_value(value[0], struct_bin_data, value[1] + bin_offset, None, addr)) else: # member is another struct/union/obj and an array temp_s = key.split("[")[1].split("]")[0] if temp_s == '' or temp_s == '0': setattr(temp_structure, key.split("[")[0], None) continue else: arr_len = int(temp_s) l = value[2] // arr_len arr = [None] * arr_len for i in range(arr_len): if attr_dict != None: arr[i] = self.__object_value(value[0], struct_bin_data, value[1] + (i * l) + bin_offset, None, addr, attr_dict[key.split("[")[0]]) else: arr[i] = self.__object_value(value[0], struct_bin_data, value[1] + (i * l) + bin_offset, None, addr) setattr(temp_structure, key.split("[")[0], arr) return temp_structure def rgetattr(self, obj, attr, *args): """ Function to get attributes. :param obj: name of the object. :type obj: object :param attr: attribute type. :type attr: str :param args: arguments. :type args: str :return: The data read from the dumps. """ def _getattr(obj, attr): return getattr(obj, attr, *args) return functools.reduce(_getattr, [obj] + attr.split('.')) def read_linkedlist(self, the_type, member, address, callback=None, callback_data=None, attr_list=None, ignore_head=True): """ Function to read linked list structure for the given structure type / member. :param the_type: structure type field. :type the_type: str :param member: member of the structure field. :type member: str :param address: address of the structure field. :type address: int :param callback: address of the structure field. :type callback: int :param callback_data: call back data of the structure field. :type callback_data: str :param attr_list: list of attributes to be read(optional) :type attr_list: list :return: The data read from the dumps. """ linked_list = [] if isinstance(address,str): address = self.read_word(address) if address == None: raise SymbolNotFound(address + "Symbol not found") if address == 0x0: return linked_list if address is None: raise InvalidInput("None address passed to read_linkedlist") offset = self.field_offset(the_type, member) first_node = address + offset size = self.sizeof(the_type) if (self.read_word(first_node) == first_node) and ignore_head: return linked_list while True: entry = self.__get_populated_object(address, the_type, size, attr_list) if callback is not None: callback(self, entry, callback_data) linked_list.append(entry) next_member_addr = self.rgetattr(entry, member) next_next_addr = self.read_word(next_member_addr) if ignore_head: if next_member_addr != 0x0 and next_next_addr != first_node: address = next_member_addr - offset else: break else: if next_member_addr != 0x0 and next_member_addr != first_node: address = next_member_addr - offset else: break return linked_list def read_parray(self, ptr_addr_or_name, count, data_type=None, attr_list=None): """ Function to read array for the given pointer address / pointer name. :param ptr_addr_or_name: address or name of the pointer. :type ptr_addr_or_name: str :param count: count of the array. :type count: int :param data_type: the structure field type :type data_type: str :param attr_list: list of attributes to be read(optional) :type attr_list: list :return: The data read from the dumps. """ ptr_address = self.resolve_virt(ptr_addr_or_name) if ptr_address is None: if isinstance(ptr_addr_or_name, str): raise SymbolNotFound(ptr_addr_or_name + " symbol not found") else: raise InvalidInput("None passed to read_parray") address = self.read_pointer(ptr_address) if address is None: raise InvalidInput("Pointer passed to read_parray points to None") if data_type is None: data_type = self.__get_datatype_from_ptr(ptr_addr_or_name) size = self.sizeof(data_type) array = [] while count >= 1: count = count - 1 array.append(self.__get_populated_object(address, data_type, size, attr_list)) address = address + size return array def read_pdatatype(self, ptr_addr_or_name, data_type=None, attr_list=None): """ Function to read data type for the given pointer address / pointer name . :param ptr_addr_or_name: address or name of the pointer. :type ptr_addr_or_name: str :param data_type: the structure field type :type data_type: str :param attr_list: list of attributes to be read(optional) :type attr_list: list :return: The data read from the dumps. """ ptr_address = self.resolve_virt(ptr_addr_or_name) if ptr_address is None: if isinstance(ptr_addr_or_name, str): raise SymbolNotFound(ptr_addr_or_name + " symbol not found") else: raise InvalidInput("None passed to read_pdatatype") address = self.read_pointer(ptr_address) if address is None: raise InvalidInput("Pointer passed to read_pdatatype points to None") if data_type is None: data_type = self.__get_datatype_from_ptr(ptr_addr_or_name) size = self.sizeof(data_type) return self.__get_populated_object(address, data_type, size, attr_list) def read_datatype(self, addr_or_name, data_type=None, attr_list=None): """ Function to read data type for the given address / name. :param addr_or_name: address or name of the pointer. :type addr_or_name: str :param data_type: the structure field type :type data_type: str :param attr_list: list of attributes to be read(optional) :type attr_list: list :return: The data read from the dumps. """ address = self.resolve_virt(addr_or_name) if address is None: if isinstance(addr_or_name,str): raise SymbolNotFound(addr_or_name + " symbol not found") else: raise InvalidInput("None passed to read_datatype") if data_type is None: if isinstance(addr_or_name, str): data_type = addr_or_name else: raise InvalidDatatype size = self.sizeof(data_type) return self.__get_populated_object(address, data_type, size, attr_list) def read_multi(self, items): """ Function to read multiple structures at a time. :param items: structure type fields from the list :type items: list :return: dictionary of data type. """ out_dict = {} for var in items: if not (isinstance(var[0], str) or isinstance(var[0], int)): raise InvalidDatatype if not (isinstance(var[1], str) or var[1] is None): raise InvalidDatatype if (var[1] is None) or (var[1].rstrip() == ""): out_dict[var[0]] = self.read_datatype(var[0]) elif var[1].rstrip() == "*": out_dict[var[0]] = self.read_pdatatype(var[0]) elif var[1].rstrip()[-1] == "*": out_dict[var[0]] = self.read_pdatatype(var[0], var[1].rstrip()[:-1].rstrip()) else: out_dict[var[0]] = self.read_datatype(var[0], var[1].rstrip()) return out_dict def pretty_print(self, clas, fop, format, indent=0): indent += 4 if isinstance(clas, list): for i in range(len(clas)): fop.write("\n" + ' ' * indent + "[{}] = (\n".format(i)) self.pretty_print(clas[i], fop, format, indent) return for k, v in clas.__dict__.items(): if '__dict__' in dir(v): fop.write(' ' * indent + k + ":\n") self.pretty_print(v, fop, format, indent) elif isinstance(v, list): fop.write(' ' * indent + k + '= (\n') indent += 4 for i in range(0, len(v)): if '__dict__' in dir(v[i]): fop.write(' ' * indent + k + "[" + str(i) + "]: \n") self.pretty_print(v[i], fop, format, indent) else: if isinstance(v[i], int) and format == "hex": fop.write(' ' * indent + '[' + str(i) + '] : ' + "0x{0:X}".format(v[i]) + "\n") else: fop.write(' ' * indent + '[' + str(i) + '] : ' + str(v[i]) + "\n") indent -= 4 else: if isinstance(v, int) and format == "hex": fop.write(' ' * indent + k + ' = ' + "0x{0:X}".format(v) + "\n") else: fop.write(' ' * indent + k + ' = ' + str(v) + "\n") def print_struct(self, struct_obj, fop, members=None, fmt_str=None, format=None): """ Function to print the complete structure or member of structure with some given format. :param struct_obj: struct object :type struct_obj: object :param fop: output file handle :type fop: file handle :param members: list of member of structure (optional argument) :type members: list :param fmt_str: format specifier for each member (optional argument) :type fmt_str: list `Example`: 1. Print Complete structure:: vpp_device = self.ramdump.read_datatype('vpp_device') self.ramdump_util.print_struct(vpp_device, fop) 2. Print member of structure without format:: vpp_device = self.ramdump.read_datatype('vpp_device') self.ramdump_util.print_struct(vpp_device, fop, ["chip_ver", "foundry_id"]) 3. Print member of structure with format:: vpp_device = self.ramdump.read_datatype('vpp_device') self.ramdump_util.print_struct(vpp_device, fop, ["chip_ver", "foundry_id"], ["0x{:08x}", "{}"]) """ if members is None: self.pretty_print(struct_obj, fop, format, 0) else: for i in range(0, len(members)): value = getattr(struct_obj, members[i]) if fmt_str is None: fop.write(members[i] + " : {}\n".format(value)) elif fmt_str[i].count("{") == 1: fop.write(fmt_str[i].format(value)) else: fop.write(fmt_str[i].format(members[i], value)) class Struct(object): """ Helper class to abstract C structs retrieval by providing a map of fields to functions on how to retrieve these Given C struct:: struct my_struct { char label[MAX_STR_SIZE]; u32 number; void *address; } You can abstract as: >>> var = Struct(ramdump, var_name, struct_name="struct my_struct", fields={'label': Struct.get_cstring, 'number': Struct.get_u32, 'address': Struct.get_pointer}) >>> var.label 'label string' >>> var.number 1234 """ _struct_name = None _fields = None def __init__(self, ramdump, base, struct_name=None, fields=None): """ :param ram_dump: Reference to the ram dump :param base: The virtual address or variable name of struct :param struct_name: Name of the structure, should start with 'struct'. Ex: 'struct my_struct' :param fields: Dictionary with key being the element name and value being a function pointer to method used to retrieve it. """ self.ramdump = ramdump self._base = self.ramdump.resolve_virt(base) self._data = {} if struct_name: self._struct_name = struct_name if fields: self._fields = fields def is_empty(self): """ :return: true if struct is empty """ return self._base == 0 or self._base is None or self._fields is None def get_address(self, key): """ :param key: struct field name :return: returns address of the named field within the struct """ return self._base + self.ramdump.field_offset(self._struct_name, key) def get_pointer(self, key): """ :param key: struct field name :return: returns the addressed pointed by field within the struct example struct:: struct { void *key; }; """ address = self.get_address(key) return self.ramdump.read_pointer(address) def get_struct_sizeof(self, key): """ :param key: struct field name :return: returns the size of a field within struct Given C struct:: struct my_struct { char key1[10]; u32 key2; }; You could do: >>> struct = Struct(ramdump, 0, struct="struct my_struct", fields={"key1": Struct.get_cstring, "key2": Struct.get_u32}) >>> struct.get_struct_sizeof(key1) 10 >>> struct.get_struct_sizeof(key2) 4 """ return self.ramdump.sizeof('((%s *) 0)->%s' % (self._struct_name, key)) def get_cstring(self, key): """ :param key: struct field name :return: returns a string that is contained within struct memory Example C struct:: struct { char key[10]; }; """ address = self.get_address(key) length = self.get_struct_sizeof(key) return self.ramdump.read_cstring(address, length) def get_cstring_from_pointer(self, key): """ :param key: struct field name :return: returns a string that is contained within struct memory Example C struct:: struct { char *key; }; """ pointer = self.get_pointer(key) return self.ramdump.read_cstring(pointer) def get_u8(self, key): """ :param key: struct field name :return: returns a u8 integer within the struct Example C struct:: struct { u8 key; }; """ address = self.get_address(key) return self.ramdump.read_byte(address) def get_u16(self, key): """ :param key: struct field name :return: returns a u16 integer within the struct Example C struct:: struct { u16 key; }; """ address = self.get_address(key) return self.ramdump.read_u16(address) def get_u32(self, key): """ :param key: struct field name :return: returns a u32 integer within the struct Example C struct:: struct { u32 key; }; """ address = self.get_address(key) return self.ramdump.read_u32(address) def get_u64(self, key): """ :param key: struct field name :return: returns a u64 integer within the struct Example C struct:: struct { u64 key; }; """ address = self.get_address(key) return self.ramdump.read_u64(address) def get_array_ptrs(self, key): """ :param key: struct field name :return: returns an array of pointers Example C struct:: struct { void *key[4]; }; """ ptr_size = self.ramdump.sizeof('void *') length = self.get_struct_sizeof(key) // ptr_size address = self.get_address(key) arr = [] for i in range(0, length - 1): ptr = self.ramdump.read_pointer(address + (ptr_size * i)) arr.append(ptr) return arr def __setattr__(self, key, value): if self._fields and key in self._fields: raise ValueError(key + "is read-only") else: super(Struct, self).__setattr__(key, value) def __getattr__(self, key): if not self.is_empty(): if key in self._data: return self._data[key] elif key in self._fields: fn = self._fields[key] value = fn(self, key) self._data[key] = value return value return None