Hangli Ge

import inspect
import hashlib
import math
from collections import Counter
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 ELFSegment


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

    def _is_empty_result(self, extracted_data) -> bool:
        """
        Override: Empty segments is an ERROR, not a valid empty case.
        A valid ELF file must have segments (program headers).
        """
        # Always return False - empty segments should be treated as an error
        return False

    def _calculate_entropy(self, data: bytes) -> float:
        """Calculate Shannon entropy of data."""
        if not data:
            return 0.0

        try:
            # Count frequency of each byte
            byte_counts = Counter(data)
            data_len = len(data)

            # Calculate entropy
            entropy = 0.0
            for count in byte_counts.values():
                if count > 0:
                    frequency = count / data_len
                    entropy -= frequency * math.log2(frequency)

            return entropy
        except Exception as e:
            self.log.error(f"Error calculating entropy: {e}")
            return 0.0

    def _map_segment_type(self, p_type_str: str) -> int:
        """Map segment type string to enum value."""
        type_map = {
            'PT_NULL': 0,
            'PT_LOAD': 1,
            'PT_DYNAMIC': 2,
            'PT_INTERP': 3,
            'PT_NOTE': 4,
            'PT_SHLIB': 5,
            'PT_PHDR': 6,
            'PT_TLS': 7
        }
        return type_map.get(p_type_str, 0)

    def _decode_segment_flags(self, flags: int) -> List[str]:
        """Decode segment flags to human-readable strings."""
        flag_strings = []

        if flags & 0x1:  # PF_X
            flag_strings.append('EXECUTE')
        if flags & 0x2:  # PF_W
            flag_strings.append('WRITE')
        if flags & 0x4:  # PF_R
            flag_strings.append('READ')

        return flag_strings if flag_strings else ['NONE']

    def _extract_segment_data(self, segment) -> ELFSegment:
        """Extract data from a single segment with granular error handling."""
        # Initialize with safe defaults
        segment_type = 0
        segment_type_str = 'unknown'
        segment_flags = 0
        segment_flags_str = ['NONE']
        segment_offset = 0
        segment_vaddr = 0
        segment_paddr = 0
        segment_filesz = 0
        segment_memsz = 0
        segment_align = 0
        segment_entropy = 0.0
        segment_sha256 = ""
        segment_md5 = ""

        # Try to get segment header
        header = None
        try:
            header = segment.header
        except Exception as e:
            self.log.warning(f"Could not access segment header: {e}")
            return ELFSegment(
                segment_type=segment_type, segment_type_str=segment_type_str,
                segment_flags=segment_flags, segment_flags_str=segment_flags_str,
                segment_offset=segment_offset, segment_vaddr=segment_vaddr,
                segment_paddr=segment_paddr, segment_filesz=segment_filesz,
                segment_memsz=segment_memsz, segment_align=segment_align,
                segment_entropy=segment_entropy, segment_sha256=segment_sha256,
                segment_md5=segment_md5
            )

        # Extract segment type
        try:
            p_type_str = header.get('p_type', 'PT_NULL')
            segment_type = self._map_segment_type(p_type_str)
            segment_type_str = p_type_str.replace('PT_', '') if p_type_str.startswith('PT_') else p_type_str
        except Exception as e:
            self.log.warning(f"Could not extract segment type: {e}")

        # Extract segment flags
        try:
            segment_flags = header.get('p_flags', 0)
            segment_flags_str = self._decode_segment_flags(segment_flags)
        except Exception as e:
            self.log.warning(f"Could not extract segment flags: {e}")

        # Extract segment addresses and sizes
        try:
            segment_offset = header.get('p_offset', 0)
        except Exception as e:
            self.log.warning(f"Could not extract segment offset: {e}")

        try:
            segment_vaddr = header.get('p_vaddr', 0)
        except Exception as e:
            self.log.warning(f"Could not extract segment vaddr: {e}")

        try:
            segment_paddr = header.get('p_paddr', 0)
        except Exception as e:
            self.log.warning(f"Could not extract segment paddr: {e}")

        try:
            segment_filesz = header.get('p_filesz', 0)
        except Exception as e:
            self.log.warning(f"Could not extract segment filesz: {e}")

        try:
            segment_memsz = header.get('p_memsz', 0)
        except Exception as e:
            self.log.warning(f"Could not extract segment memsz: {e}")

        try:
            segment_align = header.get('p_align', 0)
        except Exception as e:
            self.log.warning(f"Could not extract segment align: {e}")

        # Calculate entropy and hashes for segment data (most likely to fail)
        try:
            if segment_filesz > 0:
                segment_data = segment.data()
                if segment_data:
                    # Calculate entropy
                    segment_entropy = self._calculate_entropy(segment_data)

                    # Calculate hashes
                    segment_sha256 = hashlib.sha256(segment_data).hexdigest()
                    segment_md5 = hashlib.md5(segment_data).hexdigest()
        except Exception as e:
            self.log.warning(f"Could not read segment data for hashing: {e}")
            # Keep defaults (0.0, "", "")

        return ELFSegment(
            segment_type=segment_type,
            segment_type_str=segment_type_str,
            segment_flags=segment_flags,
            segment_flags_str=segment_flags_str,
            segment_offset=segment_offset,
            segment_vaddr=segment_vaddr,
            segment_paddr=segment_paddr,
            segment_filesz=segment_filesz,
            segment_memsz=segment_memsz,
            segment_align=segment_align,
            segment_entropy=segment_entropy,
            segment_sha256=segment_sha256,
            segment_md5=segment_md5
        )

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

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

            def extract_segments(elf):
                segments_data = []

                # Iterate through all segments with per-segment error handling
                for segment_index, segment in enumerate(elf.iter_segments()):
                    try:
                        segment_data = self._extract_segment_data(segment)
                        if segment_data:
                            segments_data.append(segment_data)
                        else:
                            self.log.warning(f"Failed to extract data for segment {segment_index}")
                    except Exception as e:
                        self.log.warning(f"Error processing segment {segment_index}: {e}")
                        # Continue processing other segments

                return segments_data

            # Check if file is valid ELF
            if not self._is_elf_file():
                self.log.error(f"No valid ELF file for {self.hash.sha256}")
                return None

            segments_data = self._with_elf_file(extract_segments)
            if segments_data is None:
                return None

            self.elf_segments = segments_data
            return self.elf_segments

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

                # Prepare data arrays for all segments
                data = []
                current_time = datetime.now(timezone.utc)
                for segment in self.elf_segments:
                    row = [
                        self.sha256,
                        self.md5,
                        self.sha1,
                        segment.segment_type,
                        segment.segment_type_str,
                        segment.segment_flags,
                        segment.segment_flags_str,
                        segment.segment_offset,
                        segment.segment_vaddr,
                        segment.segment_paddr,
                        segment.segment_filesz,
                        segment.segment_memsz,
                        segment.segment_align,
                        segment.segment_entropy,
                        segment.segment_sha256,
                        segment.segment_md5,
                        current_time
                    ]
                    data.append(row)

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'segment_type', 'segment_type_str', 'segment_flags', 'segment_flags_str',
                    'segment_offset', 'segment_vaddr', 'segment_paddr',
                    'segment_filesz', 'segment_memsz', 'segment_align',
                    'segment_entropy', 'segment_sha256', 'segment_md5',
                    'analysis_date'
                ]

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    "Enum8('NULL'=0, 'LOAD'=1, 'DYNAMIC'=2, 'INTERP'=3, 'NOTE'=4, 'SHLIB'=5, 'PHDR'=6, 'TLS'=7)",
                    'LowCardinality(String)',
                    'UInt32',
                    'Array(LowCardinality(String))',
                    'UInt64', 'UInt64', 'UInt64', 'UInt64', 'UInt64', 'UInt64',
                    'Float64',
                    'FixedString(64)', 'FixedString(32)',
                    '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_segments"