Rainer Poisel

import inspect
from typing import Any, List, Tuple
from datetime import datetime, timezone

from redb.extractors.enum import Tag
from redb.extractors.pe_extractor import PEExtractor
from redb.models.dataclasses import PEImport


class PEImportExtractor(PEExtractor):

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        pe=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            elastic_index,
            known_benign,
            known_malicious,
            pe,
        )
        self.elastic_index = self.index_prefix + "-pe_imports"
        self.log.debug(inspect.currentframe().f_code.co_name)

    def tag(self):
        return Tag.PE_IMPORT.value

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

        imports_symbols = []
        imports_lib = []
        imports_total = 0
        # pe_import = None
        try:
            directory_entry_import = getattr(self.pe, "DIRECTORY_ENTRY_IMPORT", [])
            imports_total = len(directory_entry_import)
            for entry in directory_entry_import:
                symbols = []
                tmp_import = {}
                libname = entry.dll.decode() if entry.dll else ""
                imports_lib.append(libname)
                # replace . with _ to avoid issues with elastic
                # entryname = entryname.replace(".", "_")
                for symbol in entry.imports:
                    if symbol.name:
                        symbols.append(symbol.name.decode())
                # imports_symbols[entryname] = symbols
                tmp_import[libname] = symbols
                imports_symbols.append(tmp_import)
            return PEImport(
                pe_imports_total=imports_total,
                pe_import_libraryName=imports_lib if imports_lib else None,
                pe_import_functions=imports_symbols if imports_symbols else None,
            )
        except Exception as e:
            self.log.error(f"Extract imports error {self.hash.sha256} Exception: {e}")
        return pe_import

    def extract(self):
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)
            return self._extract_imports()
        except Exception as e:
            self.log.error(f"Extract imports error {self.hash.sha256} Exception: {e}")
            return None

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return self.extract()
        elif exporter_type == "ClickHouseExporter":
            imports = self.extract()
            if (
                imports is None
                or imports.pe_imports_total == 0
                or (
                    imports.pe_import_libraryName is None
                    and imports.pe_import_functions is None
                )
            ):
                return None

            # Flatten the data - one row per function import
            data = []
            current_time = datetime.now(timezone.utc)

            for lib_funcs in imports.pe_import_functions:
                for lib, funcs in lib_funcs.items():
                    for func in funcs:
                        data.append(
                            [
                                self.sha256,  # sha256
                                self.md5,  # md5
                                self.sha1,  # sha1
                                lib,  # library_name
                                func,  # function_name
                                current_time,  # analysis_date
                            ]
                        )

            column_names = [
                "sha256",
                "md5",
                "sha1",
                "library_name",
                "function_name",
                "analysis_date",
            ]

            column_type_names = [
                "FixedString(64)",
                "FixedString(32)",
                "FixedString(40)",
                "LowCardinality(Nullable(String))",
                "LowCardinality(Nullable(String))",
                "DateTime64(3, 'UTC')",
            ]

            if not data:
                return None

            return (data, column_names, column_type_names)

    def get_clickhouse_table(self) -> str:
        return "redb_pe_imports"