Carole Lartizien

import base64
import inspect
import json
from pprint import pprint
import re

import pefile

from redb.ext.rich_header import get_rich_idVersion
from redb.extractors.enum import Tag
from redb.extractors.pe_extractor import PEExtractor
from redb.extractors.pe_extractors.pe_dotnet import PEDotNetExtractor
from redb.models.dataclasses import PE
from datetime import datetime, timezone
from typing import Any


class PEFeaturesExtractor(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.pe_features = None
        self.elastic_index = self.index_prefix + "-pe_features"
        self.log.debug(inspect.currentframe().f_code.co_name)

    def _extract_type(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        if self.pe.is_dll():
            return "DLL"
        elif self.pe.is_driver():
            return "DRIVER"
        elif self.pe.is_exe():
            return "EXE"

    def _extract_architecture(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        mt = {"0x14c": "x86", "0x0200": "Itanium", "0x8664": "x64"}
        machine_value = self.pe.FILE_HEADER.Machine
        if isinstance(machine_value, int):
            return mt.get(str(hex(machine_value)), "")
        return str(machine_value) + " => Not x86/64 or Itanium"

    def _extract_rich_header(self):
        """Extract the Rich header from the PE file

        to decode it back from base64, use:
        rh_tmp = json.loads(rh_b64_string)
        rh_decoded = {}
        for k in rh_tmp:
            if isinstance(rh_tmp[k], str):
                rh_decoded[k] = base64.b64decode(rh_tmp[k])
            else:
                rh_decoded[k] = rh_tmp[k]
        """
        self.log.debug(inspect.currentframe().f_code.co_name)
        rh = self.pe.parse_rich_header()
        rich_header_b64 = {}
        rich_header_infos = []
        if rh:
            for key in rh:
                if isinstance(rh[key], bytes):
                    rich_header_b64[key] = base64.b64encode(rh[key]).decode("utf-8")
                else:
                    rich_header_b64[key] = rh[key]
            rich_header_infos = self._get_rich_header_infos(rh)
        return json.dumps(rich_header_b64) if rich_header_b64 else None, "\n".join(
            item for item in rich_header_infos
        ) if rich_header_infos else None

    def _get_rich_header_infos(self, rh):
        """Parse Rich header information from the PE Rich Header dump"""
        self.log.debug(inspect.currentframe().f_code.co_name)

        rich_header_infos = []
        try:
            # Get list of @Comp.IDs and counts from Rich header
            # Elements in rich_fields at even indices are @Comp.IDs
            # Elements in rich_fields at odd indices are counts
            # example:                  'values': [8681481,
            #                             1,
            #                             9795593,
            #                             1]
            rich_fields = rh.get("values", [])
            if len(rich_fields) % 2 != 0:
                self.log.info(f"rich header extraction stopped for {self.hash.sha256}")
                return None

            comp_id = None
            for i in rich_fields:
                if rich_fields.index(i) % 2 == 0:
                    # even -> save value
                    comp_id = get_rich_idVersion(i)
                else:
                    # odd -> add to list
                    if comp_id:
                        rich_header_infos.append(f"{comp_id} count={i}")
                        comp_id = None
        except Exception as e:
            self.log.error(
                f"Extract rich header error {self.hash.sha256} Exception: {e}"
            )
        return rich_header_infos

    def _extract_version_info(self):
        """Extract the VS_VERSIONINFO field in a PE file

        Returns:
        vsinfo: a list of "key:value" strings from VS_INFORMATION content.
               None if no VS_INFORMATION content is present.
        """
        self.log.debug(inspect.currentframe().f_code.co_name)
        version_info = []
        try:
            if hasattr(self.pe, "VS_VERSIONINFO") and hasattr(self.pe, "FileInfo"):
                for finfo in self.pe.FileInfo:
                    for entry in finfo:
                        if hasattr(entry, "StringTable"):
                            for st_entry in entry.StringTable:
                                for key, str_entry in list(st_entry.entries.items()):
                                    version_info.append(f"{key.decode()}:{str_entry.decode()}")
        except Exception as e:
            self.log.error(
                f"Extract VersionInfo error {self.hash.sha256}  Exception: {e}"
            )
        return version_info if version_info else None

    def _extract_exports(self):
        """
        Returns:
        export_library_name: a str representing the name of the export library.
                            An empty string if no export library is present.
        exp_symbols_list: a list of function names exported as per DIRECTORY_ENTRY_EXPORT.symbols
                            An empty list if no export library is present.
        export_timestamp: a timestamp representing the time of the export.
        """
        self.log.debug(inspect.currentframe().f_code.co_name)
        export_symbols_list = []
        export_library_name = ""
        export_library_name_raw = ""
        export_timestamp = None
        try:
            if hasattr(self.pe, "DIRECTORY_ENTRY_EXPORT"):
                export_timestamp = self.pe.DIRECTORY_ENTRY_EXPORT.struct.TimeDateStamp
                export_directory = self.pe.DIRECTORY_ENTRY_EXPORT
                export_library_name_rva = export_directory.struct.Name
                try:
                    export_library_name = self.remove_non_utf8(
                        self.pe.get_string_at_rva(export_library_name_rva)
                    ).decode()
                    export_library_name_raw = self.pe.get_string_at_rva(
                        export_library_name_rva
                    ).__str__()
                except Exception as e:
                    self.log.error(
                        f"Error while getting export library name {self.pe.get_string_at_rva(export_library_name_rva)} for file: {self.hash.sha256} {e}"
                    )
                    export_library_name = "REDB_ERROR"
                if hasattr(self.pe.DIRECTORY_ENTRY_EXPORT, "symbols"):
                    for exp in self.pe.DIRECTORY_ENTRY_EXPORT.symbols:
                        export_symbols_list.append(
                            exp.name.decode() if exp.name else None
                        )
        except Exception as e:
            self.log.error(f"Extract exports error {self.hash.sha256} Exception: {e}")

        return (
            export_library_name if export_library_name else None,
            export_library_name_raw if export_library_name_raw else None,
            export_symbols_list if export_symbols_list else None,
            export_timestamp if export_timestamp else None,
        )

    def _extract_dbg_info(self):
        """Extract debug information from PE if present

        At the moment this function parses only the DEBUG_ENTRY Structure where Type
        field == 1, i.e. IMAGE_DEBUG_TYPE_CODEVIEW, where the pdb is eventually present.

        Returns:
        debug_entry: the dump of pe.DIRECTORY_ENTRY_DEBUG.struct only for the type 1
                    "IMAGE_DEBUG_TYPE_CODEVIEW". "None" if not present.
        debug_time: a int representing the epoch timestamp as present in the
                    DIRECTORY_ENTRY_DEBUG. "None" if not present.
        pdb_info: a str with the pdb path. "None" if not present.
        """
        self.log.debug(inspect.currentframe().f_code.co_name)
        dbg_timestamp = None
        dbg_timestamp_utc = ""
        dbg_pdb_info = ""
        dbg_pdb_info_raw = ""
        dbg_struct = ""
        try:
            if hasattr(self.pe, "DIRECTORY_ENTRY_DEBUG"):
                for debug_entry in self.pe.DIRECTORY_ENTRY_DEBUG:
                    if (
                        debug_entry.struct.Type
                        == pefile.DEBUG_TYPE["IMAGE_DEBUG_TYPE_CODEVIEW"]
                    ):
                        dbg_timestamp = debug_entry.struct.TimeDateStamp
                        dbg_timestamp_utc = datetime.fromtimestamp(
                            dbg_timestamp, timezone.utc
                        ).strftime("%Y-%m-%d %H:%M:%S")
                        # dbg_struct = base64.b64encode(debug_entry.entry).decode()
                        dbg_struct = debug_entry.entry.__str__()
                        if hasattr(debug_entry.entry, "PdbFileName"):
                            dbg_pdb_info = self.remove_non_utf8(
                                debug_entry.entry.PdbFileName.rstrip(b"\x00")
                            ).decode()
                            dbg_pdb_info_raw = debug_entry.entry.PdbFileName.rstrip(
                                b"\x00"
                            ).__str__()
                            # if dbg_pdb_info:
                            #     dbg_pdb_info = dbg_pdb_info.rstrip(b"\x00")
                            #     dbg_pdb_info = dbg_pdb_info.decode()

        except Exception as e:
            self.log.error(f"Extract DBG info error {self.hash.sha256} Exception: {e}")
        return (
            dbg_timestamp if dbg_timestamp else None,
            dbg_timestamp_utc if dbg_timestamp_utc else None,
            dbg_pdb_info if dbg_pdb_info else None,
            dbg_pdb_info_raw if dbg_pdb_info_raw else None,
            dbg_struct if dbg_struct else None,
        )

    def _extract_tls_info(self):
        """Check for the presence of Thread Local Storage and related extract
        callback addresses.

        Taken from the original version of PEScanner, as the python3 porting
        "ext_pescanner" does not have it.

        Returns:
        pe.DIRECTORY_ENTRY_TLS.struct: a pefile.Structure type containing the dump
                    of the TLS structure and content. "None" is no TLS is present.
        callbacks: a list containing the TLS callbacks addresse. Empty list if no
                    address is found, "None" is no TLS is present.
        """
        self.log.debug(inspect.currentframe().f_code.co_name)
        callbacks = []
        tls_dir_struct = ""
        try:
            if (
                hasattr(self.pe, "DIRECTORY_ENTRY_TLS")
                and self.pe.DIRECTORY_ENTRY_TLS
                and self.pe.DIRECTORY_ENTRY_TLS.struct
                and self.pe.DIRECTORY_ENTRY_TLS.struct.AddressOfCallBacks
            ):
                # tls_dir_struct = base64.b64encode(self.pe.DIRECTORY_ENTRY_TLS.struct).decode()
                tls_dir_struct = self.pe.DIRECTORY_ENTRY_TLS.struct.__str__()
                callback_array_rva = (
                    self.pe.DIRECTORY_ENTRY_TLS.struct.AddressOfCallBacks
                    - self.pe.OPTIONAL_HEADER.ImageBase
                )
                # Originally it was while True
                # todo while can't be used cause risky
                # what maximum range makes sense to add here?
                for idx in range(10000):
                    func = self.pe.get_dword_from_data(
                        self.pe.get_data(callback_array_rva + 4 * idx, 4), 0
                    )
                    if func == 0:
                        break
                    callbacks.append(func)
        except Exception as e:
            self.log.error(f"Extract TLS error {self.hash.sha256} Exception: {e}")
        return callbacks if callbacks else None, tls_dir_struct if tls_dir_struct else None

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

    def extract(self):
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)
            (
                export_library_name,
                export_library_name_raw,
                export_symbols_list,
                export_timestamp,
            ) = self._extract_exports()
            (
                dbg_timestamp,
                dbg_timestamp_utc,
                dbg_pdb_info,
                dbg_pdb_info_raw,
                dbg_struct,
            ) = self._extract_dbg_info()
            tls_callbacks, tls_struct = self._extract_tls_info()
            rich_header_dump, rich_header_parsed = self._extract_rich_header()

            is_dotnet = self._check_dotnet()

            number_of_resources = 0
            number_of_imports = 0
            if hasattr(self.pe, "DIRECTORY_ENTRY_RESOURCE"):
                number_of_resources = len(self.pe.DIRECTORY_ENTRY_RESOURCE.entries)
            if hasattr(self.pe, "DIRECTORY_ENTRY_IMPORT"):
                number_of_imports = len(self.pe.DIRECTORY_ENTRY_IMPORT)

            self.pe_features = PE(
                type=self._extract_type(),
                magic=hex(self.pe.OPTIONAL_HEADER.Magic),
                entry_point=hex(self.pe.OPTIONAL_HEADER.AddressOfEntryPoint),
                major_linker_version=self.pe.OPTIONAL_HEADER.MajorLinkerVersion,
                minor_linker_version=self.pe.OPTIONAL_HEADER.MinorLinkerVersion,
                target_machine=self.pe.FILE_HEADER.Machine,
                architecture=self._extract_architecture(),
                compilation_time=self.pe.FILE_HEADER.TimeDateStamp,
                compilation_time_utc=datetime.fromtimestamp(
                    self.pe.FILE_HEADER.TimeDateStamp, timezone.utc
                ).strftime("%Y-%m-%d %H:%M:%S"),
                rich_header_dump=rich_header_dump,
                rich_header_parsed=rich_header_parsed,
                dos_header=self.pe.DOS_HEADER.__str__(),
                nt_header=self.pe.NT_HEADERS.__str__(),
                optional_header=self.pe.OPTIONAL_HEADER.__str__(),
                file_header=self.pe.FILE_HEADER.__str__(),
                version_info=self._extract_version_info(),
                export_library_name=export_library_name,
                export_library_name_raw=export_library_name_raw,
                export_symbols_list=export_symbols_list,
                export_timestamp=export_timestamp,
                is_dotnet=is_dotnet,
                dbg_timestamp=dbg_timestamp,
                dbg_timestamp_utc=dbg_timestamp_utc,
                dbg_pdb_info=dbg_pdb_info,
                dbg_pdb_info_raw=dbg_pdb_info_raw,
                dbg_struct=dbg_struct,
                tls_callbacks=tls_callbacks,
                tls_struct=tls_struct,
                is_signed=self._is_signed(),
                has_overlay=self._has_overlay(),
                number_of_sections=len(self.pe.sections),
                number_of_imports=number_of_imports,
                number_of_exports=(
                    len(export_symbols_list) if export_symbols_list else 0
                ),
                number_of_resources=number_of_resources,
            )
            return self.pe_features
        except Exception as e:
            self.log.error(f"Error extracting PE features {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.pe_features
        elif exporter_type == "ClickHouseExporter":
            try:
                pe_dump = self.pe.dump_dict()   
                # Convert PE headers directly to JSON
                dos_header_raw = json.dumps(pe_dump['DOS_HEADER'])
                nt_header_raw = json.dumps(pe_dump['NT_HEADERS'])
                optional_header_raw = json.dumps(pe_dump['OPTIONAL_HEADER'])
                file_header_raw = json.dumps(pe_dump['FILE_HEADER'])

                # Convert magic and entry_point from hex strings to integers
                magic_raw = int(self.pe_features.magic, 16) if isinstance(self.pe_features.magic, str) else self.pe_features.magic
                entry_point = int(self.pe_features.entry_point, 16) if isinstance(self.pe_features.entry_point, str) else self.pe_features.entry_point

                # Convert target_machine to string if it's an integer
                target_machine = str(self.pe_features.target_machine) if isinstance(self.pe_features.target_machine, int) else self.pe_features.target_machine

                # debug struct
                dbg_struct_raw = {}
                if self.pe_features.dbg_struct:
                    if hasattr(self.pe, "DIRECTORY_ENTRY_DEBUG"):
                        for debug_entry in self.pe.DIRECTORY_ENTRY_DEBUG:
                            if (
                                debug_entry.struct.Type
                                == pefile.DEBUG_TYPE["IMAGE_DEBUG_TYPE_CODEVIEW"]
                            ):
                                dbg_struct_raw = json.dumps(debug_entry.struct.__dict__)

                # tls struct
                tls_struct_raw = {}
                if self.pe_features.tls_struct:
                    if (
                        hasattr(self.pe, "DIRECTORY_ENTRY_TLS")
                        and self.pe.DIRECTORY_ENTRY_TLS
                        and self.pe.DIRECTORY_ENTRY_TLS.struct
                        and self.pe.DIRECTORY_ENTRY_TLS.struct.AddressOfCallBacks
                    ):
                        tls_struct_raw = json.dumps(self.pe.DIRECTORY_ENTRY_TLS.struct.__dict__)

                # Ensure arrays are properly initialized
                rich_header_parsed = []
                if self.pe_features.rich_header_parsed:
                    if isinstance(self.pe_features.rich_header_parsed, str):
                        rich_header_parsed = [x for x in self.pe_features.rich_header_parsed.split("\n") if x]
                    elif isinstance(self.pe_features.rich_header_parsed, list):
                        rich_header_parsed = self.pe_features.rich_header_parsed

                version_info = self.pe_features.version_info if self.pe_features.version_info else []
                version_info_raw = {}
                if version_info:
                    for finfo in self.pe.FileInfo:
                        for entry in finfo:
                            if hasattr(entry, "StringTable"):
                                for st_entry in entry.StringTable:
                                    for key, str_entry in list(st_entry.entries.items()):
                                        version_info_raw[key.decode()] = str_entry.decode()
                version_info_raw = json.dumps(version_info_raw) if version_info_raw else "{}"

                export_symbols_list = self.pe_features.export_symbols_list if self.pe_features.export_symbols_list else []
                tls_callbacks = self.pe_features.tls_callbacks if self.pe_features.tls_callbacks else []

                # Prepare data array
                data = [[
                    self.sha256,
                    self.md5,
                    self.sha1,
                    self.pe_features.dos_header,
                    dos_header_raw,
                    self.pe_features.nt_header,
                    nt_header_raw,
                    self.pe_features.optional_header,
                    optional_header_raw,
                    self.pe_features.file_header,
                    file_header_raw,
                    magic_raw,
                    entry_point,
                    self.pe_features.major_linker_version,
                    self.pe_features.minor_linker_version,
                    target_machine,
                    self.pe_features.architecture,
                    self.pe_features.compilation_time,
                    1 if self.pe_features.is_dotnet else 0,
                    1 if self.pe_features.is_signed else 0,
                    1 if self.pe_features.has_overlay else 0,
                    self.pe_features.number_of_sections,
                    self.pe_features.number_of_imports,
                    self.pe_features.number_of_exports,
                    self.pe_features.number_of_resources,
                    self.pe_features.type,
                    self.pe_features.dbg_struct,
                    dbg_struct_raw,
                    self.pe_features.dbg_timestamp,
                    self.pe_features.dbg_pdb_info,
                    self.pe_features.dbg_pdb_info_raw,
                    self.pe_features.tls_struct,
                    tls_struct_raw,
                    self.pe_features.export_timestamp,
                    self.pe_features.export_library_name,
                    self.pe_features.export_library_name_raw,
                    self.pe_features.rich_header_dump if self.pe_features.rich_header_dump else "{}",
                    rich_header_parsed,
                    version_info,
                    version_info_raw,
                    export_symbols_list,
                    tls_callbacks,
                    datetime.now(timezone.utc)
                ]]

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'dos_header', 'dos_header_raw', 'nt_header', 'nt_header_raw', 'optional_header', 'optional_header_raw', 'file_header', 'file_header_raw',
                    'magic_raw', 'entry_point',
                    'major_linker_version', 'minor_linker_version',
                    'target_machine', 'architecture',
                    'compilation_time',
                    'is_dotnet', 'is_signed', 'has_overlay',
                    'number_of_sections', 'number_of_imports', 'number_of_exports', 'number_of_resources',
                    'type',
                    'dbg_struct', 'dbg_struct_raw', 'dbg_timestamp', 'dbg_pdb_info', 'dbg_pdb_info_raw',
                    'tls_struct', 'tls_struct_raw', 'export_timestamp', 'export_library_name', 'export_library_name_raw',
                    'rich_header_dump', 'rich_header_parsed', 'version_info', 'version_info_raw',
                    'export_symbols_list', 'tls_callbacks',
                    'analysis_date'
                ]

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    'String', 'JSON', 'String', 'JSON', 'String', 'JSON', 'String', 'JSON',
                    'UInt16', 'UInt32',
                    'UInt8', 'UInt8',
                    'LowCardinality(String)', 'Enum8(\'x86\' = 1, \'Itanium\' = 2, \'x64\' = 3)',
                    'UInt64',
                    'UInt8', 'UInt8', 'UInt8',
                    'UInt16', 'UInt16', 'UInt16', 'UInt16',
                    'Enum8(\'DLL\' = 1, \'EXE\' = 2, \'DRIVER\' = 3)',
                    'Nullable(String)', 'JSON', 'Nullable(UInt64)', 'Nullable(String)', 'Nullable(String)',
                    'Nullable(String)', 'JSON', 'Nullable(UInt64)', 'Nullable(String)', 'Nullable(String)',
                    'JSON', 'Array(Nullable(String))', 'Array(Nullable(String))', 'JSON',
                    'Array(Nullable(String))', 'Array(Nullable(UInt64))',
                    '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_pe_features"