Hamid Reza Vaezi Joze

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"