#!/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: sys.stderr.write('Import has failed probably because of the missing "future" package. Please install all the packages for ' 'interpreter {} from the $IDF_PATH/requirements.txt file.\n'.format(sys.executable)) sys.exit(1) import os import argparse import subprocess import tempfile import struct import base64 import binascii import logging import re import time from pygdbmi.gdbcontroller import GdbController, DEFAULT_GDB_TIMEOUT_SEC IDF_PATH = os.getenv('IDF_PATH') if not IDF_PATH: sys.stderr.write("IDF_PATH is not found! Set proper IDF_PATH in environment.\n") sys.exit(2) sys.path.insert(0, os.path.join(IDF_PATH, 'components', 'esptool_py', 'esptool')) try: import esptool except ImportError: sys.stderr.write("esptool is not found!\n") sys.exit(2) PARTTOOL_PY = os.path.join(IDF_PATH, "components", "partition_table", "parttool.py") ESPTOOL_PY = os.path.join(IDF_PATH, "components", "esptool_py", "esptool", "esptool.py") try: import typing except ImportError: pass # only needed for type annotations, ignore if not found __version__ = "0.4-dev" if os.name == 'nt': CLOSE_FDS = False else: CLOSE_FDS = True INVALID_CAUSE_VALUE = 0xFFFF XCHAL_EXCCAUSE_NUM = 64 # 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"), # ESP panic pseudo reasons XCHAL_EXCCAUSE_NUM + 0: ("UnknownException", "Unknown exception"), XCHAL_EXCCAUSE_NUM + 1: ("DebugException", "Unhandled debug exception"), XCHAL_EXCCAUSE_NUM + 2: ("DoubleException", "Double exception"), XCHAL_EXCCAUSE_NUM + 3: ("KernelException", "Unhandled kernel exception"), XCHAL_EXCCAUSE_NUM + 4: ("CoprocessorException", "Coprocessor exception"), XCHAL_EXCCAUSE_NUM + 5: ("InterruptWDTTimoutCPU0", "Interrupt wdt timeout on CPU0"), XCHAL_EXCCAUSE_NUM + 6: ("InterruptWDTTimoutCPU1", "Interrupt wdt timeout on CPU1"), XCHAL_EXCCAUSE_NUM + 7: ("CacheError", "Cache disabled but cached memory region accessed"), } 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) class ESPCoreDumpVersion(object): """Core dump version class """ # This class contains all version-dependent params ESP_CORE_DUMP_CHIP_ESP32 = 0 ESP_CORE_DUMP_CHIP_ESP32S2 = 2 def __init__(self, version=None): """Constructor for core dump version """ super(ESPCoreDumpVersion, self).__init__() if version is None: self.version = 0 else: self.set_version(version) @staticmethod def make_dump_ver(maj, min): return (((maj & 0xFF) << 8) | ((min & 0xFF) << 0)) def set_version(self, version): self.version = version @property def chip_ver(self): return ((self.version & 0xFFFF0000) >> 16) @property def dump_ver(self): return (self.version & 0x0000FFFF) @property def major(self): return ((self.version & 0x0000FF00) >> 8) @property def minor(self): return (self.version & 0x000000FF) class ESPCoreDumpLoader(ESPCoreDumpVersion): """Core dump loader base class """ # "legacy" stands for core dumps v0.1 (before IDF v4.1) ESP_COREDUMP_VERSION_BIN_V1 = ESPCoreDumpVersion.make_dump_ver(0, 1) ESP_COREDUMP_VERSION_BIN_V2 = ESPCoreDumpVersion.make_dump_ver(0, 2) ESP_COREDUMP_VERSION_ELF_CRC32 = ESPCoreDumpVersion.make_dump_ver(1, 0) ESP_COREDUMP_VERSION_ELF_SHA256 = ESPCoreDumpVersion.make_dump_ver(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 ESP_COREDUMP_BIN_V1_HDR_FMT = '<4L' ESP_COREDUMP_BIN_V1_HDR_SZ = struct.calcsize(ESP_COREDUMP_BIN_V1_HDR_FMT) ESP_COREDUMP_HDR_FMT = '<5L' ESP_COREDUMP_HDR_SZ = struct.calcsize(ESP_COREDUMP_HDR_FMT) ESP_COREDUMP_TSK_HDR_FMT = '<3L' ESP_COREDUMP_TSK_HDR_SZ = struct.calcsize(ESP_COREDUMP_TSK_HDR_FMT) ESP_COREDUMP_MEM_SEG_HDR_FMT = '<2L' ESP_COREDUMP_MEM_SEG_HDR_SZ = struct.calcsize(ESP_COREDUMP_MEM_SEG_HDR_FMT) ESP_COREDUMP_NOTE_HDR_FMT = '<3L' ESP_COREDUMP_NOTE_HDR_SZ = struct.calcsize(ESP_COREDUMP_NOTE_HDR_FMT) ESP_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 _extract_elf_corefile(self, off=0, exe_name=None): """ Reads the ELF formatted core dump image and parse it """ core_off = off self.set_version(self.hdr['ver']) if self.dump_ver == self.ESP_COREDUMP_VERSION_ELF_CRC32: checksum_len = self.ESP_COREDUMP_CRC_SZ elif self.dump_ver == self.ESP_COREDUMP_VERSION_ELF_SHA256: checksum_len = self.ESP_COREDUMP_SHA256_SZ else: raise ESPCoreDumpLoaderError("Core dump version '%d' is not supported!" % self.dump_ver) core_elf = ESPCoreDumpElfFile() data = self.read_data(core_off, self.hdr['tot_len'] - checksum_len - self.ESP_COREDUMP_HDR_SZ) try: self.core_elf_file.write(data) self.core_elf_file.flush() self.core_elf_file.seek(0) core_elf._read_elf_file(self.core_elf_file) 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(" None """Creates core dump ELF file """ core_off = off tcbsz_aligned = self.hdr['tcbsz'] if tcbsz_aligned % 4: tcbsz_aligned = 4 * (old_div(tcbsz_aligned,4) + 1) core_elf = ESPCoreDumpElfFile() notes = b'' core_dump_info_notes = b'' task_info_notes = b'' task_status = EspCoreDumpTaskStatus() for i in range(self.hdr['task_num']): task_status.task_index = i task_status.task_flags = EspCoreDumpTaskStatus.TASK_STATUS_CORRECT data = self.read_data(core_off, self.ESP_COREDUMP_TSK_HDR_SZ) tcb_addr,stack_top,stack_end = struct.unpack_from(self.ESP_COREDUMP_TSK_HDR_FMT, data) if stack_end > 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.ESP_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: raise ESPCoreDumpError(str(e)) 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(" None """Creates core dump ELF file """ off = 0 data = self.read_data(off, self.ESP_COREDUMP_HDR_SZ) vals = struct.unpack_from(self.ESP_COREDUMP_HDR_FMT, data) self.hdr = dict(zip(('tot_len', 'ver', 'task_num', 'tcbsz', 'segs_num'), vals)) self.core_elf_file = self.create_temp_file() self.set_version(self.hdr['ver']) if self.chip_ver == ESPCoreDumpVersion.ESP_CORE_DUMP_CHIP_ESP32S2 or self.chip_ver == ESPCoreDumpVersion.ESP_CORE_DUMP_CHIP_ESP32: if self.dump_ver == self.ESP_COREDUMP_VERSION_ELF_CRC32 or self.dump_ver == self.ESP_COREDUMP_VERSION_ELF_SHA256: self._extract_elf_corefile(off + self.ESP_COREDUMP_HDR_SZ, exe_name) elif self.dump_ver == self.ESP_COREDUMP_VERSION_BIN_V2: self._extract_bin_corefile(off + self.ESP_COREDUMP_HDR_SZ) elif self.dump_ver == self.ESP_COREDUMP_VERSION_BIN_V1: self._extract_bin_corefile(off + self.ESP_COREDUMP_BIN_V1_HDR_SZ) else: raise ESPCoreDumpLoaderError("Core dump version '0x%x' is not supported!" % (self.dump_ver)) else: raise ESPCoreDumpLoaderError("Core dump chip '0x%x' is not supported!" % (self.chip_ver)) self.core_elf_file.flush() def read_data(self, off, sz): """Reads data from raw core dump got from flash or UART """ self.core_src_file.seek(off) data = self.core_src_file.read(sz) return data def cleanup(self): def remove(f, timeout_sec): """ Removes the file. Waits `timeout_sec` for the file to stop being used """ timeout = time.time() + timeout_sec while (time.time() <= timeout): try: os.remove(f) return except OSError: time.sleep(.1) logging.warning("File \'%s\' is used by another process and can't be removed" % f) logging.debug("Cleaning up...") if self.core_elf_file: self.core_elf_file.close() self.core_elf_file = None if self.core_src_file: self.core_src_file.close() self.core_src_file = None for t in self.temp_files: remove(t, 2) class ESPCoreDumpFileLoader(ESPCoreDumpLoader): """Core dump file loader class """ def __init__(self, path, b64=False): """Constructor for core dump file loader """ super(ESPCoreDumpFileLoader, self).__init__() self._load_coredump(path, b64) def _load_coredump(self, path, b64): """Loads core dump from (raw binary or base64-encoded) file """ logging.debug("Load core dump from '%s', %s format" % (path, "b64" if b64 else "raw")) if not b64: self.core_src_file = open(path, mode="rb") else: self.core_src_file = self.create_temp_file() with open(path, 'rb') as fb64: while True: line = fb64.readline() if len(line) == 0: break data = base64.standard_b64decode(line.rstrip(b'\r\n')) self.core_src_file.write(data) self.core_src_file.flush() self.core_src_file.seek(0) class ESPCoreDumpFlashLoader(ESPCoreDumpLoader): """Core dump flash loader class """ ESP_COREDUMP_FLASH_LEN_FMT = ' 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) raise e def _load_coredump(self, off=None): """Loads core dump from flash using parttool or elftool (if offset is set) """ try: if off: logging.info("Invoke esptool to read image.") self.invoke_esptool(off=off) else: logging.info("Invoke parttool to read image.") self.invoke_parttool() except subprocess.CalledProcessError as e: if len(e.output): logging.info(e.output) logging.error("Error during the subprocess execution") def _read_core_dump_length(self, f): """Reads core dump length """ data = f.read(self.ESP_COREDUMP_FLASH_LEN_SZ) tot_len, = struct.unpack_from(self.ESP_COREDUMP_FLASH_LEN_FMT, data) return tot_len def create_corefile(self, exe_name=None): # type: (str) -> None """Checks flash coredump data integrity and creates ELF file """ data = self.read_data(0, self.ESP_COREDUMP_HDR_SZ) self.checksum_len = 0 _,coredump_ver_data,_,_,_ = struct.unpack_from(self.ESP_COREDUMP_HDR_FMT, data) self.set_version(coredump_ver_data) if self.chip_ver != ESPCoreDumpVersion.ESP_CORE_DUMP_CHIP_ESP32S2 and self.chip_ver != ESPCoreDumpVersion.ESP_CORE_DUMP_CHIP_ESP32: raise ESPCoreDumpLoaderError("Invalid core dump chip version: '%s', should be <= '0x%x'" % (self.chip_ver, self.ESP_CORE_DUMP_CHIP_ESP32S2)) if self.dump_ver == self.ESP_COREDUMP_VERSION_ELF_CRC32 or self.dump_ver == self.ESP_COREDUMP_VERSION_BIN_V1 \ or self.dump_ver == self.ESP_COREDUMP_VERSION_BIN_V2: logging.debug("Dump size = %d, crc off = 0x%x", self.dump_sz, self.dump_sz - self.ESP_COREDUMP_CRC_SZ) data = self.read_data(self.dump_sz - self.ESP_COREDUMP_CRC_SZ, self.ESP_COREDUMP_CRC_SZ) dump_crc, = struct.unpack_from(self.ESP_COREDUMP_CRC_FMT, data) data = self.read_data(0, self.dump_sz - self.ESP_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 self.dump_ver == self.ESP_COREDUMP_VERSION_ELF_SHA256: dump_sha256 = self.read_data(self.dump_sz - self.ESP_COREDUMP_SHA256_SZ, self.ESP_COREDUMP_SHA256_SZ) data = self.read_data(0, self.dump_sz - self.ESP_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)) super(ESPCoreDumpFlashLoader, self).create_corefile(exe_name) def load_aux_elf(elf_path): # type: (str) -> (ESPCoreDumpElfFile, str) """ Loads auxiliary 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 core_prepare(args): loader = None core_filename = None if not args.core: # Core file not specified, try to read core dump from flash. loader = ESPCoreDumpFlashLoader(args.off, port=args.port, baud=args.baud) elif args.core_format != "elf": # Core file specified, but not yet in ELF format. Convert it from raw or base64 into ELF. loader = ESPCoreDumpFileLoader(args.core, args.core_format == 'b64') else: # Core file is already in the ELF format core_filename = args.core # Load/convert the core file if loader: loader.create_corefile(exe_name=args.prog) core_filename = loader.core_elf_file.name if args.save_core: # We got asked to save the core file, make a copy with open(args.save_core, "w+b") as f_out: loader.core_elf_file.seek(0) f_out.write(loader.core_elf_file.read()) return core_filename, loader def dbg_corefile(args): """ Command to load core dump from file or flash and run GDB debug session with it """ rom_elf, rom_sym_cmd = load_aux_elf(args.rom_elf) core_filename, loader = core_prepare(args) p = subprocess.Popen(bufsize=0, args=[args.gdb, '--nw', # ignore .gdbinit '--core=%s' % core_filename, # core file, '-ex', rom_sym_cmd, args.prog ], stdin=None, stdout=None, stderr=None, close_fds=CLOSE_FDS ) p.wait() if loader: loader.cleanup() print('Done!') def gdbmi_filter_responses(responses, resp_message, resp_type): return list(filter(lambda rsp: rsp["message"] == resp_message and rsp["type"] == resp_type, responses)) def gdbmi_run_cmd_get_responses(p, cmd, resp_message, resp_type, multiple=True, done_message=None, done_type=None): \ # type: (GdbController, str, typing.Optional[str], str, bool, typing.Optional[str], typing.Optional[str]) -> list p.write(cmd, read_response=False) t_end = time.time() + DEFAULT_GDB_TIMEOUT_SEC filtered_response_list = [] all_responses = [] while time.time() < t_end: more_responses = p.get_gdb_response(timeout_sec=0, raise_error_on_timeout=False) filtered_response_list += filter(lambda rsp: rsp["message"] == resp_message and rsp["type"] == resp_type, more_responses) all_responses += more_responses if filtered_response_list and not multiple: break if done_message and done_type and gdbmi_filter_responses(more_responses, done_message, done_type): break if not filtered_response_list and not multiple: raise ESPCoreDumpError("Couldn't find response with message '{}', type '{}' in responses '{}'".format( resp_message, resp_type, str(all_responses) )) return filtered_response_list def gdbmi_run_cmd_get_one_response(p, cmd, resp_message, resp_type): # type: (GdbController, str, typing.Optional[str], str) -> dict return gdbmi_run_cmd_get_responses(p, cmd, resp_message, resp_type, multiple=False)[0] def gdbmi_start(gdb_path, gdb_cmds, core_filename, prog_filename): # type: (str, typing.List[str], str, str) -> GdbController """ Start GDB and get GdbController instance which wraps it """ gdb_args = ['--quiet', # inhibit dumping info at start-up '--nx', # inhibit window interface '--nw', # ignore .gdbinit '--interpreter=mi2', # use GDB/MI v2 '--core=%s' % core_filename] # core file for c in gdb_cmds: if c: gdb_args += ['-ex', c] gdb_args.append(prog_filename) res = GdbController(gdb_path=gdb_path, gdb_args=gdb_args) # Consume initial output by issuing a dummy command res.write("-data-list-register-values x pc", timeout_sec=5) return res def gdbmi_cmd_exec_console(p, gdb_cmd): # type: (GdbController, str) -> str """ Execute a generic GDB console command via MI2 """ filtered_responses = gdbmi_run_cmd_get_responses(p, "-interpreter-exec console \"%s\"" % gdb_cmd, None, "console", multiple=True, done_message="done", done_type="result") return "".join([x["payload"] for x in filtered_responses])\ .replace('\\n', '\n').replace('\\t', '\t').rstrip("\n") def gdbmi_get_thread_ids(p): # type: (GdbController) -> (str, typing.List[str]) """ Get the current thread ID and the list of all thread IDs known to GDB, as strings """ result = gdbmi_run_cmd_get_one_response(p, "-thread-list-ids", "done", "result")["payload"] current_thread_id = result["current-thread-id"] thread_ids = result["thread-ids"]["thread-id"] return current_thread_id, thread_ids def gdbmi_switch_thread(p, thr_id): # type: (GdbController, str) -> None """ Tell GDB to switch to a specific thread, given its ID """ gdbmi_run_cmd_get_one_response(p, "-thread-select %s" % thr_id, "done", "result") def gdbmi_get_thread_info(p, thr_id=None): # type: (GdbController, typing.Optional[str]) -> dict """ Get thread info dictionary for the given thread ID """ return gdbmi_run_cmd_get_one_response(p, "-thread-info" + (" %s" % thr_id) if thr_id else "", "done", "result")["payload"]["threads"][0] def gdbmi_data_evaluate_expression(p, expr): # type: (GdbController, str) -> str """ Get the value of an expression, similar to the 'print' command """ return gdbmi_run_cmd_get_one_response(p, "-data-evaluate-expression \"%s\"" % expr, "done", "result")["payload"]["value"] def gdbmi_freertos_get_task_name(p, tcb_addr): # type: (GdbController, int) -> str """ Get FreeRTOS task name given the TCB address """ try: val = gdbmi_data_evaluate_expression(p, "(char*)((TCB_t *)0x%x)->pcTaskName" % tcb_addr) except ESPCoreDumpError: return '' # Value is of form '0x12345678 "task_name"', extract the actual name result = re.search(r"\"([^']*)\"$", val) if result: return result.group(1) return '' def gdb2freertos_thread_id(gdb_target_id): # type: (str) -> int """ Convert GDB 'target ID' to the FreeRTOS TCB address """ return int(gdb_target_id.replace("process ", ""), 0) def info_corefile(args): """ Command to load core dump from file or flash and print it's data in user friendly form """ rom_elf, rom_sym_cmd = load_aux_elf(args.rom_elf) core_filename, loader = core_prepare(args) exe_elf = ESPCoreDumpElfFile(args.prog) core_elf = ESPCoreDumpElfFile(core_filename) 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)) p = gdbmi_start(args.gdb, [rom_sym_cmd], core_filename, args.prog) 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 ====================") if extra_note: extra_info = struct.unpack("<%dL" % (len(extra_note.desc) / struct.calcsize("