scripts/logging/dictionary/database_gen.py

#!/usr/bin/env python3
#
# Copyright (c) 2020 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0

"""
Dictionary-based Logging Database Generator

This takes the built Zephyr ELF binary and produces a JSON database
file for dictionary-based logging. This database is used together
with the parser to decode binary log messages.
"""

import argparse
import logging
import os
import re
import string
import struct
import sys

import dictionary_parser.log_database
from dictionary_parser.log_database import LogDatabase
from dictionary_parser.utils import extract_one_string_in_section
from dictionary_parser.utils import find_string_in_mappings

import elftools
from elftools.elf.constants import SH_FLAGS
from elftools.elf.elffile import ELFFile
from elftools.elf.descriptions import describe_ei_data
from elftools.elf.sections import SymbolTableSection
from elftools.dwarf.descriptions import (
    describe_DWARF_expr
)
from elftools.dwarf.locationlists import (
    LocationExpr, LocationParser
)


LOGGER_FORMAT = "%(name)s: %(levelname)s: %(message)s"
logger = logging.getLogger(os.path.basename(sys.argv[0]))


# Sections that contains static strings
STATIC_STRING_SECTIONS = [
    'rodata',
    '.rodata',
    'pinned.rodata',
]


# Regulation expression to match DWARF location
DT_LOCATION_REGEX = re.compile(r"\(DW_OP_addr: ([0-9a-f]+)")


# Format string for pointers (default for 32-bit pointers)
PTR_FMT = '0x%08x'


# Potential string encodings. Add as needed.
STR_ENCODINGS = [
    'ascii',
    'iso-8859-1',
]


# List of acceptable escape character
ACCEPTABLE_ESCAPE_CHARS = [
    b'\r',
    b'\n',
]


def parse_args():
    """Parse command line arguments"""
    argparser = argparse.ArgumentParser()

    argparser.add_argument("elffile", help="Zephyr ELF binary")
    argparser.add_argument("--build", help="Build ID")
    argparser.add_argument("--build-header",
                           help="Header file containing BUILD_VERSION define")
    argparser.add_argument("--debug", action="store_true",
                           help="Print extra debugging information")
    argparser.add_argument("-v", "--verbose", action="store_true",
                           help="Print more information")

    outfile_grp = argparser.add_mutually_exclusive_group(required=True)
    outfile_grp.add_argument("--json",
                             help="Output Dictionary Logging Database file in JSON")
    outfile_grp.add_argument("--syst",
                             help="Output MIPI Sys-T Collateral XML file")

    return argparser.parse_args()


def extract_elf_code_data_sections(elf):
    """Find all sections in ELF file"""
    sections = {}

    for sect in elf.iter_sections():
        # Only Allocated sections with PROGBITS are included
        # since they actually have code/data.
        #
        # On contrary, BSS is allocated but NOBITS.
        if (
            (sect['sh_flags'] & SH_FLAGS.SHF_ALLOC) == SH_FLAGS.SHF_ALLOC
            and sect['sh_type'] == 'SHT_PROGBITS'
        ):
            sections[sect.name] = {
                    'name'    : sect.name,
                    'size'    : sect['sh_size'],
                    'start'   : sect['sh_addr'],
                    'end'     : sect['sh_addr'] + sect['sh_size'] - 1,
                    'data'    : sect.data(),
                }

    return sections


def find_elf_sections(elf, sh_name):
    """Find all sections in ELF file"""
    for section in elf.iter_sections():
        if section.name == sh_name:
            ret = {
                'name'    : section.name,
                'size'    : section['sh_size'],
                'start'   : section['sh_addr'],
                'end'     : section['sh_addr'] + section['sh_size'] - 1,
                'data'    : section.data(),
            }

            return ret

    return None


def get_kconfig_symbols(elf):
    """Get kconfig symbols from the ELF file"""
    for section in elf.iter_sections():
        if isinstance(section, SymbolTableSection):
            return {sym.name: sym.entry.st_value
                    for sym in section.iter_symbols()
                       if sym.name.startswith("CONFIG_")}

    raise LookupError("Could not find symbol table")


def find_log_const_symbols(elf):
    """Extract all "log_const_*" symbols from ELF file"""
    symbol_tables = [s for s in elf.iter_sections()
                     if isinstance(s, elftools.elf.sections.SymbolTableSection)]

    ret_list = []

    for section in symbol_tables:
        if not isinstance(section, elftools.elf.sections.SymbolTableSection):
            continue

        if section['sh_entsize'] == 0:
            continue

        for symbol in section.iter_symbols():
            if symbol.name.startswith("log_const_"):
                ret_list.append(symbol)

    return ret_list


def parse_log_const_symbols(database, log_const_section, log_const_symbols, string_mappings):
    """Find the log instances and map source IDs to names"""
    if database.is_tgt_little_endian():
        formatter = "<"
    else:
        formatter = ">"

    if database.is_tgt_64bit():
        # 64-bit pointer to string
        formatter += "Q"
    else:
        # 32-bit pointer to string
        formatter += "L"

    # log instance level
    formatter += "B"

    datum_size = struct.calcsize(formatter)

    # Get the address of first log instance
    first_offset = log_const_symbols[0].entry['st_value']
    for sym in log_const_symbols:
        if sym.entry['st_value'] < first_offset:
            first_offset = sym.entry['st_value']

    first_offset -= log_const_section['start']

    # find all log_const_*
    for sym in log_const_symbols:
        # Find data offset in log_const_section for this symbol
        offset = sym.entry['st_value'] - log_const_section['start']

        idx_s = offset
        idx_e = offset + datum_size

        datum = log_const_section['data'][idx_s:idx_e]

        if len(datum) != datum_size:
            # Not enough data to unpack
            continue

        str_ptr, level = struct.unpack(formatter, datum)

        # Offset to rodata section for string
        instance_name = find_string_in_mappings(string_mappings, str_ptr)
        if instance_name is None:
            instance_name = "unknown"

        logger.info("Found Log Instance: %s, level: %d", instance_name, level)

        # source ID is simply the element index in the log instance array
        source_id = int((offset - first_offset) / sym.entry['st_size'])

        database.add_log_instance(source_id, instance_name, level, sym.entry['st_value'])


def extract_elf_information(elf, database):
    """Extract information from ELF file and store in database"""
    e_ident = elf.header['e_ident']
    elf_data = describe_ei_data(e_ident['EI_DATA'])

    if elf_data == elftools.elf.descriptions._DESCR_EI_DATA['ELFDATA2LSB']:
        database.set_tgt_endianness(LogDatabase.LITTLE_ENDIAN)
    elif elf_data == elftools.elf.descriptions._DESCR_EI_DATA['ELFDATA2MSB']:
        database.set_tgt_endianness(LogDatabase.BIG_ENDIAN)
    else:
        logger.error("Cannot determine endianness from ELF file, exiting...")
        sys.exit(1)


def process_kconfigs(elf, database):
    """Process kconfigs to extract information"""
    kconfigs = get_kconfig_symbols(elf)

    # 32 or 64-bit target
    database.set_tgt_bits(64 if "CONFIG_64BIT" in kconfigs else 32)

    # Architecture
    for name, arch in dictionary_parser.log_database.ARCHS.items():
        if arch['kconfig'] in kconfigs:
            database.set_arch(name)
            break

    # Put some kconfigs into the database
    #
    # Use 32-bit timestamp? or 64-bit?
    if "CONFIG_LOG_TIMESTAMP_64BIT" in kconfigs:
        database.add_kconfig("CONFIG_LOG_TIMESTAMP_64BIT",
                             kconfigs['CONFIG_LOG_TIMESTAMP_64BIT'])


def extract_logging_subsys_information(elf, database, string_mappings):
    """
    Extract logging subsys related information and store in database.

    For example, this extracts the list of log instances to establish
    mapping from source ID to name.
    """
    # Extract log constant section for module names
    section_log_const = find_elf_sections(elf, "log_const_sections")
    if section_log_const is None:
        # ESP32 puts "log_const_*" info log_static_section instead of log_const_sections
        section_log_const = find_elf_sections(elf, "log_static_section")

    if section_log_const is None:
        logger.error("Cannot find section 'log_const_sections' in ELF file, exiting...")
        sys.exit(1)

    # Find all "log_const_*" symbols and parse them
    log_const_symbols = find_log_const_symbols(elf)
    parse_log_const_symbols(database, section_log_const, log_const_symbols, string_mappings)


def is_die_attr_ref(attr):
    """
    Returns True if the DIE attribute is a reference.
    """
    return bool(attr.form in ('DW_FORM_ref1', 'DW_FORM_ref2',
                              'DW_FORM_ref4', 'DW_FORM_ref8',
                              'DW_FORM_ref'))


def find_die_var_base_type(compile_unit, die, is_const):
    """
    Finds the base type of a DIE and returns the name.
    If DW_AT_type is a reference, it will recursively go through
    the references to find the base type. Returns None is no
    base type is found.
    """
    # DIE is of base type. So extract the name.
    if die.tag == 'DW_TAG_base_type':
        return die.attributes['DW_AT_name'].value.decode('ascii'), is_const

    # Not a type, cannot continue
    if not 'DW_AT_type' in die.attributes:
        return None, None

    if die.tag == 'DW_TAG_const_type':
        is_const = True

    # DIE is probably a reference to another.
    # If so, check if the reference is a base type.
    type_attr = die.attributes['DW_AT_type']
    if is_die_attr_ref(type_attr):
        ref_addr = compile_unit.cu_offset + type_attr.raw_value
        ref_die = compile_unit.get_DIE_from_refaddr(ref_addr)

        return find_die_var_base_type(compile_unit, ref_die, is_const)

    # Not a base type, and not reference
    return None, None


def is_die_var_const_char(compile_unit, die):
    """
    Returns True if DIE of type variable is const char.
    """
    var_type, is_const = find_die_var_base_type(compile_unit, die, False)

    if var_type is not None and var_type.endswith('char') and is_const:
        return True

    return False


def extract_string_variables(elf):
    """
    Find all string variables (char) in all Compilation Units and
    Debug information Entry (DIE) in ELF file.
    """
    dwarf_info = elf.get_dwarf_info()
    loc_lists = dwarf_info.location_lists()
    loc_parser = LocationParser(loc_lists)

    strings = []

    # Loop through all Compilation Units and
    # Debug information Entry (DIE) to extract all string variables
    for compile_unit in dwarf_info.iter_CUs():
        for die in compile_unit.iter_DIEs():
            # Only care about variables with location information
            # and of type "char"
            if die.tag == 'DW_TAG_variable':
                if ('DW_AT_type' in die.attributes
                    and 'DW_AT_location' in die.attributes
                    and is_die_var_const_char(compile_unit, die)
                ):
                    # Extract location information, which is
                    # its address in memory.
                    loc_attr = die.attributes['DW_AT_location']
                    if loc_parser.attribute_has_location(loc_attr, die.cu['version']):
                        loc = loc_parser.parse_from_attribute(loc_attr, die.cu['version'])
                        if isinstance(loc, LocationExpr):
                            try:
                                addr = describe_DWARF_expr(loc.loc_expr,
                                                        dwarf_info.structs)

                                matcher = DT_LOCATION_REGEX.match(addr)
                                if matcher:
                                    addr = int(matcher.group(1), 16)
                                    if addr > 0:
                                        strings.append({
                                            'name': die.attributes['DW_AT_name'].value,
                                            'addr': addr,
                                            'die': die
                                        })
                            except KeyError:
                                pass

    return strings


def try_decode_string(str_maybe):
    """Check if it is a printable string"""
    for encoding in STR_ENCODINGS:
        try:
            decoded_str = str_maybe.decode(encoding)

            # Check if string is printable according to Python
            # since the parser (written in Python) will need to
            # print the string.
            #
            # Note that '\r' and '\n' are not included in
            # string.printable so they need to be checked separately.
            printable = True
            for one_char in decoded_str:
                if (one_char not in string.printable
                    and one_char not in ACCEPTABLE_ESCAPE_CHARS):
                    printable = False
                    break

            if printable:
                return decoded_str
        except UnicodeDecodeError:
            pass

    return None


def extract_strings_in_one_section(section, str_mappings):
    """Extract NULL-terminated strings in one ELF section"""
    bindata = section['data']

    if len(bindata) < 2:
        # Can't have a NULL-terminated string with fewer than 2 bytes.
        return str_mappings

    idx = 0

    # If first byte is not NULL, it may be a string.
    if bindata[0] == 0:
        start = None
    else:
        start = 0

    while idx < len(bindata):
        if start is None:
            if bindata[idx] == 0:
                # Skip NULL bytes to find next string
                idx += 1
            else:
                # Beginning of possible string
                start = idx
                idx += 1
        else:
            if bindata[idx] != 0:
                # Skipping till next NULL byte for possible string
                idx += 1
            else:
                # End of possible string
                end = idx

                if start != end:
                    str_maybe = bindata[start:end]
                    decoded_str = try_decode_string(str_maybe)

                    # Only store readable string
                    if decoded_str is not None:
                        addr = section['start'] + start

                        if addr not in str_mappings:
                            str_mappings[addr] = decoded_str

                            # Decoded string may contain un-printable characters
                            # (e.g. extended ASC-II characters) or control
                            # characters (e.g. '\r' or '\n'), so simply print
                            # the byte string instead.
                            logger.debug('Found string via extraction at ' + PTR_FMT + ': %s',
                                         addr, str_maybe)

                            # GCC-based toolchain will reuse the NULL character
                            # for empty strings. There is no way to know which
                            # one is being reused, so just treat all NULL character
                            # at the end of legitimate strings as empty strings.
                            null_addr = section['start'] + end
                            str_mappings[null_addr] = ''

                            logger.debug('Found null string via extraction at ' + PTR_FMT,
                                         null_addr)

                start = None
                idx += 1

    return str_mappings


def extract_static_strings(elf, database, section_extraction=False):
    """
    Extract static strings from ELF file using DWARF,
    and also extraction from binary data.
    """
    string_mappings = {}

    elf_sections = extract_elf_code_data_sections(elf)

    # Extract strings using ELF DWARF information
    str_vars = extract_string_variables(elf)
    for str_var in str_vars:
        for _, sect in elf_sections.items():
            one_str = extract_one_string_in_section(sect, str_var['addr'])
            if one_str is not None:
                string_mappings[str_var['addr']] = one_str
                logger.debug('Found string variable at ' + PTR_FMT + ': %s',
                             str_var['addr'], one_str)
                break

    if section_extraction:
        # Extract strings from ELF sections
        string_sections = STATIC_STRING_SECTIONS
        rawstr_map = {}

        # Some architectures may put static strings into additional sections.
        # So need to extract them too.
        arch_data = dictionary_parser.log_database.ARCHS[database.get_arch()]
        if "extra_string_section" in arch_data:
            string_sections.extend(arch_data['extra_string_section'])

        for sect_name in string_sections:
            if sect_name in elf_sections:
                rawstr_map = extract_strings_in_one_section(elf_sections[sect_name],
                                                                rawstr_map)

        for one_str in rawstr_map:
            if one_str not in string_mappings:
                string_mappings[one_str] = rawstr_map[one_str]

    return string_mappings


def main():
    """Main function of database generator"""
    args = parse_args()

    # Setup logging
    logging.basicConfig(format=LOGGER_FORMAT, level=logging.WARNING)
    if args.debug:
        logger.setLevel(logging.DEBUG)
    elif args.verbose:
        logger.setLevel(logging.INFO)

    elffile = open(args.elffile, "rb")
    if not elffile:
        logger.error("ERROR: Cannot open ELF file: %s, exiting...", args.elffile)
        sys.exit(1)

    logger.info("ELF file %s", args.elffile)

    if args.json:
        logger.info("JSON Database file %s", args.json)
        section_extraction = True

    if args.syst:
        logger.info("MIPI Sys-T Collateral file %s", args.syst)
        section_extraction = False

    elf = ELFFile(elffile)

    database = LogDatabase()

    if args.build_header:
        with open(args.build_header) as f:
            for l in f:
                match = re.match(r'\s*#define\s+BUILD_VERSION\s+(.*)', l)
                if match:
                    database.set_build_id(match.group(1))
                    break

    if args.build:
        database.set_build_id(args.build)
        logger.info("Build ID: %s", args.build)

    extract_elf_information(elf, database)

    process_kconfigs(elf, database)

    logger.info("Target: %s, %d-bit", database.get_arch(), database.get_tgt_bits())
    if database.is_tgt_little_endian():
        logger.info("Endianness: Little")
    else:
        logger.info("Endianness: Big")

    if database.is_tgt_64bit():
        global PTR_FMT
        PTR_FMT = '0x%016x'

    # Extract strings from ELF files
    string_mappings = extract_static_strings(elf, database, section_extraction)
    if len(string_mappings) > 0:
        database.set_string_mappings(string_mappings)
        logger.info("Found %d strings", len(string_mappings))

    # Extract information related to logging subsystem
    if not section_extraction:
        # The logging subsys information (e.g. log module names)
        # may require additional strings outside of those extracted
        # via ELF DWARF variables. So generate a new string mappings
        # with strings in various ELF sections.
        string_mappings = extract_static_strings(elf, database, section_extraction=True)

    extract_logging_subsys_information(elf, database, string_mappings)

    # Write database file
    if args.json:
        if not LogDatabase.write_json_database(args.json, database):
            logger.error("ERROR: Cannot open database file for write: %s, exiting...",
                         args.json)
            sys.exit(1)

    if args.syst:
        if not LogDatabase.write_syst_database(args.syst, database):
            logger.error("ERROR: Cannot open database file for write: %s, exiting...",
                         args.syst)
            sys.exit(1)

    elffile.close()


if __name__ == "__main__":
    main()