Rami Mustafa A. Mohammad

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"