Nader Sohrabi Safa

import inspect
from datetime import datetime, timezone
from typing import Any, List, Set

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 ELFImport


class ELFImportExtractor(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_imports = None
        self.elastic_index = self.index_prefix + "-elf_imports"
        self.log.debug(inspect.currentframe().f_code.co_name)

    def _get_import_libraries(self, elf) -> List[str]:
        """Extract imported libraries from dynamic section."""
        libraries = []

        try:
            # Get the dynamic section
            dynamic_section = elf.get_section_by_name('.dynamic')
            if not dynamic_section:
                return libraries

            # Extract DT_NEEDED entries (required libraries)
            for tag in dynamic_section.iter_tags():
                if tag.entry.d_tag == 'DT_NEEDED':
                    libraries.append(tag.needed)

        except Exception as e:
            self.log.error(f"Error extracting import libraries: {e}")

        return libraries

    def _get_imported_functions_from_symbols(self, elf) -> Set[str]:
        """Extract imported functions from dynamic symbol table."""
        imported_functions = set()

        try:
            # Get the dynamic symbol table
            dynsym_section = elf.get_section_by_name('.dynsym')
            if not dynsym_section or not hasattr(dynsym_section, 'iter_symbols'):
                return imported_functions

            # Look for undefined symbols (imports)
            for symbol in dynsym_section.iter_symbols():
                # Check if symbol is undefined (imported)
                if (symbol.entry.get('st_shndx', 0) == 'SHN_UNDEF' and
                    symbol.name and
                    symbol.entry.get('st_info', {}).get('bind') in ['STB_GLOBAL', 'STB_WEAK']):
                    imported_functions.add(symbol.name)

        except Exception as e:
            self.log.error(f"Error extracting imported functions from symbols: {e}")

        return imported_functions

    def _get_imported_functions_from_relocations(self, elf) -> Set[str]:
        """Extract imported functions from relocation sections."""
        imported_functions = set()

        try:
            # Look through relocation sections
            for section in elf.iter_sections():
                if hasattr(section, 'iter_relocations'):
                    try:
                        for relocation in section.iter_relocations():
                            # Get symbol associated with relocation
                            if hasattr(relocation, 'symbol') and relocation.symbol:
                                symbol_name = relocation.symbol.name
                                if symbol_name:
                                    imported_functions.add(symbol_name)
                    except Exception as e:
                        self.log.debug(f"Could not process relocations in section {section.name}: {e}")

        except Exception as e:
            self.log.error(f"Error extracting imported functions from relocations: {e}")

        return imported_functions

    def _get_plt_functions(self, elf) -> Set[str]:
        """Extract functions from PLT (Procedure Linkage Table) sections."""
        plt_functions = set()

        try:
            # Look for PLT-related sections
            plt_sections = ['.plt', '.plt.got', '.plt.sec']

            for section_name in plt_sections:
                section = elf.get_section_by_name(section_name)
                if section:
                    # PLT functions are typically associated with relocations
                    # We'll get them from the relocation analysis
                    pass

        except Exception as e:
            self.log.error(f"Error extracting PLT functions: {e}")

        return plt_functions

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

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

            def extract_data(elf):
                # Extract import libraries
                import_libraries = self._get_import_libraries(elf)

                # Extract imported functions from multiple sources
                imported_functions = set()

                # From dynamic symbols
                symbol_imports = self._get_imported_functions_from_symbols(elf)
                imported_functions.update(symbol_imports)

                # From relocations
                relocation_imports = self._get_imported_functions_from_relocations(elf)
                imported_functions.update(relocation_imports)

                # From PLT
                plt_imports = self._get_plt_functions(elf)
                imported_functions.update(plt_imports)

                # Convert to sorted lists for consistent output
                import_libraries_list = sorted(list(set(import_libraries)))
                import_functions_list = sorted(list(imported_functions))

                # Return ELFImport dataclass
                return ELFImport(
                    elf_imports_total=len(import_functions_list),
                    elf_import_libraries=import_libraries_list,
                    elf_import_functions=import_functions_list,
                )

            if not self._is_elf_file():
                return None

            result = self._with_elf_file(extract_data)
            if result is None:
                return None

            self.elf_imports = result
            return self.elf_imports

        except Exception as e:
            self.log.error(f"Error extracting ELF imports {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_imports
        elif exporter_type == "ClickHouseExporter":
            try:
                if not self.elf_imports:
                    return None

                # Prepare data array
                data = [[
                    self.sha256,
                    self.md5,
                    self.sha1,
                    self.elf_imports.elf_imports_total,
                    self.elf_imports.elf_import_libraries,
                    self.elf_imports.elf_import_functions,
                    datetime.now(timezone.utc)
                ]]

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'elf_imports_total',
                    'elf_import_libraries',
                    'elf_import_functions',
                    'analysis_date'
                ]

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    'UInt32',
                    'Array(LowCardinality(String))',
                    'Array(LowCardinality(String))',
                    '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_imports"