Van Cuong Tran

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"