Indira Thouvenin

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 ELFExport


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

    def _get_exported_functions_from_symbols(self, elf) -> Set[str]:
        """Extract exported functions from symbol tables."""
        exported_functions = set()

        try:
            # Check both static (.symtab) and dynamic (.dynsym) symbol tables
            symbol_sections = ['.symtab', '.dynsym']

            for section_name in symbol_sections:
                section = elf.get_section_by_name(section_name)
                if not section or not hasattr(section, 'iter_symbols'):
                    continue

                for symbol in section.iter_symbols():
                    # Check if symbol is exported (defined and globally visible)
                    if (symbol.name and
                        symbol.entry.get('st_shndx', 0) != 'SHN_UNDEF' and  # Not undefined
                        symbol.entry.get('st_info', {}).get('bind') in ['STB_GLOBAL', 'STB_WEAK'] and  # Global or weak binding
                        symbol.entry.get('st_info', {}).get('type') in ['STT_FUNC', 'STT_OBJECT']):  # Function or object

                        # Additional checks for meaningful exports
                        if (symbol.entry.get('st_value', 0) > 0 or  # Has a value
                            symbol.entry.get('st_size', 0) > 0):    # Has a size
                            exported_functions.add(symbol.name)

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

        return exported_functions

    def _get_exported_functions_from_dynamic_section(self, elf) -> Set[str]:
        """Extract exported functions information from dynamic section."""
        exported_functions = set()

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

            # Look for version definition sections that might indicate exports
            # This is complementary to symbol table analysis
            for tag in dynamic_section.iter_tags():
                if tag.entry.d_tag == 'DT_SONAME':
                    # If it has a SONAME, it's likely a shared library with exports
                    # The actual exports are still found in symbol tables
                    pass

        except Exception as e:
            self.log.error(f"Error extracting exports from dynamic section: {e}")

        return exported_functions

    def _filter_meaningful_exports(self, exports: Set[str]) -> Set[str]:
        """Filter out compiler-generated and internal symbols to focus on meaningful exports."""
        filtered_exports = set()

        # Common patterns to exclude (compiler-generated, internal symbols)
        exclude_patterns = [
            '_start',
            '_init',
            '_fini',
            '__libc_',
            '__gmon_start__',
            '_IO_stdin_used',
            '__data_start',
            '__bss_start',
            '_edata',
            '_end',
            '__TMC_END__',
            '_ITM_deregisterTMCloneTable',
            '_ITM_registerTMCloneTable',
            '__cxa_finalize',
            '__gxx_personality_v0',
            '_Jv_RegisterClasses'
        ]

        for export in exports:
            # Skip empty or very short names
            if not export or len(export) < 2:
                continue

            # Skip symbols that match exclude patterns
            skip = False
            for pattern in exclude_patterns:
                if export.startswith(pattern):
                    skip = True
                    break

            # Skip symbols that look like internal mangled names (but keep reasonable C++ names)
            if export.startswith('_Z') and len(export) > 50:  # Very long mangled names
                skip = True

            if not skip:
                filtered_exports.add(export)

        return filtered_exports

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

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

            def extract_data(elf):
                # Extract exported functions from multiple sources
                exported_functions = set()

                # From symbol tables (primary source)
                symbol_exports = self._get_exported_functions_from_symbols(elf)
                exported_functions.update(symbol_exports)

                # From dynamic section (supplementary)
                dynamic_exports = self._get_exported_functions_from_dynamic_section(elf)
                exported_functions.update(dynamic_exports)

                # Filter out compiler-generated and internal symbols
                meaningful_exports = self._filter_meaningful_exports(exported_functions)

                # Convert to sorted list for consistent output
                export_functions_list = sorted(list(meaningful_exports))

                # Return ELFExport dataclass
                return ELFExport(
                    elf_exports_total=len(export_functions_list),
                    elf_export_functions=export_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_exports = result
            return self.elf_exports

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

                # Prepare data array
                data = [[
                    self.sha256,
                    self.md5,
                    self.sha1,
                    self.elf_exports.elf_exports_total,
                    self.elf_exports.elf_export_functions,
                    datetime.now(timezone.utc)
                ]]

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'elf_exports_total',
                    'elf_export_functions',
                    'analysis_date'
                ]

                if not data:
                    return None

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    'UInt32',
                    'Array(LowCardinality(String))',
                    'DateTime64(3, \'UTC\')'
                ]

                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_exports"