Wei Quan

import logging
from abc import ABCMeta, abstractmethod
import inspect

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

from redb.extractors.extractor import Extractor

logger = logging.getLogger(__name__)


@abstractmethod
class ELFExtractor(Extractor, metaclass=ABCMeta):

    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
        )
        # Store ELF object if provided, otherwise we'll create it per-operation for security
        self._provided_elf = elf
        self._elf_file_valid = None  # Cache validity check

    def _with_elf_file(self, operation):
        """Safely execute an operation with an ELF file using context manager.

        Args:
            operation: A callable that takes an ELFFile object and returns a result

        Returns:
            The result of the operation, or None if an error occurred
        """
        if self._provided_elf:
            try:
                return operation(self._provided_elf)
            except ELFError as e:
                if "String Table not found" in str(e):
                    self.log.warning(f"String table missing in ELF {self.hash.sha256}: {e}")
                    return None
                else:
                    self.log.error(f"ELF format error {self.hash.sha256} Full error: {e}")
                    return None
            except Exception as e:
                self.log.error(f"Error processing ELF file {self.hash.sha256} Full error: {e}")
                return None

        try:
            with open(self.filepath, 'rb') as f:
                elf = ELFFile(f)
                if not elf:
                    raise ELFError("Empty file?")
                return operation(elf)
        except ELFError as e:
            if "String Table not found" in str(e):
                self.log.warning(f"String table missing in ELF {self.hash.sha256}: {e}")
                return None
            else:
                self.log.error(f"ELF format error {self.hash.sha256} Full error: {e}")
                return None
        except Exception as e:
            self.log.error(f"Error reading ELF file {self.hash.sha256} Full error: {e}")
            return None

    def _is_elf_file(self):
        """Check if the file is a valid ELF binary."""
        if self._elf_file_valid is not None:
            return self._elf_file_valid

        def check_elf_validity(elf):
            # pyelftools ELFFile object existing means it's valid ELF
            # Just check that we can access the header
            header = elf.header
            return header is not None

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

    def _is_64bit(self):
        """Check if the ELF binary is 64-bit."""
        def check_64bit(elf):
            return elf.header.get('e_ident', {}).get('EI_CLASS') == 'ELFCLASS64'

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

    def _is_stripped(self):
        """Check if the ELF binary is stripped (no symbol table)."""
        def check_stripped(elf):
            # Look for symbol table sections
            for section in elf.iter_sections():
                if section.name in ['.symtab', '.strtab']:
                    return False
            return True

        try:
            result = self._with_elf_file(check_stripped)
            return result if result is not None else True
        except Exception as e:
            self.log.error(f"Error checking if ELF is stripped: {e}")
            return True

    def _has_debug_info(self):
        """Check if the ELF binary contains debug information."""
        def check_debug_info(elf):
            # Look for debug sections
            debug_sections = ['.debug_info', '.debug_line', '.debug_str', '.debug_abbrev']
            for section in elf.iter_sections():
                if section.name in debug_sections:
                    return True
            return False

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

    def _get_architecture(self):
        """Get the architecture of the ELF binary."""
        def get_arch(elf):
            machine = elf.header.get('e_machine', 'EM_NONE')

            # Map common machine types to readable names
            arch_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_RISCV': 'RISC-V'
            }

            return arch_map.get(machine, machine)

        try:
            result = self._with_elf_file(get_arch)
            return result if result is not None else "unknown"
        except Exception as e:
            self.log.error(f"Error getting architecture: {e}")
            return "unknown"

    def _get_endianness(self):
        """Get the endianness of the ELF binary."""
        def get_endian(elf):
            data_encoding = elf.header.get('e_ident', {}).get('EI_DATA')
            if data_encoding == 'ELFDATA2LSB':
                return "little"
            elif data_encoding == 'ELFDATA2MSB':
                return "big"
            return "unknown"

        try:
            result = self._with_elf_file(get_endian)
            return result if result is not None else "unknown"
        except Exception as e:
            self.log.error(f"Error getting endianness: {e}")
            return "unknown"

    def _get_file_type(self):
        """Get the file type of the ELF binary."""
        def get_file_type(elf):
            etype = elf.header.get('e_type', 'ET_NONE')

            # Map file types to readable names
            type_map = {
                'ET_NONE': 'none',
                'ET_REL': 'relocatable',
                'ET_EXEC': 'executable',
                'ET_DYN': 'shared_object',
                'ET_CORE': 'core_dump'
            }

            return type_map.get(etype, etype)

        try:
            result = self._with_elf_file(get_file_type)
            return result if result is not None else "unknown"
        except Exception as e:
            self.log.error(f"Error getting file type: {e}")
            return "unknown"

    def _is_pie(self):
        """Check if the ELF binary is position-independent executable."""
        def check_pie(elf):
            # PIE binaries are typically ET_DYN type
            etype = elf.header.get('e_type', 'ET_NONE')
            if etype == 'ET_DYN':
                # Check if it has an entry point (executable) vs library
                entry_point = elf.header.get('e_entry', 0)
                return entry_point > 0
            return False

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

    def _has_stack_protection(self):
        """Check if the binary has stack protection (canaries)."""
        def check_stack_protection(elf):
            # Look for stack protection symbols
            stack_symbols = ['__stack_chk_fail', '__stack_chk_guard']

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

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

    def _has_nx_bit(self):
        """Check if the binary has NX bit (non-executable stack)."""
        def check_nx_bit(elf):
            # Look for GNU_STACK segment
            for segment in elf.iter_segments():
                if segment.header.get('p_type') == 'PT_GNU_STACK':
                    flags = segment.header.get('p_flags', 0)
                    # Check if execute flag is NOT set (NX enabled)
                    return not (flags & 0x1)  # PF_X = 0x1
            return False

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

    def _has_relro(self):
        """Check if the binary has RELRO (Relocation Read-Only)."""
        def check_relro(elf):
            # Look for GNU_RELRO segment
            for segment in elf.iter_segments():
                if segment.header.get('p_type') == 'PT_GNU_RELRO':
                    return True
            return False

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

    def _get_build_id(self):
        """Extract build ID from notes section."""
        def get_build_id(elf):
            # Look for build ID in notes sections
            for section in elf.iter_sections():
                if section.name == '.note.gnu.build-id':
                    for note in section.iter_notes():
                        if note['n_type'] == 'NT_GNU_BUILD_ID':
                            # Convert bytes to hex string
                            build_id = note['n_desc']
                            if isinstance(build_id, bytes):
                                return build_id.hex()
                            return str(build_id)
            return None

        try:
            result = self._with_elf_file(get_build_id)
            return result
        except Exception as e:
            self.log.error(f"Error getting build ID: {e}")
            return None

    def _count_sections(self):
        """Count the number of sections in the ELF file."""
        def count_sections(elf):
            return elf.header.get('e_shnum', 0)

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

    def _count_segments(self):
        """Count the number of segments (program headers) in the ELF file."""
        def count_segments(elf):
            return elf.header.get('e_phnum', 0)

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

    def _count_symbols(self):
        """Count the total number of symbols in symbol tables."""
        def count_symbols(elf):
            symbol_count = 0
            for section in elf.iter_sections():
                if hasattr(section, 'iter_symbols'):
                    symbol_count += section.num_symbols()
            return symbol_count

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

    def _get_dependencies(self):
        """Get list of dynamic dependencies."""
        def get_dependencies(elf):
            dependencies = []
            dynamic_section = elf.get_section_by_name('.dynamic')
            if dynamic_section:
                for tag in dynamic_section.iter_tags():
                    if tag.entry.d_tag == 'DT_NEEDED':
                        dependencies.append(tag.needed)
            return dependencies

        try:
            result = self._with_elf_file(get_dependencies)
            return result if result is not None else []
        except Exception as e:
            self.log.error(f"Error getting dependencies: {e}")
            return []