espressif/esp-idf
1
0
Fork 1
mirror of https://github.com/espressif/esp-idf.git synced 2025-08-05 13:44:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

core_dump: split corefile and coredump script from idf into a separate package

Browse Source
This commit is contained in:
Aleksei Apaseev
2022-02-17 15:27:34 +08:00
parent 97fb98a91b
commit 528cdc1546
22 changed files with 102 additions and 2057 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitlab/ci/host-test.yml
View File

@@ -185,6 +185,7 @@ test_idf_monitor:
- tools/test_idf_monitor/outputs/* - tools/test_idf_monitor/outputs/*
expire_in: 1 week expire_in: 1 week
script: script:
- eval $($IDF_PATH/tools/idf_tools.py export)
- cd ${IDF_PATH}/tools/test_idf_monitor - cd ${IDF_PATH}/tools/test_idf_monitor
- ./run_test_idf_monitor.py - ./run_test_idf_monitor.py
@@ -276,6 +277,7 @@ test_espcoredump:
# install CMake version specified in tools.json # install CMake version specified in tools.json
SETUP_TOOLS_LIST: "all" SETUP_TOOLS_LIST: "all"
script: script:
- eval $($IDF_PATH/tools/idf_tools.py export)
- retry_failed git clone ${IDF_COREDUMP_ELF_REPO} -b $IDF_COREDUMP_ELF_TAG - retry_failed git clone ${IDF_COREDUMP_ELF_REPO} -b $IDF_COREDUMP_ELF_TAG
- cd ${IDF_PATH}/components/espcoredump/test/ - cd ${IDF_PATH}/components/espcoredump/test/
- ./test_espcoredump.sh ${CI_PROJECT_DIR}/idf-coredump-elf - ./test_espcoredump.sh ${CI_PROJECT_DIR}/idf-coredump-elf

125
components/espcoredump/corefile/__init__.py
View File

@@ -1,125 +0,0 @@
#
# Copyright 2021 Espressif Systems (Shanghai) CO., LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
__version__ = '0.4-dev'
import abc
import os
from abc import abstractmethod
from importlib import import_module
from future.utils import with_metaclass
try:
from typing import Optional, Tuple
except ImportError:
pass
IDF_PATH = os.path.normpath(os.getenv('IDF_PATH', '.'))
XTENSA_TARGETS = ['esp32', 'esp32s2', 'esp32s3']
RISCV_TARGETS = ['esp32c3']
SUPPORTED_TARGETS = XTENSA_TARGETS + RISCV_TARGETS
class ESPCoreDumpError(RuntimeError):
pass
class ESPCoreDumpLoaderError(ESPCoreDumpError):
pass
class BaseArchMethodsMixin(with_metaclass(abc.ABCMeta)): # type: ignore
@staticmethod
@abstractmethod
def get_registers_from_stack(data, grows_down):
# type: (bytes, bool) -> Tuple[list[int], Optional[dict[int, int]]]
"""
Parse stack data, growing up stacks are not supported for now.
:param data: stack data
:param grows_down: stack grow direction
:return: return tuple (regs, exception_regs)
"""
pass
@staticmethod
@abstractmethod
def build_prstatus_data(tcb_addr, task_regs): # type: (int, list[int]) -> str
"""
Build PrStatus note section
:param tcb_addr: tcb addr
:param task_regs: registers
:return: str
"""
pass
class BaseTargetMethods(with_metaclass(abc.ABCMeta, BaseArchMethodsMixin)): # type: ignore
UNKNOWN = 'unknown'
TARGET = UNKNOWN
COREDUMP_FAKE_STACK_START = 0x20000000
COREDUMP_FAKE_STACK_LIMIT = 0x30000000
COREDUMP_MAX_TASK_STACK_SIZE = 64 * 1024
def __init__(self): # type: () -> None
if self.TARGET == self.UNKNOWN:
raise ValueError('Please use the derived child-class with valid TARGET')
self._set_attr_from_soc_header()
def _set_attr_from_soc_header(self): # type: () -> None
module = import_module('corefile.soc_headers.{}'.format(self.TARGET))
for k, v in module.__dict__.items():
if k.startswith('SOC_'):
setattr(self, k, v)
def _esp_ptr_in_dram(self, addr): # type: (int) -> bool
return self.SOC_DRAM_LOW <= addr < self.SOC_DRAM_HIGH # type: ignore
def _esp_ptr_in_iram(self, addr): # type: (int) -> bool
return self.SOC_IRAM_LOW <= addr < self.SOC_IRAM_HIGH # type: ignore
def _esp_ptr_in_rtc_slow(self, addr): # type: (int) -> bool
return self.SOC_RTC_DATA_LOW <= addr < self.SOC_RTC_DATA_HIGH # type: ignore
def _esp_ptr_in_rtc_dram_fast(self, addr): # type: (int) -> bool
return self.SOC_RTC_DRAM_LOW <= addr < self.SOC_RTC_DRAM_HIGH # type: ignore
def tcb_is_sane(self, tcb_addr, tcb_size): # type: (int, int) -> bool
for func in [self._esp_ptr_in_dram,
self._esp_ptr_in_iram,
self._esp_ptr_in_rtc_slow,
self._esp_ptr_in_rtc_dram_fast]:
res = func(tcb_addr) and func(tcb_addr + tcb_size - 1)
if res:
return True
return False
def _esp_stack_ptr_in_dram(self, addr): # type: (int) -> bool
return not (addr < self.SOC_DRAM_LOW + 0x10
or addr > self.SOC_DRAM_HIGH - 0x10
or (addr & 0xF) != 0)
def stack_is_sane(self, stack_start, stack_end): # type: (int, int) -> bool
return (self._esp_stack_ptr_in_dram(stack_start)
and self._esp_ptr_in_dram(stack_end)
and stack_start < stack_end
and (stack_end - stack_start) < self.COREDUMP_MAX_TASK_STACK_SIZE)
def addr_is_fake(self, addr): # type: (int) -> bool
return (self.COREDUMP_FAKE_STACK_START <= addr < self.COREDUMP_FAKE_STACK_LIMIT
or addr > 2 ** 31 - 1)

44
components/espcoredump/corefile/_parse_soc_header.py
View File

@@ -1,44 +0,0 @@
"""
This file is used to generate soc header constants into sub-package soc_headers
"""
import os
from ast import literal_eval
from corefile import IDF_PATH, SUPPORTED_TARGETS
def main(): # type: () -> None
constants = [
'SOC_DRAM_LOW',
'SOC_DRAM_HIGH',
'SOC_IRAM_LOW',
'SOC_IRAM_HIGH',
'SOC_RTC_DATA_LOW',
'SOC_RTC_DATA_HIGH',
'SOC_RTC_DRAM_LOW',
'SOC_RTC_DRAM_HIGH',
]
for target in SUPPORTED_TARGETS:
target_constants = {}
soc_header_fp = os.path.join(IDF_PATH, 'components/soc/{}/include/soc/soc.h'.format(target))
module_fp = os.path.join(IDF_PATH, 'components', 'espcoredump', 'corefile', 'soc_headers',
'{}.py'.format(target))
with open(soc_header_fp) as fr:
for line in fr.readlines():
for attr in constants:
if '#define {}'.format(attr) in line:
target_constants[attr] = literal_eval(line.strip().split()[-1])
for attr in constants:
if attr not in target_constants:
raise ValueError('ERROR: Attr {} is missing in {}'.format(attr, soc_header_fp))
with open(module_fp, 'w') as fw:
for k, v in target_constants.items():
fw.write('{} = {}\n'.format(k, hex(v)))
if __name__ == '__main__':
main()

372
components/espcoredump/corefile/elf.py
View File

@@ -1,372 +0,0 @@
#
# Copyright 2021 Espressif Systems (Shanghai) CO., LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import hashlib
import os
from construct import (AlignedStruct, Bytes, Const, Container, GreedyRange, Int16ul, Int32ul, Padding, Pointer,
Sequence, Struct, this)
try:
from typing import Optional
except ImportError:
pass
# Following structs are based on spec
# https://refspecs.linuxfoundation.org/elf/elf.pdf
# and source code
# IDF_PATH/components/espcoredump/include_core_dump/elf.h
ElfIdentification = Struct(
'EI_MAG' / Const(b'\x7fELF'),
'EI_CLASS' / Const(b'\x01'), # ELFCLASS32
'EI_DATA' / Const(b'\x01'), # ELFDATA2LSB
'EI_VERSION' / Const(b'\x01'), # EV_CURRENT
Padding(9),
)
ElfHeader = Struct(
'e_ident' / ElfIdentification,
'e_type' / Int16ul,
'e_machine' / Int16ul,
'e_version' / Int32ul,
'e_entry' / Int32ul,
'e_phoff' / Int32ul,
'e_shoff' / Int32ul,
'e_flags' / Int32ul,
'e_ehsize' / Int16ul,
'e_phentsize' / Int16ul,
'e_phnum' / Int16ul,
'e_shentsize' / Int16ul,
'e_shnum' / Int16ul,
'e_shstrndx' / Int16ul,
)
SectionHeader = Struct(
'sh_name' / Int32ul,
'sh_type' / Int32ul,
'sh_flags' / Int32ul,
'sh_addr' / Int32ul,
'sh_offset' / Int32ul,
'sh_size' / Int32ul,
'sh_link' / Int32ul,
'sh_info' / Int32ul,
'sh_addralign' / Int32ul,
'sh_entsize' / Int32ul,
)
ProgramHeader = Struct(
'p_type' / Int32ul,
'p_offset' / Int32ul,
'p_vaddr' / Int32ul,
'p_paddr' / Int32ul,
'p_filesz' / Int32ul,
'p_memsz' / Int32ul,
'p_flags' / Int32ul,
'p_align' / Int32ul,
)
ElfHeaderTables = Struct(
'elf_header' / ElfHeader,
'program_headers' / Pointer(this.elf_header.e_phoff, ProgramHeader[this.elf_header.e_phnum]),
'section_headers' / Pointer(this.elf_header.e_shoff, SectionHeader[this.elf_header.e_shnum]),
)
NoteSection = AlignedStruct(
4,
'namesz' / Int32ul,
'descsz' / Int32ul,
'type' / Int32ul,
'name' / Bytes(this.namesz),
'desc' / Bytes(this.descsz),
)
NoteSections = GreedyRange(NoteSection)
class ElfFile(object):
"""
Elf class to a single elf file
"""
SHN_UNDEF = 0x00
SHT_PROGBITS = 0x01
SHT_STRTAB = 0x03
SHT_NOBITS = 0x08
PT_LOAD = 0x01
PT_NOTE = 0x04
ET_CORE = 0x04
EV_CURRENT = 0x01
def __init__(self, elf_path=None, e_type=None, e_machine=None):
# type: (Optional[str], Optional[int], Optional[int]) -> None
self.e_type = e_type
self.e_machine = e_machine
self._struct = None # type: Optional[Struct]
self._model = None # type: Optional[Container]
self.sections = [] # type: list[ElfSection]
self.load_segments = [] # type: list[ElfSegment]
self.note_segments = [] # type: list[ElfNoteSegment]
if elf_path and os.path.isfile(elf_path):
self.read_elf(elf_path)
def read_elf(self, elf_path): # type: (str) -> None
"""
Read elf file, also write to ``self.model``, ``self.program_headers``,
``self.section_headers``
:param elf_path: elf file path
:return: None
"""
with open(elf_path, 'rb') as fr:
elf_bytes = fr.read()
header_tables = ElfHeaderTables.parse(elf_bytes)
self.e_type = header_tables.elf_header.e_type
self.e_machine = header_tables.elf_header.e_machine
self._struct = self._generate_struct_from_headers(header_tables)
self._model = self._struct.parse(elf_bytes)
self.load_segments = [ElfSegment(seg.ph.p_vaddr,
seg.data,
seg.ph.p_flags) for seg in self._model.load_segments]
self.note_segments = [ElfNoteSegment(seg.ph.p_vaddr,
seg.data,
seg.ph.p_flags) for seg in self._model.note_segments]
self.sections = [ElfSection(self._parse_string_table(self._model.string_table, sec.sh.sh_name),
sec.sh.sh_addr,
sec.data,
sec.sh.sh_flags) for sec in self._model.sections]
@staticmethod
def _parse_string_table(byte_str, offset): # type: (bytes, int) -> str
section_name_str = byte_str[offset:]
string_end = section_name_str.find(0x00)
if (string_end == -1):
raise ValueError('Unable to get section name from section header string table')
name = section_name_str[:string_end].decode('utf-8')
return name
def _generate_struct_from_headers(self, header_tables): # type: (Container) -> Struct
"""
Generate ``construct`` Struct for this file
:param header_tables: contains elf_header, program_headers, section_headers
:return: Struct of the whole file
"""
elf_header = header_tables.elf_header
program_headers = header_tables.program_headers
section_headers = header_tables.section_headers
assert program_headers or section_headers
string_table_sh = None
load_segment_subcons = []
note_segment_subcons = []
# Here we point back to make segments know their program headers
for i, ph in enumerate(program_headers):
args = [
'ph' / Pointer(elf_header.e_phoff + i * ProgramHeader.sizeof(), ProgramHeader),
'data' / Pointer(ph.p_offset, Bytes(ph.p_filesz)),
]
if ph.p_vaddr == 0 and ph.p_type == self.PT_NOTE:
args.append('note_secs' / Pointer(ph.p_offset, NoteSections))
note_segment_subcons.append(Struct(*args))
elif ph.p_vaddr != 0:
load_segment_subcons.append(Struct(*args))
section_subcons = []
for i, sh in enumerate(section_headers):
if sh.sh_type == self.SHT_STRTAB and i == elf_header.e_shstrndx:
string_table_sh = sh
elif sh.sh_addr != 0 and sh.sh_type == self.SHT_PROGBITS:
section_subcons.append(Struct(
'sh' / Pointer(elf_header.e_shoff + i * SectionHeader.sizeof(), SectionHeader),
'data' / Pointer(sh.sh_offset, Bytes(sh.sh_size)),
))
args = [
'elf_header' / ElfHeader,
'load_segments' / Sequence(*load_segment_subcons),
'note_segments' / Sequence(*note_segment_subcons),
'sections' / Sequence(*section_subcons),
]
if string_table_sh is not None:
args.append('string_table' / Pointer(string_table_sh.sh_offset, Bytes(string_table_sh.sh_size)))
return Struct(*args)
@property
def sha256(self): # type: () -> bytes
"""
:return: SHA256 hash of the input ELF file
"""
sha256 = hashlib.sha256()
sha256.update(self._struct.build(self._model)) # type: ignore
return sha256.digest()
class ElfSection(object):
SHF_WRITE = 0x01
SHF_ALLOC = 0x02
SHF_EXECINSTR = 0x04
SHF_MASKPROC = 0xf0000000
def __init__(self, name, addr, data, flags): # type: (str, int, bytes, int) -> None
self.name = name
self.addr = addr
self.data = data
self.flags = flags
def attr_str(self): # type: () -> str
if self.flags & self.SHF_MASKPROC:
return 'MS'
res = 'R'
res += 'W' if self.flags & self.SHF_WRITE else ' '
res += 'X' if self.flags & self.SHF_EXECINSTR else ' '
res += 'A' if self.flags & self.SHF_ALLOC else ' '
return res
def __repr__(self): # type: () -> str
return '{:>32} [Addr] 0x{:>08X}, [Size] 0x{:>08X} {:>4}' \
.format(self.name, self.addr, len(self.data), self.attr_str())
class ElfSegment(object):
PF_X = 0x01
PF_W = 0x02
PF_R = 0x04
def __init__(self, addr, data, flags): # type: (int, bytes, int) -> None
self.addr = addr
self.data = data
self.flags = flags
self.type = ElfFile.PT_LOAD
def attr_str(self): # type: () -> str
res = ''
res += 'R' if self.flags & self.PF_R else ' '
res += 'W' if self.flags & self.PF_W else ' '
res += 'E' if self.flags & self.PF_X else ' '
return res
@staticmethod
def _type_str(): # type: () -> str
return 'LOAD'
def __repr__(self): # type: () -> str
return '{:>8} Addr 0x{:>08X}, Size 0x{:>08X} Flags {:4}' \
.format(self._type_str(), self.addr, len(self.data), self.attr_str())
class ElfNoteSegment(ElfSegment):
def __init__(self, addr, data, flags): # type: (int, bytes, int) -> None
super(ElfNoteSegment, self).__init__(addr, data, flags)
self.type = ElfFile.PT_NOTE
self.note_secs = NoteSections.parse(self.data)
@staticmethod
def _type_str(): # type: () -> str
return 'NOTE'
TASK_STATUS_CORRECT = 0x00
TASK_STATUS_TCB_CORRUPTED = 0x01
TASK_STATUS_STACK_CORRUPTED = 0x02
EspTaskStatus = Struct(
'task_index' / Int32ul,
'task_flags' / Int32ul,
'task_tcb_addr' / Int32ul,
'task_stack_start' / Int32ul,
'task_stack_len' / Int32ul,
'task_name' / Bytes(16),
)
class ESPCoreDumpElfFile(ElfFile):
PT_INFO = 8266
PT_TASK_INFO = 678
PT_EXTRA_INFO = 677
CURR_TASK_MARKER = 0xdeadbeef
# ELF file machine type
EM_XTENSA = 0x5E
EM_RISCV = 0xF3
def __init__(self, elf_path=None, e_type=None, e_machine=None):
# type: (Optional[str], Optional[int], Optional[int]) -> None
_e_type = e_type or self.ET_CORE
_e_machine = e_machine or self.EM_XTENSA
super(ESPCoreDumpElfFile, self).__init__(elf_path, _e_type, _e_machine)
def add_segment(self, addr, data, seg_type, flags): # type: (int, bytes, int, int) -> None
if seg_type != self.PT_NOTE:
self.load_segments.append(ElfSegment(addr, data, flags))
else:
self.note_segments.append(ElfNoteSegment(addr, data, flags))
def dump(self, output_path): # type: (str) -> None
"""
Dump self.model into file
:param output_path: output file path
:return: None
"""
res = b''
res += ElfHeader.build({
'e_type': self.e_type,
'e_machine': self.e_machine,
'e_version': self.EV_CURRENT,
'e_entry': 0,
'e_phoff': ElfHeader.sizeof(),
'e_shoff': 0,
'e_flags': 0,
'e_ehsize': ElfHeader.sizeof(),
'e_phentsize': ProgramHeader.sizeof(),
'e_phnum': len(self.load_segments) + len(self.note_segments),
'e_shentsize': 0,
'e_shnum': 0,
'e_shstrndx': self.SHN_UNDEF,
})
offset = ElfHeader.sizeof() + (len(self.load_segments) + len(self.note_segments)) * ProgramHeader.sizeof()
_segments = self.load_segments + self.note_segments # type: ignore
for seg in _segments:
res += ProgramHeader.build({
'p_type': seg.type,
'p_offset': offset,
'p_vaddr': seg.addr,
'p_paddr': seg.addr,
'p_filesz': len(seg.data),
'p_memsz': len(seg.data),
'p_flags': seg.flags,
'p_align': 0,
})
offset += len(seg.data)
for seg in _segments:
res += seg.data
with open(output_path, 'wb') as fw:
fw.write(res)

129
components/espcoredump/corefile/gdb.py
View File

@@ -1,129 +0,0 @@
#
# Copyright 2021 Espressif Systems (Shanghai) CO., LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging
import re
import time
from pygdbmi.gdbcontroller import DEFAULT_GDB_TIMEOUT_SEC, GdbController
from . import ESPCoreDumpError
class EspGDB(object):
def __init__(self, gdb_path, gdb_cmds, core_filename, prog_filename, timeout_sec=DEFAULT_GDB_TIMEOUT_SEC):
"""
Start GDB and initialize a GdbController instance
"""
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)
self.p = GdbController(gdb_path=gdb_path, gdb_args=gdb_args)
self.timeout = timeout_sec
# Consume initial output by issuing a dummy command
self._gdbmi_run_cmd_get_responses(cmd='-data-list-register-values x pc',
resp_message=None, resp_type='console', multiple=True,
done_message='done', done_type='result')
def __del__(self):
try:
self.p.exit()
except IndexError:
logging.warning('Attempt to terminate the GDB process failed, because it is already terminated. Skip')
def _gdbmi_run_cmd_get_responses(self, cmd, resp_message, resp_type, multiple=True,
done_message=None, done_type=None):
self.p.write(cmd, read_response=False)
t_end = time.time() + self.timeout
filtered_response_list = []
all_responses = []
while time.time() < t_end:
more_responses = self.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 self._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(self, cmd, resp_message, resp_type):
return self._gdbmi_run_cmd_get_responses(cmd, resp_message, resp_type, multiple=False)[0]
def _gdbmi_data_evaluate_expression(self, expr):
""" Get the value of an expression, similar to the 'print' command """
return self._gdbmi_run_cmd_get_one_response("-data-evaluate-expression \"%s\"" % expr,
'done', 'result')['payload']['value']
def get_freertos_task_name(self, tcb_addr):
""" Get FreeRTOS task name given the TCB address """
try:
val = self._gdbmi_data_evaluate_expression('(char*)((TCB_t *)0x%x)->pcTaskName' % tcb_addr)
except (ESPCoreDumpError, KeyError):
# KeyError is raised when "value" is not in "payload"
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 run_cmd(self, gdb_cmd):
""" Execute a generic GDB console command via MI2
"""
filtered_responses = self._gdbmi_run_cmd_get_responses(cmd="-interpreter-exec console \"%s\"" % gdb_cmd,
resp_message=None, resp_type='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 get_thread_info(self):
""" Get information about all threads known to GDB, and the current thread ID """
result = self._gdbmi_run_cmd_get_one_response('-thread-info', 'done', 'result')['payload']
current_thread_id = result['current-thread-id']
threads = result['threads']
return threads, current_thread_id
def switch_thread(self, thr_id):
""" Tell GDB to switch to a specific thread, given its ID """
self._gdbmi_run_cmd_get_one_response('-thread-select %s' % thr_id, 'done', 'result')
@staticmethod
def _gdbmi_filter_responses(responses, resp_message, resp_type):
return list(filter(lambda rsp: rsp['message'] == resp_message and rsp['type'] == resp_type, responses))
@staticmethod
def gdb2freertos_thread_id(gdb_target_id):
""" Convert GDB 'target ID' to the FreeRTOS TCB address """
return int(gdb_target_id.replace('process ', ''), 0)

583
components/espcoredump/corefile/loader.py
View File

@@ -1,583 +0,0 @@
#
# Copyright 2021 Espressif Systems (Shanghai) CO., LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import base64
import binascii
import hashlib
import logging
import os
import subprocess
import sys
import tempfile
from construct import AlignedStruct, Bytes, GreedyRange, Int32ul, Padding, Struct, abs_, this
from . import ESPCoreDumpLoaderError
from .elf import (TASK_STATUS_CORRECT, TASK_STATUS_TCB_CORRUPTED, ElfFile, ElfSegment, ESPCoreDumpElfFile,
EspTaskStatus, NoteSection)
from .riscv import Esp32c3Methods
from .xtensa import Esp32Methods, Esp32S2Methods, Esp32S3Methods
try:
from typing import Optional, Tuple
except ImportError:
pass
IDF_PATH = os.getenv('IDF_PATH', '')
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')
# Following structs are based on source code
# components/espcoredump/include_core_dump/esp_core_dump_priv.h
EspCoreDumpV1Header = Struct(
'tot_len' / Int32ul,
'ver' / Int32ul,
'task_num' / Int32ul,
'tcbsz' / Int32ul,
)
EspCoreDumpV2Header = Struct(
'tot_len' / Int32ul,
'ver' / Int32ul,
'task_num' / Int32ul,
'tcbsz' / Int32ul,
'segs_num' / Int32ul,
)
CRC = Int32ul
SHA256 = Bytes(32)
TaskHeader = Struct(
'tcb_addr' / Int32ul,
'stack_top' / Int32ul,
'stack_end' / Int32ul,
)
MemSegmentHeader = Struct(
'mem_start' / Int32ul,
'mem_sz' / Int32ul,
'data' / Bytes(this.mem_sz),
)
class EspCoreDumpVersion(object):
"""Core dump version class, it contains all version-dependent params
"""
# Chip IDs should be in sync with components/esp_hw_support/include/esp_chip_info.h
ESP32 = 0
ESP32S2 = 2
ESP32S3 = 9
XTENSA_CHIPS = [ESP32, ESP32S2, ESP32S3]
ESP32C3 = 5
RISCV_CHIPS = [ESP32C3]
COREDUMP_SUPPORTED_TARGETS = XTENSA_CHIPS + RISCV_CHIPS
def __init__(self, version=None): # type: (int) -> 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(major, minor): # type: (int, int) -> int
return ((major & 0xFF) << 8) | ((minor & 0xFF) << 0)
def set_version(self, version): # type: (int) -> None
self.version = version
@property
def chip_ver(self): # type: () -> int
return (self.version & 0xFFFF0000) >> 16
@property
def dump_ver(self): # type: () -> int
return self.version & 0x0000FFFF
@property
def major(self): # type: () -> int
return (self.version & 0x0000FF00) >> 8
@property
def minor(self): # type: () -> int
return self.version & 0x000000FF
class EspCoreDumpLoader(EspCoreDumpVersion):
# "legacy" stands for core dumps v0.1 (before IDF v4.1)
BIN_V1 = EspCoreDumpVersion.make_dump_ver(0, 1)
BIN_V2 = EspCoreDumpVersion.make_dump_ver(0, 2)
ELF_CRC32 = EspCoreDumpVersion.make_dump_ver(1, 0)
ELF_SHA256 = EspCoreDumpVersion.make_dump_ver(1, 1)
def __init__(self): # type: () -> None
super(EspCoreDumpLoader, self).__init__()
self.core_src_file = None # type: Optional[str]
self.core_src_struct = None
self.core_src = None
self.core_elf_file = None # type: Optional[str]
self.header = None
self.header_struct = EspCoreDumpV1Header
self.checksum_struct = CRC
# target classes will be assigned in ``_reload_coredump``
self.target_methods = Esp32Methods()
self.temp_files = [] # type: list[str]
def _create_temp_file(self): # type: () -> str
t = tempfile.NamedTemporaryFile('wb', delete=False)
# Here we close this at first to make sure the read/write is wrapped in context manager
# Otherwise the result will be wrong if you read while open in another session
t.close()
self.temp_files.append(t.name)
return t.name
def _load_core_src(self): # type: () -> str
"""
Write core elf into ``self.core_src``,
Return the target str by reading core elf
"""
with open(self.core_src_file, 'rb') as fr: # type: ignore
coredump_bytes = fr.read()
_header = EspCoreDumpV1Header.parse(coredump_bytes) # first we use V1 format to get version
self.set_version(_header.ver)
if self.dump_ver == self.ELF_CRC32:
self.checksum_struct = CRC
self.header_struct = EspCoreDumpV2Header
elif self.dump_ver == self.ELF_SHA256:
self.checksum_struct = SHA256
self.header_struct = EspCoreDumpV2Header
elif self.dump_ver == self.BIN_V1:
self.checksum_struct = CRC
self.header_struct = EspCoreDumpV1Header
elif self.dump_ver == self.BIN_V2:
self.checksum_struct = CRC
self.header_struct = EspCoreDumpV2Header
else:
raise ESPCoreDumpLoaderError('Core dump version "0x%x" is not supported!' % self.dump_ver)
self.core_src_struct = Struct(
'header' / self.header_struct,
'data' / Bytes(this.header.tot_len - self.header_struct.sizeof() - self.checksum_struct.sizeof()),
'checksum' / self.checksum_struct,
)
self.core_src = self.core_src_struct.parse(coredump_bytes) # type: ignore
# Reload header if header struct changes after parsing
if self.header_struct != EspCoreDumpV1Header:
self.header = EspCoreDumpV2Header.parse(coredump_bytes)
if self.chip_ver in self.COREDUMP_SUPPORTED_TARGETS:
if self.chip_ver == self.ESP32:
self.target_methods = Esp32Methods() # type: ignore
elif self.chip_ver == self.ESP32S2:
self.target_methods = Esp32S2Methods() # type: ignore
elif self.chip_ver == self.ESP32C3:
self.target_methods = Esp32c3Methods() # type: ignore
elif self.chip_ver == self.ESP32S3:
self.target_methods = Esp32S3Methods() # type: ignore
else:
raise NotImplementedError
else:
raise ESPCoreDumpLoaderError('Core dump chip "0x%x" is not supported!' % self.chip_ver)
return self.target_methods.TARGET # type: ignore
def _validate_dump_file(self): # type: () -> None
if self.chip_ver not in self.COREDUMP_SUPPORTED_TARGETS:
raise ESPCoreDumpLoaderError('Invalid core dump chip version: "{}", should be <= "0x{:X}"'
.format(self.chip_ver, self.ESP32S2))
if self.checksum_struct == CRC:
self._crc_validate()
elif self.checksum_struct == SHA256:
self._sha256_validate()
def _crc_validate(self): # type: () -> None
data_crc = binascii.crc32(
EspCoreDumpV2Header.build(self.core_src.header) + self.core_src.data) & 0xffffffff # type: ignore
if data_crc != self.core_src.checksum: # type: ignore
raise ESPCoreDumpLoaderError(
'Invalid core dump CRC %x, should be %x' % (data_crc, self.core_src.crc)) # type: ignore
def _sha256_validate(self): # type: () -> None
data_sha256 = hashlib.sha256(
EspCoreDumpV2Header.build(self.core_src.header) + self.core_src.data) # type: ignore
data_sha256_str = data_sha256.hexdigest()
sha256_str = binascii.hexlify(self.core_src.checksum).decode('ascii') # type: ignore
if data_sha256_str != sha256_str:
raise ESPCoreDumpLoaderError('Invalid core dump SHA256 "{}", should be "{}"'
.format(data_sha256_str, sha256_str))
def create_corefile(self, exe_name=None, e_machine=ESPCoreDumpElfFile.EM_XTENSA):
# type: (Optional[str], int) -> None
"""
Creates core dump ELF file
"""
self._validate_dump_file()
self.core_elf_file = self._create_temp_file()
if self.dump_ver in [self.ELF_CRC32,
self.ELF_SHA256]:
self._extract_elf_corefile(exe_name, e_machine)
elif self.dump_ver in [self.BIN_V1,
self.BIN_V2]:
self._extract_bin_corefile(e_machine)
else:
raise NotImplementedError
def _extract_elf_corefile(self, exe_name=None, e_machine=ESPCoreDumpElfFile.EM_XTENSA): # type: (str, int) -> None
"""
Reads the ELF formatted core dump image and parse it
"""
with open(self.core_elf_file, 'wb') as fw: # type: ignore
fw.write(self.core_src.data) # type: ignore
core_elf = ESPCoreDumpElfFile(self.core_elf_file, e_machine=e_machine) # type: ignore
# Read note segments from core file which are belong to tasks (TCB or stack)
for seg in core_elf.note_segments:
for note_sec in seg.note_secs:
# Check for version info note
if note_sec.name == 'ESP_CORE_DUMP_INFO' \
and note_sec.type == ESPCoreDumpElfFile.PT_INFO \
and exe_name:
exe_elf = ElfFile(exe_name)
app_sha256 = binascii.hexlify(exe_elf.sha256)
coredump_sha256_struct = Struct(
'ver' / Int32ul,
'sha256' / Bytes(64) # SHA256 as hex string
)
coredump_sha256 = coredump_sha256_struct.parse(note_sec.desc[:coredump_sha256_struct.sizeof()])
if coredump_sha256.sha256 != app_sha256:
raise ESPCoreDumpLoaderError(
'Invalid application image for coredump: coredump SHA256({!r}) != app SHA256({!r}).'
.format(coredump_sha256, app_sha256))
if coredump_sha256.ver != self.version:
raise ESPCoreDumpLoaderError(
'Invalid application image for coredump: coredump SHA256 version({}) != app SHA256 version({}).'
.format(coredump_sha256.ver, self.version))
@staticmethod
def _get_aligned_size(size, align_with=4): # type: (int, int) -> int
if size % align_with:
return align_with * (size // align_with + 1)
return size
@staticmethod
def _build_note_section(name, sec_type, desc): # type: (str, int, str) -> bytes
b_name = bytearray(name, encoding='ascii') + b'\0'
return NoteSection.build({ # type: ignore
'namesz': len(b_name),
'descsz': len(desc),
'type': sec_type,
'name': b_name,
'desc': desc,
})
def _extract_bin_corefile(self, e_machine=ESPCoreDumpElfFile.EM_XTENSA): # type: (int) -> None
"""
Creates core dump ELF file
"""
coredump_data_struct = Struct(
'tasks' / GreedyRange(
AlignedStruct(
4,
'task_header' / TaskHeader,
'tcb' / Bytes(self.header.tcbsz), # type: ignore
'stack' / Bytes(abs_(this.task_header.stack_top - this.task_header.stack_end)), # type: ignore
)
),
'mem_seg_headers' / MemSegmentHeader[self.core_src.header.segs_num] # type: ignore
)
core_elf = ESPCoreDumpElfFile(e_machine=e_machine)
notes = b''
core_dump_info_notes = b''
task_info_notes = b''
coredump_data = coredump_data_struct.parse(self.core_src.data) # type: ignore
for i, task in enumerate(coredump_data.tasks):
stack_len_aligned = self._get_aligned_size(abs(task.task_header.stack_top - task.task_header.stack_end))
task_status_kwargs = {
'task_index': i,
'task_flags': TASK_STATUS_CORRECT,
'task_tcb_addr': task.task_header.tcb_addr,
'task_stack_start': min(task.task_header.stack_top, task.task_header.stack_end),
'task_stack_end': max(task.task_header.stack_top, task.task_header.stack_end),
'task_stack_len': stack_len_aligned,
'task_name': Padding(16).build({}) # currently we don't have task_name, keep it as padding
}
# Write TCB
try:
if self.target_methods.tcb_is_sane(task.task_header.tcb_addr, self.header.tcbsz): # type: ignore
core_elf.add_segment(task.task_header.tcb_addr,
task.tcb,
ElfFile.PT_LOAD,
ElfSegment.PF_R | ElfSegment.PF_W)
elif task.task_header.tcb_addr and self.target_methods.addr_is_fake(task.task_header.tcb_addr):
task_status_kwargs['task_flags'] |= TASK_STATUS_TCB_CORRUPTED
except ESPCoreDumpLoaderError as e:
logging.warning('Skip TCB {} bytes @ 0x{:x}. (Reason: {})'
.format(self.header.tcbsz, task.task_header.tcb_addr, e)) # type: ignore
# Write stack
try:
if self.target_methods.stack_is_sane(task_status_kwargs['task_stack_start'],
task_status_kwargs['task_stack_end']):
core_elf.add_segment(task_status_kwargs['task_stack_start'],
task.stack,
ElfFile.PT_LOAD,
ElfSegment.PF_R | ElfSegment.PF_W)
elif (task_status_kwargs['task_stack_start']
and self.target_methods.addr_is_fake(task_status_kwargs['task_stack_start'])):
task_status_kwargs['task_flags'] |= TASK_STATUS_TCB_CORRUPTED
core_elf.add_segment(task_status_kwargs['task_stack_start'],
task.stack,
ElfFile.PT_LOAD,
ElfSegment.PF_R | ElfSegment.PF_W)
except ESPCoreDumpLoaderError as e:
logging.warning('Skip task\'s ({:x}) stack {} bytes @ 0x{:x}. (Reason: {})'
.format(task_status_kwargs['tcb_addr'],
task_status_kwargs['stack_len_aligned'],
task_status_kwargs['stack_base'],
e))
try:
logging.debug('Stack start_end: 0x{:x} @ 0x{:x}'
.format(task.task_header.stack_top, task.task_header.stack_end))
task_regs, extra_regs = self.target_methods.get_registers_from_stack(
task.stack,
task.task_header.stack_end > task.task_header.stack_top
)
except Exception as e:
raise ESPCoreDumpLoaderError(str(e))
task_info_notes += self._build_note_section('TASK_INFO',
ESPCoreDumpElfFile.PT_TASK_INFO,
EspTaskStatus.build(task_status_kwargs))
notes += self._build_note_section('CORE',
ElfFile.PT_LOAD,
self.target_methods.build_prstatus_data(task.task_header.tcb_addr,
task_regs))
if len(core_dump_info_notes) == 0: # the first task is the crashed task
core_dump_info_notes += self._build_note_section('ESP_CORE_DUMP_INFO',
ESPCoreDumpElfFile.PT_INFO,
Int32ul.build(self.header.ver)) # type: ignore
_regs = [task.task_header.tcb_addr]
# For xtensa, we need to put the exception registers into the extra info as well
if e_machine == ESPCoreDumpElfFile.EM_XTENSA and extra_regs:
for reg_id in extra_regs:
_regs.extend([reg_id, extra_regs[reg_id]])
core_dump_info_notes += self._build_note_section(
'EXTRA_INFO',
ESPCoreDumpElfFile.PT_EXTRA_INFO,
Int32ul[len(_regs)].build(_regs)
)
if self.dump_ver == self.BIN_V2:
for header in coredump_data.mem_seg_headers:
logging.debug('Read memory segment {} bytes @ 0x{:x}'.format(header.mem_sz, header.mem_start))
core_elf.add_segment(header.mem_start, header.data, ElfFile.PT_LOAD, ElfSegment.PF_R | ElfSegment.PF_W)
# add notes
try:
core_elf.add_segment(0, notes, ElfFile.PT_NOTE, 0)
except ESPCoreDumpLoaderError as e:
logging.warning('Skip NOTES segment {:d} bytes @ 0x{:x}. (Reason: {})'.format(len(notes), 0, e))
# add core dump info notes
try:
core_elf.add_segment(0, core_dump_info_notes, ElfFile.PT_NOTE, 0)
except ESPCoreDumpLoaderError as e:
logging.warning('Skip core dump info NOTES segment {:d} bytes @ 0x{:x}. (Reason: {})'
.format(len(core_dump_info_notes), 0, e))
try:
core_elf.add_segment(0, task_info_notes, ElfFile.PT_NOTE, 0)
except ESPCoreDumpLoaderError as e:
logging.warning('Skip failed tasks info NOTES segment {:d} bytes @ 0x{:x}. (Reason: {})'
.format(len(task_info_notes), 0, e))
# dump core ELF
core_elf.e_type = ElfFile.ET_CORE
core_elf.dump(self.core_elf_file) # type: ignore
class ESPCoreDumpFlashLoader(EspCoreDumpLoader):
ESP_COREDUMP_PART_TABLE_OFF = 0x8000
def __init__(self, offset, target=None, port=None, baud=None):
# type: (int, Optional[str], Optional[str], Optional[int]) -> None
super(ESPCoreDumpFlashLoader, self).__init__()
self.port = port
self.baud = baud
self._get_core_src(offset, target)
self.target = self._load_core_src()
def _get_core_src(self, off, target=None): # type: (int, Optional[str]) -> 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, target=target)
else:
logging.info('Invoke parttool to read image.')
self._invoke_parttool()
except subprocess.CalledProcessError as e:
if e.output:
logging.info(e.output)
logging.error('Error during the subprocess execution')
def _invoke_esptool(self, off=None, target=None): # type: (Optional[int], Optional[str]) -> None
"""
Loads core dump from flash using elftool
"""
if target is None:
target = 'auto'
tool_args = [sys.executable, ESPTOOL_PY, '-c', target]
if self.port:
tool_args.extend(['-p', self.port])
if self.baud:
tool_args.extend(['-b', str(self.baud)])
self.core_src_file = self._create_temp_file()
try:
(part_offset, part_size) = self._get_core_dump_partition_info()
if not off:
off = part_offset # set default offset if not specified
logging.warning('The core dump image offset is not specified. Use partition offset: %d.', part_offset)
if part_offset != off:
logging.warning('Predefined image offset: %d does not match core dump partition offset: %d', off,
part_offset)
# Here we use V1 format to locate the size
tool_args.extend(['read_flash', str(off), str(EspCoreDumpV1Header.sizeof())])
tool_args.append(self.core_src_file) # type: ignore
# read core dump length
et_out = subprocess.check_output(tool_args)
if et_out:
logging.info(et_out.decode('utf-8'))
header = EspCoreDumpV1Header.parse(open(self.core_src_file, 'rb').read()) # type: ignore
if not header or not 0 < header.tot_len <= part_size:
logging.error('Incorrect size of core dump image: {}, use partition size instead: {}'
.format(header.tot_len, part_size))
coredump_len = part_size
else:
coredump_len = header.tot_len
# set actual size of core dump image and read it from flash
tool_args[-2] = str(coredump_len)
et_out = subprocess.check_output(tool_args)
if 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 _invoke_parttool(self): # type: () -> None
"""
Loads core dump from flash using parttool
"""
tool_args = [sys.executable, PARTTOOL_PY]
if self.port:
tool_args.extend(['--port', self.port])
tool_args.extend(['read_partition', '--partition-type', 'data', '--partition-subtype', 'coredump', '--output'])
self.core_src_file = self._create_temp_file()
try:
tool_args.append(self.core_src_file) # type: ignore
# read core dump partition
et_out = subprocess.check_output(tool_args)
if et_out:
logging.info(et_out.decode('utf-8'))
except subprocess.CalledProcessError as e:
logging.error('parttool 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 _get_core_dump_partition_info(self, part_off=None): # type: (Optional[int]) -> Tuple[int, int]
"""
Get core dump partition info using parttool
"""
logging.info('Retrieving core dump partition offset and size...')
if not part_off:
part_off = self.ESP_COREDUMP_PART_TABLE_OFF
try:
tool_args = [sys.executable, PARTTOOL_PY, '-q', '--partition-table-offset', str(part_off)]
if self.port:
tool_args.extend(['--port', self.port])
invoke_args = tool_args + ['get_partition_info', '--partition-type', 'data',
'--partition-subtype', 'coredump',
'--info', 'offset', 'size']
res = subprocess.check_output(invoke_args).strip()
(offset_str, size_str) = res.rsplit(b'\n')[-1].split(b' ')
size = int(size_str, 16)
offset = int(offset_str, 16)
logging.info('Core dump partition offset=%d, size=%d', offset, size)
except subprocess.CalledProcessError as e:
logging.error('parttool get partition info failed with err %d', e.returncode)
logging.debug("Command ran: '%s'", e.cmd)
logging.debug('Command out:')
logging.debug(e.output)
logging.error('Check if the coredump partition exists in partition table.')
raise e
return offset, size
class ESPCoreDumpFileLoader(EspCoreDumpLoader):
def __init__(self, path, is_b64=False): # type: (str, bool) -> None
super(ESPCoreDumpFileLoader, self).__init__()
self.is_b64 = is_b64
self._get_core_src(path)
self.target = self._load_core_src()
def _get_core_src(self, path): # type: (str) -> None
"""
Loads core dump from (raw binary or base64-encoded) file
"""
logging.debug('Load core dump from "%s", %s format', path, 'b64' if self.is_b64 else 'raw')
if not self.is_b64:
self.core_src_file = path
else:
self.core_src_file = self._create_temp_file()
with open(self.core_src_file, 'wb') as fw:
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'))
fw.write(data) # type: ignore

62
components/espcoredump/corefile/riscv.py
View File

@@ -1,62 +0,0 @@
#
# Copyright 2021 Espressif Systems (Shanghai) CO., LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from construct import Int16ul, Int32ul, Padding, Struct
from corefile import BaseArchMethodsMixin, BaseTargetMethods, ESPCoreDumpLoaderError
try:
from typing import Any, Optional, Tuple
except ImportError:
pass
RISCV_GP_REGS_COUNT = 32
PRSTATUS_SIZE = 204
PRSTATUS_OFFSET_PR_CURSIG = 12
PRSTATUS_OFFSET_PR_PID = 24
PRSTATUS_OFFSET_PR_REG = 72
ELF_GREGSET_T_SIZE = 128
PrStruct = Struct(
Padding(PRSTATUS_OFFSET_PR_CURSIG),
'pr_cursig' / Int16ul,
Padding(PRSTATUS_OFFSET_PR_PID - PRSTATUS_OFFSET_PR_CURSIG - Int16ul.sizeof()),
'pr_pid' / Int32ul,
Padding(PRSTATUS_OFFSET_PR_REG - PRSTATUS_OFFSET_PR_PID - Int32ul.sizeof()),
'regs' / Int32ul[RISCV_GP_REGS_COUNT],
Padding(PRSTATUS_SIZE - PRSTATUS_OFFSET_PR_REG - ELF_GREGSET_T_SIZE)
)
class RiscvMethodsMixin(BaseArchMethodsMixin):
@staticmethod
def get_registers_from_stack(data, grows_down):
# type: (bytes, bool) -> Tuple[list[int], Optional[dict[int, int]]]
regs = Int32ul[RISCV_GP_REGS_COUNT].parse(data)
if not grows_down:
raise ESPCoreDumpLoaderError('Growing up stacks are not supported for now!')
return regs, None
@staticmethod
def build_prstatus_data(tcb_addr, task_regs): # type: (int, list[int]) -> Any
return PrStruct.build({
'pr_cursig': 0,
'pr_pid': tcb_addr,
'regs': task_regs,
})
class Esp32c3Methods(BaseTargetMethods, RiscvMethodsMixin):
TARGET = 'esp32c3'

0
components/espcoredump/corefile/soc_headers/__init__.py
View File

8
components/espcoredump/corefile/soc_headers/esp32.py
View File

@@ -1,8 +0,0 @@
SOC_DRAM_LOW = 0x3ffae000
SOC_DRAM_HIGH = 0x40000000
SOC_IRAM_LOW = 0x40080000
SOC_IRAM_HIGH = 0x400a0000
SOC_RTC_DRAM_LOW = 0x3ff80000
SOC_RTC_DRAM_HIGH = 0x3ff82000
SOC_RTC_DATA_LOW = 0x50000000
SOC_RTC_DATA_HIGH = 0x50002000

8
components/espcoredump/corefile/soc_headers/esp32c3.py
View File

@@ -1,8 +0,0 @@
SOC_IRAM_LOW = 0x4037c000
SOC_IRAM_HIGH = 0x403e0000
SOC_DRAM_LOW = 0x3fc80000
SOC_DRAM_HIGH = 0x3fce0000
SOC_RTC_DRAM_LOW = 0x50000000
SOC_RTC_DRAM_HIGH = 0x50002000
SOC_RTC_DATA_LOW = 0x50000000
SOC_RTC_DATA_HIGH = 0x50002000

8
components/espcoredump/corefile/soc_headers/esp32s2.py
View File

@@ -1,8 +0,0 @@
SOC_IRAM_LOW = 0x40020000
SOC_IRAM_HIGH = 0x40070000
SOC_DRAM_LOW = 0x3ffb0000
SOC_DRAM_HIGH = 0x40000000
SOC_RTC_DRAM_LOW = 0x3ff9e000
SOC_RTC_DRAM_HIGH = 0x3ffa0000
SOC_RTC_DATA_LOW = 0x50000000
SOC_RTC_DATA_HIGH = 0x50002000

8
components/espcoredump/corefile/soc_headers/esp32s3.py
View File

@@ -1,8 +0,0 @@
SOC_DRAM_LOW = 0x3FC88000
SOC_DRAM_HIGH = 0x3FD00000
SOC_IRAM_LOW = 0x40370000
SOC_IRAM_HIGH = 0x403E0000
SOC_RTC_DRAM_LOW = 0x600FE000
SOC_RTC_DRAM_HIGH = 0x60100000
SOC_RTC_DATA_LOW = 0x50000000
SOC_RTC_DATA_HIGH = 0x50002000

281
components/espcoredump/corefile/xtensa.py
View File

@@ -1,281 +0,0 @@
#
# Copyright 2021 Espressif Systems (Shanghai) CO., LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from construct import Int16ul, Int32ul, Int64ul, Struct
from . import BaseArchMethodsMixin, BaseTargetMethods, ESPCoreDumpLoaderError
try:
from typing import Any, Optional, Tuple
except ImportError:
pass
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 ExceptionRegisters(object):
# extra regs IDs used in EXTRA_INFO note
EXCCAUSE_IDX = 0
EXCVADDR_IDX = 1
EPC1_IDX = 177
EPC2_IDX = 178
EPC3_IDX = 179
EPC4_IDX = 180
EPC5_IDX = 181
EPC6_IDX = 182
EPC7_IDX = 183
EPS2_IDX = 194
EPS3_IDX = 195
EPS4_IDX = 196
EPS5_IDX = 197
EPS6_IDX = 198
EPS7_IDX = 199
@property
def registers(self): # type: () -> dict[str, int]
return {k: v for k, v in self.__class__.__dict__.items()
if not k.startswith('__') and isinstance(v, int)}
# Following structs are based on source code
# IDF_PATH/components/espcoredump/src/core_dump_port.c
PrStatus = Struct(
'si_signo' / Int32ul,
'si_code' / Int32ul,
'si_errno' / Int32ul,
'pr_cursig' / Int16ul,
'pr_pad0' / Int16ul,
'pr_sigpend' / Int32ul,
'pr_sighold' / Int32ul,
'pr_pid' / Int32ul,
'pr_ppid' / Int32ul,
'pr_pgrp' / Int32ul,
'pr_sid' / Int32ul,
'pr_utime' / Int64ul,
'pr_stime' / Int64ul,
'pr_cutime' / Int64ul,
'pr_cstime' / Int64ul,
)
def print_exc_regs_info(extra_info): # type: (list[int]) -> None
"""
Print the register info by parsing extra_info
:param extra_info: extra info data str
:return: None
"""
exccause = extra_info[1 + 2 * ExceptionRegisters.EXCCAUSE_IDX + 1]
exccause_str = XTENSA_EXCEPTION_CAUSE_DICT.get(exccause)
if not exccause_str:
exccause_str = ('Invalid EXCCAUSE code', 'Invalid EXCAUSE description or not found.')
print('exccause 0x%x (%s)' % (exccause, exccause_str[0]))
print('excvaddr 0x%x' % extra_info[1 + 2 * ExceptionRegisters.EXCVADDR_IDX + 1])
# skip crashed_task_tcb, exccause, and excvaddr
for i in range(5, len(extra_info), 2):
if (extra_info[i] >= ExceptionRegisters.EPC1_IDX and extra_info[i] <= ExceptionRegisters.EPC7_IDX):
print('epc%d 0x%x' % ((extra_info[i] - ExceptionRegisters.EPC1_IDX + 1), extra_info[i + 1]))
# skip crashed_task_tcb, exccause, and excvaddr
for i in range(5, len(extra_info), 2):
if (extra_info[i] >= ExceptionRegisters.EPS2_IDX and extra_info[i] <= ExceptionRegisters.EPS7_IDX):
print('eps%d 0x%x' % ((extra_info[i] - ExceptionRegisters.EPS2_IDX + 2), extra_info[i + 1]))
# from "gdb/xtensa-tdep.h"
# typedef struct
# {
# 0 xtensa_elf_greg_t pc;
# 1 xtensa_elf_greg_t ps;
# 2 xtensa_elf_greg_t lbeg;
# 3 xtensa_elf_greg_t lend;
# 4 xtensa_elf_greg_t lcount;
# 5 xtensa_elf_greg_t sar;
# 6 xtensa_elf_greg_t windowstart;
# 7 xtensa_elf_greg_t windowbase;
# 8..63 xtensa_elf_greg_t reserved[8+48];
# 64 xtensa_elf_greg_t ar[64];
# } xtensa_elf_gregset_t;
REG_PC_IDX = 0
REG_PS_IDX = 1
REG_LB_IDX = 2
REG_LE_IDX = 3
REG_LC_IDX = 4
REG_SAR_IDX = 5
# REG_WS_IDX = 6
# REG_WB_IDX = 7
REG_AR_START_IDX = 64
# REG_AR_NUM = 64
# FIXME: acc to xtensa_elf_gregset_t number of regs must be 128,
# but gdb complains when it less then 129
REG_NUM = 129
# XT_SOL_EXIT = 0
XT_SOL_PC = 1
XT_SOL_PS = 2
# XT_SOL_NEXT = 3
XT_SOL_AR_START = 4
XT_SOL_AR_NUM = 4
# XT_SOL_FRMSZ = 8
XT_STK_EXIT = 0
XT_STK_PC = 1
XT_STK_PS = 2
XT_STK_AR_START = 3
XT_STK_AR_NUM = 16
XT_STK_SAR = 19
XT_STK_EXCCAUSE = 20
XT_STK_EXCVADDR = 21
XT_STK_LBEG = 22
XT_STK_LEND = 23
XT_STK_LCOUNT = 24
XT_STK_FRMSZ = 25
class XtensaMethodsMixin(BaseArchMethodsMixin):
@staticmethod
def get_registers_from_stack(data, grows_down):
# type: (bytes, bool) -> Tuple[list[int], Optional[dict[int, int]]]
extra_regs = {v: 0 for v in ExceptionRegisters().registers.values()}
regs = [0] * REG_NUM
# TODO: support for growing up stacks
if not grows_down:
raise ESPCoreDumpLoaderError('Growing up stacks are not supported for now!')
ex_struct = Struct(
'stack' / Int32ul[XT_STK_FRMSZ]
)
if len(data) < ex_struct.sizeof():
raise ESPCoreDumpLoaderError('Too small stack to keep frame: %d bytes!' % len(data))
stack = ex_struct.parse(data).stack
# Stack frame type indicator is always the first item
rc = stack[XT_STK_EXIT]
if rc != 0:
regs[REG_PC_IDX] = stack[XT_STK_PC]
regs[REG_PS_IDX] = stack[XT_STK_PS]
for i in range(XT_STK_AR_NUM):
regs[REG_AR_START_IDX + i] = stack[XT_STK_AR_START + i]
regs[REG_SAR_IDX] = stack[XT_STK_SAR]
regs[REG_LB_IDX] = stack[XT_STK_LBEG]
regs[REG_LE_IDX] = stack[XT_STK_LEND]
regs[REG_LC_IDX] = stack[XT_STK_LCOUNT]
# FIXME: crashed and some running tasks (e.g. prvIdleTask) have EXCM bit set
# and GDB can not unwind callstack properly (it implies not windowed call0)
if regs[REG_PS_IDX] & (1 << 5):
regs[REG_PS_IDX] &= ~(1 << 4)
if stack[XT_STK_EXCCAUSE] in XTENSA_EXCEPTION_CAUSE_DICT:
extra_regs[ExceptionRegisters.EXCCAUSE_IDX] = stack[XT_STK_EXCCAUSE]
else:
extra_regs[ExceptionRegisters.EXCCAUSE_IDX] = INVALID_CAUSE_VALUE
extra_regs[ExceptionRegisters.EXCVADDR_IDX] = stack[XT_STK_EXCVADDR]
else:
regs[REG_PC_IDX] = stack[XT_SOL_PC]
regs[REG_PS_IDX] = stack[XT_SOL_PS]
for i in range(XT_SOL_AR_NUM):
regs[REG_AR_START_IDX + i] = stack[XT_SOL_AR_START + i]
# nxt = stack[XT_SOL_NEXT]
return regs, extra_regs
@staticmethod
def build_prstatus_data(tcb_addr, task_regs): # type: (int, list[int]) -> Any
return PrStatus.build({
'si_signo': 0,
'si_code': 0,
'si_errno': 0,
'pr_cursig': 0, # TODO: set sig only for current/failed task
'pr_pad0': 0,
'pr_sigpend': 0,
'pr_sighold': 0,
'pr_pid': tcb_addr,
'pr_ppid': 0,
'pr_pgrp': 0,
'pr_sid': 0,
'pr_utime': 0,
'pr_stime': 0,
'pr_cutime': 0,
'pr_cstime': 0,
}) + Int32ul[len(task_regs)].build(task_regs)
class Esp32Methods(BaseTargetMethods, XtensaMethodsMixin):
TARGET = 'esp32'
class Esp32S2Methods(BaseTargetMethods, XtensaMethodsMixin):
TARGET = 'esp32s2'
class Esp32S3Methods(BaseTargetMethods, XtensaMethodsMixin):
TARGET = 'esp32s3'

348
components/espcoredump/espcoredump.py Executable file → Normal file
View File

@@ -1,334 +1,22 @@
#!/usr/bin/env python
# #
# ESP-IDF Core Dump Utility # SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
#
# SPDX-License-Identifier: Apache-2.0
#
import argparse
import logging import logging
import os import os.path
import subprocess
import sys
from shutil import copyfile
from construct import GreedyRange, Int32ul, Struct
from corefile import RISCV_TARGETS, SUPPORTED_TARGETS, XTENSA_TARGETS, __version__, xtensa
from corefile.elf import TASK_STATUS_CORRECT, ElfFile, ElfSegment, ESPCoreDumpElfFile, EspTaskStatus
from corefile.gdb import EspGDB
from corefile.loader import ESPCoreDumpFileLoader, ESPCoreDumpFlashLoader
from pygdbmi.gdbcontroller import DEFAULT_GDB_TIMEOUT_SEC
try: try:
from typing import Optional, Tuple from esp_coredump import CoreDump
except ImportError: except ImportError:
# Only used for type annotations raise ModuleNotFoundError('No module named "esp_coredump" please install esp_coredump by running '
pass '"python -m pip install esp-coredump"')
IDF_PATH = os.getenv('IDF_PATH') from esp_coredump.cli_ext import parser
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)
if os.name == 'nt':
CLOSE_FDS = False
else:
CLOSE_FDS = True
def load_aux_elf(elf_path): # type: (str) -> str def main(): # type: () -> None
"""
Loads auxiliary ELF file and composes GDB command to read its symbols.
"""
sym_cmd = ''
if os.path.exists(elf_path):
elf = ElfFile(elf_path)
for s in elf.sections:
if s.name == '.text':
sym_cmd = 'add-symbol-file %s 0x%x' % (elf_path, s.addr)
return sym_cmd
def get_core_dump_elf(e_machine=ESPCoreDumpFileLoader.ESP32):
# type: (int) -> Tuple[str, Optional[str], Optional[list[str]]]
loader = None
core_filename = None
target = None
temp_files = None
if not args.core:
# Core file not specified, try to read core dump from flash.
loader = ESPCoreDumpFlashLoader(args.off, args.chip, 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, e_machine=e_machine)
core_filename = loader.core_elf_file
if args.save_core:
# We got asked to save the core file, make a copy
copyfile(loader.core_elf_file, args.save_core)
target = loader.target
temp_files = loader.temp_files
return core_filename, target, temp_files # type: ignore
def get_target(): # type: () -> str
if args.chip != 'auto':
return args.chip # type: ignore
inst = esptool.ESPLoader.detect_chip(args.port, args.baud)
return inst.CHIP_NAME.lower().replace('-', '') # type: ignore
def get_gdb_path(target=None): # type: (Optional[str]) -> str
if args.gdb:
return args.gdb # type: ignore
if target is None:
target = get_target()
if target in XTENSA_TARGETS:
# For some reason, xtensa-esp32s2-elf-gdb will report some issue.
# Use xtensa-esp32-elf-gdb instead.
return 'xtensa-esp32-elf-gdb'
if target in RISCV_TARGETS:
return 'riscv32-esp-elf-gdb'
raise ValueError('Invalid value: {}. For now we only support {}'.format(target, SUPPORTED_TARGETS))
def get_rom_elf_path(target=None): # type: (Optional[str]) -> str
if args.rom_elf:
return args.rom_elf # type: ignore
if target is None:
target = get_target()
return '{}_rom.elf'.format(target)
def dbg_corefile(): # type: () -> Optional[list[str]]
"""
Command to load core dump from file or flash and run GDB debug session with it
"""
exe_elf = ESPCoreDumpElfFile(args.prog)
core_elf_path, target, temp_files = get_core_dump_elf(e_machine=exe_elf.e_machine)
rom_elf_path = get_rom_elf_path(target)
rom_sym_cmd = load_aux_elf(rom_elf_path)
gdb_tool = get_gdb_path(target)
p = subprocess.Popen(bufsize=0,
args=[gdb_tool,
'--nw', # ignore .gdbinit
'--core=%s' % core_elf_path, # core file,
'-ex', rom_sym_cmd,
args.prog],
stdin=None, stdout=None, stderr=None,
close_fds=CLOSE_FDS)
p.wait()
print('Done!')
return temp_files
def info_corefile(): # type: () -> Optional[list[str]]
"""
Command to load core dump from file or flash and print it's data in user friendly form
"""
exe_elf = ESPCoreDumpElfFile(args.prog)
core_elf_path, target, temp_files = get_core_dump_elf(e_machine=exe_elf.e_machine)
core_elf = ESPCoreDumpElfFile(core_elf_path)
if exe_elf.e_machine != core_elf.e_machine:
raise ValueError('The arch should be the same between core elf and exe elf')
extra_note = None
task_info = []
for seg in core_elf.note_segments:
for note_sec in seg.note_secs:
if note_sec.type == ESPCoreDumpElfFile.PT_EXTRA_INFO and 'EXTRA_INFO' in note_sec.name.decode('ascii'):
extra_note = note_sec
if note_sec.type == ESPCoreDumpElfFile.PT_TASK_INFO and 'TASK_INFO' in note_sec.name.decode('ascii'):
task_info_struct = EspTaskStatus.parse(note_sec.desc)
task_info.append(task_info_struct)
print('===============================================================')
print('==================== ESP32 CORE DUMP START ====================')
rom_elf_path = get_rom_elf_path(target)
rom_sym_cmd = load_aux_elf(rom_elf_path)
gdb_tool = get_gdb_path(target)
gdb = EspGDB(gdb_tool, [rom_sym_cmd], core_elf_path, args.prog, timeout_sec=args.gdb_timeout_sec)
extra_info = None
if extra_note:
extra_info = Struct('regs' / GreedyRange(Int32ul)).parse(extra_note.desc).regs
marker = extra_info[0]
if marker == ESPCoreDumpElfFile.CURR_TASK_MARKER:
print('\nCrashed task has been skipped.')
else:
task_name = gdb.get_freertos_task_name(marker)
print("\nCrashed task handle: 0x%x, name: '%s', GDB name: 'process %d'" % (marker, task_name, marker))
print('\n================== CURRENT THREAD REGISTERS ===================')
# Only xtensa have exception registers
if exe_elf.e_machine == ESPCoreDumpElfFile.EM_XTENSA:
if extra_note and extra_info:
xtensa.print_exc_regs_info(extra_info)
else:
print('Exception registers have not been found!')
print(gdb.run_cmd('info registers'))
print('\n==================== CURRENT THREAD STACK =====================')
print(gdb.run_cmd('bt'))
if task_info and task_info[0].task_flags != TASK_STATUS_CORRECT:
print('The current crashed task is corrupted.')
print('Task #%d info: flags, tcb, stack (%x, %x, %x).' % (task_info[0].task_index,
task_info[0].task_flags,
task_info[0].task_tcb_addr,
task_info[0].task_stack_start))
print('\n======================== THREADS INFO =========================')
print(gdb.run_cmd('info threads'))
# THREADS STACKS
threads, _ = gdb.get_thread_info()
for thr in threads:
thr_id = int(thr['id'])
tcb_addr = gdb.gdb2freertos_thread_id(thr['target-id'])
task_index = int(thr_id) - 1
task_name = gdb.get_freertos_task_name(tcb_addr)
gdb.switch_thread(thr_id)
print('\n==================== THREAD {} (TCB: 0x{:x}, name: \'{}\') ====================='
.format(thr_id, tcb_addr, task_name))
print(gdb.run_cmd('bt'))
if task_info and task_info[task_index].task_flags != TASK_STATUS_CORRECT:
print("The task '%s' is corrupted." % thr_id)
print('Task #%d info: flags, tcb, stack (%x, %x, %x).' % (task_info[task_index].task_index,
task_info[task_index].task_flags,
task_info[task_index].task_tcb_addr,
task_info[task_index].task_stack_start))
print('\n\n======================= ALL MEMORY REGIONS ========================')
print('Name Address Size Attrs')
merged_segs = []
core_segs = core_elf.load_segments
for sec in exe_elf.sections:
merged = False
for seg in core_segs:
if seg.addr <= sec.addr <= seg.addr + len(seg.data):
# sec: |XXXXXXXXXX|
# seg: |...XXX.............|
seg_addr = seg.addr
if seg.addr + len(seg.data) <= sec.addr + len(sec.data):
# sec: |XXXXXXXXXX|
# seg: |XXXXXXXXXXX...|
# merged: |XXXXXXXXXXXXXX|
seg_len = len(sec.data) + (sec.addr - seg.addr)
else:
# sec: |XXXXXXXXXX|
# seg: |XXXXXXXXXXXXXXXXX|
# merged: |XXXXXXXXXXXXXXXXX|
seg_len = len(seg.data)
merged_segs.append((sec.name, seg_addr, seg_len, sec.attr_str(), True))
core_segs.remove(seg)
merged = True
elif sec.addr <= seg.addr <= sec.addr + len(sec.data):
# sec: |XXXXXXXXXX|
# seg: |...XXX.............|
seg_addr = sec.addr
if (seg.addr + len(seg.data)) >= (sec.addr + len(sec.data)):
# sec: |XXXXXXXXXX|
# seg: |..XXXXXXXXXXX|
# merged: |XXXXXXXXXXXXX|
seg_len = len(sec.data) + (seg.addr + len(seg.data)) - (sec.addr + len(sec.data))
else:
# sec: |XXXXXXXXXX|
# seg: |XXXXXX|
# merged: |XXXXXXXXXX|
seg_len = len(sec.data)
merged_segs.append((sec.name, seg_addr, seg_len, sec.attr_str(), True))
core_segs.remove(seg)
merged = True
if not merged:
merged_segs.append((sec.name, sec.addr, len(sec.data), sec.attr_str(), False))
for ms in merged_segs:
print('%s 0x%x 0x%x %s' % (ms[0], ms[1], ms[2], ms[3]))
for cs in core_segs:
# core dump exec segments are from ROM, other are belong to tasks (TCB or stack)
if cs.flags & ElfSegment.PF_X:
seg_name = 'rom.text'
else:
seg_name = 'tasks.data'
print('.coredump.%s 0x%x 0x%x %s' % (seg_name, cs.addr, len(cs.data), cs.attr_str()))
if args.print_mem:
print('\n====================== CORE DUMP MEMORY CONTENTS ========================')
for cs in core_elf.load_segments:
# core dump exec segments are from ROM, other are belong to tasks (TCB or stack)
if cs.flags & ElfSegment.PF_X:
seg_name = 'rom.text'
else:
seg_name = 'tasks.data'
print('.coredump.%s 0x%x 0x%x %s' % (seg_name, cs.addr, len(cs.data), cs.attr_str()))
print(gdb.run_cmd('x/%dx 0x%x' % (len(cs.data) // 4, cs.addr)))
print('\n===================== ESP32 CORE DUMP END =====================')
print('===============================================================')
del gdb
print('Done!')
return temp_files
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='espcoredump.py v%s - ESP32 Core Dump Utility' % __version__)
parser.add_argument('--chip', default=os.environ.get('ESPTOOL_CHIP', 'auto'),
choices=['auto'] + SUPPORTED_TARGETS,
help='Target chip type')
parser.add_argument('--port', '-p', default=os.environ.get('ESPTOOL_PORT', esptool.ESPLoader.DEFAULT_PORT),
help='Serial port device')
parser.add_argument('--baud', '-b', type=int,
default=os.environ.get('ESPTOOL_BAUD', esptool.ESPLoader.ESP_ROM_BAUD),
help='Serial port baud rate used when flashing/reading')
parser.add_argument('--gdb-timeout-sec', type=int, default=DEFAULT_GDB_TIMEOUT_SEC,
help='Overwrite the default internal delay for gdb responses')
common_args = argparse.ArgumentParser(add_help=False)
common_args.add_argument('--debug', '-d', type=int, default=3,
help='Log level (0..3)')
common_args.add_argument('--gdb', '-g',
help='Path to gdb')
common_args.add_argument('--core', '-c',
help='Path to core dump file (if skipped core dump will be read from flash)')
common_args.add_argument('--core-format', '-t', choices=['b64', 'elf', 'raw'], default='elf',
help='File specified with "-c" is an ELF ("elf"), '
'raw (raw) or base64-encoded (b64) binary')
common_args.add_argument('--off', '-o', type=int,
help='Offset of coredump partition in flash (type "make partition_table" to see).')
common_args.add_argument('--save-core', '-s',
help='Save core to file. Otherwise temporary core file will be deleted. '
'Does not work with "-c"', )
common_args.add_argument('--rom-elf', '-r',
help='Path to ROM ELF file. Will use "<target>_rom.elf" if not specified')
common_args.add_argument('prog', help='Path to program\'s ELF binary')
operations = parser.add_subparsers(dest='operation')
operations.add_parser('dbg_corefile', parents=[common_args],
help='Starts GDB debugging session with specified corefile')
info_coredump = operations.add_parser('info_corefile', parents=[common_args],
help='Print core dump info from file')
info_coredump.add_argument('--print-mem', '-m', action='store_true',
help='Print memory dump')
args = parser.parse_args() args = parser.parse_args()
if args.debug == 0: if args.debug == 0:
@@ -343,13 +31,19 @@ if __name__ == '__main__':
log_level = logging.DEBUG log_level = logging.DEBUG
logging.basicConfig(format='%(levelname)s: %(message)s', level=log_level) logging.basicConfig(format='%(levelname)s: %(message)s', level=log_level)
print('espcoredump.py v%s' % __version__) kwargs = {k: v for k, v in vars(args).items() if v is not None}
del(kwargs['debug'])
del(kwargs['operation'])
espcoredump = CoreDump(**kwargs)
temp_core_files = None temp_core_files = None
try: try:
if args.operation == 'info_corefile': if args.operation == 'info_corefile':
temp_core_files = info_corefile() temp_core_files = espcoredump.info_corefile()
elif args.operation == 'dbg_corefile': elif args.operation == 'dbg_corefile':
temp_core_files = dbg_corefile() temp_core_files = espcoredump.dbg_corefile()
else: else:
raise ValueError('Please specify action, should be info_corefile or dbg_corefile') raise ValueError('Please specify action, should be info_corefile or dbg_corefile')
finally: finally:
@@ -359,3 +53,7 @@ if __name__ == '__main__':
os.remove(f) os.remove(f)
except OSError: except OSError:
pass pass
if __name__ == '__main__':
main()

32
components/espcoredump/test/test_espcoredump.py
View File

@@ -1,36 +1,20 @@
#!/usr/bin/env python #!/usr/bin/env python
# #
# Copyright 2018 Espressif Systems (Shanghai) PTE LTD # SPDX-FileCopyrightText: 2018-2022 Espressif Systems (Shanghai) CO LTD
# # SPDX-License-Identifier: Apache-2.0
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os import os
import sys
import unittest import unittest
try: try:
from corefile.elf import ESPCoreDumpElfFile from esp_coredump.corefile import ESPCoreDumpLoaderError
from corefile.loader import ESPCoreDumpFileLoader, ESPCoreDumpLoaderError from esp_coredump.corefile.elf import ESPCoreDumpElfFile
from esp_coredump.corefile.loader import ESPCoreDumpFileLoader
except ImportError: except ImportError:
idf_path = os.getenv('IDF_PATH') raise ModuleNotFoundError('No module named "esp_coredump" please install esp_coredump by running '
if idf_path: '"python -m pip install esp-coredump"')
sys.path.insert(0, os.path.join(idf_path, 'components', 'espcoredump'))
else:
sys.path.insert(0, '..')
from corefile.elf import ESPCoreDumpElfFile
from corefile.loader import ESPCoreDumpFileLoader, ESPCoreDumpLoaderError
SUPPORTED_TARGET = ['esp32', 'esp32s2', 'esp32c3'] SUPPORTED_TARGET = ['esp32', 'esp32s2', 'esp32c3', 'esp32s3']
class TestESPCoreDumpElfFile(unittest.TestCase): class TestESPCoreDumpElfFile(unittest.TestCase):

11
components/espcoredump/test/test_espcoredump.sh
View File

@@ -11,21 +11,22 @@ else
elf_dir=$1 elf_dir=$1
fi fi
COREDUMP_VERSION="espcoredump.py v$(python -c "import pkg_resources; print(pkg_resources.get_distribution('esp-coredump').version)")"
COREDUMP_VERSION_REGEX="espcoredump.py v([0-9])+.([0-9a-z-])+(.[0-9a-z-])?"
SUPPORTED_TARGETS=("esp32" "esp32s2" "esp32c3" "esp32s3" ) SUPPORTED_TARGETS=("esp32" "esp32s2" "esp32c3" "esp32s3" )
res=0 res=0
python -m coverage erase
for chip in "${SUPPORTED_TARGETS[@]}"; do for chip in "${SUPPORTED_TARGETS[@]}"; do
{ {
echo "run b64 decoding tests on $chip" echo "run b64 decoding tests on $chip"
python -m coverage run -a --source=corefile ../espcoredump.py --chip="$chip" --gdb-timeout-sec 5 info_corefile -m -t b64 -c "${chip}/coredump.b64" -s "${chip}/core.elf" "${elf_dir}/${chip}.elf" &>"${chip}/output" && esp-coredump --chip="$chip" --gdb-timeout-sec 5 info_corefile -m -t b64 -c "${chip}/coredump.b64" -s "${chip}/core.elf" "${elf_dir}/${chip}.elf" &>"${chip}/output" &&
sed -i -E "s/$COREDUMP_VERSION_REGEX/$COREDUMP_VERSION/" "${chip}/expected_output"
diff "${chip}/expected_output" "${chip}/output" && diff "${chip}/expected_output" "${chip}/output" &&
python -m coverage run -a --source=corefile ../espcoredump.py --chip="$chip" --gdb-timeout-sec 5 info_corefile -m -t elf -c "${chip}/core.elf" "${elf_dir}/${chip}.elf" &>"${chip}/output2" && esp-coredump --chip="$chip" --gdb-timeout-sec 5 info_corefile -m -t elf -c "${chip}/core.elf" "${elf_dir}/${chip}.elf" &>"${chip}/output2" &&
sed -E "s/$COREDUMP_VERSION_REGEX/$COREDUMP_VERSION/" "${chip}/expected_output"
diff "${chip}/expected_output" "${chip}/output2" diff "${chip}/expected_output" "${chip}/output2"
} || { } || {
echo 'The test for espcoredump has failed!' echo 'The test for espcoredump has failed!'
res=1 res=1
} }
done done
python -m coverage run -a --source=corefile ./test_espcoredump.py
python -m coverage report ../corefile/*.py ../espcoredump.py
exit $res exit $res

14
tools/ci/check_copyright_ignore.txt
View File

@@ -728,18 +728,6 @@ components/esp_wifi/src/lib_printf.c
components/esp_wifi/src/mesh_event.c components/esp_wifi/src/mesh_event.c
components/esp_wifi/src/smartconfig.c components/esp_wifi/src/smartconfig.c
components/esp_wifi/test/test_wifi_init.c components/esp_wifi/test/test_wifi_init.c
components/espcoredump/corefile/__init__.py
components/espcoredump/corefile/_parse_soc_header.py
components/espcoredump/corefile/elf.py
components/espcoredump/corefile/gdb.py
components/espcoredump/corefile/loader.py
components/espcoredump/corefile/riscv.py
components/espcoredump/corefile/soc_headers/esp32.py
components/espcoredump/corefile/soc_headers/esp32c3.py
components/espcoredump/corefile/soc_headers/esp32s2.py
components/espcoredump/corefile/soc_headers/esp32s3.py
components/espcoredump/corefile/xtensa.py
components/espcoredump/espcoredump.py
components/espcoredump/include/esp_core_dump.h components/espcoredump/include/esp_core_dump.h
components/espcoredump/include/port/riscv/esp_core_dump_summary_port.h components/espcoredump/include/port/riscv/esp_core_dump_summary_port.h
components/espcoredump/include/port/xtensa/esp_core_dump_summary_port.h components/espcoredump/include/port/xtensa/esp_core_dump_summary_port.h
@@ -754,7 +742,6 @@ components/espcoredump/src/core_dump_binary.c
components/espcoredump/src/core_dump_flash.c components/espcoredump/src/core_dump_flash.c
components/espcoredump/src/core_dump_uart.c components/espcoredump/src/core_dump_uart.c
components/espcoredump/src/port/riscv/core_dump_port.c components/espcoredump/src/port/riscv/core_dump_port.c
components/espcoredump/test/test_espcoredump.py
components/espcoredump/test_apps/main/test_core_dump.c components/espcoredump/test_apps/main/test_core_dump.c
components/fatfs/diskio/diskio.c components/fatfs/diskio/diskio.c
components/fatfs/diskio/diskio_impl.h components/fatfs/diskio/diskio_impl.h
@@ -2582,7 +2569,6 @@ tools/test_apps/system/panic/test_panic_util/test_panic_util.py
tools/test_apps/system/startup/app_test.py tools/test_apps/system/startup/app_test.py
tools/test_apps/system/startup/main/test_startup_main.c tools/test_apps/system/startup/main/test_startup_main.c
tools/test_idf_monitor/dummy.c tools/test_idf_monitor/dummy.c
tools/test_idf_monitor/run_test_idf_monitor.py
tools/test_idf_py/extra_path/some_ext.py tools/test_idf_py/extra_path/some_ext.py
tools/test_idf_py/idf_ext.py tools/test_idf_py/idf_ext.py
tools/test_idf_py/test_idf_extensions/test_ext/__init__.py tools/test_idf_py/test_idf_extensions/test_ext/__init__.py

1
tools/ci/executable-list.txt
View File

@@ -3,7 +3,6 @@ components/app_update/otatool.py
components/efuse/efuse_table_gen.py components/efuse/efuse_table_gen.py
components/efuse/test_efuse_host/efuse_tests.py components/efuse/test_efuse_host/efuse_tests.py
components/esp_wifi/test_md5/test_md5.sh components/esp_wifi/test_md5/test_md5.sh
components/espcoredump/espcoredump.py
components/espcoredump/test/test_espcoredump.py components/espcoredump/test/test_espcoredump.py
components/espcoredump/test/test_espcoredump.sh components/espcoredump/test/test_espcoredump.sh
components/espcoredump/test_apps/build_espcoredump.sh components/espcoredump/test_apps/build_espcoredump.sh

1
tools/idf_monitor_base/constants.py
View File

@@ -39,7 +39,6 @@ __version__ = '1.1'
# paths to scripts # paths to scripts
PANIC_OUTPUT_DECODE_SCRIPT = os.path.join(os.path.dirname(__file__), '..', 'gdb_panic_server.py') PANIC_OUTPUT_DECODE_SCRIPT = os.path.join(os.path.dirname(__file__), '..', 'gdb_panic_server.py')
COREDUMP_SCRIPT = os.path.join(os.path.dirname(__file__), '..', '..', 'components', 'espcoredump', 'espcoredump.py')
# regex matches an potential PC value (0x4xxxxxxx) # regex matches an potential PC value (0x4xxxxxxx)
MATCH_PCADDR = re.compile(r'0x4[0-9a-f]{7}', re.IGNORECASE) MATCH_PCADDR = re.compile(r'0x4[0-9a-f]{7}', re.IGNORECASE)

82
tools/idf_monitor_base/coredump.py
View File

@@ -1,12 +1,11 @@
import io
import os import os
import queue import queue
import subprocess
import sys
import tempfile import tempfile
from contextlib import contextmanager from contextlib import contextmanager, redirect_stdout
from typing import Generator from typing import Generator
from .constants import COREDUMP_SCRIPT, TAG_KEY from .constants import TAG_KEY
from .logger import Logger from .logger import Logger
from .output_helpers import yellow_print from .output_helpers import yellow_print
from .web_socket_client import WebSocketClient from .web_socket_client import WebSocketClient
@@ -46,46 +45,45 @@ class CoreDump:
if self._decode_coredumps != COREDUMP_DECODE_INFO: if self._decode_coredumps != COREDUMP_DECODE_INFO:
raise NotImplementedError('process_coredump: %s not implemented' % self._decode_coredumps) raise NotImplementedError('process_coredump: %s not implemented' % self._decode_coredumps)
coredump_file = None coredump_file = None
try: # On Windows, the temporary file can't be read unless it is closed.
# On Windows, the temporary file can't be read unless it is closed. # Set delete=False and delete the file manually later.
# Set delete=False and delete the file manually later. with tempfile.NamedTemporaryFile(mode='wb', delete=False) as coredump_file:
with tempfile.NamedTemporaryFile(mode='wb', delete=False) as coredump_file: coredump_file.write(self._coredump_buffer)
coredump_file.write(self._coredump_buffer) coredump_file.flush()
coredump_file.flush()
if self.websocket_client: if self.websocket_client:
self.logger.output_enabled = True
yellow_print('Communicating through WebSocket')
self.websocket_client.send({'event': 'coredump',
'file': coredump_file.name,
'prog': self.elf_file})
yellow_print('Waiting for debug finished event')
self.websocket_client.wait([('event', 'debug_finished')])
yellow_print('Communications through WebSocket is finished')
else:
cmd = [sys.executable,
COREDUMP_SCRIPT,
'info_corefile',
'--core', coredump_file.name,
'--core-format', 'b64',
self.elf_file
]
output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
self.logger.output_enabled = True
self.logger.print(output) # noqa: E999
self.logger.output_enabled = False # Will be reenabled in check_coredump_trigger_after_print
except subprocess.CalledProcessError as e:
yellow_print('Failed to run espcoredump script: {}\n{}\n\n'.format(e, e.output))
self.logger.output_enabled = True self.logger.output_enabled = True
self.logger.print(COREDUMP_UART_START + b'\n') yellow_print('Communicating through WebSocket')
self.logger.print(self._coredump_buffer) self.websocket_client.send({'event': 'coredump',
# end line will be printed in handle_serial_input 'file': coredump_file.name,
finally: 'prog': self.elf_file})
if coredump_file is not None: yellow_print('Waiting for debug finished event')
try: self.websocket_client.wait([('event', 'debug_finished')])
os.unlink(coredump_file.name) yellow_print('Communications through WebSocket is finished')
except OSError as e: else:
yellow_print('Couldn\'t remote temporary core dump file ({})'.format(e)) try:
import esp_coredump
except ImportError as e:
yellow_print('Failed to parse core dump info: '
'Module {} is not installed \n\n'.format(e.name))
self.logger.output_enabled = True
self.logger.print(COREDUMP_UART_START + b'\n')
self.logger.print(self._coredump_buffer)
# end line will be printed in handle_serial_input
else:
coredump = esp_coredump.CoreDump(core=coredump_file.name, core_format='b64', prog=self.elf_file)
f = io.StringIO()
with redirect_stdout(f):
coredump.info_corefile()
output = f.getvalue()
self.logger.output_enabled = True
self.logger.print(output.encode('utf-8'))
self.logger.output_enabled = False # Will be reenabled in check_coredump_trigger_after_print
if coredump_file is not None:
try:
os.unlink(coredump_file.name)
except OSError as e:
yellow_print('Couldn\'t remote temporary core dump file ({})'.format(e))
def _check_coredump_trigger_before_print(self, line): # type: (bytes) -> None def _check_coredump_trigger_before_print(self, line): # type: (bytes) -> None
if self._decode_coredumps == COREDUMP_DECODE_DISABLE: if self._decode_coredumps == COREDUMP_DECODE_DISABLE:

5
tools/requirements/requirements.core.txt
View File

@@ -8,16 +8,13 @@ cryptography
pyparsing pyparsing
pyelftools pyelftools
idf-component-manager idf-component-manager
esp-coredump
# esptool dependencies (see components/esptool_py/esptool/setup.py) # esptool dependencies (see components/esptool_py/esptool/setup.py)
reedsolo reedsolo
bitstring bitstring
ecdsa ecdsa
# espcoredump dependencies
construct
pygdbmi
# kconfig and menuconfig dependencies # kconfig and menuconfig dependencies
kconfiglib kconfiglib
windows-curses; sys_platform == 'win32' windows-curses; sys_platform == 'win32'

35
tools/test_idf_monitor/run_test_idf_monitor.py
View File

@@ -1,18 +1,7 @@
#!/usr/bin/env python #!/usr/bin/env python
# #
# Copyright 2018-2019 Espressif Systems (Shanghai) PTE LTD # SPDX-FileCopyrightText: 2018-2022 Espressif Systems (Shanghai) CO LTD
# # SPDX-License-Identifier: Apache-2.0
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import print_function, unicode_literals from __future__ import print_function, unicode_literals
@@ -20,6 +9,7 @@ import errno
import filecmp import filecmp
import os import os
import pty import pty
import re
import socket import socket
import subprocess import subprocess
import sys import sys
@@ -58,6 +48,22 @@ SOCKET_TIMEOUT = 30
# the test is restarted after failure (idf_monitor has to be killed): # the test is restarted after failure (idf_monitor has to be killed):
RETRIES_PER_TEST = 2 RETRIES_PER_TEST = 2
COREDUMP_VERSION_REGEX = r'espcoredump\.py v\d+\.[\d\w-]+(\.[\d\w-]+)?'
def remove_coredump_version_string(file_path):
with open(file_path, 'r') as file:
init_text = file.read()
modified_text = re.sub(COREDUMP_VERSION_REGEX, '', init_text, re.MULTILINE)
if not init_text != modified_text:
return None
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_file.write(modified_text.encode())
return temp_file.name
def monitor_timeout(process): def monitor_timeout(process):
if process.poll() is None: if process.poll() is None:
@@ -167,7 +173,10 @@ def test_iteration(runner, test):
print('\tThe client was closed successfully') print('\tThe client was closed successfully')
f1 = IN_DIR + test[2] f1 = IN_DIR + test[2]
f2 = OUT_DIR + test[2] f2 = OUT_DIR + test[2]
temp_f1, temp_f2 = remove_coredump_version_string(f1), remove_coredump_version_string(f2)
print('\tdiff {} {}'.format(f1, f2)) print('\tdiff {} {}'.format(f1, f2))
if temp_f1 and temp_f2:
f1, f2 = temp_f1, temp_f2
if filecmp.cmp(f1, f2, shallow=False): if filecmp.cmp(f1, f2, shallow=False):
print('\tTest has passed') print('\tTest has passed')
else: else: