Xi Lu

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

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


class MachOUniversalExtractor(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_universal"
        self.log.debug(inspect.currentframe().f_code.co_name)

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

    def _extract_universal_info(self):
        """Extract Universal/FAT binary architecture information using new API."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        if not self.macho:
            return None

        try:
            # Parse at Universal level first (new API requirement)
            self.macho.parse()

            # Get architectures using new API
            architectures = self.macho.get_architectures()
            if not architectures:
                return None

            # Check if this is a FAT binary
            is_fat = len(architectures) > 1

            architecture_info = []

            # Extract info for each architecture
            for arch_name in architectures:
                try:
                    # Get general info for this architecture
                    general_info = self.macho.get_general_info(arch=arch_name)

                    # Get header info for this architecture
                    header_info = self.macho.get_macho_header(arch=arch_name)

                    # Get architecture-specific MachO instance for detailed analysis
                    arch_macho = self.macho.get_macho_for_arch(arch_name)

                    # Calculate architecture slice hash (if we can access the raw data)
                    arch_sha256 = None
                    arch_md5 = None
                    arch_sha1 = None

                    # For FAT binaries, try to get slice-specific info
                    if is_fat and arch_macho:
                        try:
                            # This would require access to the slice data
                            # For now, we'll use the general file info
                            arch_sha256 = general_info.get('SHA256', '') if general_info else ''
                            arch_md5 = general_info.get('MD5', '') if general_info else ''
                            arch_sha1 = general_info.get('SHA1', '') if general_info else ''
                        except Exception as e:
                            self.log.debug(f"Could not extract slice hash for {arch_name}: {e}")

                    architecture_info.append({
                        'architecture': arch_name,
                        'arch_sha256': arch_sha256,
                        'arch_md5': arch_md5,
                        'arch_sha1': arch_sha1,
                        'cputype': header_info.get('cputype') if header_info else None,
                        'cpusubtype': header_info.get('cpusubtype') if header_info else None,
                        'filetype': header_info.get('filetype') if header_info else None
                    })

                except Exception as e:
                    self.log.warning(f"Error extracting info for architecture {arch_name}: {e}")
                    continue

            # Create Universal dataclass
            macho_universal = MachOUniversal(
                is_fat=is_fat,
                architecture_count=len(architectures),
                architectures=architectures,
                architecture_info=architecture_info,
                fat_hash=self.sha256,
                fat_md5=self.md5,
                fat_sha1=self.sha1
            )

            return macho_universal

        except Exception as e:
            self.log.error(f"Error extracting MachO Universal info: {e}")
            return None

    def _extract_fat_architecture_mappings(self):
        """Extract detailed FAT binary architecture mappings for database relationships."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        if not self.macho:
            return []

        try:
            # Parse at Universal level first
            self.macho.parse()

            # Get architectures using new API
            architectures = self.macho.get_architectures()
            if not architectures or len(architectures) <= 1:
                return []  # Not a FAT binary

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

            # For each architecture, create a mapping record
            for arch_name in architectures:
                try:
                    # Get general info
                    general_info = self.macho.get_general_info()

                    # Create mapping record for FAT binary architecture table
                    mapping = {
                        'fat_hash': self.sha256,  # SHA256 of the FAT binary
                        'architecture': arch_name,
                        'arch_sha256': general_info.get('SHA256', '') if general_info else '',  # Will need proper slice extraction
                        'arch_md5': general_info.get('MD5', '') if general_info else '',
                        'arch_sha1': general_info.get('SHA1', '') if general_info else '',
                        'arch_filename': f"{general_info.get('Filename', '')}.{arch_name}" if general_info else '',
                        'analysis_date': current_time
                    }
                    mappings.append(mapping)

                except Exception as e:
                    self.log.warning(f"Error creating mapping for architecture {arch_name}: {e}")
                    continue

            return mappings

        except Exception as e:
            self.log.error(f"Error extracting FAT architecture mappings: {e}")
            return []

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            universal_info = self._extract_universal_info()
            return universal_info
        except Exception as e:
            self.log.error(f"Error extracting MachO Universal info: {e}")
            return None

    def extract_fat_binary_basic_properties_data(self):
        """Extract data needed for creating multiple BasicProperties records for FAT binaries.

        Returns:
            Tuple: (is_fat, fat_sha256, architectures_info) where:
                - is_fat: bool indicating if this is a FAT binary
                - fat_sha256: SHA256 of the FAT wrapper
                - architectures_info: dict with arch names and their hashes
        """
        self.log.debug(inspect.currentframe().f_code.co_name)

        if not self.macho:
            return False, None, {}

        try:
            # Parse at Universal level first
            self.macho.parse()

            # Get architectures using new API
            architectures = self.macho.get_architectures()
            if not architectures or len(architectures) <= 1:
                return False, None, {}  # Not a FAT binary

            # This is a FAT binary
            architectures_info = {}

            for arch_name in architectures:
                try:
                    # Get general info for this architecture
                    general_info = self.macho.get_general_info(arch=arch_name)

                    if general_info:
                        architectures_info[arch_name] = {
                            'sha256': general_info.get('SHA256', ''),
                            'md5': general_info.get('MD5', ''),
                            'sha1': general_info.get('SHA1', ''),
                            'filename': general_info.get('Filename', ''),
                            'filesize': general_info.get('Filesize', 0)
                        }
                except Exception as e:
                    self.log.warning(f"Error extracting info for architecture {arch_name}: {e}")
                    continue

            return True, self.sha256, architectures_info

        except Exception as e:
            self.log.error(f"Error extracting FAT binary data: {e}")
            return False, None, {}

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

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

            # Get architecture info for the binary
            try:
                if universal_info.is_fat:
                    # For FAT binaries, architecture fields should be NULL since it contains multiple
                    architecture_raw = None
                    architecture_str = None
                else:
                    # For single-arch binaries, get the actual architecture info
                    header_info = self.macho.get_macho_header()
                    architecture_raw = header_info.get('cputype', 0) if header_info else 0
                    architecture_str = universal_info.architectures[0] if universal_info.architectures else None
            except Exception as e:
                self.log.warning(f"Could not get architecture info for binary: {e}")
                architecture_raw = None
                architecture_str = None

            # Main Universal binary record
            data.append([
                self.sha256,                              # sha256
                self.md5,                                 # md5
                self.sha1,                                # sha1
                None,                                     # parent_sha256 (always None for main FAT binary)
                architecture_raw,                         # architecture (raw CPU type)
                architecture_str,                         # architecture_str (human-readable)
                universal_info.is_fat,                    # is_fat
                universal_info.architecture_count,       # architecture_count
                universal_info.architectures,            # architectures (array)
                json.dumps(universal_info.architecture_info[0] if len(universal_info.architecture_info) == 1 else {"architectures": universal_info.architecture_info}) if universal_info.architecture_info else None,  # architecture_info (JSON)
                current_time,                             # analysis_date
            ])

            column_names = [
                'sha256', 'md5', 'sha1', 'parent_sha256', 'architecture', 'architecture_str',
                'is_fat', 'architecture_count', 'architectures', 'architecture_info',
                'analysis_date'
            ]

            column_type_names = [
                'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                'Nullable(FixedString(64))', 'Nullable(UInt32)', 'LowCardinality(Nullable(String))',
                'UInt8', 'UInt32', 'Array(LowCardinality(String))', 'JSON',
                'DateTime64(3, \'UTC\')'
            ]

            return (data, column_names, column_type_names)

        return None

    def prepare_fat_architecture_export_data(self) -> Any:
        """Prepare export data for the FAT binary architecture mapping table."""
        mappings = self._extract_fat_architecture_mappings()
        if not mappings:
            return None

        data = []
        for mapping in mappings:
            data.append([
                mapping['fat_hash'],
                mapping['architecture'],
                mapping['arch_sha256'],
                mapping['arch_md5'],
                mapping['arch_sha1'],
                mapping['arch_filename'],
                mapping['analysis_date'],
            ])

        column_names = [
            'fat_hash', 'architecture', 'arch_sha256', 'arch_md5', 'arch_sha1',
            'arch_filename', 'analysis_date'
        ]

        column_type_names = [
            'FixedString(64)', 'LowCardinality(String)', 'FixedString(64)',
            'FixedString(32)', 'FixedString(40)', 'String',
            'DateTime64(3, \'UTC\')'
        ]

        return (data, column_names, column_type_names)

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

    # def get_fat_architecture_table(self) -> str:
    #     """Return table name for FAT binary architecture mappings."""
    #     return "redb_fat_binary_architectures"