Mana Saleh Al Reshan

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 ELFDependency


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

    def _map_dependency_type(self, dt_tag_str: str) -> int:
        """Map dynamic tag string to enum value."""
        type_map = {
            'DT_NEEDED': 1,
            'DT_SONAME': 14,
            'DT_RPATH': 15,
            'DT_RUNPATH': 29
        }
        return type_map.get(dt_tag_str, 0)

    def _extract_dynamic_dependencies(self, elf) -> List[Dict]:
        """Extract dependencies from the dynamic section."""
        dependencies = []

        try:
            # Get the dynamic section
            dynamic_section = elf.get_section_by_name('.dynamic')
            if not dynamic_section:
                self.log.debug("No .dynamic section found")
                return dependencies

            # Iterate through dynamic tags
            for tag in dynamic_section.iter_tags():
                dt_tag = tag.entry.d_tag

                # Handle different dependency types
                if dt_tag == 'DT_NEEDED':
                    # Required library
                    dependency_name = tag.needed
                    dependencies.append(ELFDependency(
                        dependency_name=dependency_name,
                        dependency_type=self._map_dependency_type('DT_NEEDED'),
                        dependency_type_str='NEEDED'
                    ))

                elif dt_tag == 'DT_SONAME':
                    # Shared object name
                    dependency_name = tag.soname
                    dependencies.append(ELFDependency(
                        dependency_name=dependency_name,
                        dependency_type=self._map_dependency_type('DT_SONAME'),
                        dependency_type_str='SONAME'
                    ))

                elif dt_tag == 'DT_RPATH':
                    # Runtime library search path
                    dependency_name = tag.rpath
                    dependencies.append(ELFDependency(
                        dependency_name=dependency_name,
                        dependency_type=self._map_dependency_type('DT_RPATH'),
                        dependency_type_str='RPATH'
                    ))

                elif dt_tag == 'DT_RUNPATH':
                    # Runtime library search path (newer)
                    dependency_name = tag.runpath
                    dependencies.append(ELFDependency(
                        dependency_name=dependency_name,
                        dependency_type=self._map_dependency_type('DT_RUNPATH'),
                        dependency_type_str='RUNPATH'
                    ))

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

        return dependencies

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

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

            def extract_data(elf):
                # Extract dependencies - returns list of ELFDependency dataclasses
                dependencies = self._extract_dynamic_dependencies(elf)
                return dependencies

            if not self._is_elf_file():
                return None

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

            self.elf_dependencies = result
            return self.elf_dependencies

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

                # Prepare data arrays for all dependencies
                data = []
                current_time = datetime.now(timezone.utc)
                for dep in self.elf_dependencies:
                    row = [
                        self.sha256,
                        self.md5,
                        self.sha1,
                        dep.dependency_name,
                        dep.dependency_type,
                        dep.dependency_type_str,
                        current_time
                    ]
                    data.append(row)

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'dependency_name', 'dependency_type', 'dependency_type_str',
                    'analysis_date'
                ]

                if not data:
                    return None

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    'LowCardinality(String)',
                    "Enum8('NEEDED'=1, 'SONAME'=14, 'RPATH'=15, 'RUNPATH'=29)",
                    '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_dependencies"