mirror of
https://github.com/mrphrazer/r2con2021_deobfuscation
synced 2026-08-17 18:26:03 -04:00
152 lines
4.5 KiB
Python
152 lines
4.5 KiB
Python
#!/usr/bin/python3
|
|
import sys
|
|
from miasm.analysis.binary import Container
|
|
from miasm.analysis.machine import Machine
|
|
from miasm.core.locationdb import LocationDB
|
|
from miasm.expression.expression import *
|
|
from miasm.expression.simplifications import expr_simp
|
|
from miasm.ir.symbexec import SymbolicExecutionEngine
|
|
|
|
|
|
# hardcoded list of VM handlers taken from the binary
|
|
VM_HANDLERS = set([
|
|
0x129e,
|
|
0x1238,
|
|
0x126d,
|
|
0x11c4,
|
|
0x1262,
|
|
0x11a9,
|
|
0x1245,
|
|
0x11f1,
|
|
0x11e1,
|
|
0x1281,
|
|
0x1226,
|
|
])
|
|
|
|
|
|
def constraint_memory(address, num_of_bytes):
|
|
"""
|
|
Reads `num_of_bytes` from the binary at a given address
|
|
and builds symbolic formulas to pre-configure the symbolic
|
|
execution engine for concolinc execution.
|
|
"""
|
|
global container
|
|
# read bytes from binary
|
|
byte_stream = container.bin_stream.getbytes(address, num_of_bytes)
|
|
# build symbolic memory address
|
|
sym_address = ExprMem(ExprInt(address, 64), num_of_bytes * 8)
|
|
# build symbolic memory value
|
|
sym_value = ExprInt(int.from_bytes(
|
|
byte_stream, byteorder='little'), num_of_bytes * 8)
|
|
|
|
return sym_address, sym_value
|
|
|
|
def disassemble(sb, address):
|
|
"""
|
|
Callback to dump individual VM handler information,
|
|
execution context etc.
|
|
"""
|
|
# fetch concrete value of current virtual instruction pointer
|
|
vip = sb.symbols[ExprId("RDX", 64)]
|
|
|
|
# catch the individual handlers and print execution context
|
|
if int(address) == 0x129e:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x1238:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x126d:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x11c4:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x1262:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x11a9:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x1245:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x11f1:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x11e1:
|
|
# calculate address for bytecode
|
|
# address = expr_simp(vip + ExprInt(1, 64))
|
|
# read from bytecode
|
|
# constant = expr_simp(sb.symbols[ExprMem(address, 32)])
|
|
# print(f"{vip}: PUSH {constant}")
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x1281:
|
|
print(f"{vip}: handler {address}")
|
|
elif int(address) == 0x1226:
|
|
print(f"{vip}: handler {address}")
|
|
|
|
|
|
# check arguments
|
|
if len(sys.argv) != 2:
|
|
print(f"[*] Syntax: {sys.argv[0]} <file>")
|
|
exit()
|
|
|
|
# parse file path
|
|
file_path = sys.argv[1]
|
|
|
|
# address of vm entry
|
|
start_addr = 0x115a
|
|
|
|
# init symbol table
|
|
loc_db = LocationDB()
|
|
# read binary file
|
|
container = Container.from_stream(open(file_path, 'rb'), loc_db)
|
|
# get CPU abstraction
|
|
machine = Machine(container.arch)
|
|
# disassembly engine
|
|
mdis = machine.dis_engine(container.bin_stream, loc_db=loc_db)
|
|
|
|
# initialize lifter to intermediate representation
|
|
lifter = machine.lifter_model_call(mdis.loc_db)
|
|
|
|
# disassemble the function at address
|
|
asm_cfg = mdis.dis_multiblock(start_addr)
|
|
|
|
# translate asm_cfg into ira_cfg
|
|
ira_cfg = lifter.new_ircfg_from_asmcfg(asm_cfg)
|
|
|
|
# init SE engine
|
|
sb = SymbolicExecutionEngine(lifter)
|
|
|
|
# constraint bytecode -- start address and size (highest address - lowest address)
|
|
sym_address, sym_value = constraint_memory(0x4060, 0x4140 - 0x4060)
|
|
sb.symbols[sym_address] = sym_value
|
|
|
|
# constraint VM input (rdi, first function argument). The value in `ExprInt` rerpesents the function's input value.
|
|
rdi = ExprId("RDI", 64)
|
|
sb.symbols[rdi] = ExprInt(2, 64)
|
|
|
|
|
|
# init worklist
|
|
basic_block_worklist = [ExprInt(start_addr, 64)]
|
|
|
|
# worklist algorithm
|
|
while basic_block_worklist:
|
|
# get current block
|
|
current_block = basic_block_worklist.pop()
|
|
|
|
# print(f"current block: {current_block}")
|
|
|
|
# if current block is a VM handler, dump handler-specific knowledge
|
|
if current_block.is_int() and int(current_block) in VM_HANDLERS:
|
|
disassemble(sb, current_block)
|
|
|
|
# symbolical execute block -> next_block: symbolic value/address to execute
|
|
next_block = sb.run_block_at(ira_cfg, current_block, step=False)
|
|
|
|
# print(f"next block: {next_block}")
|
|
|
|
# is next block is integer or label, continue execution
|
|
if next_block.is_int() or next_block.is_loc():
|
|
basic_block_worklist.append(next_block)
|
|
|
|
# dump symbolic state
|
|
# sb.dump()
|
|
|
|
# dump VMs/functions' return value -- only works if SE runs until the end
|
|
# rax = ExprId("RAX", 64)
|
|
# value = sb.symbols[rax]
|
|
# print(f"VM return value: {value}")
|