Weilai Xiang

#!/usr/bin/env python3
"""
Local MachO extractors test - test actual extractors without server connections
"""

import sys
import os
import logging
import hashlib
from datetime import datetime, timezone
from typing import Dict, Any, Optional
from unittest.mock import Mock, patch

# Add the redb directory to the path so we can import the extractors
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'redb'))

# Mock settings to avoid database connections
with patch.dict('os.environ', {'REDB_ENV': 'test'}):
    # Import the actual extractors
    from redb.extractors.macho_extractors.macho_features import MachOFeaturesExtractor
    from redb.extractors.macho_extractors.macho_segments import MachOSegmentExtractor
    from redb.extractors.macho_extractors.macho_imports import MachOImportExtractor
    from redb.extractors.macho_extractors.macho_exports import MachOExportExtractor
    from redb.extractors.macho_extractors.macho_dylibs import MachODylibExtractor
    from redb.extractors.macho_extractors.macho_signature import MachOSignatureExtractor
    from redb.extractors.macho_extractors.macho_universal import MachOUniversalExtractor

def setup_logging():
    """Setup basic logging configuration."""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s'
    )
    return logging.getLogger(__name__)

class MockLogger:
    """Mock logger for testing without RedB dependencies."""
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def debug(self, msg):
        self.logger.debug(msg)
    
    def info(self, msg):
        self.logger.info(msg)
    
    def warning(self, msg):
        self.logger.warning(msg)
    
    def error(self, msg):
        self.logger.error(msg)

class MockExporter:
    """Mock exporter that does nothing."""
    def export(self, data):
        pass

def create_mock_extractor(extractor_class, filepath):
    """Create an extractor instance with mocked dependencies."""
    log = MockLogger()
    
    # Mock the exporters to avoid database connections
    mock_exporters = [MockExporter()]
    
    # Create the extractor with mocked dependencies
    extractor = extractor_class(
        filepath=filepath,
        log=log,
        exporters=mock_exporters,
        index_prefix="test",
        elastic_index="test_macho",
        known_benign=False,
        known_malicious=False
    )
    
    return extractor

def test_extractor(extractor_class, filepath, extractor_name):
    """Test a specific extractor."""
    log = setup_logging()
    log.info(f"Testing {extractor_name}...")
    
    try:
        # Create the extractor with mocked dependencies
        extractor = create_mock_extractor(extractor_class, filepath)
        
        if not extractor.macho:
            log.error(f"{extractor_name}: No MachO object created")
            return False

        # Test the actual extract method
        result = extractor.extract()
        
        if result:
            log.info(f"{extractor_name}: Successfully extracted data")
            print(f"\n{'='*60}")
            print(f"=== {extractor_name.upper()} RESULTS ===")
            print(f"{'='*60}")
            
            if isinstance(result, list):
                print(f"📊 Extracted {len(result)} items")
                print()
                for i, item in enumerate(result):
                    print(f"📦 Item {i+1}:")
                    print(f"   {'─'*40}")
                    # Handle dataclass objects in lists
                    if hasattr(item, '__dataclass_fields__'):
                        from dataclasses import asdict
                        item_dict = asdict(item)
                        for key, value in item_dict.items():
                            if isinstance(value, (list, dict)):
                                if isinstance(value, list):
                                    print(f"   🔹 {key}: List with {len(value)} items")
                                    print(f"      {value}")
                                elif isinstance(value, dict):
                                    print(f"   🔹 {key}: Dict with {len(value)} keys")
                                    for k, v in value.items():
                                        print(f"      {k}: {v}")
                            else:
                                print(f"   🔹 {key}: {value}")
                    else:
                        # Regular dict
                        for key, value in item.items():
                            if isinstance(value, (list, dict)):
                                if isinstance(value, list):
                                    print(f"   🔹 {key}: List with {len(value)} items")
                                    print(f"      {value}")
                                elif isinstance(value, dict):
                                    print(f"   🔹 {key}: Dict with {len(value)} keys")
                                    for k, v in value.items():
                                        print(f"      {k}: {v}")
                            else:
                                print(f"   🔹 {key}: {value}")
                    print()
            else:
                print("📊 Extracted data:")
                print()
                # Handle dataclass objects
                if hasattr(result, '__dataclass_fields__'):
                    # It's a dataclass, use dataclasses.asdict
                    from dataclasses import asdict
                    result_dict = asdict(result)
                    for key, value in result_dict.items():
                        if isinstance(value, (list, dict)):
                            if isinstance(value, list):
                                print(f"🔹 {key}: List with {len(value)} items")
                                print(f"   {value}")
                            elif isinstance(value, dict):
                                print(f"🔹 {key}: Dict with {len(value)} keys")
                                for k, v in value.items():
                                    print(f"   {k}: {v}")
                        else:
                            print(f"🔹 {key}: {value}")
                        print()
                else:
                    # It's a regular dict
                    for key, value in result.items():
                        if isinstance(value, (list, dict)):
                            if isinstance(value, list):
                                print(f"🔹 {key}: List with {len(value)} items")
                                print(f"   {value}")
                            elif isinstance(value, dict):
                                print(f"🔹 {key}: Dict with {len(value)} keys")
                                for k, v in value.items():
                                    print(f"   {k}: {v}")
                        else:
                            print(f"🔹 {key}: {value}")
                        print()
        else:
            log.warning(f"{extractor_name}: No data extracted")
            print(f"\n❌ {extractor_name}: No data extracted")
        
        return True
        
    except Exception as e:
        log.error(f"{extractor_name}: Error during extraction: {e}")
        import traceback
        traceback.print_exc()
        return False

def main():
    """Main function."""
    if len(sys.argv) < 2:
        print("Usage: python test_macho_extractors_local.py <macho_file> [extractor_name]")
        print("\nAvailable extractors:")
        print("  features    - Basic MachO header and metadata")
        print("  segments    - Segment information and analysis")
        print("  imports     - Imported functions and libraries")
        print("  exports     - Exported symbols")
        print("  dylibs      - Dynamic library dependencies")
        print("  signature   - Code signing information")
        print("  universal   - FAT/Universal binary information")
        print("  all         - Test all extractors")
        sys.exit(1)
    
    filepath = sys.argv[1]
    extractor_name = sys.argv[2] if len(sys.argv) > 2 else "all"
    
    if not os.path.exists(filepath):
        print(f"File not found: {filepath}")
        sys.exit(1)
    
    # Define extractors
    extractors = {
        'features': (MachOFeaturesExtractor, "MachO Features"),
        'segments': (MachOSegmentExtractor, "MachO Segments"),
        'imports': (MachOImportExtractor, "MachO Imports"),
        'exports': (MachOExportExtractor, "MachO Exports"),
        'dylibs': (MachODylibExtractor, "MachO Dylibs"),
        'signature': (MachOSignatureExtractor, "MachO Code Signature"),
        'universal': (MachOUniversalExtractor, "MachO Universal/FAT"),
    }
    
    if extractor_name == "all":
        print(f"Testing all extractors with file: {filepath}")
        success_count = 0
        for name, (extractor_class, display_name) in extractors.items():
            if test_extractor(extractor_class, filepath, display_name):
                success_count += 1
            print("-" * 50)
        
        print(f"\n✅ {success_count}/{len(extractors)} extractors completed successfully")
        
    elif extractor_name in extractors:
        extractor_class, display_name = extractors[extractor_name]
        success = test_extractor(extractor_class, filepath, display_name)
        
        if success:
            print(f"\n✅ {display_name} testing completed successfully")
        else:
            print(f"\n❌ {display_name} testing failed")
            sys.exit(1)
    else:
        print(f"Unknown extractor: {extractor_name}")
        print("Available extractors:", ", ".join(extractors.keys()) + ", all")
        sys.exit(1)

if __name__ == "__main__":
    main()