Jan G. P. W. Clevers

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"