Hanlin Cai

import logging
from abc import ABCMeta, abstractmethod
import inspect

import magic
import pefile
from dotnetfile import DotNetPE

from redb.extractors.extractor import Extractor

logger = logging.getLogger(__name__)


@abstractmethod
class PEExtractor(Extractor, metaclass=ABCMeta):

    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
        )
        self.pe = pe if pe else self._generate_pefile_object()
        self.dotnet = None

    def _generate_pefile_object(self):
        pe = None
        try:
            pe = pefile.PE(self.filepath)
            if not pe:
                raise pefile.PEFormatError("Empty file?")
        except pefile.PEFormatError as e:
            self.log.error(f"Format error {self.hash.sha256} Full error : {e}")
        return pe

    def _generate_dotnetfile_object(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        dotnet = None
        error = None
        try:
            dotnet = DotNetPE(self.filepath)
            if not dotnet:
                raise Exception("Empty file?")
        except Exception as e:
            self.log.error(
                f"Format error dotnet file {self.hash.sha256} Full error : {e}"
            )
            error = e
        return dotnet, error

    def _check_dotnet(self):
        try:
            file_type = magic.from_buffer(self.binary)
            if ".Net" in file_type:
                return True
            for entry in self.pe.OPTIONAL_HEADER.DATA_DIRECTORY:
                # IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR is typically 14
                if (
                    entry.name == "IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR"
                    and entry.Size > 0
                ):
                    return True
            return False
        except AttributeError as e:
            self.log.error(
                f"AttributeError error dotnet file {self.hash.sha256} Full error : {e}"
            )
            return False

    def _is_signed(self):
        address = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[
            pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_SECURITY"]
        ].VirtualAddress
        if address == 0:
            return False
        return True

    def _has_overlay(self):
        return bool(self.pe.get_overlay())