Hancheng Yu

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 ELFRelocation


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

    def _get_relocation_type_string(self, reloc_type: int, machine_arch: str) -> str:
        """Convert relocation type number to human-readable string based on architecture."""
        # This is a simplified mapping - real implementation would need comprehensive
        # architecture-specific relocation type mappings

        common_types = {
            0: "R_NONE",
            1: "R_DIRECT",
            2: "R_PC_RELATIVE",
            3: "R_GOT",
            4: "R_PLT",
            5: "R_COPY",
            6: "R_GLOB_DAT",
            7: "R_JMP_SLOT",
            8: "R_RELATIVE"
        }

        # Architecture-specific mappings could be added here
        if machine_arch == "x86_64":
            x86_64_types = {
                1: "R_X86_64_64",
                2: "R_X86_64_PC32",
                3: "R_X86_64_GOT32",
                4: "R_X86_64_PLT32",
                5: "R_X86_64_COPY",
                6: "R_X86_64_GLOB_DAT",
                7: "R_X86_64_JUMP_SLOT",
                8: "R_X86_64_RELATIVE"
            }
            return x86_64_types.get(reloc_type, f"R_X86_64_{reloc_type}")
        elif machine_arch == "x86":
            i386_types = {
                1: "R_386_32",
                2: "R_386_PC32",
                3: "R_386_GOT32",
                4: "R_386_PLT32",
                5: "R_386_COPY",
                6: "R_386_GLOB_DAT",
                7: "R_386_JMP_SLOT",
                8: "R_386_RELATIVE"
            }
            return i386_types.get(reloc_type, f"R_386_{reloc_type}")

        return common_types.get(reloc_type, f"R_UNKNOWN_{reloc_type}")

    def _extract_relocation_data(self, relocation, section_name: str, machine_arch: str) -> Dict:
        """Extract data from a single relocation entry."""
        try:
            # Get relocation offset
            relocation_offset = relocation.entry.get('r_offset', 0)

            # Get relocation type
            relocation_type = relocation.entry.get('r_info_type', 0)

            # Get symbol index
            relocation_symbol_index = relocation.entry.get('r_info_sym', 0)

            # Get addend (only present in RELA sections)
            relocation_addend = None
            if hasattr(relocation.entry, 'r_addend'):
                relocation_addend = relocation.entry.get('r_addend', 0)

            # Get symbol name if available
            relocation_symbol_name = ""
            if hasattr(relocation, 'symbol') and relocation.symbol:
                relocation_symbol_name = relocation.symbol.name or f"<symbol_{relocation_symbol_index}>"
            else:
                relocation_symbol_name = f"<symbol_{relocation_symbol_index}>"

            # Get type string mapping
            relocation_type_str = self._get_relocation_type_string(relocation_type, machine_arch)

            return ELFRelocation(
                relocation_offset=relocation_offset,
                relocation_type=relocation_type,
                relocation_type_str=relocation_type_str,
                relocation_symbol_index=relocation_symbol_index,
                relocation_symbol_name=relocation_symbol_name,
                relocation_section=section_name,
                relocation_addend=relocation_addend
            )

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

    def _extract_relocations_from_section(self, section, machine_arch: str) -> List[Dict]:
        """Extract all relocations from a relocation section."""
        relocations = []

        try:
            if not hasattr(section, 'iter_relocations'):
                return relocations

            section_name = section.name or f"<unnamed_section>"

            for relocation in section.iter_relocations():
                reloc_data = self._extract_relocation_data(relocation, section_name, machine_arch)
                if reloc_data:
                    relocations.append(reloc_data)

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

        return relocations

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

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

            def extract_data(elf):
                # Get architecture for relocation type mapping
                machine_arch = self._get_architecture()
                all_relocations = []

                # Iterate through all sections looking for relocation sections with per-section error handling
                for section_index, section in enumerate(elf.iter_sections()):
                    try:
                        # Check if this is a relocation section (.rel or .rela)
                        if (section.name and
                            (section.name.startswith('.rel') or section.name.startswith('.rela')) and
                            hasattr(section, 'iter_relocations')):

                            section_relocations = self._extract_relocations_from_section(section, machine_arch)
                            all_relocations.extend(section_relocations)
                            self.log.debug(f"Extracted {len(section_relocations)} relocations from section {section.name}")
                    except Exception as e:
                        section_name = getattr(section, 'name', f'section_{section_index}')
                        self.log.warning(f"Error processing relocation section {section_name}: {e}")
                        # Continue processing other sections

                return all_relocations

            if not self._is_elf_file():
                return None

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

            self.elf_relocations = result
            return self.elf_relocations

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

                # Prepare data arrays for all relocations
                data = []
                current_time = datetime.now(timezone.utc)
                for reloc in self.elf_relocations:
                    row = [
                        self.sha256,
                        self.md5,
                        self.sha1,
                        reloc.relocation_offset,
                        reloc.relocation_type,
                        reloc.relocation_type_str,
                        reloc.relocation_symbol_index,
                        reloc.relocation_symbol_name,
                        reloc.relocation_addend,
                        reloc.relocation_section,
                        current_time
                    ]
                    data.append(row)

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'relocation_offset', 'relocation_type', 'relocation_type_str',
                    'relocation_symbol_index', 'relocation_symbol_name',
                    'relocation_addend', 'relocation_section',
                    'analysis_date'
                ]

                if not data:
                    return None

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    'UInt64', 'UInt32', 'LowCardinality(String)',
                    'UInt32', 'LowCardinality(String)',
                    'Nullable(Int64)', '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_relocations"