Xiao Zheng

"""
MachO Code Signature Extractor using LIEF + asn1crypto

Parses LC_CODE_SIGNATURE to extract:
- Code Directory (identifier, team_id, flags, hashes)
- Entitlements (XML plist)
- X.509 Certificates from CMS/PKCS#7 signature
- Signature verification status
"""

import hashlib
import inspect
import json
import plistlib
import struct
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple

import lief

lief.disable_leak_warning()
from asn1crypto import cms, x509

from redb.extractors.enum import Tag
from redb.extractors.macho_extractor import MachOExtractor
from redb.models.dataclasses import (
    MachOCertificate,
    MachOCodeDirectory,
    MachOCodeSigningInfo,
    MachOEntitlement,
)


# Code Signature Magic Values
CSMAGIC_REQUIREMENT = 0xFADE0C00
CSMAGIC_REQUIREMENTS = 0xFADE0C01
CSMAGIC_CODEDIRECTORY = 0xFADE0C02
CSMAGIC_EMBEDDED_SIGNATURE = 0xFADE0CC0  # SuperBlob
CSMAGIC_DETACHED_SIGNATURE = 0xFADE0CC1
CSMAGIC_BLOBWRAPPER = 0xFADE0B01  # CMS Signature
CSMAGIC_EMBEDDED_ENTITLEMENTS = 0xFADE7171
CSMAGIC_EMBEDDED_ENTITLEMENTS_DER = 0xFADE7172

# Code Directory Flags
CS_FLAGS = {
    0x00000001: "CS_VALID",
    0x00000002: "CS_ADHOC",
    0x00000004: "CS_GET_TASK_ALLOW",
    0x00000008: "CS_INSTALLER",
    0x00000010: "CS_FORCED_LV",
    0x00000020: "CS_INVALID_ALLOWED",
    0x00000100: "CS_HARD",
    0x00000200: "CS_KILL",
    0x00000400: "CS_CHECK_EXPIRATION",
    0x00000800: "CS_RESTRICT",
    0x00001000: "CS_ENFORCEMENT",
    0x00002000: "CS_REQUIRE_LV",
    0x00004000: "CS_ENTITLEMENTS_VALIDATED",
    0x00008000: "CS_NVRAM_UNRESTRICTED",
    0x00010000: "CS_RUNTIME",  # Hardened Runtime
    0x00020000: "CS_LINKER_SIGNED",
}

# Hash Types
HASH_TYPES = {
    0: "NO_HASH",
    1: "SHA1",
    2: "SHA256",
    3: "SHA256_TRUNCATED",
    4: "SHA384",
    5: "SHA512",
}

# Slot Types (used in SuperBlob index)
CSSLOT_CODEDIRECTORY = 0  # Primary Code Directory
CSSLOT_INFOSLOT = 1
CSSLOT_REQUIREMENTS = 2
CSSLOT_RESOURCEDIR = 3
CSSLOT_APPLICATION = 4
CSSLOT_ENTITLEMENTS = 5
CSSLOT_DER_ENTITLEMENTS = 7
CSSLOT_SIGNATURESLOT = 0x10000  # CMS Signature
CSSLOT_ALTERNATE_CODEDIRECTORIES = 0x1000  # First alternate CD (0x1000, 0x1001, ...)


class MachOSignatureExtractor(MachOExtractor):
    """
    Extracts code signature information from Mach-O binaries using LIEF.
    Parses certificates using asn1crypto for full X.509 details.
    """

    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_signature"
        self.log.debug(inspect.currentframe().f_code.co_name)
        self._lief_binary = None
        self._lief_fat = None

    def tag(self):
        return [
            Tag.MACHO_SIGNATURE.value,
            Tag.MACHO_CODE_DIRECTORY.value,
            Tag.MACHO_CERTIFICATE.value,
            Tag.MACHO_ENTITLEMENTS.value
        ]

    def _get_lief_binary(self, arch_index: int = 0) -> Optional[lief.MachO.Binary]:
        """Get LIEF binary for a specific architecture index."""
        try:
            if self._lief_fat is None:
                config = lief.MachO.ParserConfig()
                config.parse_dyld_exports = False
                config.parse_dyld_bindings = False
                config.parse_dyld_rebases = False
                self._lief_fat = lief.MachO.parse(self.filepath, config)

            if self._lief_fat is None:
                return None

            if arch_index < self._lief_fat.size:
                return self._lief_fat.at(arch_index)
            return None
        except Exception as e:
            self.log.error(f"Error parsing MachO with LIEF: {e}")
            return None

    def _parse_superblob(self, data: bytes) -> Dict[int, bytes]:
        """
        Parse the SuperBlob structure to extract individual blobs.

        SuperBlob structure:
        - magic: uint32 (0xfade0cc0)
        - length: uint32
        - count: uint32
        - index[count]: (type: uint32, offset: uint32)
        - blobs...

        Returns dict keyed by slot type (not magic), so we can handle
        multiple Code Directories (primary at slot 0, alternates at 0x1000+).
        """
        blobs = {}
        if len(data) < 12:
            return blobs

        magic, length, count = struct.unpack(">III", data[:12])

        if magic != CSMAGIC_EMBEDDED_SIGNATURE:
            self.log.warning(f"Unexpected SuperBlob magic: {hex(magic)}")
            return blobs

        # Parse blob index
        offset = 12
        blob_index = []
        for _ in range(count):
            if offset + 8 > len(data):
                break
            slot_type, blob_offset = struct.unpack(">II", data[offset:offset + 8])
            blob_index.append((slot_type, blob_offset))
            offset += 8

        # Extract each blob, keyed by slot type
        for slot_type, blob_offset in blob_index:
            if blob_offset + 8 > len(data):
                continue
            blob_magic, blob_length = struct.unpack(">II", data[blob_offset:blob_offset + 8])
            if blob_offset + blob_length <= len(data):
                blobs[slot_type] = data[blob_offset:blob_offset + blob_length]

        return blobs

    def _parse_code_directory(self, cd_data: bytes) -> Optional[MachOCodeDirectory]:
        """
        Parse CodeDirectory blob.

        CodeDirectory structure:
        - magic: uint32 (0xfade0c02)
        - length: uint32
        - version: uint32
        - flags: uint32
        - hashOffset: uint32
        - identOffset: uint32
        - nSpecialSlots: uint32
        - nCodeSlots: uint32
        - codeLimit: uint32
        - hashSize: uint8
        - hashType: uint8
        - platform: uint8
        - pageSize: uint8
        - spare2: uint32
        - scatterOffset: uint32 (v >= 0x20100)
        - teamOffset: uint32 (v >= 0x20200)
        ...
        """
        if len(cd_data) < 44:
            return None

        try:
            magic, length, version, flags = struct.unpack(">IIII", cd_data[:16])

            if magic != CSMAGIC_CODEDIRECTORY:
                return None

            hash_offset, ident_offset = struct.unpack(">II", cd_data[16:24])
            n_special_slots, n_code_slots, code_limit = struct.unpack(">III", cd_data[24:36])
            hash_size, hash_type, platform, page_size_log2 = struct.unpack(">BBBB", cd_data[36:40])

            # Extract identifier string
            identifier = None
            if ident_offset > 0 and ident_offset < len(cd_data):
                end = cd_data.find(b'\x00', ident_offset)
                if end > ident_offset:
                    identifier = cd_data[ident_offset:end].decode('utf-8', errors='replace')

            # Extract team ID (version >= 0x20200)
            team_id = None
            if version >= 0x20200 and len(cd_data) >= 52:
                team_offset = struct.unpack(">I", cd_data[48:52])[0]
                if team_offset > 0 and team_offset < len(cd_data):
                    end = cd_data.find(b'\x00', team_offset)
                    if end > team_offset:
                        team_id = cd_data[team_offset:end].decode('utf-8', errors='replace')

            # Parse flags
            flags_str = [name for flag, name in CS_FLAGS.items() if flags & flag]

            # Calculate CD hash
            cd_hash = hashlib.sha256(cd_data).hexdigest()

            return MachOCodeDirectory(
                identifier=identifier,
                team_id=team_id,
                flags=flags,
                cd_hash=cd_hash,
            )
        except Exception as e:
            self.log.error(f"Error parsing CodeDirectory: {e}")
            return None

    def _parse_entitlements(self, ent_data: bytes) -> Optional[Dict]:
        """
        Parse Entitlements blob (XML plist).

        Returns parsed entitlements dict.
        """
        if len(ent_data) < 8:
            return None

        try:
            magic, length = struct.unpack(">II", ent_data[:8])

            # Only accept XML plist magic - DER requires separate parser
            if magic != CSMAGIC_EMBEDDED_ENTITLEMENTS:
                return None

            # Extract plist data (after header)
            plist_data = ent_data[8:]

            # Parse plist
            try:
                entitlements = plistlib.loads(plist_data)
                return entitlements
            except Exception:
                # Try to strip null bytes
                plist_data = plist_data.rstrip(b'\x00')
                entitlements = plistlib.loads(plist_data)
                return entitlements

        except Exception as e:
            self.log.debug(f"Error parsing entitlements: {e}")
            return None

    def _parse_entitlements_der(self, ent_data: bytes) -> Optional[Dict]:
        """Best-effort parse of DER-encoded entitlements.

        DER-only binaries are rare (standard codesign embeds both XML and DER).
        Uses a layered approach: structured ASN.1 first, string extraction fallback.
        """
        if len(ent_data) < 8:
            return None

        try:
            magic, length = struct.unpack(">II", ent_data[:8])
            if magic != CSMAGIC_EMBEDDED_ENTITLEMENTS_DER:
                return None

            der_data = ent_data[8:]

            # Layer 1: Try structured ASN.1 parsing via asn1crypto
            try:
                from asn1crypto import core

                entitlements = {}
                seq = core.Sequence.load(der_data)
                for item in seq:
                    try:
                        if len(item) >= 2:
                            key = str(item[0].native)
                            val_element = item[1]
                            tag = val_element.tag
                            if tag == 0:      # BOOLEAN
                                entitlements[key] = val_element.contents != b'\x00'
                            elif tag == 1:    # UTF8String
                                entitlements[key] = val_element.native
                            elif tag == 2:    # INTEGER
                                entitlements[key] = val_element.native
                            elif tag == 3:    # SET OF (array of strings)
                                entitlements[key] = [str(v.native) for v in val_element]
                            else:
                                entitlements[key] = val_element.native
                    except Exception:
                        continue
                if entitlements:
                    return entitlements
            except Exception as e:
                self.log.debug(f"ASN.1 structured parse failed, trying string extraction: {e}")

            # Layer 2: Regex extraction of known entitlement key strings from raw DER
            # Gets keys but not typed values — still useful for detection/triage
            import re
            text = der_data.decode('utf-8', errors='ignore')
            keys = re.findall(
                r'(com\.apple\.[a-zA-Z0-9._-]+'
                r'|application-identifier'
                r'|get-task-allow'
                r'|platform-application'
                r'|team-identifier'
                r'|keychain-access-groups)',
                text
            )
            if keys:
                # Deduplicate while preserving order
                seen = set()
                entitlements = {}
                for key in keys:
                    if key not in seen:
                        seen.add(key)
                        entitlements[key] = True  # Assume boolean (most common type)
                self.log.info(
                    f"DER entitlements: ASN.1 parse failed, extracted {len(entitlements)} "
                    f"keys via string matching (values defaulted to True)"
                )
                return entitlements

            return None

        except Exception as e:
            self.log.debug(f"Error parsing DER entitlements: {e}")
            return None

    def _parse_cms_signature(self, cms_data: bytes) -> Tuple[List[MachOCertificate], Optional[str], bool]:
        """
        Parse CMS/PKCS#7 signature blob to extract certificates.

        Returns tuple of (certificates_list, signing_time, is_verified).
        """
        certificates = []
        signing_time = None
        has_cms_signature = False

        if len(cms_data) < 8:
            return certificates, signing_time, has_cms_signature

        try:
            magic, length = struct.unpack(">II", cms_data[:8])

            if magic != CSMAGIC_BLOBWRAPPER:
                return certificates, signing_time, has_cms_signature

            # CMS data starts after the blob header
            der_data = cms_data[8:]

            # Parse CMS ContentInfo
            content_info = cms.ContentInfo.load(der_data)

            if content_info['content_type'].native != 'signed_data':
                return certificates, signing_time, is_verified

            signed_data = content_info['content']

            # Extract certificates
            if signed_data['certificates']:
                cert_order = 0
                for cert_choice in signed_data['certificates']:
                    try:
                        if cert_choice.name == 'certificate':
                            cert = cert_choice.chosen
                            macho_cert = self._extract_certificate_info(cert, cert_order)
                            if macho_cert:
                                certificates.append(macho_cert)
                                cert_order += 1
                    except Exception as e:
                        self.log.debug(f"Error parsing certificate: {e}")
                        continue

            # Determine certificate types (leaf, intermediate, root)
            self._classify_certificates(certificates)

            # Extract signing time from signer info
            if signed_data['signer_infos']:
                for signer_info in signed_data['signer_infos']:
                    if signer_info['signed_attrs']:
                        for attr in signer_info['signed_attrs']:
                            if attr['type'].native == 'signing_time':
                                signing_time = attr['values'][0].native
                                if isinstance(signing_time, datetime):
                                    signing_time = signing_time.strftime("%Y-%m-%d %H:%M:%S")
                                break

            # Note: This only indicates presence of CMS signature structure, not
            # cryptographic verification. Full verification would require checking
            # against Apple's CA chain.
            has_cms_signature = len(certificates) > 0

        except Exception as e:
            self.log.error(f"Error parsing CMS signature: {e}")

        return certificates, signing_time, has_cms_signature

    def _extract_certificate_info(self, cert: x509.Certificate, order: int) -> Optional[MachOCertificate]:
        """Extract certificate information from an X.509 certificate."""
        try:
            tbs = cert['tbs_certificate']

            # Serial number
            serial = tbs['serial_number'].native
            serial_hex = format(serial, 'x').upper()

            # Subject and Issuer
            subject = self._format_name(tbs['subject'])
            issuer = self._format_name(tbs['issuer'])

            # Extract common name, organization, and OU from subject
            common_name = None
            organization = None
            organizational_unit = None
            try:
                for rdn in tbs['subject'].chosen:
                    for attr in rdn:
                        oid = attr['type'].native
                        value = attr['value'].native
                        if oid == 'common_name':
                            common_name = value
                        elif oid == 'organization_name':
                            organization = value
                        elif oid == 'organizational_unit_name':
                            organizational_unit = value
            except Exception:
                pass

            # Validity
            validity = tbs['validity']
            valid_from = validity['not_before'].native
            valid_to = validity['not_after'].native

            if isinstance(valid_from, datetime):
                valid_from = valid_from.strftime("%Y-%m-%d %H:%M:%S")
            if isinstance(valid_to, datetime):
                valid_to = valid_to.strftime("%Y-%m-%d %H:%M:%S")

            # Algorithm
            algorithm = tbs['signature']['algorithm'].native

            # Thumbprint (SHA-1 - industry standard for certificate thumbprints)
            cert_der = cert.dump()
            thumbprint = hashlib.sha1(cert_der).hexdigest().upper()

            return MachOCertificate(
                subject=subject,
                issuer=issuer,
                serial_number=serial_hex,
                not_before=valid_from,
                not_after=valid_to,
                thumbprint=thumbprint,
                certificate_type="unknown",  # Will be classified later
                order=order,
                algorithm=algorithm,
                common_name=common_name,
                organization=organization,
                organizational_unit=organizational_unit,
            )
        except Exception as e:
            self.log.error(f"Error extracting certificate info: {e}")
            return None

    def _format_name(self, name) -> str:
        """Format X.509 Name as a readable string."""
        try:
            parts = []
            for rdn in name.chosen:
                for attr in rdn:
                    oid = attr['type'].native
                    value = attr['value'].native
                    # Map common OIDs to abbreviations
                    oid_map = {
                        'common_name': 'CN',
                        'organization_name': 'O',
                        'organizational_unit_name': 'OU',
                        'country_name': 'C',
                        'state_or_province_name': 'ST',
                        'locality_name': 'L',
                        'email_address': 'E',
                        'user_id': 'UID',
                    }
                    abbrev = oid_map.get(oid, oid)
                    parts.append(f"{abbrev}={value}")
            return ", ".join(parts)
        except Exception:
            return str(name.human_friendly) if hasattr(name, 'human_friendly') else str(name)

    def _classify_certificates(self, certificates: List[MachOCertificate]):
        """
        Classify certificates as leaf, intermediate, or root using
        chain relationships:
        1. Root: self-signed (subject == issuer)
        2. Intermediate: issued another cert in the chain
        3. Leaf: everything else (end-entity)
        """
        if not certificates:
            return

        for cert in certificates:
            if cert.subject == cert.issuer:
                cert.certificate_type = "root"
            # Check if this cert issued any other cert in the chain
            elif any(c.issuer == cert.subject and c != cert for c in certificates):
                cert.certificate_type = "intermediate"
            else:
                cert.certificate_type = "leaf"

    def _extract_code_signature(self, arch_index: int = 0, arch_name: str = None) -> Optional[MachOCodeSigningInfo]:
        """Extract code signature information for a specific architecture."""
        self.log.debug(f"Extracting code signature for arch index {arch_index}")

        binary = self._get_lief_binary(arch_index)
        if binary is None:
            return None

        # Check if binary has code signature
        if not binary.has_code_signature:
            return MachOCodeSigningInfo(
                _id=self.sha256,
                signing_type="unsigned",
                has_signature=False,
                number_of_certificates=0,
                arch_identifier=arch_name,
            )

        # Get raw code signature data
        code_sig = binary.code_signature
        if code_sig is None:
            return None

        # LIEF's code_sig.data only returns the load command, not the actual signature
        # We need to read the signature data directly from the file
        # For FAT binaries, data_offset is relative to the slice, not the file
        sig_data = None
        try:
            with open(self.filepath, 'rb') as f:
                # fat_offset is 0 for single-arch binaries, slice offset for FAT
                file_offset = code_sig.data_offset + binary.fat_offset
                f.seek(file_offset)
                sig_data = f.read(code_sig.data_size)
        except Exception as e:
            self.log.error(f"Error reading code signature data from file: {e}")
            return None

        if not sig_data:
            return None

        # Parse SuperBlob to get individual blobs (keyed by slot type)
        blobs = self._parse_superblob(sig_data)

        # Parse all Code Directories (primary + alternates)
        code_directories = []

        # Primary Code Directory at slot 0
        if CSSLOT_CODEDIRECTORY in blobs:
            cd = self._parse_code_directory(blobs[CSSLOT_CODEDIRECTORY])
            if cd:
                cd.slot_type = CSSLOT_CODEDIRECTORY
                code_directories.append(cd)

        # Alternate Code Directories at slots 0x1000, 0x1001, ...
        for slot_type in sorted(blobs.keys()):
            if slot_type >= CSSLOT_ALTERNATE_CODEDIRECTORIES and slot_type < CSSLOT_SIGNATURESLOT:
                cd = self._parse_code_directory(blobs[slot_type])
                if cd:
                    cd.slot_type = slot_type
                    code_directories.append(cd)

        # Use primary CD (or first available) for signing info
        primary_cd = code_directories[0] if code_directories else None

        # Parse Entitlements - prefer XML plist, fall back to DER for edge cases
        entitlements = None
        if CSSLOT_ENTITLEMENTS in blobs:
            entitlements = self._parse_entitlements(blobs[CSSLOT_ENTITLEMENTS])
        if entitlements is None and CSSLOT_DER_ENTITLEMENTS in blobs:
            # DER-only binary (rare — non-standard signing tool like ldid)
            self.log.warning("Only DER entitlements found, no XML plist slot")
            entitlements = self._parse_entitlements_der(blobs[CSSLOT_DER_ENTITLEMENTS])

        # Parse CMS Signature (certificates)
        certificates = []
        signing_time = None
        has_cms_signature = False

        if CSSLOT_SIGNATURESLOT in blobs:
            certificates, signing_time, has_cms_signature = self._parse_cms_signature(blobs[CSSLOT_SIGNATURESLOT])

        # Determine signing type
        signing_type = "unsigned"
        if primary_cd:
            if primary_cd.flags and (primary_cd.flags & 0x00000002):  # CS_ADHOC
                signing_type = "ad-hoc"
            elif len(certificates) > 0:
                signing_type = "certificate"
            elif primary_cd.identifier:
                # Has CodeDirectory but no CMS = ad-hoc
                signing_type = "ad-hoc"

        # Check for hardened runtime and library validation flags
        has_hardened_runtime = False
        has_library_validation = False
        if primary_cd and primary_cd.flags:
            has_hardened_runtime = bool(primary_cd.flags & 0x00010000)  # CS_RUNTIME
            has_library_validation = bool(primary_cd.flags & 0x00002000)  # CS_REQUIRE_LV

        return MachOCodeSigningInfo(
            _id=self.sha256,
            signing_type=signing_type,
            has_signature=has_cms_signature,
            number_of_certificates=len(certificates),
            code_directories=code_directories,
            entitlements=entitlements,
            x509_certificates=certificates,
            date_signed=signing_time,
            arch_identifier=arch_name,
            has_hardened_runtime=has_hardened_runtime,
            has_library_validation=has_library_validation,
        )

    def extract(self) -> Optional[Any]:
        """Extract code signature information from the Mach-O binary."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        try:
            # Get LIEF FAT binary to check architecture count
            config = lief.MachO.ParserConfig()
            config.parse_dyld_exports = False
            config.parse_dyld_bindings = False
            config.parse_dyld_rebases = False
            self._lief_fat = lief.MachO.parse(self.filepath, config)

            if self._lief_fat is None:
                self.log.error("Failed to parse MachO with LIEF")
                return None

            # Get architecture names from machofile for consistency
            arch_names = []
            if self.macho:
                try:
                    arch_names = self.macho.get_architectures()
                except Exception:
                    pass

            # If no arch names from machofile, generate them
            if not arch_names:
                arch_names = [f"arch_{i}" for i in range(self._lief_fat.size)]

            if self._lief_fat.size > 1:
                # FAT/Universal binary - extract for each architecture
                results = []
                for i in range(self._lief_fat.size):
                    arch_name = arch_names[i] if i < len(arch_names) else f"arch_{i}"
                    signature = self._extract_code_signature(i, arch_name)
                    if signature:
                        results.append(signature)
                return results if results else None
            else:
                # Single architecture
                arch_name = arch_names[0] if arch_names else None
                return self._extract_code_signature(0, arch_name)

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

    def prepare_export_data(self, exporter_type: str) -> Any:
        """Prepare data for export to Elasticsearch or ClickHouse."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        if exporter_type == "ElasticsearchExporter":
            return self.extract()

        elif exporter_type == "ClickHouseExporter":
            result = self.extract()
            if result is None:
                return None

            # Normalize to list
            signatures = result if isinstance(result, list) else [result]

            current_time = datetime.now(timezone.utc)

            # Prepare signature data
            signatures_data = []
            certificates_data = []
            entitlements_data = []

            for sig in signatures:
                if sig is None:
                    continue

                # Get architecture-specific hash if available
                arch_sha256 = self.sha256

                # If FAT binary, try to get arch-specific sha256
                if self.macho and sig.arch_identifier:
                    try:
                        arch_info = self.macho.get_general_info(arch=sig.arch_identifier)
                        if arch_info:
                            arch_sha256 = arch_info.get('SHA256', self.sha256)
                    except Exception as e:
                        self.log.debug(f"Could not get arch-specific info: {e}")

                # Get primary CD fields for signatures table
                primary_cd = sig.code_directories[0] if sig.code_directories else None
                cd_identifier = primary_cd.identifier if primary_cd else None
                cd_team_id = primary_cd.team_id if primary_cd else None
                cd_flags = primary_cd.flags if primary_cd else None
                cd_hash = primary_cd.cd_hash if primary_cd else None

                signatures_data.append([
                    arch_sha256,
                    sig.signing_type,
                    sig.has_signature,
                    sig.number_of_certificates,
                    cd_identifier,
                    cd_team_id,
                    cd_flags,
                    cd_hash,
                    sig.has_hardened_runtime,
                    sig.has_library_validation,
                    sig.date_signed,
                    current_time,
                ])

                # Prepare certificate data
                if sig.x509_certificates:
                    for cert in sig.x509_certificates:
                        certificates_data.append([
                            arch_sha256,
                            cert.certificate_type,
                            cert.algorithm,
                            cert.issuer,
                            cert.subject,
                            cert.serial_number,
                            cert.thumbprint,
                            datetime.strptime(cert.not_before, "%Y-%m-%d %H:%M:%S") if cert.not_before else None,
                            datetime.strptime(cert.not_after, "%Y-%m-%d %H:%M:%S") if cert.not_after else None,
                            current_time,
                        ])

                # Prepare entitlements data
                if sig.entitlements:
                    for ent_key, ent_value in sig.entitlements.items():
                        # Determine type and set appropriate value columns
                        if isinstance(ent_value, bool):
                            ent_type = 'boolean'
                            value_bool = 1 if ent_value else 0
                            value_string = None
                            value_array = []
                        elif isinstance(ent_value, list):
                            ent_type = 'array'
                            value_bool = None
                            value_string = None
                            # Convert all list items to strings
                            value_array = [str(v) for v in ent_value]
                        else:
                            ent_type = 'string'
                            value_bool = None
                            value_string = str(ent_value)
                            value_array = []

                        entitlements_data.append([
                            arch_sha256,
                            ent_key,
                            ent_type,
                            value_bool,
                            value_string,
                            value_array,
                            current_time,
                        ])

            return {
                'multi_table': True,
                'signatures': {
                    'table': 'redb_macho_signatures',
                    'data': signatures_data,
                    'column_names': [
                        'sha256',
                        'signing_type', 'has_signature', 'number_of_certificates',
                        'cd_identifier', 'cd_team_id', 'cd_flags', 'cd_hash',
                        'has_hardened_runtime', 'has_library_validation',
                        'date_signed', 'analysis_date'
                    ],
                    'column_type_names': [
                        'FixedString(64)',
                        "Enum8('unsigned' = 0, 'ad-hoc' = 1, 'certificate' = 2)", 'UInt8', 'UInt32',
                        'Nullable(String)', 'Nullable(String)', 'Nullable(UInt32)', 'Nullable(String)',
                        'UInt8', 'UInt8',
                        'Nullable(DateTime64(3, \'UTC\'))', 'DateTime64(3, \'UTC\')'
                    ]
                },
                'certificates': {
                    'table': 'redb_macho_certificates',
                    'data': certificates_data,
                    'column_names': [
                        'sha256', 'certificate_type', 'certificate_algorithm',
                        'certificate_issuer', 'certificate_subject', 'certificate_serial_number',
                        'certificate_thumbprint', 'certificate_valid_from', 'certificate_valid_to',
                        'analysis_date'
                    ],
                    'column_type_names': [
                        'FixedString(64)',
                        "Enum8('leaf' = 1, 'intermediate' = 2, 'root' = 3)", 'LowCardinality(String)',
                        'String', 'String', 'String', 'FixedString(40)',
                        'DateTime64(3, \'UTC\')', 'DateTime64(3, \'UTC\')', 'DateTime64(3, \'UTC\')'
                    ]
                },
                'entitlements': {
                    'table': 'redb_macho_entitlements',
                    'data': entitlements_data,
                    'column_names': [
                        'sha256', 'entitlement_key', 'entitlement_type',
                        'value_bool', 'value_string', 'value_array',
                        'analysis_date'
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'LowCardinality(String)',
                        "Enum8('boolean' = 1, 'string' = 2, 'array' = 3)",
                        'Nullable(UInt8)', 'Nullable(String)', 'Array(String)',
                        'DateTime64(3, \'UTC\')'
                    ]
                }
            }

        return None

    def get_clickhouse_table(self) -> str:
        """Return primary ClickHouse table name."""
        return "redb_macho_signatures"

    def get_clickhouse_tables(self) -> Dict[str, str]:
        """Return all ClickHouse tables for multi-table export."""
        return {
            'signatures': 'redb_macho_signatures',
            'certificates': 'redb_macho_certificates',
            'entitlements': 'redb_macho_entitlements'
        }