Hanh Le

import hashlib
import inspect
import base64
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 MachOSegment


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

    def _is_empty_result(self, extracted_data) -> bool:
        """
        Override: Empty segments is an ERROR, not a valid empty case.
        A valid MachO file must have segments (at minimum __PAGEZERO, __TEXT).
        """
        # Always return False - empty segments should be treated as an error
        return False

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

    def _extract_segments_for_arch(self, arch_name):
        """Extract segment information for a specific architecture."""
        self.log.debug(f"Extracting segments for architecture: {arch_name}")
        segments = []

        try:
            # Get segments using new API with architecture parameter
            segments_data = self.macho.get_segments(arch=arch_name)
            if not segments_data:
                return segments

            # Extract segments for this architecture
            for segment in segments_data:
                try:
                    segment_name = segment.get('segname', 'Unknown')

                    # Calculate segment hash
                    segment_data = self._get_segment_data(segment)
                    if segment_data:
                        seg_sha256 = hashlib.sha256(segment_data).hexdigest()
                    else:
                        seg_sha256 = ""

                    # Use entropy already calculated by machofile module, rounded to 3 decimal places
                    seg_entropy = round(segment.get('entropy', 0.0), 3)

                    # Create segment dataclass with architecture info
                    macho_segment = MachOSegment(
                        segment_name=segment_name,
                        segment_vaddr=segment.get('vaddr', 0),
                        segment_vsize=segment.get('vsize', 0),
                        segment_offset=segment.get('offset', 0),
                        segment_size=segment.get('size', 0),
                        segment_max_vm_protection=segment.get('max_vm_protection', 0),
                        segment_initial_vm_protection=segment.get('initial_vm_protection', 0),
                        segment_nsects=segment.get('nsects', 0),
                        segment_flags=segment.get('flags', 0),
                        segment_entropy=seg_entropy,
                        segment_sha256=seg_sha256,
                    )
                    # Add architecture info to the segment
                    macho_segment.architecture = arch_name
                    segments.append(macho_segment)

                except Exception as e:
                    self.log.warning(
                        f'Unable to process segment "{segment.get("segname", "Unknown")}" for architecture {arch_name} in {self.hash.sha256}: {e}'
                    )
                    continue

            return segments

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

    def _extract_segments(self):
        """Extract segment information from all architectures in the MachO binary."""
        self.log.debug(inspect.currentframe().f_code.co_name)
        segments = []

        if not self.macho:
            return segments

        try:
            # Get architectures using new API (already parsed in base class)
            architectures = self.macho.get_architectures()
            if not architectures:
                return segments

            # Process each architecture
            for arch_name in architectures:
                arch_segments = self._extract_segments_for_arch(arch_name)
                segments.extend(arch_segments)

            return segments

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

    def _get_segment_data(self, segment):
        """Get the raw data for a segment."""
        try:
            offset = segment.get('offset', 0)
            size = segment.get('size', 0)
            
            if size == 0:
                return None
            
            # Read segment data from file
            with open(self.filepath, 'rb') as f:
                f.seek(offset)
                return f.read(size)
                
        except Exception as e:
            self.log.warning(f"Error reading segment data: {e}")
            return None

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            segments = self._extract_segments()
            return segments
        except Exception as e:
            self.log.error(f"Error extracting MachO segments: {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

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

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

                # Get architecture-specific sha256 and header info
                try:
                    arch_general_info = self.macho.get_general_info(arch=arch_name)
                    arch_sha256 = arch_general_info.get('SHA256', self.sha256)

                    # Get raw architecture value
                    arch_header_raw = self.macho.get_macho_header(arch=arch_name)
                    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

                for segment in segments:
                    data.append([
                        arch_sha256,                          # sha256 (architecture-specific)
                        segment.segment_name,                 # segment_name
                        segment.segment_vaddr,                # segment_vaddr
                        segment.segment_vsize,                # segment_vsize
                        segment.segment_offset,               # segment_offset
                        segment.segment_size,                 # segment_size
                        segment.segment_max_vm_protection,    # segment_max_vm_protection
                        segment.segment_initial_vm_protection, # segment_initial_vm_protection
                        segment.segment_nsects,               # segment_nsects
                        segment.segment_flags,                # segment_flags
                        segment.segment_entropy,              # segment_entropy
                        segment.segment_sha256,               # segment_sha256
                        current_time,                         # analysis_date
                    ])

            column_names = [
                'sha256',
                'segment_name', 'segment_vaddr', 'segment_vsize', 'segment_offset',
                'segment_size', 'segment_max_vm_protection', 'segment_initial_vm_protection',
                'segment_nsects', 'segment_flags', 'segment_entropy', 'segment_sha256',
                'analysis_date'
            ]

            if not data:
                return None

            column_type_names = [
                'FixedString(64)',
                'String', 'UInt64', 'UInt64', 'UInt64',
                'UInt64', 'UInt32', 'UInt32',
                'UInt32', 'UInt32', 'Float64', 'FixedString(64)',
                'DateTime64(3, \'UTC\')'
            ]

            return (data, column_names, column_type_names)

        return None

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