import os
import struct
import sys
from typing import List
from construct import (
Bytes,
Computed,
IfThenElse,
Int8ul,
Int16ul,
Int32ul,
Int64ul,
Struct,
this,
)
ELF_MAGIC = b"\x7fELF"
MIN_FILE_SIZE = 64
ELF_HEADER_SIZE_32 = 52
ELF_HEADER_SIZE_64 = 64
PROG_HEADER_SIZE_32 = 32
PROG_HEADER_SIZE_64 = 56
SECTION_HEADER_SIZE_32 = 40
SECTION_HEADER_SIZE_64 = 64
ELFCLASS32 = 1
ELFCLASS64 = 2
ELFDATA2LSB = 1
ELFDATA2MSB = 2
ET_CORE = 4
PT_LOAD = 1
PT_NOTE = 4
COREDUMP_MAGIC = 0x434F5245
NUTTX_NOTE_NAME = b"NuttX\x00"
SHN_UNDEF = 0
SHN_LORESERVE = 0xFF00
class ElfHeaderParser:
"""Parses ELF file headers and provides validation capabilities"""
ELF_HEADER_STRUCT = Struct(
"ei_magic" / Bytes(4),
"ei_class" / Int8ul,
"ei_data" / Int8ul,
"ei_version" / Int8ul,
"ei_osabi" / Int8ul,
"ei_abiversion" / Int8ul,
"ei_pad" / Bytes(7),
"e_type" / Int16ul,
"e_machine" / Int16ul,
"e_version" / Int32ul,
"e_entry"
/ IfThenElse(
this.ei_class == ELFCLASS32, Int32ul, Int64ul
),
"e_phoff"
/ IfThenElse(
this.ei_class == ELFCLASS32, Int32ul, Int64ul
),
"e_shoff"
/ IfThenElse(
this.ei_class == ELFCLASS32, Int32ul, Int64ul
),
"e_flags" / Int32ul,
"e_ehsize" / Int16ul,
"e_phentsize" / Int16ul,
"e_phnum" / Int16ul,
"e_shentsize" / Int16ul,
"e_shnum" / Int16ul,
"e_shstrndx" / Int16ul,
"endian"
/ Computed(
lambda ctx: "<" if ctx.ei_data == ELFDATA2LSB else ">"
),
"is_core" / Computed(lambda ctx: ctx.e_type == ET_CORE),
)
def __init__(self, data):
self.data = data
self.header = None
def parse_header(self) -> List[str]:
"""Parse ELF header and return list of issues"""
issues = []
if self.data[0:4] != ELF_MAGIC:
return [
f"Invalid ELF magic number (expected 0x7f454c46, got 0x{self.data[0:4].hex()})"
]
try:
self.header = self.ELF_HEADER_STRUCT.parse(self.data)
except Exception as e:
issues.append(f"ELF header parsing failed: {str(e)}")
return issues
def validate_header_sizes(self) -> List[str]:
"""Validate critical size fields in ELF header"""
issues = []
if not self.header:
return ["Cannot validate sizes: header parsing failed"]
expected_sizes = {
ELFCLASS32: {
"ehsize": ELF_HEADER_SIZE_32,
"phentsize": PROG_HEADER_SIZE_32,
"shentsize": SECTION_HEADER_SIZE_32,
},
ELFCLASS64: {
"ehsize": ELF_HEADER_SIZE_64,
"phentsize": PROG_HEADER_SIZE_64,
"shentsize": SECTION_HEADER_SIZE_64,
},
}
if self.header.ei_class not in expected_sizes:
return [f"Unknown ELF class: {self.header.ei_class}"]
exp_sizes = expected_sizes.get(self.header.ei_class)
if self.header.e_ehsize != exp_sizes["ehsize"]:
issues.append(
f"Invalid ELF header size: expected {exp_sizes['ehsize']} bytes, "
f"actual {self.header.e_ehsize} bytes"
)
if self.header.e_phentsize != exp_sizes["phentsize"]:
issues.append(
f"Invalid program header entry size: expected {exp_sizes['phentsize']} bytes, "
f"actual {self.header.e_phentsize} bytes"
)
if (
self.header.e_shnum > 0 or not self.header.is_core
) and self.header.e_shentsize != exp_sizes["shentsize"]:
issues.append(
f"Invalid section header entry size: expected {exp_sizes['shentsize']} bytes, "
f"actual {self.header.e_shentsize} bytes"
)
return issues
def validate_header_tables(self, file_size) -> List[str]:
"""Validate program and section header tables are within file bounds"""
issues = []
if not self.header:
return ["Cannot validate tables: header parsing failed"]
if self.header.e_phnum > 0:
prog_table_size = self.header.e_phnum * self.header.e_phentsize
prog_table_end = self.header.e_phoff + prog_table_size
if prog_table_end > file_size:
issues.append(
f"Program header table exceeds file: "
f"offset={self.header.e_phoff}, size={prog_table_size}, "
f"file_size={file_size}"
)
if self.header.e_shnum > 0:
sect_table_size = self.header.e_shnum * self.header.e_shentsize
sect_table_end = self.header.e_shoff + sect_table_size
if sect_table_end > file_size:
issues.append(
f"Section header table exceeds file: "
f"offset={self.header.e_shoff}, size={sect_table_size}, "
f"file_size={file_size}"
)
if (
self.header.e_shstrndx != SHN_UNDEF
and self.header.e_shstrndx < SHN_LORESERVE
):
if self.header.e_shstrndx >= self.header.e_shnum:
issues.append(
f"Invalid section string table index: {self.header.e_shstrndx} >= "
f"{self.header.e_shnum} (number of sections)"
)
return issues
def validate(self, file_size) -> List[str]:
"""Unified ELF header parsing and validation"""
issues = []
issues.extend(self.parse_header())
if not self.header:
return issues
issues.extend(self.validate_header_sizes())
issues.extend(self.validate_header_tables(file_size))
return issues
def validate_segment_integrity(core_file, header, file_size) -> List[str]:
"""Validate that all LOAD segments have complete content in core files"""
issues = []
if header.ei_class == ELFCLASS32:
PROG_HEADER_STRUCT = 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,
)
else:
PROG_HEADER_STRUCT = Struct(
"p_type" / Int32ul,
"p_flags" / Int32ul,
"p_offset" / Int64ul,
"p_vaddr" / Int64ul,
"p_paddr" / Int64ul,
"p_filesz" / Int64ul,
"p_memsz" / Int64ul,
"p_align" / Int64ul,
)
try:
core_file.seek(header.e_phoff)
for i in range(header.e_phnum):
phdr_data = core_file.read(header.e_phentsize)
if len(phdr_data) != header.e_phentsize:
issues.append(f"Failed to read program header #{i}")
continue
try:
seg_header = PROG_HEADER_STRUCT.parse(phdr_data)
seg_type = seg_header.p_type
seg_offset = seg_header.p_offset
seg_vaddr = seg_header.p_vaddr
seg_filesz = seg_header.p_filesz
except Exception as e:
issues.append(f"Failed to parse program header #{i}: {str(e)}")
continue
if seg_type == PT_LOAD and seg_filesz > 0:
seg_end = seg_offset + seg_filesz
if seg_offset >= file_size:
issues.append(
f"Segment #{i} at 0x{seg_vaddr:08x} is completely missing: "
f"starts at offset {seg_offset} beyond file end {file_size}"
)
elif seg_end > file_size:
truncated = seg_end - file_size
available = file_size - seg_offset
issues.append(
f"Segment #{i} at 0x{seg_vaddr:08x} is truncated: "
f"declared {seg_filesz} bytes, available {available} bytes, "
f"missing {truncated} bytes (file size: {file_size} bytes)"
)
except Exception as e:
issues.append(f"Error validating segments: {str(e)}")
return issues
def validate_core_file(core_file, header, file_size) -> str:
"""Validate nuttx core dump file, return error string or empty string"""
if header.e_phnum <= 0:
return "No program headers found in core file"
try:
last_ph_position = header.e_phoff + (header.e_phnum - 1) * header.e_phentsize
core_file.seek(last_ph_position)
phdr_data = core_file.read(header.e_phentsize)
if len(phdr_data) != header.e_phentsize:
return "Failed to read last program header"
seg_type = struct.unpack(header.endian + "I", phdr_data[0:4])[0]
if seg_type != PT_NOTE:
return ""
if header.ei_class == ELFCLASS32:
seg_offset = struct.unpack(header.endian + "I", phdr_data[4:8])[0]
seg_filesz = struct.unpack(header.endian + "I", phdr_data[16:20])[0]
else:
seg_offset = struct.unpack(header.endian + "Q", phdr_data[8:16])[0]
seg_filesz = struct.unpack(header.endian + "Q", phdr_data[32:40])[0]
if seg_offset + seg_filesz > file_size:
return f"Note segment exceeds file: offset={seg_offset}, size={seg_filesz}, file_size={file_size}"
core_file.seek(seg_offset)
note_header = core_file.read(12)
if len(note_header) != 12:
return "Note segment header incomplete (expected 12 bytes)"
name_size, desc_size, note_type = struct.unpack(
f"{header.endian}III", note_header
)
if not name_size:
return "Note name size cannot be zero"
if not desc_size:
return "Note description size cannot be zero"
total_size = 12 + name_size + desc_size
if total_size > seg_filesz:
return f"Note segment too small: expected {total_size} bytes, actual {seg_filesz} bytes"
core_file.seek(seg_offset + 12)
name_data = core_file.read(name_size)
if len(name_data) != name_size:
return f"Failed to read note name (expected {name_size} bytes)"
if not name_data.startswith(NUTTX_NOTE_NAME):
return ""
if note_type != COREDUMP_MAGIC:
return f"Invalid NuttX note type: expected 0x{COREDUMP_MAGIC:08X}, actual 0x{note_type:08X}"
except (OSError, struct.error) as e:
return f"Error validating core dump: {str(e)}"
return ""
def check_elf_integrity(filename) -> List[str]:
"""Perform ELF file integrity check"""
issues = []
try:
file_size = os.path.getsize(filename)
if file_size < MIN_FILE_SIZE:
return [f"File too small (min {MIN_FILE_SIZE} bytes required)"]
with open(filename, "rb") as f:
data = f.read(MIN_FILE_SIZE)
if len(data) < MIN_FILE_SIZE:
return [
f"Failed to read ELF header: read {len(data)} bytes (expected {MIN_FILE_SIZE} bytes)"
]
parser = ElfHeaderParser(data)
issues.extend(parser.validate(file_size))
if parser.header and parser.header.is_core:
issues.extend(validate_segment_integrity(f, parser.header, file_size))
if error := validate_core_file(f, parser.header, file_size):
issues.append(error)
return issues
except OSError as e:
return [f"File operation error: {str(e)}"]
except Exception as e:
return [f"Unexpected error: {str(e)}"]
def main():
if len(sys.argv) != 2:
print(f"Usage: {sys.argv[0]} <file>")
print(f"Example: {sys.argv[0]} example.elf")
sys.exit(1)
file_path = sys.argv[1]
if not os.path.isfile(file_path):
print(f"Error: File '{file_path}' does not exist")
sys.exit(1)
issues = check_elf_integrity(file_path)
passed = len(issues) == 0
print(f"Check Integrity: {'PASSED' if passed else 'FAILED'}")
if not passed:
print(f"Found {len(issues)} issues:")
for i, issue in enumerate(issues, 1):
print(f" [{i}] {issue}")
sys.exit(1)
sys.exit(0)
if __name__ == "__main__":
main()