Walther von Hahn

import inspect
from redb.ext.spoof_check import (
    Result,
    checksum_test,
    duplicate_test,
    import_count_test,
    linker_test,
)
from redb.extractors.enum import Tag
from redb.extractors.pe_extractor import PEExtractor
from redb.models.dataclasses import (
    DotNetInconsistencyTests,
    PEInconsistencyTests,
)
from datetime import datetime, timezone
from typing import Any


class PEInconstistencyTestsExtractor(PEExtractor):
    """Collection of functions to perform features inconsistency tests

    A Test where the result is True means that there is an inconsistency.
    At the moments it runs a series of inconsistency tests on PE metadata from
    - spoof_check
    - pescanner
    - dotnetfile
    """

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        pe=None,
        dotnet=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            elastic_index,
            known_benign,
            known_malicious,
            pe,
        )
        self.dotnet = dotnet if dotnet else None
        self.pe_inconsistency_tests = None
        self.dotnet_inconsistency_tests = None
        self.elastic_index = self.index_prefix + "-pe_inconsistency_tests"

    def tag(self):
        return [Tag.PE_INCONSISTENCY_TESTS.value, Tag.DOTNET_INCONSISTENCY_TESTS.value]

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        tests_performed = False

        # Handle PE rich header tests
        try:
            rich_header = self.pe.parse_rich_header()
            if rich_header:
                self.pe_inconsistency_tests = PEInconsistencyTests(
                    test_rich_header_checksum=checksum_test(self.pe, rich_header) == Result.INVALID,
                    test_rich_header_duplicate=duplicate_test(self.pe, rich_header) == Result.INVALID,
                    test_rich_header_linker=linker_test(self.pe, rich_header) == Result.INVALID,
                    test_rich_header_import_count=import_count_test(self.pe, rich_header) == Result.INVALID,
                )
                tests_performed = True
            else:
                self.pe_inconsistency_tests = PEInconsistencyTests(
                    test_rich_header_checksum=None,
                    test_rich_header_duplicate=None,
                    test_rich_header_linker=None,
                    test_rich_header_import_count=None,
                )
        except Exception as e:
            self.log.error(f"Error processing rich header tests for {self.hash.sha256}: {e}")
            self.pe_inconsistency_tests = None

            # self.export_to_elastic([self.pe_inconsistency_tests])

        # Handle .NET tests
        try:
            if self._check_dotnet():
                if not self.dotnet:
                    self.dotnet, self.error = self._generate_dotnetfile_object()
                if self.error:
                    self.log.error(f"Error generating .NET object {self.hash.sha256}: {self.error}")
                self.dotnet_inconsistency_tests = DotNetInconsistencyTests(
                    test_dotnet_data_dir_hidden=self.dotnet.AntiMetadataAnalysis.is_dotnet_data_directory_hidden,
                    test_dotnet_extra_data=self.dotnet.AntiMetadataAnalysis.has_metadata_table_extra_data,
                    test_dotnet_fake_types=self.dotnet.AntiMetadataAnalysis.has_self_referenced_typeref_entries,
                    test_dotnet_invalid_type_ref=self.dotnet.AntiMetadataAnalysis.has_invalid_typeref_entries,
                    test_dotnet_fake_datastreams=self.dotnet.AntiMetadataAnalysis.has_fake_data_streams,
                    test_dotnet_extra_module_table=self.dotnet.AntiMetadataAnalysis.module_table_has_multiple_rows,
                    test_dotnet_extra_assembly_table=self.dotnet.AntiMetadataAnalysis.assembly_table_has_multiple_rows,
                    test_dotnet_invalid_strings_stream=self.dotnet.AntiMetadataAnalysis.has_invalid_strings_stream_entries,
                    test_dotnet_streams_mixed_case=self.dotnet.AntiMetadataAnalysis.has_mixed_case_stream_names,
                    test_dotnet_method_def_invalid_table=self.dotnet.AntiMetadataAnalysis.has_invalid_methoddef_entries,
                    test_dotnet_max_len_exceeding_strings=self.dotnet.AntiMetadataAnalysis.has_max_len_exceeding_strings,
                )
                tests_performed = True
        except Exception as e:
            self.log.error(f"Error processing .NET tests for {self.hash.sha256}: {e}")
            self.dotnet_inconsistency_tests = None

        # self.export_to_elastic([self.dotnet_inconsistency_tests])

        # If no tests were performed, return False to skip database insertion
        if not tests_performed:
            self.log.info("No inconsistency tests were performed.")
            return False

        return True

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return [self.pe_inconsistency_tests, self.dotnet_inconsistency_tests]
        elif exporter_type == "ClickHouseExporter":
            current_time = datetime.now(timezone.utc)

            # For PE tests: if no rich header (all True), store NULL instead
            has_rich_header = any([
                hasattr(self.pe_inconsistency_tests, 'test_rich_header_checksum'),
                hasattr(self.pe_inconsistency_tests, 'test_rich_header_duplicate'),
                hasattr(self.pe_inconsistency_tests, 'test_rich_header_linker'),
                hasattr(self.pe_inconsistency_tests, 'test_rich_header_import_count')
            ])
            
            pe_tests = [
                None if not has_rich_header else self.pe_inconsistency_tests.test_rich_header_checksum,
                None if not has_rich_header else self.pe_inconsistency_tests.test_rich_header_duplicate,
                None if not has_rich_header else self.pe_inconsistency_tests.test_rich_header_linker,
                None if not has_rich_header else self.pe_inconsistency_tests.test_rich_header_import_count,
            ]
            
            # For .NET tests: if not a .NET file, store NULL instead of False
            dotnet_tests = [
                self.dotnet_inconsistency_tests.test_dotnet_data_dir_hidden if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_extra_data if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_fake_types if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_invalid_type_ref if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_fake_datastreams if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_extra_module_table if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_extra_assembly_table if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_invalid_strings_stream if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_streams_mixed_case if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_method_def_invalid_table if self.dotnet_inconsistency_tests else None,
                self.dotnet_inconsistency_tests.test_dotnet_max_len_exceeding_strings if self.dotnet_inconsistency_tests else None,
            ]
            
            data = [[
                self.sha256,
                self.md5,
                self.sha1,
                *pe_tests,
                *dotnet_tests,
                current_time
            ]]

            column_names = [
                'sha256', 'md5', 'sha1',
                'test_rich_header_checksum', 'test_rich_header_duplicate', 'test_rich_header_linker', 'test_rich_header_import_count',
                'test_dotnet_data_dir_hidden', 'test_dotnet_extra_data',
                'test_dotnet_fake_types', 'test_dotnet_invalid_type_ref',
                'test_dotnet_fake_datastreams', 'test_dotnet_extra_module_table',
                'test_dotnet_extra_assembly_table', 'test_dotnet_invalid_strings_stream',
                'test_dotnet_streams_mixed_case', 'test_dotnet_method_def_invalid_table',
                'test_dotnet_max_len_exceeding_strings', 'analysis_date'
            ]
            
            column_type_names = [
                'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                'Nullable(Boolean)', 'Nullable(Boolean)', 'Nullable(Boolean)', 'Nullable(Boolean)',
                'Nullable(Boolean)', 'Nullable(Boolean)', 'Nullable(Boolean)', 'Nullable(Boolean)',
                'Nullable(Boolean)', 'Nullable(Boolean)', 'Nullable(Boolean)', 'Nullable(Boolean)',
                'Nullable(Boolean)', 'Nullable(Boolean)', 'Nullable(Boolean)',
                'DateTime64(3, \'UTC\')'
            ]

            if not data:
                return None

            return (data, column_names, column_type_names)

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