Manuchehr Soleimani

import hashlib
import inspect
import json
import os
import time
import threading
from datetime import datetime, timezone
from typing import Dict, Any, Optional

from redb.extractors.enum import Tag
from redb.extractors.extractor import Extractor
import magic
import pefile
import ppdeep
import tlsh

# Import our BinjaDecompiler (conditional)
from redb.extractors.decompiler.bninja.decompiler import BinaryNinjaDecompiler

class DecompileBinja(Extractor):
    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        filetype=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            elastic_index,
            known_benign,
            known_malicious,
        )
        self.log.debug(inspect.currentframe().f_code.co_name)

        # Check if Binary Ninja is available
       # if not BINARYNINJA_AVAILABLE:
        #     self.log.error("Binary Ninja is not available in this container")
        #    raise ImportError(
        #        "Binary Ninja module not found - not available in feature extraction container"
        #    )
        self.analysis_results = None
        self.binja_decompiler = None
        self.filetype = filetype

        # Convert TIMEOUT to integer with a default of 1200 seconds (20 minutes)
        try:
            self.BINJA_TIMEOUT = int(os.getenv("BINJA_TIMEOUT", "1200"))
        except ValueError:
            self.log.warning(
                "Invalid BINJA_TIMEOUT value, using default of 1200 seconds"
            )
            self.BINJA_TIMEOUT = 1200

        # Convert TIMEOUT to integer with a default of 1200 seconds (20 minutes)
        try:
            self.DECOMPILE_EXTRACTOR_TIMEOUT = int(
                os.getenv("DECOMPILE_EXTRACTOR_TIMEOUT", "2580")
            )
        except ValueError:
            self.log.warning(
                "Invalid DECOMPILE_EXTRACTOR_TIMEOUT value, using default of 2580 seconds"
            )
            self.DECOMPILE_EXTRACTOR_TIMEOUT = 2580

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cleanup_run()

    def calculate_md5(self, input_str):
        """Calculate MD5 hash of a string."""
        return hashlib.md5(input_str.encode("utf-8")).hexdigest()

    def is_dotnet(self):
        """Check if the binary is a .NET assembly.

        Returns:
            bool: True if the file is a .NET assembly, False otherwise
        """
        try:
            if self.filetype == "pebin":
                file_type = magic.from_buffer(self.binary)
                if ".Net" in file_type:
                    return True
                pe = pefile.PE(self.filepath)
                for entry in pe.OPTIONAL_HEADER.DATA_DIRECTORY:
                    # IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR is typically 14
                    if (
                        entry.name == "IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR"
                        and entry.Size > 0
                    ):
                        return True
                return False
        except AttributeError as e:
            self.log.error(
                f"AttributeError error dotnet file {self.hash.sha256} Full error : {e}"
            )
            return False

    def cleanup_run(self):
        """Clean up after analysis."""
        try:
            # BinaryNinjaDecompiler uses context manager pattern (__enter__/__exit__)
            # Cleanup happens automatically when exiting the 'with' block
            self.binja_decompiler = None

            # Force garbage collection
            import gc

            gc.collect()

        except Exception as e:
            self.log.error(f"Error in cleanup: {e}")

    def analyze_binary(self) -> Optional[Dict[str, Any]]:
        """Run Binary Ninja analysis and return results."""
        self.log.debug("Starting binary analysis")

        try:
            # Use BinaryNinjaDecompiler as a context manager to ensure proper setup/cleanup
            with BinaryNinjaDecompiler(
                filepath=self.filepath,
                timeout=self.BINJA_TIMEOUT,
                log=self.log,
                exporters=self.exporters,
                index_prefix=self.index_prefix,
                filetype=self.filetype,
            ) as decompiler:
                self.binja_decompiler = decompiler

                if decompiler.extract():
                    # Store results before context manager exits
                    results = decompiler.analysis_results
                    return results
                else:
                    self.log.error("BinaryNinjaDecompiler extraction failed")
                    return None

        except Exception as e:
            self.log.error(f"Error in Binary Ninja analysis: {e}")
            import traceback
            self.log.error(f"Traceback: {traceback.format_exc()}")
            return None

        finally:
            self.cleanup_run()

    def extract(self):
        """Extract and process all analysis results."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        # Create a flag to track if extraction completed
        extraction_completed = False
        extraction_result = False
        extraction_error = None

        # Define the extraction process as a separate function
        def do_extraction():
            nonlocal extraction_completed, extraction_result, extraction_error
            try:
                results = self.analyze_binary()
                if not results:
                    extraction_result = False
                else:
                    self.analysis_results = results
                    extraction_result = True
            except Exception as e:
                extraction_error = e
                extraction_result = False
            finally:
                extraction_completed = True

        # Start extraction in a separate thread
        extraction_thread = threading.Thread(target=do_extraction)
        extraction_thread.daemon = True
        extraction_thread.start()

        # Wait for the extraction to complete or timeout
        start_time = time.time()
        while (
            not extraction_completed
            and (time.time() - start_time) < self.DECOMPILE_EXTRACTOR_TIMEOUT
        ):
            time.sleep(1)

        if not extraction_completed:
            self.log.error(
                f"Extraction timed out after {self.DECOMPILE_EXTRACTOR_TIMEOUT} seconds"
            )
            # Force cleanup
            self.cleanup_run()
            return None

        if extraction_error:
            self.log.error(f"Error in extraction: {extraction_error}")
            return None

        # Return the actual analysis results, not just a boolean
        return self.analysis_results if extraction_result else None

    def prepare_export_data(self, exporter_type: str) -> Any:
        """Prepare data for database export."""
        self.log.debug(inspect.currentframe().f_code.co_name)
        if not self.analysis_results:
            return None

        # # Delegate to the BinjaDecompiler for consistent export formatting
        # if self.binja_decompiler:
        #     return self.binja_decompiler.prepare_export_data(exporter_type)
        # else:
        #     self.log.error("BinjaDecompiler not available for export preparation")
        #     return None

        if exporter_type == "ClickHouseExporter":
            now = datetime.now(timezone.utc)

            def prepare_array_field(value, array_type):
                """Helper to prepare array fields with proper null handling"""
                if value is None:
                    return []
                return value

            # Add a helper function to handle empty strings
            def ensure_not_empty(value, default="UNKNOWN"):
                """Ensure a string value is not empty"""
                if value is None or value == "":
                    return default
                return value

            def ssdeep_disassembly(func):
                try:
                    if len(func) > 1:
                        return ppdeep.hash(func)
                    return ""
                except Exception as e:
                    self.log.error(f"Error in disassembly ssdeep hash calculation: {e}")
                    return ""

            def tlsh_disassembly(func):
                try:
                    if len(func) >= 50:
                        return tlsh.hash(func.encode("utf-8"))
                    return ""
                except Exception as e:
                    self.log.error(f"Error in disassembly tlsh hash calculation: {e}")
                    return ""

            return {
                "multi_table": True,
                "decompiled_content": {
                    "table": "code_binja_decompiled_functions_content",
                    "data": [
                        [
                            f["decompiled_function_hash"],
                            f["decompiled_function"],
                            f["function_type"],
                            now,
                        ]
                        for f in self.analysis_results["decompiled"]
                    ],
                    "column_names": [
                        "decompiled_function_hash",
                        "decompiled_function",
                        "function_type",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "String",
                        "Enum8('USER'=1, 'LIBRARY'=2, 'THUNK'=3, 'EXTERNAL'=4, 'UNKNOWN'=5)",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "decompiled_refs": {
                    "table": "code_binja_decompiled_functions_references",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            self.analysis_results["sha1"],
                            self.analysis_results["md5"],
                            f["decompiled_function_hash"],
                            f["disassembled_function_hash"],  # New linking field
                            f["decompiled_function_name"],
                            f["decompiled_function_prototype"],
                            f["decompiled_function_address"],
                            now,
                        ]
                        for f in self.analysis_results["decompiled"]
                    ],
                    "column_names": [
                        "sha256",
                        "sha1",
                        "md5",
                        "decompiled_function_hash",
                        "disassembled_function_hash",  # New linking field
                        "decompiled_function_name",
                        "decompiled_function_prototype",
                        "decompiled_function_address",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(40)",
                        "FixedString(32)",
                        "FixedString(64)",
                        "Nullable(FixedString(64))",  # New linking field
                        "LowCardinality(String)",
                        "LowCardinality(String)",
                        "UInt64",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "disassembled_content": {
                    "table": "code_binja_disassembled_functions_content",
                    "data": [
                        [
                            f["disassembled_function_hash"],
                            f.get("disassembled_function", ""),
                            f.get("disassembled_function_no_addresses", ""),
                            f.get("function_type", "UNKNOWN"),
                            f.get("instructions_count", 0),
                            prepare_array_field(
                                f.get("instructions_types"), "LowCardinality(String)"
                            ),
                            f.get("control_flow_count", 0),
                            prepare_array_field(
                                f.get("memory_access_pattern"), "LowCardinality(String)"
                            ),
                            prepare_array_field(
                                f.get("register_usage"), "LowCardinality(String)"
                            ),
                            f.get("data_references_count", 0),
                            f.get("max_block_size", 0),
                            f.get("num_calls", 0),
                            f.get("stack_size", 0),
                            now,
                        ]
                        for f in self.analysis_results["disassembled"]
                    ],
                    "column_names": [
                        "disassembled_function_hash",
                        "disassembled_function",
                        "disassembled_function_no_addresses",
                        "function_type",
                        "instructions_count",
                        "instructions_types",
                        "control_flow_count",
                        "memory_access_pattern",
                        "register_usage",
                        "data_references_count",
                        "max_block_size",
                        "num_calls",
                        "stack_size",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "String",
                        "String",
                        "Enum8('USER'=1, 'LIBRARY'=2, 'THUNK'=3, 'EXTERNAL'=4, 'UNKNOWN'=5)",
                        "UInt32",
                        "Array(LowCardinality(String))",
                        "UInt32",
                        "Array(LowCardinality(String))",
                        "Array(LowCardinality(String))",
                        "UInt32",
                        "Nullable(UInt32)",
                        "Nullable(UInt32)",
                        "Nullable(Int32)",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "disassembled_refs": {
                    "table": "code_binja_disassembled_functions_references",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            self.analysis_results["sha1"],
                            self.analysis_results["md5"],
                            f["disassembled_function_hash"],
                            f["decompiled_function_hash"],
                            f["disassembled_function_name"],
                            f["disassembled_function_address"],
                            ssdeep_disassembly(
                                f.get("disassembled_function_no_addresses", "")
                            ),
                            tlsh_disassembly(
                                f.get("disassembled_function_no_addresses", "")
                            ),
                            now,
                        ]
                        for f in self.analysis_results["disassembled"]
                    ],
                    "column_names": [
                        "sha256",
                        "sha1",
                        "md5",
                        "disassembled_function_hash",
                        "decompiled_function_hash",
                        "disassembled_function_name",
                        "disassembled_function_address",
                        "ssdeep_disassembly",
                        "tlsh_disassembly",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(40)",
                        "FixedString(32)",
                        "FixedString(64)",
                        "Nullable(FixedString(64))",
                        "LowCardinality(String)",
                        "UInt64",
                        "Nullable(String)",
                        "Nullable(FixedString(72))",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "cfg_blocks": {
                    "table": "code_binja_cfg_blocks",
                    "data": [
                        [
                            b["block_id"],
                            self.analysis_results["sha256"],
                            self.analysis_results["sha1"],
                            self.analysis_results["md5"],
                            b["function_address"],
                            b["block_start_address"],
                            b["block_end_address"],
                            b["block_size"],
                            b["instructions_count"],
                            b["block_instructions"],  # MD5 hash of instructions
                            b["predecessor_blocks"],
                            b["successor_blocks"],
                            b["depth"],
                            b["position"],
                            b["branch_type"],
                            b["block_type"],
                            b["flags"],
                            b["dominators"],
                            b["post_dominators"],
                            now,
                        ]
                        # Flatten: iterate through all functions, then all blocks in each function
                        for func_cfg in self.analysis_results["cfg"]
                        if func_cfg is not None  # Handle None from failed extractions
                        for b in func_cfg["blocks"]
                    ],
                    "column_names": [
                        "block_id",
                        "sha256",
                        "sha1",
                        "md5",
                        "function_address",
                        "block_start_address",
                        "block_end_address",
                        "block_size",
                        "instructions_count",
                        "block_instructions_hash",
                        "predecessor_blocks",
                        "successor_blocks",
                        "depth",
                        "position",
                        "branch_type",
                        "block_type",
                        "flags",
                        "dominators",
                        "post_dominators",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",  # block_id (SHA256)
                        "FixedString(64)",  # sha256
                        "FixedString(40)",  # sha1
                        "FixedString(32)",  # md5
                        "UInt64",  # function_address
                        "UInt64",  # block_start_address
                        "UInt64",  # block_end_address
                        "UInt32",  # block_size
                        "UInt16",  # instructions_count
                        "FixedString(32)",  # block_instructions_hash (MD5)
                        "Array(UInt64)",  # predecessor_blocks
                        "Array(UInt64)",  # successor_blocks
                        "UInt16",  # depth
                        "UInt16",  # position
                        "Enum8('DIRECT'=1, 'CONDITIONAL'=2, 'CALL'=3, 'RETURN'=4, 'FALLTHROUGH'=5, 'INDIRECT'=6, 'UNKNOWN'=7)",  # branch_type
                        "Enum8('CODE'=1, 'DATA'=2, 'THUNK'=3)",  # block_type
                        "Array(String)",  # flags (EntryBlock, ExitBlock, LoopBlock)
                        "Array(UInt16)",  # dominators
                        "Array(UInt16)",  # post_dominators
                        "DateTime64(3, 'UTC')",  # analysis_date
                    ],
                },
                "function_analysis_errors": {
                    "table": "function_analysis_errors_binja",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            f["function_name"],
                            f["function_address"],
                            f.get("error_location", "unknown"),
                            f.get("error_message", ""),
                            f.get("error_details", ""),
                            f.get("error_type", "unknown"),
                            self.calculate_md5(
                                f"{f['error_message']}{f['function_name']}{f['function_address']}{f['error_location']}"
                            ),
                            "new",
                            now,
                        ]
                        for f in self.analysis_results["errors"]
                    ],
                    "column_names": [
                        "sha256",
                        "function_name",
                        "function_address",
                        "error_location",
                        "error_message",
                        "error_details",
                        "error_type",
                        "error_hash",
                        "status",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "Nullable(String)",
                        "UInt64",
                        "LowCardinality(String)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "FixedString(32)",
                        "Enum8('new'=1, 'investigating'=2, 'fixed'=3, 'wontfix'=4)",
                        "DateTime64(3, 'UTC')",
                    ],
                },
            }

    def tag(self) -> str:
        """Return the tag for this extractor."""
        return Tag.DECOMPILED.value

    def get_clickhouse_table(self) -> str:
        """Not used directly as we're handling multiple tables."""
        pass


if __name__ == "__main__":
    # Setup basic logging
    import logging

    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger("DecompileBinja")

    # Parse command line arguments
    import argparse

    parser = argparse.ArgumentParser(description="Binary Ninja Decompiler Wrapper")
    parser.add_argument("filepath", help="Path to the binary file to analyze")
    parser.add_argument(
        "--output", "-o", help="Output JSON file path (default: stdout)"
    )
    parser.add_argument(
        "--timeout",
        "-t",
        type=int,
        default=1200,
        help="Analysis timeout in seconds (default: 1200)",
    )
    args = parser.parse_args()

    # Create and run the extractor
    with DecompileBinja(args.filepath, logger) as extractor:
        success = extractor.extract()

        if not success:
            logger.error("Analysis failed")
            exit(1)

        # Output results
        if args.output:
            with open(args.output, "w") as f:
                json.dump(extractor.analysis_results, f)
            logger.info(f"Results written to {args.output}")
        else:
            print(json.dumps(extractor.analysis_results))