diff --git a/scripts/elf_task_schedules.py b/scripts/elf_task_schedules.py new file mode 100755 index 0000000..9639053 --- /dev/null +++ b/scripts/elf_task_schedules.py @@ -0,0 +1,333 @@ +#!/usr/bin/env python3 + +import argparse +import base64 +import sys +from dataclasses import dataclass +from pathlib import Path + +from elftools.dwarf.descriptions import describe_form_class +from elftools.dwarf.dwarf_expr import DWARFExprParser +from elftools.dwarf.locationlists import LocationEntry +from elftools.elf.elffile import ELFFile + +STRUCT_NAME = "task_schedule" + + +@dataclass +class Candidate: + name: str + address: int + size: int + element_size: int + file_offset: int + source: str | None + + +@dataclass +class SkippedCandidate: + name: str + reason: str + source: str | None + + +def die_name(die): + attr = die.attributes.get("DW_AT_name") + if attr is None: + return None + return attr.value.decode("utf-8", errors="replace") + + +def type_die(die): + if "DW_AT_type" not in die.attributes: + return None + return die.get_DIE_from_attribute("DW_AT_type") + + +def resolve_type(die): + while die is not None and die.tag in ( + "DW_TAG_const_type", + "DW_TAG_volatile_type", + "DW_TAG_restrict_type", + "DW_TAG_atomic_type", + "DW_TAG_typedef", + ): + die = type_die(die) + return die + + +def type_size(die): + die = resolve_type(die) + if die is None: + return None + if "DW_AT_byte_size" in die.attributes: + return die.attributes["DW_AT_byte_size"].value + if die.tag == "DW_TAG_array_type": + return array_size(die) + return None + + +def subrange_count(subrange_die): + if "DW_AT_count" in subrange_die.attributes: + return subrange_die.attributes["DW_AT_count"].value + if "DW_AT_upper_bound" not in subrange_die.attributes: + return None + + upper = subrange_die.attributes["DW_AT_upper_bound"].value + lower = subrange_die.attributes.get("DW_AT_lower_bound") + lower = lower.value if lower is not None else 0 + return upper - lower + 1 + + +def array_size(array_die): + if "DW_AT_byte_size" in array_die.attributes: + return array_die.attributes["DW_AT_byte_size"].value + + elem = type_size(type_die(array_die)) + if elem is None: + return None + + count = 1 + for child in array_die.iter_children(): + if child.tag != "DW_TAG_subrange_type": + continue + sub_count = subrange_count(child) + if sub_count is None: + return None + count *= sub_count + return elem * count + + +def array_element_size(array_die): + return type_size(type_die(array_die)) + + +def is_task_schedule_array(var_die): + var_type = resolve_type(type_die(var_die)) + if var_type is None or var_type.tag != "DW_TAG_array_type": + return False + + elem_type = resolve_type(type_die(var_type)) + return elem_type is not None and elem_type.tag == "DW_TAG_structure_type" and die_name(elem_type) == STRUCT_NAME + + +def source_location(die, dwarf_info): + file_attr = die.attributes.get("DW_AT_decl_file") + if file_attr is None: + return None + + lineprog = dwarf_info.line_program_for_CU(die.cu) + if lineprog is None: + return None + + file_index = file_attr.value - 1 + if file_index < 0 or file_index >= len(lineprog["file_entry"]): + return None + + entry = lineprog["file_entry"][file_index] + filename = entry.name.decode("utf-8", errors="replace") + include_dirs = lineprog.header.include_directory + directory = "" + if entry.dir_index != 0: + directory = include_dirs[entry.dir_index - 1].decode("utf-8", errors="replace") + + line = die.attributes.get("DW_AT_decl_line") + path = str(Path(directory) / filename) if directory else filename + return f"{path}:{line.value}" if line is not None else path + + +def address_from_expr(expr, dwarf_info): + parser = DWARFExprParser(dwarf_info.structs) + operations = parser.parse_expr(expr) + if len(operations) != 1: + return None + + op = operations[0] + if op.op_name == "DW_OP_addr": + return op.args[0] + + # Some toolchains emit indexed address operations in split-DWARF-like forms. + if op.op_name in ("DW_OP_addrx", "DW_OP_GNU_addr_index"): + try: + return dwarf_info.get_addr(op.args[0]) + except AttributeError: + return None + + return None + + +def variable_address(var_die, dwarf_info): + loc_attr = var_die.attributes.get("DW_AT_location") + if loc_attr is None: + return None + + form_class = describe_form_class(loc_attr.form) + if form_class == "exprloc": + return address_from_expr(loc_attr.value, dwarf_info) + + if form_class != "loclist": + return None + + loc_lists = dwarf_info.location_lists() + loc_list = loc_lists.get_location_list_at_offset(loc_attr.value, die=var_die) + addresses = set() + for entry in loc_list: + if isinstance(entry, LocationEntry): + address = address_from_expr(entry.loc_expr, dwarf_info) + if address is not None: + addresses.add(address) + + return addresses.pop() if len(addresses) == 1 else None + + +def file_offset_for_address(elf, address, size): + for segment in elf.iter_segments(): + if segment.header.p_type != "PT_LOAD": + continue + + start = segment.header.p_vaddr + file_end = start + segment.header.p_filesz + if start <= address and address + size <= file_end: + return segment.header.p_offset + (address - start) + + for section in elf.iter_sections(): + if section.header.sh_type == "SHT_NOBITS" or section.header.sh_addr == 0: + continue + + start = section.header.sh_addr + end = start + section.header.sh_size + if start <= address and address + size <= end: + return section.header.sh_offset + (address - start) + + return None + + +def symbol_file_offsets(elf, name, size): + offsets = [] + for section in elf.iter_sections(): + if section.header.sh_type not in ("SHT_SYMTAB", "SHT_DYNSYM"): + continue + + for symbol in section.iter_symbols(): + if symbol.name != name: + continue + if symbol["st_info"]["type"] != "STT_OBJECT": + continue + if isinstance(symbol["st_shndx"], str): + continue + + target = elf.get_section(symbol["st_shndx"]) + if target.header.sh_type == "SHT_NOBITS": + continue + + sym_size = symbol["st_size"] + if sym_size != 0 and sym_size < size: + continue + + if target.header.sh_addr == 0: + offset = target.header.sh_offset + symbol["st_value"] + else: + offset = target.header.sh_offset + symbol["st_value"] - target.header.sh_addr + + section_start = target.header.sh_offset + section_end = section_start + target.header.sh_size + if section_start <= offset and offset + size <= section_end: + offsets.append(offset) + + return sorted(set(offsets)) + + +def find_candidates(elf, dwarf_info, name_filter): + candidates = [] + skipped = [] + + for cu in dwarf_info.iter_CUs(): + for die in cu.iter_DIEs(): + if die.tag != "DW_TAG_variable" or not is_task_schedule_array(die): + continue + + name = die_name(die) or "" + source = source_location(die, dwarf_info) + if name_filter is not None and name != name_filter: + continue + + array_die = resolve_type(type_die(die)) + size = array_size(array_die) + if size is None: + skipped.append(SkippedCandidate(name, "unknown array size", source)) + continue + + element_size = array_element_size(array_die) + if element_size is None: + skipped.append(SkippedCandidate(name, "unknown array element size", source)) + continue + + address = variable_address(die, dwarf_info) + if address is None: + skipped.append(SkippedCandidate(name, "no absolute ELF address", source)) + continue + + file_offset = file_offset_for_address(elf, address, size) + if file_offset is None: + offsets = symbol_file_offsets(elf, name, size) + if len(offsets) == 1: + file_offset = offsets[0] + else: + skipped.append(SkippedCandidate(name, "not backed by initialized ELF bytes", source)) + continue + + candidates.append(Candidate(name, address, size, element_size, file_offset, source)) + + return candidates, skipped + + +def describe_candidate(candidate): + source = f" ({candidate.source})" if candidate.source else "" + return f"{candidate.name} @ 0x{candidate.address:x}, {candidate.size} bytes{source}" + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Print the raw hex bytes for the single initialized 'struct task_schedule' array in a Zephyr ELF.", + allow_abbrev=False, + ) + parser.add_argument("elf", type=Path, help="Path to zephyr.elf") + parser.add_argument( + "--name", + help="Require a specific variable name when the ELF contains multiple initialized arrays", + ) + parser.add_argument("--base64", action="store_true", help="Output base64 data instead of hex") + args = parser.parse_args() + + with args.elf.open("rb") as f: + elf = ELFFile(f) + if not elf.has_dwarf_info(): + sys.exit(f"{args.elf}: no DWARF debug info found") + + dwarf_info = elf.get_dwarf_info() + candidates, skipped = find_candidates(elf, dwarf_info, args.name) + + if len(candidates) != 1: + details = [] + if candidates: + details.append("initialized candidates:") + details.extend(f" - {describe_candidate(c)}" for c in candidates) + if skipped: + details.append("skipped candidates:") + details.extend(f" - {s.name}: {s.reason}" + (f" ({s.source})" if s.source else "") for s in skipped) + hint = " Use --name to select one." if len(candidates) > 1 else "" + sys.exit( + f"expected exactly one initialized struct {STRUCT_NAME} array, " + f"found {len(candidates)}.{hint}\n" + "\n".join(details) + ) + + candidate = candidates[0] + f.seek(candidate.file_offset) + data = f.read(candidate.size) + + array_elements = [data[i : i + candidate.element_size] for i in range(0, len(data), candidate.element_size)] + + def fmt(val: bytes): + return base64.b64encode(val).decode("utf-8") if args.base64 else val.hex() + + print("\n".join(f"{1001 + i:3d}: {fmt(val)}" for i, val in enumerate(array_elements))) diff --git a/src/infuse_iot/tools/cloud.py b/src/infuse_iot/tools/cloud.py index b92e930..dd5ea42 100644 --- a/src/infuse_iot/tools/cloud.py +++ b/src/infuse_iot/tools/cloud.py @@ -5,6 +5,7 @@ __author__ = "Jordan Yates" __copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd" +import base64 import glob import sys from typing import Any @@ -193,6 +194,11 @@ def add_parser(cls, parser): kv_parser.set_defaults(command_fn=cls.kv_state) kv_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID") kv_parser.add_argument("--schedules", action="store_true", help="Display task schedules") + kv_display = kv_parser.add_mutually_exclusive_group() + kv_display.add_argument("--hex", action="store_true", help="Display values as hex strings instead of decoding") + kv_display.add_argument( + "--base64", action="store_true", help="Display values as base64 strings instead of decoding" + ) dfu_parser = tool_parser.add_parser("dfu", help="Manage device firmware upgrades") dfu_parser.set_defaults(command_fn=cls.dfu) @@ -285,12 +291,13 @@ def val_or_na(value) -> str: print(tabulate(table)) - def _kv_display(self, table: list[tuple[str, Any]], name_base: str, dictionary: dict): + def _kv_display(self, table: list[tuple[str, str, Any]], key_val: str, name_base: str, dictionary: dict): for name, value in dictionary.items(): if isinstance(value, dict): - self._kv_display(table, f"{name_base}.{name}", value) + self._kv_display(table, key_val, f"{name_base}.{name}", value) else: - table.append((f"{name_base}.{name}", value)) + table.append((key_val, f"{name_base}.{name}", value)) + key_val = "" def kv_state(self, client: Client): id_int = int(self.args.id, 0) @@ -301,9 +308,10 @@ def kv_state(self, client: Client): print(f"Unable to query KV state for {id_str}") return - table: list[tuple[str, Any]] = [] + table: list[tuple[str, str, Any]] = [] for element in kv_state: - key = element.key_name if isinstance(element.key_name, str) else str(element.key_id) + key = element.key_name if isinstance(element.key_name, str) else "" + key_id = str(element.key_id) # Don't display task schedules unless requested if key == "TASK_SCHEDULES" and not self.args.schedules: @@ -311,14 +319,19 @@ def kv_state(self, client: Client): if isinstance(element.data, Unset): if element.crc: - table.append((key, f"Write-only (CRC: 0x{element.crc:08x})")) + table.append((key_id, key, f"Write-only (CRC: 0x{element.crc:08x})")) else: - table.append((key, "Not set")) + table.append((key_id, key, "Not set")) else: - if isinstance(element.decoded, Unset): - table.append((key, element.data)) + if self.args.hex: + table.append((key_id, key, base64.b64decode(element.data).hex())) + elif self.args.base64: + table.append((key_id, key, element.data)) else: - self._kv_display(table, key, element.decoded.additional_properties) + if isinstance(element.decoded, Unset): + table.append((key_id, key, element.data)) + else: + self._kv_display(table, key_id, key, element.decoded.additional_properties) print(tabulate(table))