Oleksii Duda

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

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 ELFSymbol


class ELFSymbolExtractor(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_symbols = []
        self.elastic_index = self.index_prefix + "-elf_symbols"
        self.log.debug(inspect.currentframe().f_code.co_name)

    def _map_symbol_type(self, st_type_str: str) -> int:
        """Map symbol type string to enum value."""
        type_map = {
            'STT_NOTYPE': 0,
            'STT_OBJECT': 1,
            'STT_FUNC': 2,
            'STT_SECTION': 3,
            'STT_FILE': 4,
            'STT_COMMON': 5,
            'STT_TLS': 6
        }
        return type_map.get(st_type_str, 0)

    def _map_symbol_bind(self, st_bind_str: str) -> int:
        """Map symbol binding string to enum value."""
        bind_map = {
            'STB_LOCAL': 0,
            'STB_GLOBAL': 1,
            'STB_WEAK': 2
        }
        return bind_map.get(st_bind_str, 0)

    def _map_symbol_visibility(self, st_vis_str: str) -> int:
        """Map symbol visibility string to enum value."""
        vis_map = {
            'STV_DEFAULT': 0,
            'STV_INTERNAL': 1,
            'STV_HIDDEN': 2,
            'STV_PROTECTED': 3
        }
        return vis_map.get(st_vis_str, 0)

    def _extract_symbol_data(self, symbol, is_dynamic: bool = False) -> Dict:
        """Extract data from a single symbol."""
        try:
            # Get symbol name (handle empty names)
            symbol_name = symbol.name if symbol.name else f"<unnamed_{symbol.entry.get('st_name', 0)}>"

            # Get symbol properties
            symbol_value = symbol.entry.get('st_value', 0)
            symbol_size = symbol.entry.get('st_size', 0)

            # Handle section index - can be integer or special string like 'SHN_UNDEF'
            st_shndx_raw = symbol.entry.get('st_shndx', 0)
            if isinstance(st_shndx_raw, str):
                # Map special section index strings to integers
                shndx_map = {
                    'SHN_UNDEF': 0,
                    'SHN_ABS': 65521,  # 0xFFF1
                    'SHN_COMMON': 65522,  # 0xFFF2
                    'SHN_XINDEX': 65535,  # 0xFFFF
                }
                symbol_section_index = shndx_map.get(st_shndx_raw, 0)
                symbol_section_index_str = st_shndx_raw.replace('SHN_', '') if st_shndx_raw.startswith('SHN_') else st_shndx_raw
            else:
                symbol_section_index = st_shndx_raw
                symbol_section_index_str = str(st_shndx_raw)

            # Get symbol type and map to enum
            st_type_str = symbol.entry.get('st_info', {}).get('type', 'STT_NOTYPE')
            symbol_type_enum = self._map_symbol_type(st_type_str)
            symbol_type_str = st_type_str.replace('STT_', '') if st_type_str.startswith('STT_') else st_type_str

            # Get symbol binding and map to enum
            st_bind_str = symbol.entry.get('st_info', {}).get('bind', 'STB_LOCAL')
            symbol_bind_enum = self._map_symbol_bind(st_bind_str)
            symbol_bind_str = st_bind_str.replace('STB_', '') if st_bind_str.startswith('STB_') else st_bind_str

            # Get symbol visibility and map to enum
            st_vis_str = symbol.entry.get('st_other', {}).get('visibility', 'STV_DEFAULT')
            symbol_visibility_enum = self._map_symbol_visibility(st_vis_str)
            symbol_visibility_str = st_vis_str.replace('STV_', '') if st_vis_str.startswith('STV_') else st_vis_str

            return ELFSymbol(
                symbol_name=symbol_name,
                symbol_value=symbol_value,
                symbol_size=symbol_size,
                symbol_type=symbol_type_enum,
                symbol_type_str=symbol_type_str,
                symbol_bind=symbol_bind_enum,
                symbol_bind_str=symbol_bind_str,
                symbol_visibility=symbol_visibility_enum,
                symbol_visibility_str=symbol_visibility_str,
                symbol_section_index=symbol_section_index,
                symbol_section_index_str=symbol_section_index_str,
                is_dynamic=1 if is_dynamic else 0
            )

        except Exception as e:
            self.log.error(f"Error extracting symbol data: {e}")
            return None

    def _extract_symbols_from_section(self, elf, section_name: str, is_dynamic: bool = False) -> List[Dict]:
        """Extract symbols from a specific symbol table section."""
        symbols = []

        try:
            section = elf.get_section_by_name(section_name)
            if not section:
                self.log.debug(f"No {section_name} section found")
                return symbols

            if not hasattr(section, 'iter_symbols'):
                self.log.debug(f"Section {section_name} is not a symbol table")
                return symbols

            # Iterate through symbols in the section with per-symbol error handling
            for symbol_index, symbol in enumerate(section.iter_symbols()):
                try:
                    symbol_data = self._extract_symbol_data(symbol, is_dynamic)
                    if symbol_data:
                        symbols.append(symbol_data)
                except Exception as e:
                    self.log.warning(f"Error processing symbol {symbol_index} in {section_name}: {e}")
                    # Continue with other symbols

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

        return symbols

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

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

            def extract_data(elf):
                all_symbols = []

                # Extract static symbols from .symtab with individual error handling
                try:
                    static_symbols = self._extract_symbols_from_section(elf, '.symtab', is_dynamic=False)
                    all_symbols.extend(static_symbols)
                    self.log.debug(f"Extracted {len(static_symbols)} static symbols from .symtab")
                except Exception as e:
                    self.log.warning(f"Error extracting static symbols from .symtab: {e}")

                # Extract dynamic symbols from .dynsym with individual error handling
                try:
                    dynamic_symbols = self._extract_symbols_from_section(elf, '.dynsym', is_dynamic=True)
                    all_symbols.extend(dynamic_symbols)
                    self.log.debug(f"Extracted {len(dynamic_symbols)} dynamic symbols from .dynsym")
                except Exception as e:
                    self.log.warning(f"Error extracting dynamic symbols from .dynsym: {e}")

                return all_symbols

            if not self._is_elf_file():
                return None

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

            self.elf_symbols = result
            return self.elf_symbols

        except Exception as e:
            self.log.error(f"Error extracting ELF symbols {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_symbols
        elif exporter_type == "ClickHouseExporter":
            try:
                # Return valid empty structure if no symbols (e.g., stripped binary)
                # None is reserved for actual errors

                # Prepare data arrays for all symbols
                data = []
                current_time = datetime.now(timezone.utc)
                for symbol in self.elf_symbols:
                    row = [
                        self.sha256,
                        self.md5,
                        self.sha1,
                        symbol.symbol_name,
                        symbol.symbol_value,
                        symbol.symbol_size,
                        symbol.symbol_type,
                        symbol.symbol_type_str,
                        symbol.symbol_bind,
                        symbol.symbol_bind_str,
                        symbol.symbol_visibility,
                        symbol.symbol_visibility_str,
                        symbol.symbol_section_index,
                        symbol.symbol_section_index_str,
                        symbol.is_dynamic,
                        current_time
                    ]
                    data.append(row)

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'symbol_name', 'symbol_value', 'symbol_size',
                    'symbol_type', 'symbol_type_str',
                    'symbol_bind', 'symbol_bind_str',
                    'symbol_visibility', 'symbol_visibility_str',
                    'symbol_section_index', 'symbol_section_index_str', 'is_dynamic',
                    'analysis_date'
                ]

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    'LowCardinality(String)', 'UInt64', 'UInt64',
                    "Enum8('NOTYPE'=0, 'OBJECT'=1, 'FUNC'=2, 'SECTION'=3, 'FILE'=4, 'COMMON'=5, 'TLS'=6)",
                    'LowCardinality(String)',
                    "Enum8('LOCAL'=0, 'GLOBAL'=1, 'WEAK'=2)",
                    'LowCardinality(String)',
                    "Enum8('DEFAULT'=0, 'INTERNAL'=1, 'HIDDEN'=2, 'PROTECTED'=3)",
                    'LowCardinality(String)',
                    'UInt16', 'LowCardinality(String)', '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_symbols"