Wei Li

import inspect
import json
from datetime import datetime, timezone
from typing import Any

from elftools.elf.elffile import ELFFile
from elftools.common.exceptions import ELFError

from redb.extractors.enum import Tag
from redb.extractors.elf_extractor import ELFExtractor
from redb.models.dataclasses import ELFFeatures


class ELFFeaturesExtractor(ELFExtractor):

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        elf=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            elastic_index,
            known_benign,
            known_malicious,
            elf,
        )
        self.elf_features = None
        self.elastic_index = self.index_prefix + "-elf_features"
        self.log.debug(inspect.currentframe().f_code.co_name)

    def _extract_header_data(self):
        """Extract ELF header data using pyelftools."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        def extract_header(elf):
            # Initialize result with safe defaults
            result = {
                'ei_class': 0, 'ei_data': 0, 'ei_version': 0, 'ei_osabi': 0, 'ei_abiversion': 0,
                'e_type': 0, 'e_machine': 0, 'e_version': 0, 'e_entry': 0, 'e_flags': 0,
                'ei_class_str': 'unknown', 'ei_data_str': 'unknown', 'ei_osabi_str': 'unknown',
                'e_type_str': 'unknown', 'e_machine_str': 'unknown'
            }

            # Try to get the main header - if this fails, we can't extract anything
            try:
                header = elf.header
            except Exception as e:
                self.log.warning(f"Could not access ELF header: {e}")
                return result

            # Try to get e_ident section
            ident = None
            try:
                ident = header.get('e_ident', {})
            except Exception as e:
                self.log.warning(f"Could not access e_ident: {e}")

            # Extract EI_CLASS (32/64 bit)
            try:
                if ident:
                    ei_class_raw = ident.get('EI_CLASS', 'ELFCLASSNONE')
                    result['ei_class'] = 2 if ei_class_raw == 'ELFCLASS64' else 1 if ei_class_raw == 'ELFCLASS32' else 0
                    result['ei_class_str'] = '64-bit' if ei_class_raw == 'ELFCLASS64' else '32-bit' if ei_class_raw == 'ELFCLASS32' else 'unknown'
            except Exception as e:
                self.log.warning(f"Could not extract EI_CLASS: {e}")

            # Extract EI_DATA (endianness)
            try:
                if ident:
                    ei_data_raw = ident.get('EI_DATA', 'ELFDATANONE')
                    result['ei_data'] = 1 if ei_data_raw == 'ELFDATA2LSB' else 2 if ei_data_raw == 'ELFDATA2MSB' else 0
                    result['ei_data_str'] = 'Little-endian' if ei_data_raw == 'ELFDATA2LSB' else 'Big-endian' if ei_data_raw == 'ELFDATA2MSB' else 'unknown'
            except Exception as e:
                self.log.warning(f"Could not extract EI_DATA: {e}")

            # Extract EI_VERSION
            try:
                if ident:
                    ei_version_raw = ident.get('EI_VERSION', 0)
                    # Convert string like 'EV_CURRENT' to integer
                    if isinstance(ei_version_raw, str):
                        version_map = {'EV_NONE': 0, 'EV_CURRENT': 1}
                        result['ei_version'] = version_map.get(ei_version_raw, 1)
                    else:
                        result['ei_version'] = ei_version_raw
            except Exception as e:
                self.log.warning(f"Could not extract EI_VERSION: {e}")

            # Extract EI_OSABI
            try:
                if ident:
                    ei_osabi_raw = ident.get('EI_OSABI', 'ELFOSABI_SYSV')
                    osabi_map = {
                        'ELFOSABI_SYSV': 0, 'ELFOSABI_HPUX': 1, 'ELFOSABI_NETBSD': 2,
                        'ELFOSABI_LINUX': 3, 'ELFOSABI_SOLARIS': 6, 'ELFOSABI_FREEBSD': 9
                    }
                    result['ei_osabi'] = osabi_map.get(ei_osabi_raw, 0)
                    result['ei_osabi_str'] = ei_osabi_raw.replace('ELFOSABI_', '') if ei_osabi_raw and ei_osabi_raw.startswith('ELFOSABI_') else str(ei_osabi_raw) if ei_osabi_raw else 'unknown'
            except Exception as e:
                self.log.warning(f"Could not extract EI_OSABI: {e}")

            # Extract EI_ABIVERSION
            try:
                if ident:
                    result['ei_abiversion'] = ident.get('EI_ABIVERSION', 0)
            except Exception as e:
                self.log.warning(f"Could not extract EI_ABIVERSION: {e}")

            # Extract e_type (file type)
            try:
                e_type_raw = header.get('e_type', 'ET_NONE')
                type_map = {
                    'ET_NONE': 0, 'ET_REL': 1, 'ET_EXEC': 2, 'ET_DYN': 3, 'ET_CORE': 4
                }
                result['e_type'] = type_map.get(e_type_raw, 0)
                result['e_type_str'] = e_type_raw.replace('ET_', '') if e_type_raw and e_type_raw.startswith('ET_') else str(e_type_raw) if e_type_raw else 'unknown'
            except Exception as e:
                self.log.warning(f"Could not extract e_type: {e}")

            # Extract e_machine (architecture)
            try:
                e_machine_raw = header.get('e_machine', 0)
                # Map machine strings to integers and human-readable names
                # https://github.com/eliben/pyelftools/blob/main/elftools/elf/enums.py
                # https://github.com/torvalds/linux/blob/master/include/uapi/linux/elf-em.h
                # https://refspecs.linuxfoundation.org/elf/elf.pdf
                machine_int_map = {
                    'EM_NONE': 0, 'EM_M32': 1, 'EM_SPARC': 2, 'EM_386': 3,
                    'EM_68K': 4, 'EM_88K': 5, 'EM_860': 7, 'EM_MIPS': 8,
                    'EM_S370': 9, 'EM_MIPS_RS3_LE': 10, 'EM_PARISC': 15,
                    'EM_VPP500': 17, 'EM_SPARC32PLUS': 18, 'EM_960': 19,
                    'EM_PPC': 20, 'EM_PPC64': 21, 'EM_S390': 22, 'EM_V800': 36,
                    'EM_FR20': 37, 'EM_RH32': 38, 'EM_RCE': 39, 'EM_ARM': 40,
                    'EM_ALPHA': 41, 'EM_SH': 42, 'EM_SPARCV9': 43, 'EM_TRICORE': 44,
                    'EM_ARC': 45, 'EM_H8_300': 46, 'EM_H8_300H': 47, 'EM_H8S': 48,
                    'EM_H8_500': 49, 'EM_IA_64': 50, 'EM_MIPS_X': 51, 'EM_COLDFIRE': 52,
                    'EM_68HC12': 53, 'EM_X86_64': 62, 'EM_AARCH64': 183, 'EM_RISCV': 243,
                }
                machine_str_map = {
                    'EM_386': 'x86', 'EM_X86_64': 'x86_64', 'EM_ARM': 'ARM', 'EM_AARCH64': 'ARM64',
                    'EM_MIPS': 'MIPS', 'EM_PPC': 'PowerPC', 'EM_PPC64': 'PowerPC64',
                    'EM_SPARC': 'SPARC', 'EM_SPARCV9': 'SPARC64', 'EM_RISCV': 'RISC-V',
                    'EM_IA_64': 'IA-64', 'EM_S390': 'S390', 'EM_SH': 'SuperH',
                }
                if isinstance(e_machine_raw, str):
                    result['e_machine'] = machine_int_map.get(e_machine_raw, 0)
                    result['e_machine_str'] = machine_str_map.get(e_machine_raw, e_machine_raw.replace('EM_', '') if e_machine_raw.startswith('EM_') else e_machine_raw)
                else:
                    result['e_machine'] = e_machine_raw
                    result['e_machine_str'] = str(e_machine_raw)
            except Exception as e:
                self.log.warning(f"Could not extract e_machine: {e}")

            # Extract e_version
            try:
                e_version_raw = header.get('e_version', 0)
                # Convert string like 'EV_CURRENT' to integer
                if isinstance(e_version_raw, str):
                    version_map = {'EV_NONE': 0, 'EV_CURRENT': 1}
                    result['e_version'] = version_map.get(e_version_raw, 1)
                else:
                    result['e_version'] = e_version_raw
            except Exception as e:
                self.log.warning(f"Could not extract e_version: {e}")

            # Extract e_entry (entry point)
            try:
                result['e_entry'] = header.get('e_entry', 0)
            except Exception as e:
                self.log.warning(f"Could not extract e_entry: {e}")

            # Extract e_flags
            try:
                result['e_flags'] = header.get('e_flags', 0)
            except Exception as e:
                self.log.warning(f"Could not extract e_flags: {e}")

            return result

        try:
            return self._with_elf_file(extract_header)
        except Exception as e:
            self.log.error(f"Error extracting header data {self.hash.sha256}: {e}")
            return None

    def _count_dynamic_symbols(self):
        """Count symbols in dynamic symbol table."""
        def count_dynsym(elf):
            dynsym_section = elf.get_section_by_name('.dynsym')
            if dynsym_section and hasattr(dynsym_section, 'num_symbols'):
                return dynsym_section.num_symbols()
            return 0

        try:
            result = self._with_elf_file(count_dynsym)
            return result if result is not None else 0
        except Exception as e:
            self.log.error(f"Error counting dynamic symbols: {e}")
            return 0

    def _count_relocations(self):
        """Count total relocations across all sections."""
        def count_relocs(elf):
            reloc_count = 0
            for section in elf.iter_sections():
                if hasattr(section, 'iter_relocations'):
                    try:
                        reloc_count += section.num_relocations()
                    except:
                        # Some sections might not support num_relocations()
                        for _ in section.iter_relocations():
                            reloc_count += 1
            return reloc_count

        try:
            result = self._with_elf_file(count_relocs)
            return result if result is not None else 0
        except Exception as e:
            self.log.error(f"Error counting relocations: {e}")
            return 0

    def _has_gnu_hash(self):
        """Check if GNU hash table is present."""
        def check_gnu_hash(elf):
            gnu_hash_section = elf.get_section_by_name('.gnu.hash')
            return gnu_hash_section is not None

        try:
            result = self._with_elf_file(check_gnu_hash)
            return bool(result)
        except Exception as e:
            self.log.error(f"Error checking GNU hash: {e}")
            return False

    def _has_fortify(self):
        """Check if binary has fortify source protection."""
        def check_fortify(elf):
            # Look for fortified function symbols
            fortify_symbols = ['__printf_chk', '__sprintf_chk', '__snprintf_chk',
                             '__strcpy_chk', '__strcat_chk', '__memcpy_chk']

            for section in elf.iter_sections():
                if hasattr(section, 'iter_symbols'):
                    for symbol in section.iter_symbols():
                        if symbol.name in fortify_symbols:
                            return True
            return False

        try:
            result = self._with_elf_file(check_fortify)
            return bool(result)
        except Exception as e:
            self.log.error(f"Error checking fortify: {e}")
            return False

    def tag(self):
        return Tag.ELF_FEATURES.value if hasattr(Tag, 'ELF_FEATURES') else "elf_features"

    def extract(self):
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)

            # Check if file is valid ELF
            if not self._is_elf_file():
                self.log.error(f"No valid ELF file for {self.hash.sha256}")
                return None

            # Extract header data (now always returns a result, even with defaults)
            header_data = self._extract_header_data()
            if not header_data:
                self.log.error(f"Complete failure to extract any header data for {self.hash.sha256}")
                return None

            # Extract binary properties
            is_64bit = 1 if self._is_64bit() else 0
            is_stripped = 1 if self._is_stripped() else 0
            is_pie = 1 if self._is_pie() else 0
            has_canary = 1 if self._has_stack_protection() else 0
            has_nx = 1 if self._has_nx_bit() else 0
            has_relro = 1 if self._has_relro() else 0
            has_fortify = 1 if self._has_fortify() else 0

            # Extract counts
            number_of_segments = self._count_segments()
            number_of_sections = self._count_sections()
            number_of_symbols = self._count_symbols()
            number_of_dynamic_symbols = self._count_dynamic_symbols()
            number_of_relocations = self._count_relocations()
            number_of_dependencies = len(self._get_dependencies())

            # Extract build information
            build_id = self._get_build_id()
            gnu_hash_present = 1 if self._has_gnu_hash() else 0
            has_debug_info = 1 if self._has_debug_info() else 0

            self.elf_features = ELFFeatures(
                # ELF Header data
                ei_class=header_data['ei_class'],
                ei_data=header_data['ei_data'],
                ei_version=header_data['ei_version'],
                ei_osabi=header_data['ei_osabi'],
                ei_abiversion=header_data['ei_abiversion'],
                e_type=header_data['e_type'],
                e_machine=header_data['e_machine'],
                e_version=header_data['e_version'],
                e_entry=header_data['e_entry'],
                e_flags=header_data['e_flags'],
                # Human readable values
                ei_class_str=header_data['ei_class_str'],
                ei_data_str=header_data['ei_data_str'],
                ei_osabi_str=header_data['ei_osabi_str'],
                e_type_str=header_data['e_type_str'],
                e_machine_str=header_data['e_machine_str'],
                # Binary properties
                is_64bit=is_64bit,
                is_stripped=is_stripped,
                is_pie=is_pie,
                has_canary=has_canary,
                has_nx=has_nx,
                has_relro=has_relro,
                has_fortify=has_fortify,
                # Counts
                number_of_segments=number_of_segments,
                number_of_sections=number_of_sections,
                number_of_symbols=number_of_symbols,
                number_of_dynamic_symbols=number_of_dynamic_symbols,
                number_of_relocations=number_of_relocations,
                number_of_dependencies=number_of_dependencies,
                # Build information
                build_id=build_id,
                gnu_hash_present=gnu_hash_present,
                has_debug_info=has_debug_info,
            )

            return self.elf_features

        except Exception as e:
            self.log.error(f"Error extracting ELF features {self.hash.sha256}: {e}")
            return None

    def prepare_export_data(self, exporter_type: str) -> Any:
        self.log.debug(inspect.currentframe().f_code.co_name)

        if exporter_type == "ElasticsearchExporter":
            return self.elf_features
        elif exporter_type == "ClickHouseExporter":
            try:
                if not self.elf_features:
                    return None

                # Prepare data array matching the schema
                data = [[
                    self.sha256,
                    self.md5,
                    self.sha1,

                    # ELF Header data
                    self.elf_features.ei_class,
                    self.elf_features.ei_data,
                    self.elf_features.ei_version,
                    self.elf_features.ei_osabi,
                    self.elf_features.ei_abiversion,
                    self.elf_features.e_type,
                    self.elf_features.e_machine,
                    self.elf_features.e_version,
                    self.elf_features.e_entry,
                    self.elf_features.e_flags,

                    # Human readable values
                    self.elf_features.ei_class_str,
                    self.elf_features.ei_data_str,
                    self.elf_features.ei_osabi_str,
                    self.elf_features.e_type_str,
                    self.elf_features.e_machine_str,

                    # Binary properties
                    self.elf_features.is_64bit,
                    self.elf_features.is_stripped,
                    self.elf_features.is_pie,
                    self.elf_features.has_canary,
                    self.elf_features.has_nx,
                    self.elf_features.has_relro,
                    self.elf_features.has_fortify,

                    # Counts
                    self.elf_features.number_of_segments,
                    self.elf_features.number_of_sections,
                    self.elf_features.number_of_symbols,
                    self.elf_features.number_of_dynamic_symbols,
                    self.elf_features.number_of_relocations,
                    self.elf_features.number_of_dependencies,

                    # Build information
                    self.elf_features.build_id,
                    self.elf_features.gnu_hash_present,
                    self.elf_features.has_debug_info,

                    # Analysis metadata
                    datetime.now(timezone.utc)
                ]]

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'ei_class', 'ei_data', 'ei_version', 'ei_osabi', 'ei_abiversion',
                    'e_type', 'e_machine', 'e_version', 'e_entry', 'e_flags',
                    'ei_class_str', 'ei_data_str', 'ei_osabi_str', 'e_type_str', 'e_machine_str',
                    'is_64bit', 'is_stripped', 'is_pie', 'has_canary', 'has_nx', 'has_relro', 'has_fortify',
                    'number_of_segments', 'number_of_sections', 'number_of_symbols',
                    'number_of_dynamic_symbols', 'number_of_relocations', 'number_of_dependencies',
                    'build_id', 'gnu_hash_present', 'has_debug_info',
                    'analysis_date'
                ]

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    "Enum8('32-bit'=1, '64-bit'=2)",
                    "Enum8('Little-endian'=1, 'Big-endian'=2)",
                    'UInt8',
                    "Enum8('SYSV'=0, 'HPUX'=1, 'NETBSD'=2, 'LINUX'=3, 'SOLARIS'=6, 'FREEBSD'=9)",
                    'UInt8',
                    "Enum8('NONE'=0, 'REL'=1, 'EXEC'=2, 'DYN'=3, 'CORE'=4)",
                    'UInt16', 'UInt32', 'UInt64', 'UInt32',
                    'LowCardinality(String)', 'LowCardinality(String)', 'LowCardinality(String)',
                    'LowCardinality(String)', 'LowCardinality(String)',
                    'UInt8', 'UInt8', 'UInt8', 'UInt8', 'UInt8', 'UInt8', 'UInt8',
                    'UInt16', 'UInt16', 'UInt32', 'UInt32', 'UInt32', 'UInt16',
                    'Nullable(String)', 'UInt8', 'UInt8',
                    'DateTime64(3, \'UTC\')'
                ]

                if not data:
                    return None

                return (data, column_names, column_type_names)

            except Exception as e:
                self.log.error(f"Error preparing export data: {e}")
                raise

    def get_clickhouse_table(self) -> str:
        return "redb_elf_features"