Isabelle Sivignon

from dataclasses import asdict
from typing import Any
from datetime import datetime, timezone

import hashlib
import inspect
from struct import pack

from magika import Magika
from signify.fingerprinter import AuthenticodeFingerprinter
import ppdeep
import tlsh

import pefile
from elftools.elf.elffile import ELFFile
from elftools.common.exceptions import ELFError

from redb.extractors.enum import Tag
from redb.models.dataclasses import Hashes
from redb.extractors.extractor import Extractor


class HashExtractor(Extractor):

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        macho=None,
    ):
        super().__init__(
            filepath, log, exporters, index_prefix, elastic_index, known_benign, known_malicious
        )
        self.hashes = None
        self.elastic_index = self.index_prefix + "-hashes"
        self.filetype = Magika().identify_bytes(self.binary).output.label
        self.pe = None
        self.elf = None
        self.macho = macho
        if self.filetype == "pebin":  # else None
            try:
                self.pe = pefile.PE(self.filepath)
            except Exception as e:
                self.log.error(f"Failed to initialize PE file object: {e}")
                self.pe = None
        elif self.filetype == "elf":
            # ELF file will be created when needed in _extract_elf_hashes
            pass

    def _extract_hashes(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        
        # Initialize hash values with None
        md5 = None
        sha1 = None
        sha256 = None
        ssdeep_hash = None
        tlsh_hash = None
        
        # Calculate basic hashes with error handling
        try:
            md5 = hashlib.md5(self.binary).hexdigest()
        except Exception as e:
            self.log.error(f"Failed to calculate MD5 hash: {e}")
            
        try:
            sha1 = hashlib.sha1(self.binary).hexdigest()
        except Exception as e:
            self.log.error(f"Failed to calculate SHA1 hash: {e}")
            
        try:
            sha256 = hashlib.sha256(self.binary).hexdigest()
        except Exception as e:
            self.log.error(f"Failed to calculate SHA256 hash: {e}")
            
        try:
            ssdeep_hash = ppdeep.hash_from_file(self.filepath)
        except Exception as e:
            self.log.error(f"Failed to calculate ssdeep hash: {e}")
            
        try:
            tlsh_hash = tlsh.hash(self.binary)
        except Exception as e:
            self.log.error(f"Failed to calculate TLSH hash: {e}")
        
        # Create Hashes object with available values
        self.hashes = Hashes(
            md5 or "",
            sha1 or "",
            sha256 or "",
            ssdeep_hash or "",
            tlsh_hash or "",
        )

        if self.filetype == "pebin" and self.pe is not None:
            self.log.debug("Computing PEBIN related hash values")
            
            # Authentihash
            try:
                with open(self.filepath, 'rb') as f:
                    fingerprinter = AuthenticodeFingerprinter(f)
                    fingerprinter.add_authenticode_hashers(hashlib.sha256)
                    self.hashes.authentihash = fingerprinter.hash()['sha256'].hex()
            except Exception as e:
                self.log.error(f"Failed to calculate Authentihash: {e}")
                self.hashes.authentihash = None
                
            # Imphash
            try:
                self.hashes.imphash = self.pe.get_imphash()
            except Exception as e:
                self.log.error(f"Failed to calculate Imphash: {e}")
                self.hashes.imphash = None
                
            # Rich header hashes
            try:
                if self.pe.parse_rich_header():
                    self.log.debug("Computing RichHeader related hash values")
                    
                    richhash = self._compute_richhash()
                    if richhash:
                        self.hashes.richhash = richhash
                        
                    richpe_hash = self._compute_richpe_hash()
                    if richpe_hash:
                        self.hashes.richpe_hash = richpe_hash
                        
                    richpv_result = self._compute_richpv_hash()
                    if richpv_result:
                        self.hashes.richpv_hash, self.hashes.richpv_hash_sorted = richpv_result
            except Exception as e:
                self.log.error(f"Failed to calculate Rich header hashes: {e}")
        elif self.filetype == "pebin" and self.pe is None:
            self.log.warning("PE file type detected but PE object initialization failed - skipping PE-specific hashes")

        elif self.filetype == "elf":
            self.log.debug("Computing ELF related hash values")
            self._extract_elf_hashes()
        elif self.filetype == "macho":
            self.log.debug("Computing Mach-O related hash values")
            self._extract_macho_hashes()
        elif self.filetype == "apk":
            self.log.debug("Computing APK related hash values")
            self._extract_apk_hashes()
        else:
            pass
        self.log.debug(f"Hashes dump: {asdict(self.hashes)}")
        return self.hashes

    def _compute_richhash(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            rich_header = self.pe.parse_rich_header()
            if not rich_header:
                return ""
            data = rich_header["clear_data"]
            return hashlib.md5(data).hexdigest().lower()
        except:
            return ""

    def _compute_richpe_hash(self):
        """
        Computes the RichPE hash given a file path or PE object.

        RichPE hash is includes RichHeader CompID and count, as well as
        fields from IMAGE_FILE_HEADER and IMAGE_OPTIONAL_HEADER

        Parameters:
        input: it can be either a file path or a PE object

        Returns:
        richpe_hash: md5 hash of the RichPE value
        None: if no Rich Header present
        """
        try:
            # Attempt to parse Rich header
            self.log.debug(inspect.currentframe().f_code.co_name)
            rich_header = self.pe.parse_rich_header()
            if rich_header is None:
                return None

            # 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
            rich_fields = rich_header.get("values", None)
            if not rich_fields or len(rich_fields) % 2 != 0:
                return None

            md5 = hashlib.md5()

            # Update hash using @Comp.IDs and masked counts from Rich header
            while len(rich_fields):
                compid = rich_fields.pop(0)
                count = rich_fields.pop(0)
                mask = 2 ** (count.bit_length() // 2 + 1) - 1
                count |= mask
                md5.update(pack("<L", compid))
                md5.update(pack("<L", count))

            # Update hash using metadata from the PE header
            md5.update(pack("<L", self.pe.FILE_HEADER.Machine))
            md5.update(pack("<L", self.pe.FILE_HEADER.Characteristics))
            md5.update(pack("<L", self.pe.OPTIONAL_HEADER.Subsystem))
            md5.update(pack("<B", self.pe.OPTIONAL_HEADER.MajorLinkerVersion))
            md5.update(pack("<B", self.pe.OPTIONAL_HEADER.MinorLinkerVersion))
            md5.update(pack("<L", self.pe.OPTIONAL_HEADER.MajorOperatingSystemVersion))
            md5.update(pack("<L", self.pe.OPTIONAL_HEADER.MinorOperatingSystemVersion))
            md5.update(pack("<L", self.pe.OPTIONAL_HEADER.MajorImageVersion))
            md5.update(pack("<L", self.pe.OPTIONAL_HEADER.MinorImageVersion))
            md5.update(pack("<L", self.pe.OPTIONAL_HEADER.MajorSubsystemVersion))
            md5.update(pack("<L", self.pe.OPTIONAL_HEADER.MinorSubsystemVersion))

            return md5.hexdigest()
        except Exception as e:
            self.log.error(f"Failed to compute RichPE hash: {e}")
            return None

    def _compute_richpv_hash(self):
        """
        Compute the RichPV hash values, sorted and unsorted.
        RichPV excludes the most volatile Rich Header field from the MD5 input data,
        the Product Count (pC) field.

        Returns:
        richpv_hash_unsorted: md5 hash of the RichPV value unsorted
        richpv_hash_sorted: md5 hash of the RichPV value sorted
        None: if no Rich Header present
        """
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)
            # Attempt to parse Rich header
            rich_header = self.pe.parse_rich_header()
            if rich_header is None:
                return None

            # 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
            rich_fields = rich_header.get("values", None)
            if not rich_fields or len(rich_fields) % 2 != 0:
                return None

            md5 = hashlib.md5()
            md5_sorted = hashlib.md5()
            sorted_vector = []

            # Update hash using @Comp.IDs only
            for i in range(0, len(rich_fields), 2):
                compid = rich_fields[i]
                md5.update(pack("<L", compid))
                sorted_vector.append(compid)

            sorted_vector.sort()
            for compid in sorted_vector:
                md5_sorted.update(pack("<L", compid))

            return md5.hexdigest(), md5_sorted.hexdigest()
        except Exception as e:
            self.log.error(f"Failed to compute RichPV hash: {e}")
            return None

    def _extract_elf_hashes(self):
        """Extract ELF specific similarity hashes."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        try:
            with open(self.filepath, 'rb') as f:
                elf = ELFFile(f)
                if not elf:
                    return

                # Generate all similarity hashes
                import_hash = self._generate_elf_import_hash(elf)
                export_hash = self._generate_elf_export_hash(elf)
                section_hash = self._generate_elf_section_hash(elf)
                symbol_hash = self._generate_elf_symbol_hash(elf)
                dynamic_hash = self._generate_elf_dynamic_hash(elf)

                # Only set hashes if they have meaningful values
                if import_hash:
                    self.hashes.import_hash = import_hash
                if export_hash:
                    self.hashes.export_hash = export_hash
                if section_hash:
                    self.hashes.section_hash = section_hash
                if symbol_hash:
                    self.hashes.symhash = symbol_hash
                if dynamic_hash:
                    self.hashes.dynamic_hash = dynamic_hash

        except Exception as e:
            self.log.error(f"Failed to extract ELF hashes: {e}")

    def _generate_elf_import_hash(self, elf) -> str:
        """Generate MD5 hash of sorted, deduplicated imported symbol names."""
        try:
            imported_symbols = set()

            # Get dynamic symbol table
            dynsym_section = elf.get_section_by_name('.dynsym')
            if dynsym_section and hasattr(dynsym_section, 'iter_symbols'):
                for symbol in dynsym_section.iter_symbols():
                    # Look for undefined symbols (imports)
                    if (symbol.entry.get('st_shndx', 0) == 'SHN_UNDEF' and
                        symbol.name and
                        symbol.entry.get('st_info', {}).get('bind') in ['STB_GLOBAL', 'STB_WEAK']):
                        imported_symbols.add(symbol.name)

            # Also check relocations for additional imports
            for section in elf.iter_sections():
                if hasattr(section, 'iter_relocations'):
                    try:
                        for relocation in section.iter_relocations():
                            if hasattr(relocation, 'symbol') and relocation.symbol and relocation.symbol.name:
                                imported_symbols.add(relocation.symbol.name)
                    except:
                        pass

            # Sort and concatenate
            sorted_imports = sorted(list(imported_symbols))

            # Return None if no imports found
            if not sorted_imports:
                return None

            imports_string = '|'.join(sorted_imports)

            # Generate MD5 hash
            return hashlib.md5(imports_string.encode('utf-8')).hexdigest()

        except Exception as e:
            self.log.error(f"Error generating ELF import hash: {e}")
            return None

    def _generate_elf_export_hash(self, elf) -> str:
        """Generate MD5 hash of sorted, deduplicated exported symbol names."""
        try:
            exported_symbols = set()

            # Check both static and dynamic symbol tables
            symbol_sections = ['.symtab', '.dynsym']

            for section_name in symbol_sections:
                section = elf.get_section_by_name(section_name)
                if not section or not hasattr(section, 'iter_symbols'):
                    continue

                for symbol in section.iter_symbols():
                    # Check if symbol is exported (defined and globally visible)
                    if (symbol.name and
                        symbol.entry.get('st_shndx', 0) != 'SHN_UNDEF' and
                        symbol.entry.get('st_info', {}).get('bind') in ['STB_GLOBAL', 'STB_WEAK'] and
                        symbol.entry.get('st_info', {}).get('type') in ['STT_FUNC', 'STT_OBJECT']):
                        exported_symbols.add(symbol.name)

            # Sort and concatenate
            sorted_exports = sorted(list(exported_symbols))

            # Return None if no exports found
            if not sorted_exports:
                return None

            exports_string = '|'.join(sorted_exports)

            # Generate MD5 hash
            return hashlib.md5(exports_string.encode('utf-8')).hexdigest()

        except Exception as e:
            self.log.error(f"Error generating ELF export hash: {e}")
            return None

    def _generate_elf_section_hash(self, elf) -> str:
        """Generate MD5 hash of section layout (names, types, flags sequence)."""
        try:
            section_info = []

            for section in elf.iter_sections():
                header = section.header
                section_name = section.name or "<unnamed>"
                section_type = header.get('sh_type', 'SHT_NULL')
                section_flags = header.get('sh_flags', 0)

                # Create a consistent representation
                section_repr = f"{section_name}:{section_type}:{section_flags}"
                section_info.append(section_repr)

            # Return None if no meaningful sections found
            if not section_info:
                return None

            # Join all section info
            sections_string = '|'.join(section_info)

            # Generate MD5 hash
            return hashlib.md5(sections_string.encode('utf-8')).hexdigest()

        except Exception as e:
            self.log.error(f"Error generating ELF section hash: {e}")
            return None

    def _generate_elf_symbol_hash(self, elf) -> str:
        """Generate MD5 hash of sorted symbol names and types."""
        try:
            symbol_info = set()

            # Check both static and dynamic symbol tables
            symbol_sections = ['.symtab', '.dynsym']

            for section_name in symbol_sections:
                section = elf.get_section_by_name(section_name)
                if not section or not hasattr(section, 'iter_symbols'):
                    continue

                for symbol in section.iter_symbols():
                    if symbol.name:
                        symbol_type = symbol.entry.get('st_info', {}).get('type', 'STT_NOTYPE')
                        symbol_bind = symbol.entry.get('st_info', {}).get('bind', 'STB_LOCAL')

                        # Create a consistent representation
                        symbol_repr = f"{symbol.name}:{symbol_type}:{symbol_bind}"
                        symbol_info.add(symbol_repr)

            # Sort and concatenate
            sorted_symbols = sorted(list(symbol_info))

            # Return None if no symbols found
            if not sorted_symbols:
                return None

            symbols_string = '|'.join(sorted_symbols)

            # Generate MD5 hash
            return hashlib.md5(symbols_string.encode('utf-8')).hexdigest()

        except Exception as e:
            self.log.error(f"Error generating ELF symbol hash: {e}")
            return None

    def _generate_elf_dynamic_hash(self, elf) -> str:
        """Generate MD5 hash of dynamic section entries (DT_* tags)."""
        try:
            dynamic_info = []

            # Get the dynamic section
            dynamic_section = elf.get_section_by_name('.dynamic')
            if not dynamic_section:
                return None

            # Extract dynamic tags and their values
            for tag in dynamic_section.iter_tags():
                dt_tag = tag.entry.d_tag

                # Create a representation based on tag type
                if dt_tag == 'DT_NEEDED':
                    dynamic_info.append(f"DT_NEEDED:{tag.needed}")
                elif dt_tag == 'DT_SONAME':
                    dynamic_info.append(f"DT_SONAME:{tag.soname}")
                elif dt_tag == 'DT_RPATH':
                    dynamic_info.append(f"DT_RPATH:{tag.rpath}")
                elif dt_tag == 'DT_RUNPATH':
                    dynamic_info.append(f"DT_RUNPATH:{tag.runpath}")
                else:
                    # For other tags, use the tag name and value
                    dt_value = tag.entry.d_val if hasattr(tag.entry, 'd_val') else 0
                    dynamic_info.append(f"{dt_tag}:{dt_value}")

            # Sort to ensure consistent ordering
            dynamic_info.sort()
            dynamics_string = '|'.join(dynamic_info)

            # Generate MD5 hash
            return hashlib.md5(dynamics_string.encode('utf-8')).hexdigest()

        except Exception as e:
            self.log.error(f"Error generating ELF dynamic hash: {e}")
            return None

    def _extract_macho_hashes(self):
        """Extract Mach-O specific similarity hashes using machofile API.

        For FAT binaries, this extracts hashes for the current file being processed
        (either the FAT container or an individual slice). The machofile library
        handles the architecture-specific extraction when an arch parameter is provided.

        For slices: We parse the slice file directly since it's a standalone Mach-O.
        For FAT container: We use the provided macho object with combined/fat hashes.
        """
        self.log.debug(inspect.currentframe().f_code.co_name)

        try:
            # If we have a pre-parsed macho object (FAT container or single-arch with passed object)
            if self.macho:
                architectures = self.macho.get_architectures()
                is_fat = len(architectures) > 1

                if is_fat:
                    # For FAT container, get combined hashes (key may be 'fat' or 'combined')
                    all_hashes = self.macho.get_similarity_hashes()
                    if all_hashes:
                        # Try 'fat' first, then 'combined' for backwards compatibility
                        similarity_hashes = all_hashes.get('fat', all_hashes.get('combined', {}))
                    else:
                        similarity_hashes = {}
                else:
                    # Single-arch with pre-parsed object
                    similarity_hashes = self.macho.get_similarity_hashes(arch=architectures[0]) if architectures else {}
            else:
                # No pre-parsed object - parse the file (slice case)
                import machofile
                macho = machofile.UniversalMachO(self.filepath)
                macho.parse()

                architectures = macho.get_architectures()
                if architectures:
                    # For a slice, there's only one architecture
                    similarity_hashes = macho.get_similarity_hashes(arch=architectures[0]) or {}
                else:
                    similarity_hashes = {}

            # Set the hash values on the Hashes object
            if similarity_hashes:
                if similarity_hashes.get('dylib_hash'):
                    self.hashes.macho_dylib_hash = similarity_hashes['dylib_hash']
                if similarity_hashes.get('import_hash'):
                    self.hashes.macho_import_hash = similarity_hashes['import_hash']
                if similarity_hashes.get('export_hash'):
                    self.hashes.macho_export_hash = similarity_hashes['export_hash']
                if similarity_hashes.get('entitlement_hash'):
                    self.hashes.macho_entitlement_hash = similarity_hashes['entitlement_hash']
                if similarity_hashes.get('symhash'):
                    self.hashes.macho_symhash = similarity_hashes['symhash']

        except Exception as e:
            self.log.error(f"Failed to extract Mach-O hashes: {e}")

    def _extract_apk_hashes(self):
        """Extract APK specific similarity hashes (permhash)."""
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            from permhash.functions import permhash_apk
            ph = permhash_apk(self.filepath)
            if ph:
                self.hashes.permhash = ph
        except Exception as e:
            self.log.error(f"Failed to extract APK hashes: {e}")

    def extract(self):
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)
            self._extract_hashes()
            return self.hashes
        except Exception as e:
            self.log.error(f"Error extracting hashes: {e}")
            return None

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return self.hashes
        elif exporter_type == "ClickHouseExporter":
            # Safely get hash values with fallbacks for None values
            sha256 = getattr(self.hashes, 'sha256', None) or ""
            md5 = getattr(self.hashes, 'md5', None) or ""
            sha1 = getattr(self.hashes, 'sha1', None) or ""
            ssdeep_hash = getattr(self.hashes, 'ssdeep_hash', None) or ""
            tlsh_hash = getattr(self.hashes, 'tlsh_hash', None) or ""
            authentihash = getattr(self.hashes, 'authentihash', None)
            imphash = getattr(self.hashes, 'imphash', None)
            impfuzzy = getattr(self.hashes, 'impfuzzy', None)
            typerefhash = getattr(self.hashes, 'typerefhash', None)
            richhash = getattr(self.hashes, 'richhash', None)
            richpe_hash = getattr(self.hashes, 'richpe_hash', None)
            richpv_hash = getattr(self.hashes, 'richpv_hash', None)
            richpv_hash_sorted = getattr(self.hashes, 'richpv_hash_sorted', None)
            # ELF hashes
            import_hash = getattr(self.hashes, 'import_hash', None)
            export_hash = getattr(self.hashes, 'export_hash', None)
            section_hash = getattr(self.hashes, 'section_hash', None)
            symbol_hash = getattr(self.hashes, 'symhash', None)
            dynamic_hash = getattr(self.hashes, 'dynamic_hash', None)
            # Mach-O hashes
            macho_dylib_hash = getattr(self.hashes, 'macho_dylib_hash', None)
            macho_import_hash = getattr(self.hashes, 'macho_import_hash', None)
            macho_export_hash = getattr(self.hashes, 'macho_export_hash', None)
            macho_entitlement_hash = getattr(self.hashes, 'macho_entitlement_hash', None)
            macho_symhash = getattr(self.hashes, 'macho_symhash', None)
            # APK hashes
            permhash = getattr(self.hashes, 'permhash', None)

            data = [[
                sha256,
                md5,
                sha1,
                ssdeep_hash,
                tlsh_hash,
                authentihash,
                imphash,
                impfuzzy,
                typerefhash,
                richhash,
                richpe_hash,
                richpv_hash,
                richpv_hash_sorted,
                import_hash,
                export_hash,
                section_hash,
                symbol_hash,
                dynamic_hash,
                macho_dylib_hash,
                macho_import_hash,
                macho_export_hash,
                macho_entitlement_hash,
                macho_symhash,
                permhash,
                datetime.now(timezone.utc)
            ]]

            column_names = [
                'sha256', 'md5', 'sha1', 'ssdeep_hash', 'tlsh_hash',
                'authentihash', 'imphash', 'impfuzzy', 'typerefhash',
                'richhash', 'richpe_hash', 'richpv_hash', 'richpv_hash_sorted',
                'import_hash', 'export_hash', 'section_hash', 'symbol_hash', 'dynamic_hash',
                'macho_dylib_hash', 'macho_import_hash', 'macho_export_hash',
                'macho_entitlement_hash', 'macho_symhash',
                'permhash',
                'analysis_date'
            ]

            column_type_names = [
                # sha256, md5, sha1
                'String', 'String', 'String',
                # ssdeep_hash, tlsh_hash
                'Nullable(String)', 'Nullable(String)',
                # authentihash, imphash, impfuzzy, typerefhash
                'Nullable(String)', 'Nullable(String)',
                'Nullable(String)', 'Nullable(String)',
                # richhash, richpe_hash, richpv_hash, richpv_hash_sorted
                'Nullable(String)', 'Nullable(String)',
                'Nullable(String)', 'Nullable(String)',
                # import_hash, export_hash, section_hash, symbol_hash, dynamic_hash
                'Nullable(FixedString(32))', 'Nullable(FixedString(32))',
                'Nullable(FixedString(32))', 'Nullable(FixedString(32))',
                'Nullable(FixedString(32))',
                # macho_dylib_hash, macho_import_hash, macho_export_hash, macho_entitlement_hash, macho_symhash
                'Nullable(FixedString(32))', 'Nullable(FixedString(32))',
                'Nullable(FixedString(32))', 'Nullable(FixedString(32))',
                'Nullable(FixedString(32))',
                # permhash (APK)
                'Nullable(FixedString(64))',
                # analysis_date
                'DateTime64(3, \'UTC\')'
            ]

            return (data, column_names, column_type_names)

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

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