Weibin Gu

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"