hlil_decoder.py

#!/usr/bin/env python
import os.path
import logging
from termcolor import colored
from io import BytesIO

from binaryninja import * 
log_to_stderr(LogLevel.ErrorLog)
print(core_version())

h = "5c9b30d502e2f103f089607ce699520f88154e3d7988a9db801f2a2a4378bf41"

if os.path.exists(f"./testcases/{h}.bndb"):
    print("Loading existing DB")
    fm = binaryninja.FileMetadata()
    db = fm.open_existing_database(f"./testcases/{h}.bndb")

    if db is None:
        raise Exception("Database doesn't exist")

    bv = db.get_view_of_type('PE') or db.get_view_of_type('ELF')
elif os.path.exists(f"./testcases/{h}"):
    print("Creating DB")
    bv = binaryninja.BinaryViewType.get_view_of_file_with_options(f"./testcases/{h}")
else:
    raise Exception("File doesn't exist...")

morestack_noctxt_sym = bv.get_symbols_by_name("runtime.morestack_noctxt") or bv.get_symbols_by_name("_runtime.morestack_noctxt") or bv.get_symbols_by_name("runtime_morestack_noctxt") or bv.get_symbols_by_name("_runtime_morestack_noctxt")
slicebytetostring_sym = bv.get_symbols_by_name("runtime.slicebytetostring") or bv.get_symbols_by_name("_runtime.slicebytetostring") or bv.get_symbols_by_name("runtime_slicebytetostring") or bv.get_symbols_by_name("_runtime_slicebytetostring")

morestack_noctxt = bv.get_function_at(morestack_noctxt_sym[0].address)
slicebytetostring = bv.get_function_at(slicebytetostring_sym[0].address)

deobfuscate_candidates = set()

print("{} functions".format(len(bv.functions)))

for func in bv.functions:
    func_callees = func.callees
    if (len(func_callees)) == 2:
        if morestack_noctxt in func_callees and slicebytetostring in func_callees:
            print(func.name)
            for ins in func.instructions:
                ins_type = ins[0][0]
                if ins_type.text != "xor":
                    continue
                src = ins[0][2]
                dst  = ins[0][4]
                if src == "eax" and dst == "eax":
                    continue
                
                deobfuscate_candidates.add(func)
                break

logging.getLogger().setLevel(logging.INFO)

def handle_deref(bv, deref_size, deref_addr):
    br = binaryninja.BinaryReader(bv)
    br.seek(deref_addr)
    data = br.read(deref_size)
    return data

for candidate in deobfuscate_candidates:
    logging.debug(candidate.name)
    
    instrn = candidate.hlil.instructions
    
    # string based state machine goes brrrrrrrr
    state = "start"
    assigning_to = None
    data = BytesIO()
    things = []
    for ins in instrn:
        logging.debug("{} {} | {}".format(hex(ins.address), colored(ins, "red"), colored(ins.operation, "green")))

        if state == "loading" and (ins.operation != binaryninja.HighLevelILOperation.HLIL_DEREF and ins.operation != binaryninja.HighLevelILOperation.HLIL_ASSIGN):
            logging.debug("Finished")
            state = "start"
            data.seek(0)
            things.append(data.getvalue())
            data = BytesIO()
        
        if state == "start" and ins.operation == binaryninja.HighLevelILOperation.HLIL_VAR_INIT:
            if ins.operands[1].value == 0:
                continue
            
            state = "loading"
            logging.debug("Init")
            logging.debug(ins.dest.storage)
            if ins.src.operation == binaryninja.HighLevelILOperation.HLIL_CONST:
                const_len = ins.src.expr_type.width
                logging.debug("Init Const len:{}".format(const_len))

                #print(ins.src.value)
                #print(ins.src.value.value)
                #print(ins.src.expr_type)
                #print(const_len)
                #print(ins.src.value.value)
                #print(type(ins.src.value.value))
                #print(ins.src.expr_type.signed)
                data.write(ins.src.value.value.to_bytes(const_len, byteorder='little', signed=ins.src.value.value < 0))
            elif ins.src.operation == HighLevelILOperation.HLIL_DEREF:
                logging.debug("Init Deref")
                #print(ins.dest)
                deref_size = ins.src.expr_type.width
                deref_addr = ins.src.src.value.value
                logging.debug("Dereference {} bytes from {}".format(deref_size, hex(deref_addr)))
                data.write(handle_deref(bv, deref_size, deref_addr))
            else:
                raise Exception("unhandled VAR_INIT src load")

            
        elif state == "loading" and ins.operation == binaryninja.HighLevelILOperation.HLIL_DEREF:
            deref_size = ins.src.expr_type.return_value.width
            deref_addr = ins.src.value.value
            logging.debug("HLIL_DEREF Dereference {} bytes from {}".format(deref_size, hex(deref_addr)))
            #print(ins.)
            data.write(handle_deref(bv, deref_size, deref_addr))
            
        elif state == "loading" and ins.operation == binaryninja.HighLevelILOperation.HLIL_ASSIGN:
            # TODO: Catch an assign after a deref, remove this hack and do it properly
            if ins.operands[1].value == 0:
                continue
            logging.debug("Assign")
            
            if ins.dest.operation == binaryninja.HighLevelILOperation.HLIL_STRUCT_FIELD:
                offset = ins.dest.offset
                logging.debug("OFFSET YO")
                data.seek(offset)

            if ins.src.operation == binaryninja.HighLevelILOperation.HLIL_CONST:
                data.write(ins.src.value.value.to_bytes(const_len, byteorder='little', signed=ins.src.value.value < 0))
            elif ins.src.operation == HighLevelILOperation.HLIL_DEREF:
                deref_size = ins.src.expr_type.width
                deref_addr = ins.src.src.value.value
                logging.debug("HLIL_ASSIGN Dereference {} bytes from {}".format(deref_size, hex(deref_addr)))
                data.write(handle_deref(bv, deref_size, deref_addr))
            else:
                raise Exception("unhandled ASSIGN load")
            
            # We might have been writing to an offset so let's jump to the end to counter-act that
            data.seek(data.getbuffer().nbytes)
    
    if len(things) != 2:
        logging.info("{} = {}".format(colored(candidate.name, "yellow"), colored("failed", "red")))
    else:
        thing1 = things[0]
        thing2 = things[1]
        logging.info("{} = {}".format(colored(candidate.name, "yellow"), colored(bytes(a ^ b for (a, b) in zip(thing1, thing2)), "white")))
    logging.info("----------------------------")