Mandicou Ba

import time

from binaryninja import (
    MediumLevelILOperation as MLIL_OP,
)

try:
    from ..function_type import FunctionTypeAnalysis
    from ..similarity.minhasher import MinHasher, TokenKind
    from ..utils.hashes import calculate_sha256, calculate_tlsh
    from .medium_level_normalization import MediumLevelNormalization
except ImportError:
    from redb.extractors.decompiler.bninja.analysis.medium_level_normalization import MediumLevelNormalization
    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


_MLIL_CALL_OPS = (
    MLIL_OP.MLIL_CALL,
    MLIL_OP.MLIL_CALL_SSA,
    MLIL_OP.MLIL_CALL_UNTYPED,
    MLIL_OP.MLIL_CALL_UNTYPED_SSA,
    MLIL_OP.MLIL_TAILCALL,
    MLIL_OP.MLIL_TAILCALL_SSA,
    MLIL_OP.MLIL_TAILCALL_UNTYPED,
    MLIL_OP.MLIL_TAILCALL_UNTYPED_SSA,
)

_MLIL_CONTROL_FLOW_OPS = (
    MLIL_OP.MLIL_IF,
    MLIL_OP.MLIL_GOTO,
    MLIL_OP.MLIL_JUMP,
    MLIL_OP.MLIL_JUMP_TO,
    MLIL_OP.MLIL_RET,
    MLIL_OP.MLIL_RET_HINT,
    MLIL_OP.MLIL_NORET,
) + _MLIL_CALL_OPS


class MediumLevelAnalysis:
    def __init__(self, function, bv, logger):
        self.function = function
        self.name = function.name
        self.start = function.start
        self.mlil_func = function.mlil
        self.bv = bv
        self.logger = logger
        self.errors = []

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

        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 _collect_mlil_skeleton_and_typed(self):
        mlil = self.mlil_func
        if not mlil:
            return [], [], [], []

        start = self.start
        norm = MediumLevelNormalization()

        skeleton = []
        skeleton_with_addr = []
        typed = []
        typed_with_addr = []

        for il in mlil.instructions:
            skel_norm = norm.normalize_instruction_all_levels(il)
            typed_norm = norm.normalize_instr_with_operands(il)

            skeleton.append(skel_norm)
            typed.append(typed_norm)

            offset = il.address - start
            if offset < 0:
                offset = 0

            skeleton_with_addr.append((offset, skel_norm))
            typed_with_addr.append((offset, typed_norm))

        return skeleton, skeleton_with_addr, typed, typed_with_addr

    def analyze(self):
        (
            instr_skeleton,
            body_mlil_skeleton_vector,
            instr_typed,
            body_mlil_typed_vector,
        ) = self._collect_mlil_skeleton_and_typed()

        instr_skeleton_str = str(instr_skeleton)
        sha256_skeleton = calculate_sha256(instr_skeleton_str)
        tlsh_skeleton = calculate_tlsh(instr_skeleton_str)

        instr_typed_str = str(instr_typed)
        sha256_typed = calculate_sha256(instr_typed_str)
        tlsh_typed = calculate_tlsh(instr_typed_str)

        seed = 0xdeadbeef
        minhash_mlil_skeleton = MinHasher(seed, self.mlil_func, TokenKind.MLIL).calculateMinHash()
        minhash_mlil_typed = MinHasher(seed, self.mlil_func, TokenKind.TYPED_MLIL).calculateMinHash()

        medium_level_json = {
            "function_address": self.start,
            "body_mlil_skeleton_vector": body_mlil_skeleton_vector,
            "sha256_mlil_skeleton": sha256_skeleton,
            "tlsh_mlil_skeleton": tlsh_skeleton,
            "minhash_mlil_skeleton": minhash_mlil_skeleton,
            "body_mlil_typed_vector": body_mlil_typed_vector,
            "sha256_mlil_typed": sha256_typed,
            "tlsh_mlil_typed": tlsh_typed,
            "minhash_mlil_typed": minhash_mlil_typed,
        }

        return medium_level_json, self.errors