Hamza Zidoum

import inspect
import json
from datetime import datetime, timezone
from typing import Any

from redb.extractors.enum import Tag
from redb.extractors.macho_extractor import MachOExtractor
from redb.models.dataclasses import MachO


class MachOFeaturesExtractor(MachOExtractor):

    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,
            macho,
        )
        self.elastic_index = self.index_prefix + "-macho_features"
        self.log.debug(inspect.currentframe().f_code.co_name)

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

    def _extract_macho_features_for_arch(self, arch_name):
        """Extract basic MachO features for a specific architecture."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        if not self.macho:
            return None

        try:
            # Get general info using new API
            general_info = self.macho.get_general_info()
            if not general_info:
                return None

            # Get header info for this specific architecture
            header_raw = self.macho.get_macho_header(arch=arch_name)
            header_formatted = self.macho.get_macho_header(arch=arch_name, formatted=True)

            if not header_raw or not header_formatted:
                return None

            # Get entry point using new API
            entry_point = None
            try:
                entry_point_info = self.macho.get_entry_point(arch=arch_name)
                if entry_point_info:
                    if isinstance(entry_point_info, dict):
                        entry_point = entry_point_info.get('entryoff') or entry_point_info.get('entry_address')
                    else:
                        entry_point = entry_point_info
            except Exception as e:
                self.log.debug(f"No entry point found: {e}")

            # Get UUID using new API
            uuid = None
            try:
                uuid_info = self.macho.get_uuid(arch=arch_name)
                if uuid_info:
                    uuid = str(uuid_info)
            except Exception as e:
                self.log.debug(f"No UUID found: {e}")

            # Get version info using new API with formatting
            version_info = None
            version_info_str = None
            try:
                version_info = self.macho.get_version_info(arch=arch_name)
                version_info_str = self.macho.get_version_info(arch=arch_name, formatted=True)
            except Exception as e:
                self.log.debug(f"No version info found: {e}")

            # Get segments using new API
            segments = self.macho.get_segments(arch=arch_name)
            number_of_segments = len(segments) if segments else 0

            # Get dylib info using new API
            dylib_names = self.macho.get_dylib_names(arch=arch_name)
            number_of_dylibs = len(dylib_names) if dylib_names else 0

            # Count imports using new API
            imported_functions = self.macho.get_imported_functions(arch=arch_name)
            number_of_imports = sum(len(funcs) for funcs in imported_functions.values()) if imported_functions else 0

            # Count exports using new API
            exported_symbols = self.macho.get_exported_symbols(arch=arch_name)
            number_of_exports = sum(len(symbols) for symbols in exported_symbols.values()) if exported_symbols else 0

            # Get unique load command types using new API
            load_commands_set = self.macho.get_load_commands_set(arch=arch_name, formatted=True)
            if isinstance(load_commands_set, set):
                load_commands_set = sorted(list(load_commands_set))

            # Get architectures for listing purposes
            architectures = self.macho.get_architectures()

            # Determine binary properties
            is_64bit = header_raw.get('magic') in [0xFEEDFACF, 0xCFFAEDFE]  # MH_MAGIC_64, MH_CIGAM_64
            is_signed = self._is_signed()
            
            # Create MachO dataclass
            macho_features = MachO(
                # Raw values from new API
                magic=header_raw.get('magic', 0),
                cputype=header_raw.get('cputype', 0),
                cpusubtype=header_raw.get('cpusubtype', 0),
                filetype=header_raw.get('filetype', 0),
                ncmds=header_raw.get('ncmds', 0),
                sizeofcmds=header_raw.get('sizeofcmds', 0),
                flags=header_raw.get('flags', 0),
                architecture=header_raw.get('cputype', 0),  # Use cputype as architecture
                architectures=[arch_info.get('cputype', 0) for arch_info in [self.macho.get_macho_header(arch=arch) for arch in architectures] if arch_info],
                # Human-readable values from formatted API
                magic_str=header_formatted.get('magic', ''),
                cputype_str=header_formatted.get('cputype', ''),
                cpusubtype_str=header_formatted.get('cpusubtype', ''),
                filetype_str=header_formatted.get('filetype', ''),
                flags_str=header_formatted.get('flags', '').split(', ') if header_formatted.get('flags') else [],
                architecture_str=header_formatted.get('cputype', ''),
                architectures_str=[arch_info.get('cputype', '') for arch_info in [self.macho.get_macho_header(arch=arch, formatted=True) for arch in architectures] if arch_info],
                # Additional properties
                is_64bit=is_64bit,
                is_signed=is_signed,
                number_of_load_commands=header_raw.get('ncmds', 0),
                load_commands_set=load_commands_set,
                number_of_segments=number_of_segments,
                number_of_dylibs=number_of_dylibs,
                number_of_imports=number_of_imports,
                number_of_exports=number_of_exports,
                entry_point=entry_point,
                uuid=uuid,
                version_info=version_info,
                version_info_str=version_info_str
            )
            
            return macho_features
            
        except Exception as e:
            self.log.error(f"Error extracting MachO features: {e}")
            return None

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            if not self.macho:
                return None
            architectures = self.macho.get_architectures()
            if not architectures:
                return None

            # For FAT binaries, return features for all architectures
            # For single arch, return just the single result
            if len(architectures) > 1:
                # FAT binary - return list of features for each architecture
                results = []
                for arch_name in architectures:
                    features = self._extract_macho_features_for_arch(arch_name)
                    if features:
                        # Add architecture identifier to the result
                        if hasattr(features, '__dict__'):
                            # If it's a dataclass, we can access its dict
                            features.arch_identifier = arch_name
                        results.append(features)
                return results
            else:
                # Single architecture - return single result
                return self._extract_macho_features_for_arch(architectures[0])
        except Exception as e:
            self.log.error(f"Error extracting MachO features: {e}")
            return None

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return self.extract()
        elif exporter_type == "ClickHouseExporter":
            if not self.macho:
                return None

            # Get architectures (macho is already parsed in __init__)
            try:
                architectures = self.macho.get_architectures()
            except Exception as e:
                self.log.error(f"Could not get architectures: {e}")
                return None

            data = []
            current_time = datetime.now(timezone.utc)

            # Loop through each architecture (1 for single, multiple for FAT)
            for arch_name in architectures:
                # Extract features for this specific architecture
                macho_features = self._extract_macho_features_for_arch(arch_name)
                if not macho_features:
                    continue

                # Get architecture-specific sha256
                try:
                    arch_general_info = self.macho.get_general_info(arch=arch_name)
                    arch_sha256 = arch_general_info.get('SHA256', self.sha256)
                except Exception as e:
                    self.log.warning(f"Could not get arch-specific hash for {arch_name}: {e}")
                    arch_sha256 = self.sha256

                # Architecture field (kept in features table for ORDER BY)
                architecture = macho_features.cputype    # Raw CPU type value

                data.append([
                    arch_sha256,                          # sha256 (architecture-specific)
                    architecture,                         # architecture (raw CPU type)
                    # Raw values
                    macho_features.magic,                 # magic
                    macho_features.cputype,               # cputype
                    macho_features.cpusubtype,            # cpusubtype
                    macho_features.filetype,              # filetype
                    macho_features.ncmds,                 # ncmds
                    macho_features.sizeofcmds,            # sizeofcmds
                    macho_features.flags,                 # flags
                    # Human-readable values
                    macho_features.magic_str,             # magic_str
                    macho_features.cputype_str,           # cputype_str
                    macho_features.cpusubtype_str,        # cpusubtype_str
                    macho_features.filetype_str,          # filetype_str
                    macho_features.flags_str,             # flags_str
                    # Other fields
                    macho_features.is_64bit,              # is_64bit
                    macho_features.is_signed,             # is_signed
                    macho_features.entry_point if macho_features.entry_point else None,           # entry_point
                    macho_features.uuid if macho_features.uuid else None,  # uuid
                    json.dumps(macho_features.version_info) if macho_features.version_info else "{}",  # version_info
                    json.dumps(macho_features.version_info_str) if macho_features.version_info_str else "{}",  # version_info_str
                    macho_features.load_commands_set,     # load_commands_set
                    macho_features.number_of_segments,    # number_of_segments
                    macho_features.number_of_dylibs,      # number_of_dylibs
                    macho_features.number_of_imports,     # number_of_imports
                    macho_features.number_of_exports,     # number_of_exports
                    current_time,                         # analysis_date
                ])

            column_names = [
                'sha256', 'architecture',
                # Raw values
                'magic', 'cputype', 'cpusubtype', 'filetype', 'ncmds', 'sizeofcmds', 'flags',
                # Human-readable values
                'magic_str', 'cputype_str', 'cpusubtype_str', 'filetype_str', 'flags_str',
                # Other fields
                'is_64bit', 'is_signed',
                'entry_point', 'uuid', 'version_info', 'version_info_str', 'load_commands_set',
                'number_of_segments', 'number_of_dylibs', 'number_of_imports', 'number_of_exports',
                'analysis_date'
            ]

            if not data:
                return None

            column_type_names = [
                'FixedString(64)', 'Nullable(UInt32)',
                # Raw values
                'UInt32', 'UInt32', 'UInt32', 'UInt32', 'UInt32', 'UInt32', 'UInt32',
                # Human-readable values
                'LowCardinality(String)', 'LowCardinality(String)', 'LowCardinality(String)', 'LowCardinality(String)', 'Array(LowCardinality(String))',
                # Other fields
                'UInt8', 'UInt8',
                'Nullable(UInt64)', 'Nullable(String)', 'JSON', 'JSON', 'Array(Nullable(String))',
                'UInt32', 'UInt32', 'UInt32', 'UInt32', 'DateTime64(3, \'UTC\')'
            ]

            return (data, column_names, column_type_names)
        
        return None

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