Kaixiang Peng

import time

from binaryninja import (
    LowLevelILInstruction,
)
from binaryninja import (
    LowLevelILOperation as LLIL_OP,
)
from binaryninja.lowlevelil import (
    LowLevelILAdd,
    LowLevelILConst,
    LowLevelILConstPtr,
    LowLevelILLoad,
    LowLevelILLsl,
    LowLevelILMul,
    LowLevelILPop,
    LowLevelILPush,
    LowLevelILReg,
    LowLevelILStore,
    LowLevelILSub,
)

# Support both package and standalone imports
try:
    from ..function_type import FunctionTypeAnalysis
    from ..similarity.minhasher import MinHasher, TokenKind
    from ..utils.hashes import calculate_sha256, calculate_tlsh
    from .low_level_normalization import LowLevelNormalization

except ImportError:
    # Fallback to absolute imports (for multiprocessing spawned processes)
    from redb.extractors.decompiler.bninja.analysis.low_level_normalization import LowLevelNormalization
    from redb.extractors.decompiler.bninja.similarity.minhasher import MinHasher
    from redb.extractors.decompiler.bninja.function_type import FunctionTypeAnalysis
    from redb.extractors.decompiler.bninja.utils.hashes import calculate_sha256, calculate_tlsh

class LowLevelAnalysis:
    def __init__(self, function, bv, logger):
        self.function = function
        self.name = function.name
        self.start = function.start
        self.llil_func = function.llil
        self.bv = bv
        self.logger = logger
        self.errors = []

    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.logger.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 count_control_flow_instructions(self):
        if self.llil_func is None:
            return 0

        count = 0
        for basic_block in self.llil_func.basic_blocks:
            for ins in basic_block:
                op = ins.operation
                if op in (
                    LLIL_OP.LLIL_IF,
                    LLIL_OP.LLIL_GOTO,
                    LLIL_OP.LLIL_JUMP,
                    LLIL_OP.LLIL_JUMP_TO,
                    LLIL_OP.LLIL_CALL,
                    LLIL_OP.LLIL_CALL_SSA,
                ):
                    count += 1

        return count

    def collect_memory_patterns(self):
        """ """
        patterns = set()

        try:
            llil = self.llil_func
            arch = self.bv.arch
            sp_name = arch.stack_pointer if arch and arch.stack_pointer else "sp"

            def analyze_addr(addr_expr, might_be_direct):
                """
                Visit the expression for the address and understand whether it has a direct, scaled, base offset, etc.
                access to memory
                """
                found = {
                    "direct": False,
                    "scaled": False,
                    "base_off": False,
                    "stack": False,
                    "string": False,
                }

                def addr_cb(n):
                    # stack (SP/BP-like)
                    match n:
                        case LowLevelILReg(src=reg):
                            if reg == sp_name:
                                found["stack"] = True

                        case LowLevelILConst() | LowLevelILConstPtr():
                            if might_be_direct:
                                found["direct"] = True

                        # base +/- const
                        case (
                            LowLevelILAdd(left=l, right=r)
                            | LowLevelILSub(left=l, right=r)
                        ):
                            l_is_reg = isinstance(l, LowLevelILReg)
                            r_is_reg = isinstance(r, LowLevelILReg)
                            l_is_cst = isinstance(
                                l, (LowLevelILConst, LowLevelILConstPtr)
                            )
                            r_is_cst = isinstance(
                                r, (LowLevelILConst, LowLevelILConstPtr)
                            )
                            if (l_is_reg and r_is_cst) or (r_is_reg and l_is_cst):
                                found["base_off"] = True

                        # scaled index (index*scale) or shift (index << k)
                        case LowLevelILMul(left=l, right=r):
                            if (
                                isinstance(l, LowLevelILReg)
                                and isinstance(r, LowLevelILConst)
                            ) or (
                                isinstance(r, LowLevelILReg)
                                and isinstance(l, LowLevelILConst)
                            ):
                                found["scaled"] = True

                        case LowLevelILLsl(left, right):
                            if isinstance(left, LowLevelILReg) and isinstance(
                                right, LowLevelILConst
                            ):
                                found["scaled"] = True

                    return None

                _ = list(addr_expr.traverse(addr_cb))

                if found["direct"]:
                    patterns.add("MEM_DIRECT")
                if found["scaled"]:
                    patterns.add("MEM_SCALED_INDEX")
                if found["base_off"]:
                    patterns.add("MEM_BASE_OFFSET")
                if found["stack"]:
                    patterns.add("MEM_STACK")
                elif found["string"]:
                    patterns.add("MEM_STRING")

            def func_cb(i):
                match i:
                    case LowLevelILPush(src=addr):
                        analyze_addr(addr, False)
                    case LowLevelILPop(src=addr):
                        analyze_addr(addr, False)
                    case LowLevelILLoad(src=addr):
                        analyze_addr(addr, True)
                    case LowLevelILStore(dest=addr, src=_):
                        analyze_addr(addr, True)

                return None

            # complete visit for the single instruction
            _ = list(llil.traverse(func_cb))

            return sorted(patterns)

        except Exception as e:
            self.log_error(
                "Failed to collect memory patterns via LLIL.traverse",
                self.name,
                self.start,
                e,
                "collect_memory_patterns",
            )
            return []

    def collect_register_usage(self):
        """
        Collect frequencies for register usage
        """
        try:
            llil = self.llil_func

            register_usage = {}

            def inc(reg, kind):
                if reg is None:
                    return
                entry = register_usage.setdefault(reg, {"reads": 0, "writes": 0})
                entry[kind] += 1

            if not llil:
                return {}, 0, 0

            for top_il in llil.instructions:
                registers_read = self.function.get_regs_read_by(
                    top_il.address, self.bv.arch
                )
                registers_write = self.function.get_regs_written_by(
                    top_il.address, self.bv.arch
                )

                for reg_read in registers_read:
                    inc(reg_read, "reads")

                for reg_write in registers_write:
                    inc(reg_write, "writes")

            total_reads = sum(entry["reads"] for entry in register_usage.values())
            total_writes = sum(entry["writes"] for entry in register_usage.values())

            return register_usage, total_reads, total_writes

        except Exception as e:
            self.log_error(
                "Failed to collect register usage via LLIL.traverse",
                self.name,
                self.start,
                e,
                "collect_register_usage",
            )
            return {}, 0, 0

    def _classify_address(self, bv, addr):
        """
        Classification of the address
        """
        info = {
            "address": addr,
            "section": None,
            "segment_writable": None,
            "symbol": None,
            "kind": None,  # "string", "function_ptr", "data_var", "symbol", "unknown"
            "datatype": None,  # es. "char *", "int32_t", "my_struct", ...
            "note": None,
        }

        # section / segment
        sec = bv.get_section_at(addr)
        seg = bv.get_segment_at(addr)
        if sec:
            info["section"] = sec.name
        if seg:
            info["segment_writable"] = bool(seg.writable)

        sym = bv.get_symbol_at(addr)
        if sym:
            info["symbol"] = sym.full_name

        # function pointer
        try:
            fns = list(bv.get_functions_at(addr))
        except Exception:
            # some versions have get_function_at(addr) that returns a single object or None
            fns = [bv.get_function_at(addr)] if hasattr(bv, "get_function_at") else []
        fns = [f for f in fns if f]
        if fns:
            info["kind"] = "function_ptr"
            info["datatype"] = "func"
            info["note"] = f"points to function {fns[0].name}"
            return info

        # string
        sref = bv.get_string_at(addr)
        if sref:
            info["kind"] = "string"
            # sref.type:
            info["datatype"] = (
                getattr(sref, "type", None).__class__.__name__
                if hasattr(sref, "type")
                else "string"
            )
            return info

        # data typed variable
        dv = bv.get_data_var_at(addr)
        if dv:
            info["kind"] = "data_var"
            info["datatype"] = str(dv.type) if getattr(dv, "type", None) else None
            if getattr(dv, "name", None):
                info["symbol"] = dv.name if not info["symbol"] else info["symbol"]
            return info

        # only symbol (no data var)
        if sym and not info["kind"]:
            info["kind"] = "symbol"
            return info

        # unknown
        info["kind"] = "unknown"
        return info

    def count_data_references(self):
        """Count the number of data references in a function using LLIL."""
        count = 0

        if self.llil_func is None:
            return 0

        try:
            # Iterate LLIL basic blocks and instructions
            for instr in self.llil_func.instructions:
                instr_str = str(instr)
                logged = False
                src = None

                # Check for constant dereferencing or symbolic refs
                if hasattr(instr, "src"):
                    src = instr.src
                    if isinstance(src, (LowLevelILConstPtr, LowLevelILConst)):
                        count += 1
                        logged = True

                if not logged and "_" in instr_str:
                    count += 1

                # If src is a pointer constant, check if it lands in a writable data segment
                if src is not None and isinstance(src, LowLevelILConstPtr):
                    addr = src.constant
                    segment = self.bv.get_segment_at(addr)
                    if segment and segment.writable:
                        count += 1

        except Exception as e:
            self.logger.warning(
                f"Failed to use LLIL for counting data references in "
                f"{self.name} at {self.start}: {e}"
            )
        return count

    def compute_num_calls(self):
        c = 0

        if not self.llil_func:
            return 0

        for instr in self.llil_func.instructions:
            if instr.operation in (LLIL_OP.LLIL_CALL, LLIL_OP.LLIL_TAILCALL):
                c += 1
        return c

    def compute_max_block_size(self):
        """Compute the maximum basic block size in a function."""
        if self.llil_func is None:
            return 0

        max_size = 0
        for block in self.llil_func.basic_blocks:
            try:
                # Count instructions in this block using the direct length approach
                # This avoids UTF-8 decoding issues entirely
                block_size = block.instruction_count
                max_size = max(max_size, block_size)
            except Exception as e:
                self.log_error(
                    f"[HandledError] computing max block size: {e}",
                    self.name,
                    self.start,
                    e,
                    "compute_max_block_size",
                )
        return max_size

    def estimate_stack_size(self):
        """Estimate the stack size used by a function."""
        try:
            # Binary Ninja provides a stack adjustment value for functions
            # Need to convert OffsetWithConfidence to a plain integer
            stack_adjust = self.function.stack_adjustment
            if hasattr(stack_adjust, "value"):  # Handle OffsetWithConfidence objects
                return stack_adjust.value
            return stack_adjust
        except Exception as e:
            self.log_error(
                "Failed to estimate stack size",
                self.name,
                self.start,
                e,
                "estimate_stack_size",
            )
            return -1

    def collect_instruction_types(self):
        def iter_llil_tree(root_il):
            stack = [root_il]
            while stack:
                il_single_op = stack.pop()
                if not isinstance(il_single_op, LowLevelILInstruction):
                    continue
                yield il_single_op
                for il_operand in il_single_op.operands:
                    if isinstance(il_operand, LowLevelILInstruction):
                        stack.append(il_operand)
                    elif isinstance(il_operand, (list, tuple)):
                        for sub in il_operand:
                            if isinstance(sub, LowLevelILInstruction):
                                stack.append(sub)

        type_frequencies = {}
        try:
            if self.llil_func is None:
                return type_frequencies

            il_func = self.llil_func

            for top_il in il_func.instructions:
                for il in iter_llil_tree(top_il):
                    op = getattr(il, "operation", None)
                    if op is None:
                        continue

                    category = str(op)

                    if category in type_frequencies:
                        type_frequencies[category] += 1
                    else:
                        type_frequencies[category] = 1

        except Exception as e:
            self.log_error(
                "Failed to collect LLIL instruction types",
               self.name,
                self.start,
                e,
                "collect_instruction_types_llil",
            )

        return type_frequencies

    def _collect_low_level_with_type(self):
        llil = self.llil_func
        if not llil:
            return [], []

        start = self.start

        low_level = LowLevelNormalization()

        instrs_with_addr = []

        for il in llil.instructions:
            norm = low_level.normalize_instr_with_operands(il)

            # Clamp negative offsets to 0 for UInt32 compatibility.
            # Negative offsets (instruction before function start) may occur with
            # overlapping functions or tail-calls in obfuscated/malware binaries.
            # Multiple instructions at offset 0 indicates this anomaly and can be
            # queried to identify such samples easily than by checking logs.
            # Triggered by 590ecad54cd9e1c8681509420ad56edde8b064ffbf884ce6cd8dd28eebb95ae1
            offset = il.address - start
            if offset < 0:
                offset = 0

            instrs_with_addr.append((offset, norm))

        return instrs_with_addr

    def _collect_low_level_and_with_addr(self):
        llil = self.llil_func
        if not llil:
            return [], []

        start = self.start

        low_level = LowLevelNormalization()

        instrs = []
        instrs_with_addr = []

        for il in llil.instructions:
            norm = low_level.normalize_instruction_all_levels(il)

            instrs.append(norm)

            # Clamp negative offsets to 0 for UInt32 compatibility.
            # Negative offsets (instruction before function start) may occur with
            # overlapping functions or tail-calls in obfuscated/malware binaries.
            # Multiple instructions at offset 0 indicates this anomaly and can be
            # queried to identify such samples easily than by checking logs.
            # Triggered by 590ecad54cd9e1c8681509420ad56edde8b064ffbf884ce6cd8dd28eebb95ae1
            offset = il.address - start
            if offset < 0:
                offset = 0

            instrs_with_addr.append((offset, norm))

        return instrs, instrs_with_addr


    def analyze(self):
        registers_uses, total_reads, total_written = self.collect_register_usage()
        instr_low_level, body_llil_vector = self._collect_low_level_and_with_addr()
        instr_low_level_str = str(instr_low_level)
        instructions_low_level = calculate_sha256(instr_low_level_str)
        instructions_low_level_tlsh = calculate_tlsh(instr_low_level_str)

        instr_typed_llil = self._collect_low_level_with_type()
        instr_typed_low_level_str = str(instr_typed_llil)
        instructions_typed_low_level = calculate_sha256(instr_typed_low_level_str)
        instructions_typed_low_level_tlsh = calculate_tlsh(instr_typed_low_level_str)

        seed = 0xdeadbeef
        minhash_llil_skeleton = MinHasher(seed, self.llil_func, TokenKind.LLIL).calculateMinHash()
        minhash_llil_typed = MinHasher(seed, self.llil_func, TokenKind.TYPED_LLIL).calculateMinHash()

        low_level_json = {
            "function_address": self.start,
            "function_type": FunctionTypeAnalysis(self.function)
            .get_function_type()
            .name,
            "body_llil_vector": body_llil_vector,
            "sha256_llil": instructions_low_level,
            "tlsh_llil": instructions_low_level_tlsh,
            "minhash_llil_skeleton": minhash_llil_skeleton,
            "instructions_types_llil": list(self.collect_instruction_types()),
            "instruction_typed_llil": instructions_typed_low_level,
            "tlsh_instruction_typed_llil": instructions_typed_low_level_tlsh,
            "minhash_llil_typed": minhash_llil_typed,
            "control_flow_count_llil": self.count_control_flow_instructions(),
            "memory_access_pattern_llil": self.collect_memory_patterns(),
            "register_usage": registers_uses,
            "total_reg_reads": total_reads,
            "total_reg_written": total_written,
            "data_references_count": self.count_data_references(),
            "max_block_size": self.compute_max_block_size(),
            "num_calls": self.compute_num_calls(),
            "stack_size": self.estimate_stack_size(),
        }

        return low_level_json, self.errors