Xavier Litrico

import inspect
import pefile
from dotnetfile import DotNetPE

import base64
from hashlib import md5
from pprint import pprint

import magic
import pefile

from redb.extractors.enum import Tag
from redb.extractors.pe_extractor import PEExtractor
from redb.models.dataclasses import PEDotNet
from datetime import datetime, timezone
import json
from typing import Any


class PEDotNetExtractor(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_dotnet"
        self.dotnet, self.error = self._generate_dotnetfile_object()
        self.log.debug(inspect.currentframe().f_code.co_name)

    def _decode_assembly_flags(self, flag_value):
        self.log.debug(inspect.currentframe().f_code.co_name)
        ASSEMBLY_FLAGS = {
            0x00000001: "COMIMAGE_FLAGS_ILONLY",
            0x00000002: "COMIMAGE_FLAGS_32BITREQUIRED",
            0x00000004: "COMIMAGE_FLAGS_IL_LIBRARY",
            0x00000008: "COMIMAGE_FLAGS_STRONGNAMESIGNED",
            0x00000010: "COMIMAGE_FLAGS_NATIVE_ENTRYPOINT",
            0x00010000: "COMIMAGE_FLAGS_TRACKDEBUGDATA",
            0x00020000: "COMIMAGE_FLAGS_32BITPREFERRED",
        }
        flags = []
        for bit, name in ASSEMBLY_FLAGS.items():
            if flag_value & bit:
                flags.append(name)
        return flags

    def _normalize_data(self, data):
        self.log.debug(inspect.currentframe().f_code.co_name)
        if isinstance(data, str):
            return {"value": data}
        return data

    def _compute_md5(self, data):
        self.log.debug(inspect.currentframe().f_code.co_name)
        if isinstance(data, str):
            # If it's a string, encode it to first
            return md5(data.encode("raw_unicode_escape")).hexdigest()
        elif not isinstance(data, bytes):
            # If it's neither string nor bytes, convert it to string and then to bytes
            return md5(str(data).encode("raw_unicode_escape")).hexdigest()
        else:
            return md5(data).hexdigest()

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

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        metadata = {}

        if self.error:
            metadata["Corrupted"] = self.error.args[0]
            dotnet = PEDotNet(metadata)
            self.dotnet = dotnet  # Store for later use in prepare_export_data
            return dotnet  # Return the extracted data instead of exporting directly

        metadata["info"] = {}
        try:
            available_tables = self.dotnet.existent_metadata_tables()
        except Exception as e:
            self.log.error(f"Error getting metadata tables for {self.hash.sha256}: {e}")
            available_tables = []
            metadata["info"]["PotentiallyCorrupted"] = True

        # GUID extraction
        try:
            input_string = self.dotnet.guid_stream_guids[0].string_representation
            
            formatted_uuid = (
                f"{input_string[6:8]}{input_string[4:6]}{input_string[2:4]}{input_string[0:2]}"
                f"-{input_string[10:12]}{input_string[8:10]}"
                f"-{input_string[14:16]}{input_string[12:14]}"
                f"-{input_string[16:20]}"
                f"-{input_string[20:]}"
            )
            metadata["info"]["ModuleVersionID"] = formatted_uuid
        except Exception as e:
            self.log.error(f"Error extracting GUID for {self.hash.sha256}: {e}")

        # CLR Version
        try:
            metadata["info"]["CLRversion"] = self.dotnet.get_runtime_target_version()
        except Exception as e:
            self.log.error(f"Error getting CLR version for {self.hash.sha256}: {e}")

        # Module information
        try:
            if self.dotnet.metadata_table_exists("Module"):
                metadata["info"]["AssemblyModuleName"] = self.dotnet.Module.get_module_name()
        except Exception as e:
            self.log.error(f"Error getting module info for {self.hash.sha256}: {e}")

        # Basic properties
        try:
            metadata["info"].update({
                "IsNativeNgenImage": self.dotnet.is_native_image(),
                "IsMixedAssembly": self.dotnet.is_mixed_assembly(),
                "IsWindowsFormsApp": self.dotnet.is_windows_forms_app(),
                "EntryPointToken": self.dotnet.clr_header.EntryPointToken.value,
                "HasNativeEntryPoint": self.dotnet.has_native_entry_point(),
            })
        except Exception as e:
            self.log.error(f"Error getting basic properties for {self.hash.sha256}: {e}")

        # Assembly flags
        try:
            metadata["info"]["AssemblyFlags"] = self._decode_assembly_flags(
                self.dotnet.clr_header.Flags.value
            )
        except Exception as e:
            self.log.error(f"Error getting assembly flags for {self.hash.sha256}: {e}")

        # More Metadata/Info CLR Header
        try:
            metadata["info"]["NumOfStreams"] = self.dotnet.get_number_of_streams()
            # [TODO] The following two on VT are "RVA entry point"
            #  and "Resources va" respectively. To verify.
            metadata["info"][
                "ResourcesDirectoryAddress"
            ] = self.dotnet.clr_header.ResourcesDirectoryAddress.value
            metadata["info"][
                "ResourcesDirectorySize"
            ] = self.dotnet.clr_header.ResourcesDirectorySize.value
            metadata["info"][
                "StrongNameSignatureAddress"
            ] = self.dotnet.clr_header.StrongNameSignatureAddress.value
            metadata["info"][
                "StrongNameSignatureSize"
            ] = self.dotnet.clr_header.StrongNameSignatureSize.value
            metadata["info"][
                "CodeManagerTableAddress"
            ] = self.dotnet.clr_header.CodeManagerTableAddress.value
            metadata["info"][
                "CodeManagerTableSize"
            ] = self.dotnet.clr_header.CodeManagerTableSize.value
            metadata["info"][
                "VTableFixupsAddress"
            ] = self.dotnet.clr_header.VTableFixupsAddress.value
            metadata["info"][
                "VTableFixupsSize"
            ] = self.dotnet.clr_header.VTableFixupsSize.value
            metadata["info"][
                "ExportAddressTableJumpsAddress"
            ] = self.dotnet.clr_header.ExportAddressTableJumpsAddress.value
            metadata["info"][
                "ExportAddressTableJumpsSize"
            ] = self.dotnet.clr_header.ExportAddressTableJumpsSize.value
            metadata["info"][
                "ManagedNativeHeaderAddress"
            ] = self.dotnet.clr_header.ManagedNativeHeaderAddress.value
            metadata["info"][
                "ManagedNativeHeaderSize"
            ] = self.dotnet.clr_header.ManagedNativeHeaderSize.value
        except Exception as e:
            self.log.error(f"Error getting more metadata/info clr header for {self.hash.sha256}: {e}")

        # Streams section
        try:
            metadata["Streams"] = {}
            stream_names = self.dotnet.get_stream_names()
            for name in stream_names:
                indx = stream_names.index(name)
                stream_header = self.dotnet.dotnet_stream_headers[indx]
                stream_address = self.dotnet.dotnet_streams[indx].address
                metadata["Streams"][name] = {}
                metadata["Streams"][name]["raw_name"] = base64.b64encode(
                    stream_header.Name.value
                ).decode()
                # metadata["Streams"][name]["raw_name"] = name
                metadata["Streams"][name]["size"] = stream_header.Size.value
                # Retrieve the stream data directly using the given address and size
                stream_data = self.pe.get_data(stream_address, stream_header.Size.value)
                metadata["Streams"][name]["entropy"] = (
                    "%.2f" % pefile.SectionStructure.entropy_H(self.pe, stream_data)
                )
                metadata["Streams"][name]["md5"] = md5(stream_data).hexdigest()
        except Exception as e:
            self.log.error(f"Error getting streams for {self.hash.sha256}: {e}")

        # ManifestResource
        try:
            metadata["ManifestResource"] = ()
            if "ManifestResource" in available_tables:
                metadata["info"]["HasManifestResource"] = True
                metadata["ManifestResource"] = (
                    self.dotnet.ManifestResource.get_resource_names()
                )
        except Exception as e:
            self.log.error(f"Error getting manifest resource for {self.hash.sha256}: {e}")

        # ExternalAssemblyRef
        try:
            metadata["ExternalAssemblyRef"] = {}
            if "AssemblyRef" in available_tables:
                metadata["ExternalAssemblyRef"][
                    "names"
                ] = self.dotnet.AssemblyRef.get_assemblyref_names()
                metadata["ExternalAssemblyRef"][
                    "cultures"
                ] = self.dotnet.AssemblyRef.get_assemblyref_cultures()
        except Exception as e:
            self.log.error(f"Error getting external assembly ref for {self.hash.sha256}: {e}")

        # AssemblyData
        try:
            metadata["AssemblyData"] = {}
            if "Assembly" in available_tables:
                metadata["AssemblyData"][
                    "name"
                ] = self.dotnet.Assembly.get_assembly_name()
                # It's officially a one-row table, but there are ITW files with more than one row.
                # It stores information about the current assembly. It is documented in ECMA-335.
                assembly_table = self.dotnet.metadata_tables_lookup[
                    "Assembly"
                ].table_rows[0]
                metadata["AssemblyData"][
                    "MajorVersion"
                ] = assembly_table.MajorVersion.value
                metadata["AssemblyData"][
                    "MinorVersion"
                ] = assembly_table.MinorVersion.value
                metadata["AssemblyData"]["hashalgid"] = assembly_table.HashAlgId.value
                metadata["AssemblyData"][
                    "BuildNumber"
                ] = assembly_table.BuildNumber.value
                metadata["AssemblyData"][
                    "RevisionNumber"
                ] = assembly_table.RevisionNumber.value
                metadata["AssemblyData"][
                    "culture"
                ] = self.dotnet.Assembly.get_assembly_culture()
        except Exception as e:
            self.log.error(f"Error getting assembly data for {self.hash.sha256}: {e}")

        # TypeDef handling
        try:
            metadata["TypeDef"] = {}
            if "TypeDef" in available_tables:
                metadata["info"]["HasTypeDef"] = True
                # Get basic type definitions
                try:
                    # Get all types (ANY visibility)
                    type_defs = self.dotnet.TypeDef.get_type_names()
                    metadata["TypeDef"]["names"] = type_defs if type_defs else []
                    
                    # Get detailed type information with methods
                    types_with_methods = self.dotnet.TypeDef.get_type_names_with_methods()
                    if types_with_methods:
                        metadata["TypeDef"]["detailed"] = []
                        for type_info in types_with_methods:
                            type_data = {
                                "type": type_info.Type,
                                "namespace": type_info.Namespace,
                                "flags": type_info.Flags,
                                "methods": []
                            }
                            for method in type_info.Methods:
                                method_data = {
                                    "name": method.Name,
                                    "flags": method.Flags
                                }
                                if method.Signature:
                                    method_data["signature"] = {
                                        "parameters": method.Signature.get("parameter", []),
                                        "return_type": method.Signature.get("return", ""),
                                        "has_this": method.Signature.get("hasthis", False)
                                    }
                                type_data["methods"].append(method_data)
                            metadata["TypeDef"]["detailed"].append(type_data)
                except AttributeError as e:
                    self.log.warning(f"TypeDef table exists but attributes missing for {self.hash.sha256}: {e}")
                    # metadata["info"]["TypeDefParsingError"] = f"Missing attributes: {str(e)}"
                except Exception as e:
                    self.log.error(f"Error parsing TypeDef table for {self.hash.sha256}: {e}")
                    # metadata["info"]["TypeDefParsingError"] = str(e)
        except Exception as e:
            self.log.error(f"Error handling TypeDef section for {self.hash.sha256}: {e}")
            metadata["TypeDef"] = {"names": [], "detailed": []}
            # metadata["info"]["TypeDefParsingError"] = str(e)

        # ExternalUnmanagedModules
        try:
            metadata["ExternalUnmanagedModules"] = ()
            if "ModuleRef" in available_tables:
                metadata["info"]["HasModuleRef"] = True
                metadata["ExternalUnmanagedModules"] = (
                    self.dotnet.ModuleRef.get_unmanaged_module_names(
                        self.dotnet.Type.UnmanagedModules.RAW
                    )
                )
        except Exception as e:
            self.log.error(f"Error getting external unmanaged modules for {self.hash.sha256}: {e}")

        # ExternalUnmanagedMethods
        try:
            metadata["ExternalUnmanagedMethods"] = ()
            if "ImplMap" in available_tables:
                metadata["info"]["HasImplMap"] = True
                metadata["ExternalUnmanagedMethods"] = (
                    self.dotnet.ImplMap.get_unmanaged_functions()
                )
        except Exception as e:
            self.log.error(f"Error getting external unmanaged methods for {self.hash.sha256}: {e}")

        # Event
        try:
            metadata["Event"] = ()
            if "Event" in available_tables:
                metadata["info"]["HasEvent"] = True
                metadata["Event"] = self.dotnet.Event.get_event_names()
        except Exception as e:
            self.log.error(f"Error getting event for {self.hash.sha256}: {e}")

        # Resources
        try:
            metadata["Resources"] = []
            if self.dotnet.has_resources():
                tmp_resources_list = []
                resource_data = self.dotnet.get_resources()
                for data in resource_data:
                    tmp_dict = {}
                    for resource_item in data.items():
                        if resource_item[0] == "SubResources":
                            if resource_item[1]:
                                tmp_dict["SubResources"] = {}
                                for sub_resource in resource_item[1]:
                                    for sub_resource_item in sub_resource.items():
                                        if sub_resource_item[0] == "Data":
                                            # Check if data is a sequence type that supports len()
                                            if hasattr(sub_resource_item[1], '__len__') and len(sub_resource_item[1]) > 100:
                                                tmp_dict["SubResources"]["Data"] = {}
                                                try:
                                                    # tmp_dict["SubResources"]["Data"][
                                                    #     "md5"
                                                    # ] = md5(
                                                    #     sub_resource_item[1].encode()
                                                    # ).hexdigest()
                                                    tmp_dict["SubResources"]["Data"][
                                                        "md5"
                                                    ] = self._compute_md5(
                                                        sub_resource_item[1]
                                                    )
                                                except Exception as e:
                                                    self.log.error(
                                                        f"Error in dotnet resources"
                                                        f" extraction {self.hash.sha256}: {e}"
                                                    )
                                                guess = magic.from_buffer(
                                                    sub_resource_item[1], mime=True
                                                )
                                                tmp_dict["SubResources"]["Data"][
                                                    "MimeType"
                                                ] = (guess if guess else None)
                                            else:
                                                normalized_data = self._normalize_data(
                                                    sub_resource_item[1].decode()
                                                    if isinstance(
                                                        sub_resource_item[1], bytes
                                                    )
                                                    else sub_resource_item[1]
                                                )
                                                tmp_dict["SubResources"][
                                                    sub_resource_item[0]
                                                ] = normalized_data
                                        else:
                                            tmp_dict["SubResources"][
                                                sub_resource_item[0]
                                            ] = (
                                                sub_resource_item[1].decode()
                                                if isinstance(
                                                    sub_resource_item[1], bytes
                                                )
                                                else sub_resource_item[1]
                                            )
                        elif resource_item[0] == "Data":
                            # Check if data is a sequence type that supports len()
                            if hasattr(resource_item[1], '__len__') and len(resource_item[1]) > 100:
                                tmp_dict["Data"] = {}
                                tmp_dict["Data"]["md5"] = self._compute_md5(
                                    resource_item[1]
                                )
                                guess = magic.from_buffer(resource_item[1], mime=True)
                                tmp_dict["Data"]["MimeType"] = guess if guess else None
                            else:
                                normalized_data = self._normalize_data(
                                    resource_item[1].decode()
                                    if isinstance(resource_item[1], bytes)
                                    else resource_item[1]
                                )
                                tmp_dict["Data"] = normalized_data
                        else:
                            tmp_dict[resource_item[0]] = (
                                resource_item[1].decode()
                                if isinstance(resource_item[1], bytes)
                                else resource_item[1]
                            )
                    tmp_resources_list.append(tmp_dict)
                metadata["Resources"] = tmp_resources_list
        except Exception as e:
            self.log.error(f"Error getting resources for {self.hash.sha256}: {e}")
            metadata["Resources"] = []  # Ensure we always have a valid list even on error

        dotnet = PEDotNet(metadata)
        self.dotnet = dotnet  # Store for later use in prepare_export_data
        return dotnet  # Return the extracted data instead of exporting directly


    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return self.dotnet
        elif exporter_type == "ClickHouseExporter":
            # Convert metadata to JSON string
            metadata_json = json.dumps(self.dotnet.dotnet)
            
            data = [[
                self.sha256,
                self.md5,
                self.sha1,
                metadata_json,
                datetime.now(timezone.utc)
            ]]
            
            column_names = [
                'sha256', 'md5', 'sha1', 'dotnet', 'analysis_date'
            ]
            
            if not data:
                return None

            column_type_names = [
                'String', 'String', 'String', 'JSON', 'DateTime64(3, \'UTC\')'
            ]

            return (data, column_names, column_type_names)

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