#!/usr/bin/env python # # ESP32 core dump Utility from __future__ import print_function from __future__ import unicode_literals from __future__ import division from hashlib import sha256 import sys try: from builtins import zip from builtins import str from builtins import range from past.utils import old_div from builtins import object except ImportError: print('Import has failed probably because of the missing "future" package. Please install all the packages for ' 'interpreter {} from the $IDF_PATH/requirements.txt file.'.format(sys.executable)) sys.exit(1) import os import argparse import subprocess import tempfile import struct import errno import base64 import binascii import logging import re idf_path = os.getenv('IDF_PATH') if idf_path: sys.path.insert(0, os.path.join(idf_path, 'components', 'esptool_py', 'esptool')) try: import esptool except ImportError: print("esptool is not found! Set proper $IDF_PATH in environment.") sys.exit(2) __version__ = "0.4-dev" if os.name == 'nt': CLOSE_FDS = False else: CLOSE_FDS = True INVALID_CAUSE_VALUE = 0xFFFF # Exception cause dictionary to get translation of exccause register # From 4.4.1.5 table 4-64 Exception Causes of Xtensa # Instruction Set Architecture (ISA) Reference Manual xtensa_exception_cause_dict = { 0: ("IllegalInstructionCause", "Illegal instruction"), 1: ("SyscallCause", "SYSCALL instruction"), 2: ("InstructionFetchErrorCause", "Processor internal physical address or data error during instruction fetch. (See EXCVADDR for more information)"), 3: ("LoadStoreErrorCause", "Processor internal physical address or data error during load or store. (See EXCVADDR for more information)"), 4: ("Level1InterruptCause", "Level-1 interrupt as indicated by set level-1 bits in the INTERRUPT register"), 5: ("AllocaCause", "MOVSP instruction, if caller`s registers are not in the register file"), 6: ("IntegerDivideByZeroCause", "QUOS: QUOU, REMS: or REMU divisor operand is zero"), 8: ("PrivilegedCause", "Attempt to execute a privileged operation when CRING ? 0"), 9: ("LoadStoreAlignmentCause", "Load or store to an unaligned address. (See EXCVADDR for more information)"), 12: ("InstrPIFDataErrorCause", "PIF data error during instruction fetch. (See EXCVADDR for more information)"), 13: ("LoadStorePIFDataErrorCause", "Synchronous PIF data error during LoadStore access. (See EXCVADDR for more information)"), 14: ("InstrPIFAddrErrorCause", "PIF address error during instruction fetch. (See EXCVADDR for more information)"), 15: ("LoadStorePIFAddrErrorCause", "Synchronous PIF address error during LoadStore access. (See EXCVADDR for more information)"), 16: ("InstTLBMissCause", "Error during Instruction TLB refill. (See EXCVADDR for more information)"), 17: ("InstTLBMultiHitCause", "Multiple instruction TLB entries matched. (See EXCVADDR for more information)"), 18: ("InstFetchPrivilegeCause", "An instruction fetch referenced a virtual address at a ring level less than CRING. (See EXCVADDR for more information)"), 20: ("InstFetchProhibitedCause", "An instruction fetch referenced a page mapped with an attribute that does not permit instruction fetch (EXCVADDR)."), 24: ("LoadStoreTLBMissCause", "Error during TLB refill for a load or store. (See EXCVADDR for more information)"), 25: ("LoadStoreTLBMultiHitCause", "Multiple TLB entries matched for a load or store. (See EXCVADDR for more information)"), 26: ("LoadStorePrivilegeCause", "A load or store referenced a virtual address at a ring level less than CRING. (See EXCVADDR for more information)"), 28: ("LoadProhibitedCause", "A load referenced a page mapped with an attribute that does not permit loads. (See EXCVADDR for more information)"), 29: ("StoreProhibitedCause", "A store referenced a page mapped with an attribute that does not permit stores [Region Protection Option or MMU Option]."), 32: ("Coprocessor0Disabled", "Coprocessor 0 instruction when cp0 disabled"), 33: ("Coprocessor1Disabled", "Coprocessor 1 instruction when cp1 disabled"), 34: ("Coprocessor2Disabled", "Coprocessor 2 instruction when cp2 disabled"), 35: ("Coprocessor3Disabled", "Coprocessor 3 instruction when cp3 disabled"), 36: ("Coprocessor4Disabled", "Coprocessor 4 instruction when cp4 disabled"), 37: ("Coprocessor5Disabled", "Coprocessor 5 instruction when cp5 disabled"), 38: ("Coprocessor6Disabled", "Coprocessor 6 instruction when cp6 disabled"), 39: ("Coprocessor7Disabled", "Coprocessor 7 instruction when cp7 disabled"), INVALID_CAUSE_VALUE: ("InvalidCauseRegister", "Invalid EXCCAUSE register value or current task is broken and was skipped")} class ESPCoreDumpError(RuntimeError): """Core dump runtime error class """ def __init__(self, message): """Constructor for core dump error """ super(ESPCoreDumpError, self).__init__(message) class BinStruct(object): """Binary structure representation Subclasses must specify actual structure layout using 'fields' and 'format' members. For example, the following subclass represents structure with two fields: f1 of size 2 bytes and 4 bytes f2. Little endian. class SomeStruct(BinStruct): fields = ("f1", "f2") format = " 0: self._read_sections(f, shoff, shstrndx) if phnum > 0: self._read_program_segments(f, phoff, phentsize, phnum) def _read_sections(self, f, section_header_offs, shstrndx): """Reads core dump sections from ELF file """ f.seek(section_header_offs) section_header = f.read() LEN_SEC_HEADER = 0x28 if len(section_header) == 0: raise ESPCoreDumpError("No section header found at offset %04x in ELF file." % section_header_offs) if len(section_header) % LEN_SEC_HEADER != 0: logging.warning('Unexpected ELF section header length %04x is not mod-%02x' % (len(section_header),LEN_SEC_HEADER)) # walk through the section header and extract all sections section_header_offsets = range(0, len(section_header), LEN_SEC_HEADER) def read_section_header(offs): name_offs,sec_type,flags,lma,sec_offs,size = struct.unpack_from("= ps.addr and addr < (ps.addr + seg_len): raise ESPCoreDumpError("Can not add overlapping region [%x..%x] to ELF file. Conflict with existing [%x..%x]." % (addr, addr + data_sz - 1, ps.addr, ps.addr + seg_len - 1)) if (addr + data_sz) > ps.addr and (addr + data_sz) <= (ps.addr + seg_len): raise ESPCoreDumpError("Can not add overlapping region [%x..%x] to ELF file. Conflict with existing [%x..%x]." % (addr, addr + data_sz - 1, ps.addr, ps.addr + seg_len - 1)) # append self.program_segments.append(ESPCoreDumpSegment(addr, data, type, flags)) def add_aux_segment(self, data, type, flags): """Adds new note segment """ self.aux_segments.append(ESPCoreDumpSegment(0, data, type, flags)) def write_program_headers(self, f, off, segs): for seg in segs: phdr = Elf32ProgramHeader() phdr.p_type = seg.type phdr.p_offset = off phdr.p_vaddr = seg.addr phdr.p_paddr = phdr.p_vaddr # TODO phdr.p_filesz = len(seg.data) phdr.p_memsz = phdr.p_filesz # TODO phdr.p_flags = seg.flags phdr.p_align = 0 # TODO f.write(phdr.dump()) off += phdr.p_filesz return off def dump(self, f): """Write core dump contents to file """ # TODO: currently dumps only program segments. # dumping sections is not supported yet # write ELF header ehdr = Elf32FileHeader() ehdr.e_type = self.e_type ehdr.e_machine = self.e_machine ehdr.e_entry = 0 ehdr.e_phoff = ehdr.sizeof() ehdr.e_shoff = 0 ehdr.e_flags = 0 ehdr.e_phentsize = Elf32ProgramHeader().sizeof() ehdr.e_phnum = len(self.program_segments) + len(self.aux_segments) ehdr.e_shentsize = 0 ehdr.e_shnum = 0 ehdr.e_shstrndx = self.SHN_UNDEF f.write(ehdr.dump()) # write program header table cur_off = ehdr.e_ehsize + ehdr.e_phnum * ehdr.e_phentsize cur_off = self.write_program_headers(f, cur_off, self.program_segments) cur_off = self.write_program_headers(f, cur_off, self.aux_segments) # write program segments for segment in self.program_segments: f.write(segment.data) # write aux program segments for segment in self.aux_segments: f.write(segment.data) class ESPCoreDumpLoaderError(ESPCoreDumpError): """Core dump loader error class """ def __init__(self, message): """Constructor for core dump loader error """ super(ESPCoreDumpLoaderError, self).__init__(message) def esp_core_dump_ver(chip, maj, min): return (((chip & 0xFFFF) << 16) | ((maj & 0xFF) << 8) | ((min & 0xFF) << 0)) class ESPCoreDumpLoader(object): """Core dump loader base class """ # TODO: add class for core dump version and move all version-dependent params to it ESP_CORE_DUMP_CHIP_ESP32 = 0 # "legacy" stands for core dumps v0.1 (before IDF v4.1) ESP32_COREDUMP_VERSION_BIN_V1 = esp_core_dump_ver(ESP_CORE_DUMP_CHIP_ESP32, 0, 1) ESP32_COREDUMP_VERSION_BIN_V2 = esp_core_dump_ver(ESP_CORE_DUMP_CHIP_ESP32, 0, 2) ESP32_COREDUMP_VERSION_ELF_CRC32 = esp_core_dump_ver(ESP_CORE_DUMP_CHIP_ESP32, 1, 0) ESP32_COREDUMP_VERSION_ELF_SHA256 = esp_core_dump_ver(ESP_CORE_DUMP_CHIP_ESP32, 1, 1) ESP_CORE_DUMP_INFO_TYPE = 8266 ESP_CORE_DUMP_TASK_INFO_TYPE = 678 ESP_CORE_DUMP_EXTRA_INFO_TYPE = 677 ESP_COREDUMP_CURR_TASK_MARKER = 0xdeadbeef ESP32_COREDUMP_BIN_V1_HDR_FMT = '<4L' ESP32_COREDUMP_BIN_V1_HDR_SZ = struct.calcsize(ESP32_COREDUMP_BIN_V1_HDR_FMT) ESP32_COREDUMP_HDR_FMT = '<5L' ESP32_COREDUMP_HDR_SZ = struct.calcsize(ESP32_COREDUMP_HDR_FMT) ESP32_COREDUMP_TSK_HDR_FMT = '<3L' ESP32_COREDUMP_TSK_HDR_SZ = struct.calcsize(ESP32_COREDUMP_TSK_HDR_FMT) ESP32_COREDUMP_MEM_SEG_HDR_FMT = '<2L' ESP32_COREDUMP_MEM_SEG_HDR_SZ = struct.calcsize(ESP32_COREDUMP_MEM_SEG_HDR_FMT) ESP32_COREDUMP_NOTE_HDR_FMT = '<3L' ESP32_COREDUMP_NOTE_HDR_SZ = struct.calcsize(ESP32_COREDUMP_NOTE_HDR_FMT) ESP32_COREDUMP_CRC_FMT = ' 0x40000000) def stack_is_sane(self, sp): """Check stack address if it is correct """ return not(sp < 0x3ffae010 or sp > 0x3fffffff) def addr_is_fake(self, addr): """Check if address is in fake area """ return ((addr < 0x3f3fffff and addr >= 0x20000000) or addr >= 0x80000000) def remove_tmp_file(self, fname): """Silently removes temporary file """ try: os.remove(fname) except OSError as e: if e.errno != errno.ENOENT: logging.warning("Failed to remove temp file '%s' (%d)!" % (fname, e.errno)) def cleanup(self): """Cleans up loader resources """ if self.fcore: self.fcore.close() if self.fcore_name: self.remove_tmp_file(self.fcore_name) def _extract_elf_corefile(self, core_fname=None, off=0, exe_name=None): """ Reads the ELF formatted core dump image and parse it """ core_off = off if self.hdr['ver'] == self.ESP32_COREDUMP_VERSION_ELF_CRC32: checksum_len = self.ESP32_COREDUMP_CRC_SZ elif self.hdr['ver'] == self.ESP32_COREDUMP_VERSION_ELF_SHA256: checksum_len = self.ESP32_COREDUMP_SHA256_SZ else: raise ESPCoreDumpLoaderError("Core dump version '%d' is not supported!" % self.hdr['ver']) core_elf = ESPCoreDumpElfFile() data = self.read_data(core_off, self.hdr['tot_len'] - checksum_len - self.ESP32_COREDUMP_HDR_SZ) with open(core_fname, 'w+b') as fce: try: fce.write(data) fce.flush() fce.seek(0) core_elf._read_elf_file(fce) if exe_name: exe_elf = ESPCoreDumpElfFile(exe_name) # Read note segments from core file which are belong to tasks (TCB or stack) for ns in core_elf.aux_segments: if ns.type != ESPCoreDumpElfFile.PT_NOTE: continue note_read = 0 while note_read < len(ns.data): note = Elf32NoteDesc("", 0, None) note_read += note.read(ns.data[note_read:]) # Check for version info note if 'ESP_CORE_DUMP_INFO' == note.name and note.type == self.ESP_CORE_DUMP_INFO_TYPE and exe_name: app_sha256 = binascii.hexlify(exe_elf.sha256()) n_ver_len = struct.calcsize(" stack_top: stack_len = stack_end - stack_top stack_base = stack_top else: stack_len = stack_top - stack_end stack_base = stack_end stack_len_aligned = stack_len if stack_len_aligned % 4: stack_len_aligned = 4 * (old_div(stack_len_aligned,4) + 1) core_off += self.ESP32_COREDUMP_TSK_HDR_SZ logging.debug("Read TCB %d bytes @ 0x%x" % (tcbsz_aligned, tcb_addr)) data = self.read_data(core_off, tcbsz_aligned) task_status.task_tcb_addr = tcb_addr try: if self.tcb_is_sane(tcb_addr, tcbsz_aligned): if self.hdr['tcbsz'] != tcbsz_aligned: core_elf.add_program_segment(tcb_addr, data[:self.hdr['tcbsz'] - tcbsz_aligned], ESPCoreDumpElfFile.PT_LOAD, ESPCoreDumpSegment.PF_R | ESPCoreDumpSegment.PF_W) else: core_elf.add_program_segment(tcb_addr, data, ESPCoreDumpElfFile.PT_LOAD, ESPCoreDumpSegment.PF_R | ESPCoreDumpSegment.PF_W) # task_status.task_name = bytearray("%s\0" % task_name_str, encoding='ascii') elif tcb_addr and self.addr_is_fake(tcb_addr): task_status.task_flags |= EspCoreDumpTaskStatus.TASK_STATUS_TCB_CORRUPTED except ESPCoreDumpError as e: logging.warning("Skip TCB %d bytes @ 0x%x. (Reason: %s)" % (tcbsz_aligned, tcb_addr, e)) core_off += tcbsz_aligned logging.debug("Read stack %d bytes @ 0x%x" % (stack_len_aligned, stack_base)) data = self.read_data(core_off, stack_len_aligned) if stack_len != stack_len_aligned: data = data[:stack_len - stack_len_aligned] task_status.task_stack_start = stack_base task_status.task_stack_len = stack_len_aligned try: if self.stack_is_sane(stack_base): core_elf.add_program_segment(stack_base, data, ESPCoreDumpElfFile.PT_LOAD, ESPCoreDumpSegment.PF_R | ESPCoreDumpSegment.PF_W) elif stack_base and self.addr_is_fake(stack_base): task_status.task_flags |= EspCoreDumpTaskStatus.TASK_STATUS_STACK_CORRUPTED core_elf.add_program_segment(stack_base, data, ESPCoreDumpElfFile.PT_LOAD, ESPCoreDumpSegment.PF_R | ESPCoreDumpSegment.PF_W) except ESPCoreDumpError as e: logging.warning("Skip task's (%x) stack %d bytes @ 0x%x. (Reason: %s)" % (tcb_addr, stack_len_aligned, stack_base, e)) core_off += stack_len_aligned try: logging.debug("Stack start_end: 0x%x @ 0x%x" % (stack_top, stack_end)) task_regs,extra_regs = self._get_registers_from_stack(data, stack_end > stack_top) except Exception as e: logging.error(e) return None task_info_notes += Elf32NoteDesc("TASK_INFO", self.ESP_CORE_DUMP_TASK_INFO_TYPE, task_status.dump()).dump() prstatus = XtensaPrStatus() prstatus.pr_cursig = 0 # TODO: set sig only for current/failed task prstatus.pr_pid = tcb_addr note = Elf32NoteDesc("CORE", 1, prstatus.dump() + struct.pack("<%dL" % len(task_regs), *task_regs)).dump() notes += note if ESPCoreDumpElfFile.REG_EXCCAUSE_IDX in extra_regs and len(core_dump_info_notes) == 0: # actually there will be only one such note - for crashed task core_dump_info_notes += Elf32NoteDesc("ESP_CORE_DUMP_INFO", self.ESP_CORE_DUMP_INFO_TYPE, struct.pack(" part_size: logging.error("Incorrect size of core dump image: %d, use partition size instead: %d", self.dump_sz, part_size) self.dump_sz = part_size # set actual size of core dump image and read it from flash tool_args[-2] = str(self.dump_sz) et_out = subprocess.check_output(tool_args) if len(et_out): logging.info(et_out.decode('utf-8')) except subprocess.CalledProcessError as e: logging.error("esptool script execution failed with err %d" % e.returncode) logging.debug("Command ran: '%s'" % e.cmd) logging.debug("Command out:") logging.debug(e.output) if self.fcore_name: f.close() self.remove_tmp_file(self.fcore_name) raise e return f def _load_coredump(self, off=None): """Loads core dump from flash using parttool or elftool (if offset is set) """ tool_path = None try: if off: tool_path = '' logging.info("Invoke esptool to read image.") f = self.invoke_esptool(tool_path=tool_path, off=off) else: tool_path = '' logging.info("Invoke parttool to read image.") f = self.invoke_parttool(tool_path=tool_path) except subprocess.CalledProcessError as e: if len(e.output): logging.info(e.output) logging.warning("System path is not set. Try to use predefined path.") if off: tool_path = self.get_tool_path(use_esptool=True) f = self.invoke_esptool(tool_path=tool_path, off=off) else: tool_path = self.get_tool_path(use_esptool=False) f = self.invoke_parttool(tool_path=tool_path) return f def _read_core_dump_length(self, f): """Reads core dump length """ data = f.read(self.ESP32_COREDUMP_FLASH_LEN_SZ) tot_len, = struct.unpack_from(self.ESP32_COREDUMP_FLASH_LEN_FMT, data) return tot_len def create_corefile(self, core_fname=None, exe_name=None, rom_elf=None): """Checks flash coredump data integrity and creates ELF file """ data = self.read_data(0, self.ESP32_COREDUMP_HDR_SZ) self.checksum_len = 0 _,coredump_ver,_,_,_ = struct.unpack_from(self.ESP32_COREDUMP_HDR_FMT, data) if coredump_ver == self.ESP32_COREDUMP_VERSION_ELF_CRC32 or coredump_ver == self.ESP32_COREDUMP_VERSION_BIN_V1 \ or coredump_ver == self.ESP32_COREDUMP_VERSION_BIN_V2: logging.debug("Dump size = %d, crc off = 0x%x", self.dump_sz, self.dump_sz - self.ESP32_COREDUMP_CRC_SZ) data = self.read_data(self.dump_sz - self.ESP32_COREDUMP_CRC_SZ, self.ESP32_COREDUMP_CRC_SZ) dump_crc, = struct.unpack_from(self.ESP32_COREDUMP_CRC_FMT, data) data = self.read_data(0, self.dump_sz - self.ESP32_COREDUMP_CRC_SZ) data_crc = binascii.crc32(data) & 0xffffffff if dump_crc != data_crc: raise ESPCoreDumpLoaderError("Invalid core dump CRC %x, should be %x" % (data_crc, dump_crc)) elif coredump_ver == self.ESP32_COREDUMP_VERSION_ELF_SHA256: dump_sha256 = self.read_data(self.dump_sz - self.ESP32_COREDUMP_SHA256_SZ, self.ESP32_COREDUMP_SHA256_SZ) data = self.read_data(0, self.dump_sz - self.ESP32_COREDUMP_SHA256_SZ) data_sha256 = sha256(data) data_sha256_str = data_sha256.hexdigest() dump_sha256_str = binascii.hexlify(dump_sha256).decode('ascii') if dump_sha256_str != data_sha256_str: raise ESPCoreDumpLoaderError("Invalid core dump SHA256 '%s', should be '%s'" % (dump_sha256_str, data_sha256_str)) return super(ESPCoreDumpFlashLoader, self).create_corefile(core_fname, exe_name) class GDBMIOutRecordHandler(object): """GDB/MI output record handler base class """ TAG = '' def __init__(self, f, verbose=False): """Base constructor for GDB/MI output record handler """ self.verbose = verbose def execute(self, ln): """Base method to execute GDB/MI output record handler function """ if self.verbose: logging.debug("%s.execute: [[%s]]" % (self.__class__.__name__, ln)) class GDBMIOutStreamHandler(GDBMIOutRecordHandler): """GDB/MI output stream handler class """ def __init__(self, f, verbose=False): """Constructor for GDB/MI output stream handler """ super(GDBMIOutStreamHandler, self).__init__(None, verbose) self.func = f def execute(self, ln): """Executes GDB/MI output stream handler function """ GDBMIOutRecordHandler.execute(self, ln) if self.func: # remove TAG / quotes and replace c-string \n with actual NL self.func(ln[1:].strip('"').replace('\\n', '\n').replace('\\t', '\t')) class GDBMIResultHandler(GDBMIOutRecordHandler): """GDB/MI result handler class """ TAG = '^' RC_DONE = 'done' RC_RUNNING = 'running' RC_CONNECTED = 'connected' RC_ERROR = 'error' RC_EXIT = 'exit' def __init__(self, verbose=False): """Constructor for GDB/MI result handler """ super(GDBMIResultHandler, self).__init__(None, verbose) self.result_class = '' self.result_str = '' def _parse_rc(self, ln, rc): """Parses result code """ rc_str = "{0}{1}".format(self.TAG, rc) if not ln.startswith(rc_str): return False self.result_class = rc if len(ln) > len(rc_str): self.result_str = ln[len(rc_str):] if self.result_str.startswith(','): self.result_str = self.result_str[1:] else: logging.error("Invalid result format: '%s'" % ln) else: self.result_str = '' return True def execute(self, ln): """Executes GDB/MI result handler function """ GDBMIOutRecordHandler.execute(self, ln) if self._parse_rc(ln, self.RC_DONE): return if self._parse_rc(ln, self.RC_RUNNING): return if self._parse_rc(ln, self.RC_CONNECTED): return if self._parse_rc(ln, self.RC_ERROR): return if self._parse_rc(ln, self.RC_EXIT): return logging.error("Unknown GDB/MI result: '%s'" % ln) class GDBMIThreadListIdsHandler(GDBMIResultHandler): """GDB/MI thread-list-ids handler class """ def __init__(self, verbose=False): """Constructor for GDB/MI result handler """ super(GDBMIThreadListIdsHandler, self).__init__(verbose) self.threads = [] self.current_thread = '' def execute(self, ln): """Executes GDB/MI thread-list-ids handler function """ GDBMIResultHandler.execute(self, ln) if self.result_class != self.RC_DONE: return # simple parsing method result = re.search(r'thread-ids\s*=\s*\{([^\{\}]*)\}', self.result_str) if result: for tid in re.finditer(r'thread-id="(\d+)"', result.group(1)): self.threads.append(tid.group(1)) result = re.search(r'current-thread-id="(\d+)"', self.result_str) if result: self.current_thread = result.group(1) class GDBMIThreadSelectHandler(GDBMIResultHandler): """GDB/MI thread-select handler class """ def execute(self, ln): """Executes GDB/MI thread-select handler function """ GDBMIResultHandler.execute(self, ln) if self.result_class != self.RC_DONE: return class GDBMIThreadInfoHandler(GDBMIResultHandler): """GDB/MI thread-info handler class """ def __init__(self, verbose=False): """Constructor for GDB/MI result handler """ super(GDBMIThreadInfoHandler, self).__init__(verbose) self.current = False self.id = '' self.target_id = '' self.details = '' self.name = '' self.frame = '' self.state = '' self.core = '' def execute(self, ln): """Executes GDB/MI thread-info handler function """ GDBMIResultHandler.execute(self, ln) if self.result_class != self.RC_DONE: return # simple parsing method result = re.search(r'id="(\d+)"', self.result_str) if result: self.id = result.group(1) result = re.search(r'current="\*"', self.result_str) if result: self.current = True result = re.search(r'target-id="([^"]+)"', self.result_str) if result: self.target_id = result.group(1) class GDBMIDataEvalHandler(GDBMIResultHandler): """GDB/MI data-evaluate-expression handler class """ def __init__(self, verbose=False): """Constructor for GDB/MI result handler """ super(GDBMIDataEvalHandler, self).__init__(verbose) self.value = '' def execute(self, ln): """Executes GDB/MI data-evaluate-expression handler function """ GDBMIResultHandler.execute(self, ln) if self.result_class != self.RC_DONE: return # simple parsing method if self.verbose: logging.debug("GDBMIDataEvalHandler: result '%s'", self.result_str) pos = 0 r = re.compile(r'([a-zA-Z_]+)=(.+)\,') while True: m = r.search(self.result_str, pos=pos) if not m: break if m.group(1) == 'value': if self.verbose: logging.debug("GDBMIDataEvalHandler: found value = '%s'", m.group(2)) self.value = self.result.group(1) return pos = m.end(2) + 1 res_str = self.result_str[pos:] res_str = res_str.replace(r'\"', '\'') m = re.search(r'value="([^"]+)"', res_str) if m: if self.verbose: logging.debug("GDBMIDataEvalHandler: found value = '%s'", m.group(1)) self.value = m.group(1) class GDBMIStreamConsoleHandler(GDBMIOutStreamHandler): """GDB/MI console stream handler class """ TAG = '~' def load_aux_elf(elf_path): """ Loads auxilary ELF file and composes GDB command to read its symbols """ elf = None sym_cmd = '' if os.path.exists(elf_path): elf = ESPCoreDumpElfFile(elf_path) for s in elf.sections: if s.name == '.text': sym_cmd = 'add-symbol-file %s 0x%x' % (elf_path, s.addr) return (elf, sym_cmd) def dbg_corefile(args): """ Command to load core dump from file or flash and run GDB debug session with it """ global CLOSE_FDS loader = None rom_elf,rom_sym_cmd = load_aux_elf(args.rom_elf) if not args.core: loader = ESPCoreDumpFlashLoader(args.off, port=args.port, baud=args.baud) core_fname = loader.create_corefile(args.save_core, exe_name=args.prog, rom_elf=rom_elf) if not core_fname: logging.error("Failed to create corefile!") loader.cleanup() return else: core_fname = args.core if args.core_format and args.core_format != 'elf': loader = ESPCoreDumpFileLoader(core_fname, args.core_format == 'b64') core_fname = loader.create_corefile(args.save_core, exe_name=args.prog, rom_elf=rom_elf) if not core_fname: logging.error("Failed to create corefile!") loader.cleanup() return p = subprocess.Popen(bufsize=0, args=[args.gdb, '--nw', # ignore .gdbinit '--core=%s' % core_fname, # core file, '-ex', rom_sym_cmd, args.prog ], stdin=None, stdout=None, stderr=None, close_fds=CLOSE_FDS ) p.wait() if loader: if not args.core and not args.save_core: loader.remove_tmp_file(core_fname) loader.cleanup() print('Done!') def info_corefile(args): """ Command to load core dump from file or flash and print it's data in user friendly form """ global CLOSE_FDS def gdbmi_console_stream_handler(ln): sys.stdout.write(ln) sys.stdout.flush() def gdbmi_read2prompt(f, out_handlers=None): while True: ln = f.readline().decode('utf-8').rstrip(' \r\n') if ln == '(gdb)': break elif len(ln) == 0: break elif out_handlers: for h in out_handlers: if ln.startswith(out_handlers[h].TAG): out_handlers[h].execute(ln) break def gdbmi_start(handlers, gdb_cmds): gdb_args = [args.gdb, '--quiet', # inhibit dumping info at start-up '--nx', # inhibit window interface '--nw', # ignore .gdbinit '--interpreter=mi2', # use GDB/MI v2 '--core=%s' % core_fname] # core file for c in gdb_cmds: gdb_args += ['-ex', c] gdb_args.append(args.prog) p = subprocess.Popen(bufsize=0, args=gdb_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, close_fds=CLOSE_FDS) gdbmi_read2prompt(p.stdout, handlers) return p def gdbmi_cmd_exec(p, handlers, gdbmi_cmd): for t in handlers: handlers[t].result_class = None p.stdin.write(bytearray("%s\n" % gdbmi_cmd, encoding='utf-8')) gdbmi_read2prompt(p.stdout, handlers) if not handlers[GDBMIResultHandler.TAG].result_class or handlers[GDBMIResultHandler.TAG].result_class == GDBMIResultHandler.RC_EXIT: logging.error("GDB exited (%s / %s)!" % (handlers[GDBMIResultHandler.TAG].result_class, handlers[GDBMIResultHandler.TAG].result_str)) p.wait() logging.error("Problem occured! GDB exited, restart it.") p = gdbmi_start(handlers, []) elif handlers[GDBMIResultHandler.TAG].result_class != GDBMIResultHandler.RC_DONE: logging.error("GDB/MI command failed (%s / %s)!" % (handlers[GDBMIResultHandler.TAG].result_class, handlers[GDBMIResultHandler.TAG].result_str)) return p def gdbmi_getinfo(p, handlers, gdb_cmd): return gdbmi_cmd_exec(p, handlers, "-interpreter-exec console \"%s\"" % gdb_cmd) def gdbmi_get_thread_ids(p): handlers = {} result = GDBMIThreadListIdsHandler(verbose=False) handlers[GDBMIResultHandler.TAG] = result handlers[GDBMIStreamConsoleHandler.TAG] = GDBMIStreamConsoleHandler(None, verbose=False) p = gdbmi_cmd_exec(p, handlers, "-thread-list-ids") return p,result.threads,result.current_thread def gdbmi_switch_thread(p, thr_id): handlers = {} result = GDBMIThreadSelectHandler(verbose=False) handlers[GDBMIResultHandler.TAG] = result handlers[GDBMIStreamConsoleHandler.TAG] = GDBMIStreamConsoleHandler(None, verbose=False) return gdbmi_cmd_exec(p, handlers, "-thread-select %s" % thr_id) def gdbmi_get_thread_info(p, thr_id): handlers = {} result = GDBMIThreadInfoHandler(verbose=False) handlers[GDBMIResultHandler.TAG] = result handlers[GDBMIStreamConsoleHandler.TAG] = GDBMIStreamConsoleHandler(None, verbose=False) if thr_id: cmd = "-thread-info %s" % thr_id else: cmd = "-thread-info" p = gdbmi_cmd_exec(p, handlers, cmd) return p,result def gdbmi_data_evaluate_expression(p, expr): handlers = {} result = GDBMIDataEvalHandler(verbose=False) handlers[GDBMIResultHandler.TAG] = result handlers[GDBMIStreamConsoleHandler.TAG] = GDBMIStreamConsoleHandler(None, verbose=False) p = gdbmi_cmd_exec(p, handlers, "-data-evaluate-expression \"%s\"" % expr) return p,result def gdbmi_freertos_get_task_name(p, tcb_addr): p,res = gdbmi_data_evaluate_expression(p, "(char*)((TCB_t *)0x%x)->pcTaskName" % tcb_addr) result = re.match("0x[a-fA-F0-9]+[^']*'([^']*)'", res.value) if result: return p,result.group(1) return p,'' def gdb2freertos_thread_id(gdb_thread_id): return int(gdb_thread_id.replace("process ", ""), 0) loader = None rom_elf,rom_sym_cmd = load_aux_elf(args.rom_elf) if not args.core: loader = ESPCoreDumpFlashLoader(args.off, port=args.port, baud=args.baud) core_fname = loader.create_corefile(args.save_core, exe_name=args.prog, rom_elf=rom_elf) if not core_fname: logging.error("Failed to create corefile!") loader.cleanup() return else: core_fname = args.core if args.core_format and args.core_format != 'elf': loader = ESPCoreDumpFileLoader(core_fname, args.core_format == 'b64') core_fname = loader.create_corefile(args.save_core, exe_name=args.prog, rom_elf=rom_elf) if not core_fname: logging.error("Failed to create corefile!") loader.cleanup() return exe_elf = ESPCoreDumpElfFile(args.prog) core_elf = ESPCoreDumpElfFile(core_fname) merged_segs = [] core_segs = core_elf.program_segments for s in exe_elf.sections: merged = False for ps in core_segs: if ps.addr <= s.addr and ps.addr + len(ps.data) >= s.addr: # sec: |XXXXXXXXXX| # seg: |...XXX.............| seg_addr = ps.addr if ps.addr + len(ps.data) <= s.addr + len(s.data): # sec: |XXXXXXXXXX| # seg: |XXXXXXXXXXX...| # merged: |XXXXXXXXXXXXXX| seg_len = len(s.data) + (s.addr - ps.addr) else: # sec: |XXXXXXXXXX| # seg: |XXXXXXXXXXXXXXXXX| # merged: |XXXXXXXXXXXXXXXXX| seg_len = len(ps.data) merged_segs.append((s.name, seg_addr, seg_len, s.attr_str(), True)) core_segs.remove(ps) merged = True elif ps.addr >= s.addr and ps.addr <= s.addr + len(s.data): # sec: |XXXXXXXXXX| # seg: |...XXX.............| seg_addr = s.addr if (ps.addr + len(ps.data)) >= (s.addr + len(s.data)): # sec: |XXXXXXXXXX| # seg: |..XXXXXXXXXXX| # merged: |XXXXXXXXXXXXX| seg_len = len(s.data) + (ps.addr + len(ps.data)) - (s.addr + len(s.data)) else: # sec: |XXXXXXXXXX| # seg: |XXXXXX| # merged: |XXXXXXXXXX| seg_len = len(s.data) merged_segs.append((s.name, seg_addr, seg_len, s.attr_str(), True)) core_segs.remove(ps) merged = True if not merged: merged_segs.append((s.name, s.addr, len(s.data), s.attr_str(), False)) handlers = {} handlers[GDBMIResultHandler.TAG] = GDBMIResultHandler(verbose=False) handlers[GDBMIStreamConsoleHandler.TAG] = GDBMIStreamConsoleHandler(None, verbose=False) p = gdbmi_start(handlers, [rom_sym_cmd]) extra_note = None task_info = [] for seg in core_elf.aux_segments: if seg.type != ESPCoreDumpElfFile.PT_NOTE: continue note_read = 0 while note_read < len(seg.data): note = Elf32NoteDesc("", 0, None) note_read += note.read(seg.data[note_read:]) if note.type == ESPCoreDumpLoader.ESP_CORE_DUMP_EXTRA_INFO_TYPE and 'EXTRA_INFO' in note.name: extra_note = note if note.type == ESPCoreDumpLoader.ESP_CORE_DUMP_TASK_INFO_TYPE and 'TASK_INFO' in note.name: task_info_struct = EspCoreDumpTaskStatus(buf=note.desc) task_info.append(task_info_struct) print("===============================================================") print("==================== ESP32 CORE DUMP START ====================") handlers[GDBMIResultHandler.TAG].result_class = None handlers[GDBMIStreamConsoleHandler.TAG].func = gdbmi_console_stream_handler if extra_note: extra_info = struct.unpack("<%dL" % (len(extra_note.desc) / struct.calcsize("