Hanchen Xia

import inspect
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 MachOImport


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

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

    def _extract_imports(self, arch_name=None):
        """Extract import information from the MachO binary for a specific architecture.

        Handles machofile v2026.2.4+ API where get_imported_functions() returns:
        Dict[str, List[Dict]] where each dict has {'name': str, 'sources': [str, ...]}
        """
        self.log.debug(inspect.currentframe().f_code.co_name)

        if not self.macho:
            return None

        try:
            # Get imported functions using API for specific architecture
            imported_functions = self.macho.get_imported_functions(arch=arch_name)
            if not imported_functions:
                return None

            # Keep the library→functions mapping (like PE does)
            library_names = []
            imports_with_mapping = []  # List of {library: [(name, source), ...]}

            for dylib_name, functions in imported_functions.items():
                # v2026.2.4+: dylib_name is already str, but handle bytes for compatibility
                if isinstance(dylib_name, bytes):
                    dylib_name = dylib_name.decode('utf-8', errors='replace')
                library_names.append(dylib_name)

                # Process function entries
                func_list = []
                for func_entry in functions:
                    # v2026.2.4+: func_entry is {'name': str, 'sources': [str, ...]}
                    if isinstance(func_entry, dict):
                        func_name = func_entry.get('name', '')
                        # Join sources if multiple, take first if single
                        sources = func_entry.get('sources', [])
                        import_source = sources[0] if sources else None
                        func_list.append((func_name, import_source))
                    else:
                        # Legacy format: func_entry is str or bytes
                        if isinstance(func_entry, bytes):
                            func_entry = func_entry.decode('utf-8', errors='replace')
                        func_list.append((func_entry, None))

                imports_with_mapping.append({dylib_name: func_list})

            # Count total functions
            total_functions = sum(len(list(d.values())[0]) for d in imports_with_mapping)

            # Create import dataclass with mapping preserved
            macho_import = MachOImport(
                macho_imports_total=total_functions,
                macho_import_libraryName=library_names if library_names else None,
                macho_import_functions=imports_with_mapping if imports_with_mapping else None
            )

            return macho_import

        except Exception as e:
            self.log.error(f"Error extracting MachO imports for arch {arch_name}: {e}")
            return None

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            # Get architectures (macho is already parsed in base class)
            architectures = self.macho.get_architectures()
            if len(architectures) > 1:
                # FAT binary - return list of imports for each architecture
                results = []
                for arch_name in architectures:
                    imports = self._extract_imports(arch_name)
                    if imports:
                        imports.arch_identifier = arch_name
                        results.append(imports)
                return results
            else:
                # Single architecture - return single result
                return self._extract_imports(architectures[0] if architectures else None)
        except Exception as e:
            self.log.error(f"Error extracting MachO imports: {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 base class)
            try:
                architectures = self.macho.get_architectures()
                is_fat = len(architectures) > 1
            except Exception as e:
                self.log.error(f"Could not get architectures: {e}")
                return None

            # Flatten the data - one row per function import (like PE imports)
            data = []
            current_time = datetime.now(timezone.utc)

            # Loop through each architecture (1 for single, multiple for FAT)
            for arch_name in architectures:
                # Get architecture-specific sha256
                try:
                    arch_general_info = self.macho.get_general_info(arch=arch_name)
                    arch_header_raw = self.macho.get_macho_header(arch=arch_name)
                    arch_sha256 = arch_general_info.get('SHA256', self.sha256)
                    arch_cputype_raw = arch_header_raw.get('cputype', 0) if arch_header_raw else 0
                except Exception as e:
                    self.log.warning(f"Could not get arch-specific data for {arch_name}: {e}")
                    arch_sha256 = self.sha256
                    arch_cputype_raw = 0

                # Get imports for this architecture
                macho_import = self._extract_imports(arch_name)
                if not macho_import or not macho_import.macho_import_functions:
                    continue

                # Flatten to one row per (library, function) pair
                for lib_funcs in macho_import.macho_import_functions:
                    for lib, funcs in lib_funcs.items():
                        for func_name, import_source in funcs:
                            data.append([
                                arch_sha256,        # sha256 (arch-specific)
                                lib,                # library_name
                                func_name,          # function_name
                                import_source,      # import_source (chained_fixups, bind_opcodes, symtab)
                                current_time,       # analysis_date
                            ])

            column_names = [
                'sha256',
                'library_name', 'function_name', 'import_source',
                'analysis_date'
            ]

            if not data:
                return None

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

            return (data, column_names, column_type_names)

        return None

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