Han Shu

import os
import time
import json

from .analysis.medium_level import MediumLevelAnalysis

# disable the plugins set by user for binary ninja
os.environ["BN_DISABLE_USER_PLUGINS"] = "True"
import traceback

# Binary Ninja imports (conditional)
try:
    import binaryninja
    from binaryninja import mainthread, Symbol
    from binaryninja.enums import SymbolType
    BINARYNINJA_AVAILABLE = True
except Exception:
    BINARYNINJA_AVAILABLE = False
    binaryninja = None
    mainthread = None
    Symbol = None
    SymbolType = None

# Support both package and standalone imports
try:
    # Package import (when imported from redb)
    from .utils.hashes import calculate_sha256, calculate_tlsh
    from .utils.license import set_license
    from .utils.logging import setup_default_logger
    from .analysis.strings import StringAnalysis

    # Only import modules that depend on Binary Ninja when available
    if BINARYNINJA_AVAILABLE:
        from .analysis.cfg import CFGAnalysis
        from .analysis.disassembly import DisassemblyAnalysis
        from .analysis.low_level import LowLevelAnalysis
        from .arch.creator import ArchitectureCreator
        from .custom_options import register_custom_analysis_options
        from .function_type import FunctionTypeAnalysis, FunctionType
        from .analysis.scores import ObfuscationScores

except ImportError:
    # Fallback to absolute imports (for multiprocessing spawned processes)
    from redb.extractors.decompiler.bninja.utils.hashes import calculate_sha256, calculate_tlsh
    from redb.extractors.decompiler.bninja.utils.license import set_license
    from redb.extractors.decompiler.bninja.utils.logging import setup_default_logger

    # Only import modules that depend on Binary Ninja when available
    if BINARYNINJA_AVAILABLE:
        from redb.extractors.decompiler.bninja.analysis.cfg import CFGAnalysis
        from redb.extractors.decompiler.bninja.analysis.disassembly import DisassemblyAnalysis
        from redb.extractors.decompiler.bninja.analysis.low_level import LowLevelAnalysis
        from redb.extractors.decompiler.bninja.arch.creator import ArchitectureCreator
        from redb.extractors.decompiler.bninja.custom_options import register_custom_analysis_options
        from redb.extractors.decompiler.bninja.function_type import FunctionTypeAnalysis


class BinaryNinjaDecompiler:
    """A Binary Ninja-based decompiler that replicates the functionality of GhidraDecompilerScript.
    This class extracts decompiled code, disassembly with multiple normalization levels,
    and control flow graph information from binary files.
    """

    MIN_FUNCTION_SIZE = 10  # instructions
    MIN_BLOCK_SIZE = 4  # instructions
    INVALID_STACK_SIZE = -1

    def __init__(
        self,
        filepath,
        timeout,
        log=None,
        exporters=None,
        index_prefix=None,
        filetype=None,
        goresym=None,
        decompile_modules=None,
    ):
        """Initialize the Binary Ninja decompiler.

        Args:
            filepath: Path to the binary file to analyze
            log: Logger object (optional)
            timeout: Maximum time in seconds for analysis (default: 1200)
            exporters: List of exporters for the results (optional)
            index_prefix: Prefix for elastic index (optional)
            filetype: Type of the file (optional)

        """
        self.filepath = filepath
        self.log = log if log else setup_default_logger("BninjaDecompiler")
        self.BNINJA_TIMEOUT = timeout
        self.bv = None
        self.analysis_results = None
        self.errors = []
        self.exporters = exporters
        self.index_prefix = index_prefix
        self.filetype = filetype
        self.goresym = goresym
        self.decompile_modules = decompile_modules or {"all"}

        set_license(binaryninja)

        # Map to track instruction categorization
        mainthread.set_worker_thread_count(3)
        register_custom_analysis_options(binaryninja)

    def log_error(
        self, message, function_name, address, exception=None, error_location="unknown"
    ):
        """Log an error during processing."""
        error_msg = f"Error in function {function_name} at {address}: {message}"
        if exception:
            error_msg += f" - {str(exception)}"

        self.log.error(error_msg)

        # Add to errors list
        error = {
            "function_name": function_name,
            "function_address": str(address),
            "error_location": error_location,
            "error_message": message,
            "error_details": str(exception) if exception else "",
            "error_type": type(exception).__name__ if exception else "Unknown",
            "timestamp": int(time.time() * 1000),
        }
        self.errors.append(error)

    def __enter__(self):
        """Context manager entry point."""
        self.log.info(f"Opening binary file: {self.filepath}")
        binaryninja.BinaryViewType.add_binaryview_initial_analysis_completion_event(
            self.on_analysis_complete
        )

        #self.bv = binaryninja.load(self.filepath, update_analysis=False)
        self.bv = binaryninja.load(self.filepath, update_analysis=True)
        if self.bv is None:
            raise ValueError(f"Failed to open file: {self.filepath}")

        self.log.info("Waiting for analysis to complete...")

        self.log.debug(f"Binja analysis complete: {len(list(self.bv.functions))} functions")

        # set the architecture
        # todo: personalize this for other architectures
        self.arch = ArchitectureCreator("x86").get()

        # apply goresym
        if self.goresym is not None:
            self.__apply_goresym()
        return self

    def on_analysis_complete(self, bv):
        # Request an additional update after analysis is complete to ensure IL generation
        self.bv = bv
        self.bv.update_analysis()

        return

    def __apply_goresym(self):
        file = self.goresym
        data = None
        try:
            data = json.loads(open(file, 'r').read())
        except Exception as e:
            self.log_error(
                "Failed to open file from goresym: ",
                file,
                e,
                "analyze_binary",
            )

        if data is None:
            return

        self.bv.begin_undo_actions()
        if data.get('UserFunctions') is not None:
            user_functions = data['UserFunctions']
            for func in user_functions:
                try:
                    start = int(func['Start'])
                    name = func['FullName']
                    if self.bv.get_function_at(start) is None:
                        self.bv.create_user_function(start)

                    sym = Symbol(SymbolType.FunctionSymbol, start, name, name, name)
                    self.bv.define_user_symbol(sym)
                except Exception as e:
                    self.log.warning(f"Failed to apply GoReSym symbol for UserFunction {func.get('FullName', 'unknown')} at {func.get('Start', 'unknown')}: {e}")

        if data.get('StdFunctions') is not None:
            standard_functions = data['StdFunctions']
            for func in standard_functions:
                try:
                    start = int(func['Start'])
                    name = func['FullName']
                    if self.bv.get_function_at(start) is None:
                        self.bv.create_user_function(start)

                    sym = Symbol(SymbolType.FunctionSymbol, start, name, name, name)
                    self.bv.define_user_symbol(sym)
                except Exception as e:
                    self.log.warning(f"Failed to apply GoReSym symbol for StdFunction {func.get('FullName', 'unknown')} at {func.get('Start', 'unknown')}: {e}")

        self.bv.commit_undo_actions()
        return

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit point - clean up resources."""
        if self.bv:
            # Make sure to cancel any pending analysis
            # (if we have no pending analysis, binary ninja will log an error)

            # todo(@nicolo): investigate
            # if hasattr(self.bv, "abort_analysis"):
            #    self.bv.abort_analysis()
            self.bv.file.close()

        self.log.info("Cleanup completed successfully")
        return

    def tag(self):
        """Return the tag for this extractor."""
        return "DECOMPILED"

    def _module_selected(self, module_name):
        """Check if a decompiler sub-module is selected."""
        return "all" in self.decompile_modules or module_name in self.decompile_modules

    def analyze_binary(self):
        """Run Binary Ninja analysis and return results.

        Respects self.decompile_modules to selectively run/skip sub-modules:
        - strings: independent, skipped if not selected
        - decompilation: leaf module, skipped if not selected
        - disassembly: always runs (backbone — provides hash linkage for all others)
        - llil: leaf module, skipped if not selected
        - cfg: leaf module, skipped if not selected

        Only selected modules' results are appended to the results dict for DB insertion.
        Disassembly is always computed for linkage but only inserted when selected.
        """
        try:
            results = {
                "decompiled": [],
                "disassembled": [],
                "cfg": [],
                "llil": [],
                "errors": [],
                "strings": [],
                "mlil": []
            }

            run_all = "all" in self.decompile_modules
            run_strings = run_all or "strings" in self.decompile_modules
            run_decompilation = run_all or "decompilation" in self.decompile_modules
            run_disassembly = run_all or "disassembly" in self.decompile_modules
            run_llil = run_all or "llil" in self.decompile_modules
            run_cfg = run_all or "cfg" in self.decompile_modules

            # Determine if we need the per-function loop at all
            need_per_function = run_decompilation or run_disassembly or run_llil or run_cfg

            #functions_list = list(filter(is_not_ext_lib_function, self.bv.functions))
            functions_list = list(filter(is_lib_or_thunk, self.bv.functions))
            functions_list = list(filter(self.is_too_few_blocks, functions_list))

            # Strings extraction — independent of per-function analysis
            if run_strings:
                results["strings"] = StringAnalysis(self.bv, functions_list).analyze()

            if not need_per_function:
                return results

            for function in functions_list:
                try:
                    # Decompilation (HLIL) — leaf module, skip if not selected
                    hlil_json = self.extract_hlil(function) if run_decompilation else None

                    # Disassembly — always compute (provides hash linkage for others)
                    disass_json = self.extract_disasm(function)

                    # CFG — leaf module, skip if not selected
                    cfg_json = self.extract_cfg(function) if run_cfg else None

                    # LLIL — leaf module, skip if not selected
                    lowlevel_json = self.extract_lowlevel(function) if run_llil else None

                    # we run mlil only if we have cfg
                    mlil_json = self.extract_mediumlevel(function) if run_llil else None

                    # Calculate fuzzy hashes for disassembly using utils.hashes
                    if disass_json:
                        disass_no_addr = disass_json.get("disassembled_function_no_addresses", "")
                        disass_json["tlsh_disassembly"] = calculate_tlsh(disass_no_addr)

                    if hlil_json and disass_json:
                        hlil_json["disassembled_function_hash"] = disass_json[
                            "disassembled_function_hash"
                        ]
                        disass_json["decompiled_function_hash"] = hlil_json[
                            "decompiled_function_hash"
                        ]

                        results["decompiled"].append(hlil_json)
                        if run_disassembly:
                            results["disassembled"].append(disass_json)

                    elif hlil_json:
                        hlil_json["disassembled_function_hash"] = None
                        results["decompiled"].append(hlil_json)
                    elif disass_json:
                        disass_json["decompiled_function_hash"] = None
                        if run_disassembly:
                            results["disassembled"].append(disass_json)

                    # Add bi-directional linkage between LLIL and disassembly with fuzzy hashes
                    # LLIL fuzzy hashes (tlsh_llil) are already calculated in lowlevel_json
                    if lowlevel_json and disass_json:
                        # Add disassembly info to LLIL
                        lowlevel_json["disassembled_function_hash"] = disass_json["disassembled_function_hash"]
                        lowlevel_json["tlsh_disassembly"] = disass_json.get("tlsh_disassembly")

                        # Add LLIL fuzzy hashes to disassembly for easy export access
                        disass_json["tlsh_llil"] = lowlevel_json.get("tlsh_llil")
                        disass_json["minhash"] = lowlevel_json.get("minhash")

                    elif lowlevel_json:
                        lowlevel_json["disassembled_function_hash"] = None
                        lowlevel_json["tlsh_disassembly"] = None
                    elif disass_json:
                        # No LLIL available for this disassembly
                        disass_json["tlsh_llil"] = None
                        disass_json["minhash"] = None

                    if lowlevel_json:
                        results["llil"].append(lowlevel_json)
                        results["mlil"].append(mlil_json)

                    # Add CFG linkage with disassembled_function_hash
                    # Also add cyclomatic_complexity to disass_json for similarity metrics export
                    if cfg_json and disass_json:
                        cfg_json["disassembled_function_hash"] = disass_json["disassembled_function_hash"]
                        disass_json["cyclomatic_complexity"] = cfg_json.get("cyclomatic_complexity")
                        results["cfg"].append(cfg_json)
                    elif cfg_json:
                        cfg_json["disassembled_function_hash"] = None
                        results["cfg"].append(cfg_json)

                    # Ensure cyclomatic_complexity is set even if no cfg_json
                    if disass_json and "cyclomatic_complexity" not in disass_json:
                        disass_json["cyclomatic_complexity"] = None

                except Exception as e:
                    self.log_error(
                        "Failed to process function: ",
                        function.name,
                        function.start,
                        e,
                        "analyze_binary",
                    )

            return results

        except Exception as e:
            self.log.error(f"Error in binary analysis: {str(e)}")
            return None

    def extract_mediumlevel(self, function):
        middle_level = MediumLevelAnalysis(function, self.bv, self.log)
        middle_level_result, errors = middle_level.analyze()

        for error in errors:
            self.errors.append(error)

        return middle_level_result


    def extract_lowlevel(self, function):
        low_level = LowLevelAnalysis(function, self.bv, self.log)
        disassembly_json, errors = low_level.analyze()

        for error in errors:
            self.errors.append(error)

        return disassembly_json

    def extract_cfg(self, function):
        try:
            llil = function.llil if hasattr(function, 'llil') else None
            cfg = CFGAnalysis(function, llil_function=llil).extract_function_cfg()
            return cfg
        except Exception as e:
            self.log_error(
                "Fatal error in CFG extraction",
                function.name,
                function.start,
                e,
                "extract_disassembly",
            )
            return None

    def extract_disasm(self, function):
        try:
            disass_analysis = DisassemblyAnalysis(
                self.arch, function, self.bv, self.log
            )

            # Create disassembly JSON
            disassembly_json, errors = disass_analysis.get_json()

            for error in errors:
                self.errors.append(error)

            return disassembly_json

        except Exception as e:
            self.log_error(
                "Fatal error in disassembly extraction",
                function.name,
                function.start,
                e,
                "extract_disassembly",
            )
            return None

    def extract_hlil(self, function):
        """Extract HLIL from a function."""
        try:
            # Access function.hlil directly - this will either return the HLIL or raise an exception
            # Removed hlil_if_available check as it was causing race conditions
            if function.hlil is None:
                return None

            if len(function.hlil.basic_blocks) == 0:
                return None

            # if function has one basic block, then compare the len of the instructions against minimum of our functions
            if len(function.hlil.basic_blocks) == 1:
                block = function.basic_blocks[0]
                if len(list(block.disassembly_text)) < self.MIN_FUNCTION_SIZE:
                    return None

            # Get decompiled code
            function_prototype = str(function)
            decompiled_code = str(function.hlil)
            if not decompiled_code or decompiled_code.strip() == "":
                self.log.warning(f"Empty decompilation result for {function.name}")
                return None

            callers = []
            for caller_site in function.caller_sites:
                if caller_site.hlil:
                    callers.append(str(caller_site.hlil))

            calls = []
            for call in function.call_sites:
                if call.hlil:
                    calls.append(str(call.hlil))

            analysis_score = ObfuscationScores(function.hlil)
            flattened_score = analysis_score.flattened_score()
            mba_score = analysis_score.MBA_score()

            if decompiled_code:
                # Create json object
                function_json = {
                    "decompiled_function_hash": calculate_sha256(decompiled_code),
                    "decompiled_function": decompiled_code,
                    "decompiled_function_name": function.name,
                    "decompiled_function_prototype": function_prototype,
                    "decompiled_function_address": function.start,
                    "function_type": FunctionTypeAnalysis(function)
                    .get_function_type()
                    .name,
                    "functions_caller": list(callers),
                    "functions_call": list(calls),
                    "flattened_score": flattened_score,
                    "mba_score": mba_score,
                }
                return function_json
            else:
                return None

        except Exception as e:
            self.log.warning(
                f"Failed to get HLIL for {function.name} at {function.start}: {str(e)}"
            )

        return None

    def extract(self) -> bool:
        """Extract and process all analysis results.

        Returns:
            bool: True if extraction was successful, False otherwise

        """
        self.log.info(f"Starting binary analysis on {self.filepath}")
        try:
            results = self.analyze_binary()
            if not results:
                self.log.error("Analysis failed to produce results")
                return False

            self.analysis_results = results
            self.log.info(
                f"Successfully analyzed binary: {len(results['decompiled'])} decompiled functions, "
                f"{len(results['disassembled'])} disassembled functions, "
                #f"{len(results['cfg'])} basic blocks, "
                f"{len(results['errors'])} errors"
            )
            return True

        except Exception as e:
            self.log.error(f"Error in extraction: {str(e)}")
            return False

    def is_too_few_blocks(self, function):
        if function is None:
            return False

        if function.basic_blocks is None:
            return False

        # when binary ninja does not wait for the analysis, it creates function stubs where basic blocks array
        # is not populated yet. Therefore, we disable this heuristic.
        #if len(function.basic_blocks) == 0:
        #    return False

        # if function has one basic block, then compare the len of the instructions against minimum of our functions
        if len(function.basic_blocks) == 1:
            block = function.basic_blocks[0]
            if len(list(block.disassembly_text)) < self.MIN_FUNCTION_SIZE:
                return False

        return True

def is_lib_or_thunk(function):
    function_type = FunctionTypeAnalysis(function).get_function_type()
    return not (function_type == FunctionType.THUNK or function_type == FunctionType.EXTERNAL or function_type == FunctionType.LIBRARY)