De-flattening Mutated OLLVM Control Flow with angr Symbolic Execution
About the author. I’m Colin Xiao — an independent reverse engineer, mostly working solo on contract jobs. I dig into compiled code of every kind, including but not limited to: Android and iOS apps, x86/x64/ARM native binaries, packers and obfuscators, DRM and license checks, network protocols and APIs, crypto and signature algorithms, malware, game binaries, and embedded/IoT firmware. I also handle large-scale data collection and vulnerability research. If you have any project in mind, I’d be glad to take it on — whatever it is, as long as it’s within my wheelhouse. Just reach out. You can reach me on X at @colinxiao_lab or on Telegram at https://t.me/ColinXiao666.
This research is published for educational and defensive purposes. The target sample is anonymized throughout.
OLLVM has always been one of the strongest levers in program protection: it shreds the original control flow and drives execution through a dispatcher. There’s plenty written about the basics, so I’ll assume you already know what a prologue, a successor, and a dispatcher are. What I want to cover here are the details everyone else skips — enough that you can write your own scripts for different mutated OLLVM variants using one general recipe.
On the static side there are IDA plugins like d810, but static approaches break down against mutated OLLVM — double loop heads, convergence blocks merging with loop heads, and so on. Which raises the obvious thought: if it’s control-flow obfuscation, why not just execute it dynamically once and recover the real execution order?
Finding the real blocks
The first step is to find all the real blocks — the actual code blocks — through a fixed pipeline. The classic diagram:
1. Find the loop head.
A -> B -> C -> D -> BIf execution returns to B, then B is the loop head, and BFS finds it:
# ida python
def find_loop_head(start_ea):
loop_heads = set()
queue = deque() # BFS queue
blcok = get_basic_block(start_ea) # get the basic block containing the start address
queue.append((blcok,[]))
while len(queue) > 0:
cur_block, path = queue.popleft()
if cur_block.start_ea in path:
loop_heads.add(cur_block.start_ea) # found a loop head
continue
path = path + [cur_block.start_ea] # update path
queue.extend((s, path) for s in cur_block.succs()) # enqueue successors
all_loop_heads = list(loop_heads)
all_loop_heads.sort() # sort ascending so the main loop head comes first
print("[+]Find loop heads:",[hex(lh) for lh in all_loop_heads]," -- total:",len(all_loop_heads))
return all_loop_headsThe loop head has two predecessors: the prologue (which sets up the stack and initializes variables) and the convergence block. How do you tell which is the convergence block? It has many predecessors — and those predecessors are exactly what we care about: the real blocks.
ps: the prologue is also a real block, don’t forget it.
2. Find the convergence block.
def find_converge_addr(loop_head_addr):
converge_addr = 0
block = get_basic_block(loop_head_addr) # loop head
preds = block.preds() # get predecessor blocks
pred_list = list(preds)
if len(pred_list) == 2: # standard ollvm: the loop head has two predecessors — one prologue block and one convergence block
for pred in pred_list:
tmp_list = list(pred.preds())
if len(tmp_list) > 1: # a block with multiple predecessors is the convergence block
converge_addr = pred.start_ea
print("[+]Find converge_addr:",hex(converge_addr))
return converge_addrWhatever predecessor isn’t the convergence block is the prologue:
for loop_head_addr in loop_heads:
loop_head_block = get_basic_block(loop_head_addr)
converge_addr = find_converge_addr(loop_head_addr)
real_blocks = []
# find the prologue
loop_head_preds = list(loop_head_block.preds())
loop_head_preds_addr = [b.start_ea for b in loop_head_preds]
if loop_head_addr != converge_addr:
loop_head_preds_addr.remove(converge_addr)
print("prologue block:",[hex(x) for x in loop_head_preds_addr])3. Find the ret block.
Easy — a block with no successors is the ret block:
def find_ret_block(blocks):
for block in blocks:
succs = list(block.succs()) # get successor blocks
succs_list = list(succs)
end_ea = block.end_ea # end_ea points to the address after the last instruction of the block
last_inst_ea = idc.prev_head(end_ea) # get the address of the last instruction
mnem = idc.print_insn_mnem(last_inst_ea) # get the instruction mnemonic
if len(succs_list) == 0: # no successor blocks
if mnem == "retn": # last instruction is ret
ori_ret_block = block
# walk upward to find a better ret block
while True:
tmp_block = block.preds()
pred_list = list(tmp_block)
if len(pred_list) == 1: # single predecessor
block = pred_list[0]
if get_block_size(block) == 4: # single-instruction block
continue
else:
break
else: # multiple predecessors or none
break
# handle the sub-dispatcher case
block2 = block
num = 0
i = 0
while True:
i += 1
succs_block = block2.succs()
for succ in succs_block:
child_succs = succ.succs()
succ_list = list(child_succs)
if len(succ_list) != 0:
block2 = succ
num += 1
if num > 2:
block = ori_ret_block
break
if i > 2:
break
print("[+]ret block",hex(block.start_ea))
return block.start_ea
# the ret block may have predecessors, so I count them together to be safe. you don't strictly need all of this — feel free to just eyeball the ret block yourself in some cases.4. Find all real blocks.
def find_all_real_blocks(fun_ea):
blocks = idaapi.FlowChart(idaapi.get_func(fun_ea))
loop_heads = find_loop_head(fun_ea)
all_real_blocks = []
for loop_head_addr in loop_heads:
loop_head_block = get_basic_block(loop_head_addr)
converge_addr = find_converge_addr(loop_head_addr)
real_blocks = []
# find the prologue
loop_head_preds = list(loop_head_block.preds())
loop_head_preds_addr = [b.start_ea for b in loop_head_preds]
if loop_head_addr != converge_addr:
loop_head_preds_addr.remove(converge_addr)
print("prologue block:",[hex(x) for x in loop_head_preds_addr])
real_blocks.extend(loop_head_preds_addr)
converge_block = get_basic_block(converge_addr)
list_preds = list(converge_block.preds())
for pred in list_preds:
end_ea = pred.end_ea
last_inst_ea = idc.prev_head(end_ea)
mnem = idc.print_insn_mnem(last_inst_ea)
size = get_block_size(pred)
if size > 5: # larger than a single-instruction block and not a jump instruction
start_ea = pred.start_ea
real_blocks.append(start_ea)
real_blocks.sort() # sort; first is the prologue block
all_real_blocks.append(real_blocks)
print("sub loop head and its real blocks", [hex(child_block_ea) for child_block_ea in real_blocks])
ret_addr = find_ret_block(blocks)
all_real_blocks.append(ret_addr)
print("all_real_blocks:",all_real_blocks)
all_real_block_list = []
for real_blocks in all_real_blocks:
if isinstance(real_blocks,list):
all_real_block_list.extend(real_blocks)
else:
all_real_block_list.append(real_blocks)
print(f"\nall real blocks done, count: {len(all_real_block_list)}")
print(all_real_block_list)The multiple-loop-head case is handled here too — adjust as needed. I’ll use the RoarCTF 2019 polyre challenge as the running example. The IDA run produces:
[+]Find loop heads: ['0x40063f'] -- total: 1
[+]Find converge_addr: 0x4020cc
prologue block: ['0x400620']
sub loop head and its real blocks ['0x400620', '0x401121', '0x401198', '0x4011de', '0x40124f', '0x40125e', '0x4012a4', '0x4012f6', '0x401305', '0x401326', '0x40136c', '0x4013b2', '0x4013cf', '0x4013ef', '0x401435', '0x401481', '0x401490', '0x4014ae', '0x4014d2', '0x4014e8', '0x4014f7', '0x401506', '0x401521', '0x401567', '0x4015b6', '0x4015c5', '0x4015d4', '0x4015ed', '0x4015fc', '0x401642', '0x401691', '0x4016a0', '0x4016e6', '0x401739', '0x401748', '0x401765', '0x4017ab', '0x4017fc', '0x40180b', '0x401830', '0x401849', '0x401861', '0x4018a7', '0x4018fa', '0x401909', '0x401926', '0x401940', '0x401960', '0x40197d', '0x40199b', '0x4019e1', '0x401a3d', '0x401a4c', '0x401a73', '0x401a8d', '0x401ad3', '0x401b25', '0x401b34', '0x401b4e', '0x401b5d', '0x401b75', '0x401bbb', '0x401c0d', '0x401c1c', '0x401c2b', '0x401c46', '0x401c69', '0x401caf', '0x401d03', '0x401d12', '0x401d2d', '0x401d45', '0x401d54', '0x401d9a', '0x401e00', '0x401e0f', '0x401e2d', '0x401e73', '0x401eb9', '0x401ed6', '0x401efa', '0x401f09', '0x401f2d', '0x401f3c', '0x401f60', '0x401f97', '0x401fa6', '0x401fb5', '0x401fcd', '0x401fe5', '0x401ff4', '0x40200c', '0x40201b', '0x402033', '0x40204d', '0x402072', '0x402096', '0x4020b3', '0x4020c2']
[+]ret block 0x401f54
all_real_blocks: [[4195872, 4198689, 4198808, 4198878, 4198991, 4199006, 4199076, 4199158, 4199173, 4199206, 4199276, 4199346, 4199375, 4199407, 4199477, 4199553, 4199568, 4199598, 4199634, 4199656, 4199671, 4199686, 4199713, 4199783, 4199862, 4199877, 4199892, 4199917, 4199932, 4200002, 4200081, 4200096, 4200166, 4200249, 4200264, 4200293, 4200363, 4200444, 4200459, 4200496, 4200521, 4200545, 4200615, 4200698, 4200713, 4200742, 4200768, 4200800, 4200829, 4200859, 4200929, 4201021, 4201036, 4201075, 4201101, 4201171, 4201253, 4201268, 4201294, 4201309, 4201333, 4201403, 4201485, 4201500, 4201515, 4201542, 4201577, 4201647, 4201731, 4201746, 4201773, 4201797, 4201812, 4201882, 4201984, 4201999, 4202029, 4202099, 4202169, 4202198, 4202234, 4202249, 4202285, 4202300, 4202336, 4202391, 4202406, 4202421, 4202445, 4202469, 4202484, 4202508, 4202523, 4202547, 4202573, 4202610, 4202646, 4202675, 4202690], 4202324]Recovering the execution order
This is the crux of the article: finding each block’s successor. For every real block, the rule is:
prologue → main dispatcher → jump straight to the real block we want → keep executing and see where it lands.
Two things matter:
- You must execute the prologue first — it’s the initialization, the stack setup.
- Jump straight to the real block you’re after, instead of executing the whole program every time until you reach it — I solve that with a hook.
You might wonder how OLLVM implements something as common as if/branching. This is exactly the detail most writeups skip. On x86 it’s the cmovxx family; on ARM64 it’s csel — a topic for another day. Take one real block:
loc_40199B:
mov eax, ds:dword_603054
mov ecx, ds:dword_603058
mov edx, eax
sub edx, 1
imul eax, edx
and eax, 1
cmp eax, 0
setz sil
cmp ecx, 0Ah
setl dil
or sil, dil
test sil, 1
mov eax, 0F37184F0h
mov ecx, 0A105D2C4h
cmovnz ecx, eax
mov [rbp+var_114], ecx
jmp loc_4020CCcmovxx changes a register’s value, which influences the dispatcher upstream and steers execution to different blocks. The catch: angr won’t split a cmovxx into two branches — it just accumulates a constraint. So I manually split into two states: one where the move executes and one where it doesn’t, each landing in a different block. I store the result like this:
{'0xaaaa':['0xbbb','0xccc']}Left is ZF=1, right is ZF=0 — that ordering matters when we patch later.
import logging
import angr
from tqdm import tqdm
logging.getLogger('angr').setLevel(logging.ERROR)
def capstone_decode_cmovxx(insn):
operands = insn.op_str.replace(" ", "").split(",")
dst_reg = operands[0]
src_reg = operands[1]
print(f"cmovxx parse result: dst_reg:{dst_reg}, src_reg:{src_reg}")
return dst_reg, src_reg
def find_state_succ_cmovxx(proj, base, local_state, flag, real_blocks, real_block_addr, path):
# only called when find_block_succ identifies a cmov
ins = local_state.block().capstone.insns[0]
dst_reg, src_reg = capstone_decode_cmovxx(ins)
# logic fix:
# flag == True -> ZF=1 (Zero) -> cmovnz (Not Zero) condition false -> no move -> Pass
# flag == False -> ZF=0 (Not Zero) -> cmovnz (Not Zero) condition true -> move
if not flag: # need to execute the move
try:
# fix: state.regs has no .get(); use getattr
src_val = getattr(local_state.regs, src_reg)
setattr(local_state.regs, dst_reg, src_val)
except Exception as e:
print(f"register access error: {e}")
# key fix: manually skip this cmov instruction so angr doesn't execute it again
local_state.regs.ip += ins.size
sm = proj.factory.simgr(local_state)
while(len(sm.active)):
for active_state in sm.active:
try:
ins_offset = active_state.addr - base
if ins_offset in real_blocks:
value = path[real_block_addr]
if ins_offset not in value:
value.append(ins_offset)
return ins_offset
except:
pass
sm.step(num_inst=1)
def find_block_succ(proj, base, func_offset, state, real_block_addr, real_blocks, path):
msm = proj.factory.simgr(state) # build the simulator
while len(msm.active):
for active_state in msm.active:
#print(active_state.block().capstone.insns[0])
offset = active_state.addr - base
#print("current offset:", hex(offset),"searching for real block:", hex(real_block_addr))
if offset == real_block_addr: # found the real block
print("found real block:", hex(real_block_addr))
mstate = active_state.copy() # copy the state to prep for successor discovery
msm2 = proj.factory.simgr(mstate)
msm2.step(num_inst=1) # advance into the next instruction in the block, to avoid mixing with the outer state
while len(msm2.active):
for mactive_state in msm2.active:
#print(mactive_state.block().capstone.insns[0])
ins_offset = mactive_state.addr - base
if ins_offset in real_blocks: # branchless block (or unconditional jump)
# on an unconditional jump, when at least two paths reach real blocks at once, take the non-ret real block
msm2_len = len(msm2.active)
if msm2_len > 1:
tmp_addrs = []
for s in msm2.active:
moffset = s.addr - base
tmp_value = path[real_block_addr]
if moffset in real_blocks and moffset not in tmp_value:
tmp_addrs.append(moffset)
if len(tmp_addrs) > 1:
print("at least two paths reached real blocks simultaneously:", [hex(tmp_addr) for tmp_addr in tmp_addrs])
ret_addr = real_blocks[len(real_blocks) - 1]
if ret_addr in tmp_addrs:
tmp_addrs.remove(ret_addr)
ins_offset = tmp_addrs[0]
print("after two paths reached real blocks, chose:", hex(ins_offset))
value = path[real_block_addr]
if ins_offset not in value:
value.append(ins_offset)
print(f"unconditional jump relation:{hex(real_block_addr)}-->{hex(ins_offset)}")
return
# possibly a cmovnz branch instruction
ins = mactive_state.block().capstone.insns[0]
if ins.mnemonic == 'cmovnz' or ins.mnemonic == 'cmovne':
print("found cmovnz/cmovne, splitting branches:", hex(ins_offset))
#split the two cases
state_true = mactive_state.copy()
state_true_succ_addr = find_state_succ_cmovxx(proj, base, state_true, True, real_blocks, real_block_addr, path)
state_false = mactive_state.copy()
state_false_succ_addr = find_state_succ_cmovxx(proj, base, state_false, False, real_blocks, real_block_addr, path)
if state_true_succ_addr is None or state_false_succ_addr is None:
print("cmovnz/cmovne error at:", hex(ins_offset))
print(f"cmovnz/cmovne successor error:{hex(real_block_addr)}-->{hex(state_true_succ_addr) if state_true_succ_addr is not None else state_true_succ_addr},"
f"{hex(state_false_succ_addr) if state_false_succ_addr is not None else state_false_succ_addr}")
return "erro"
#cmovne
print(f"cmovnz/cmovne branch relation:{hex(real_block_addr)}-->{hex(state_true_succ_addr)} zf = 1, {hex(state_false_succ_addr)} zf != 1")
#print(f"csel branch relation:{hex(real_block_addr)}-->{hex(state_true_succ_addr)},{hex(state_false_succ_addr)}")
return
#opposite of cmovnz
if ins.mnemonic == 'cmovz' or ins.mnemonic == 'cmove':
print("found cmovz/cmove, splitting branches:", hex(ins_offset))
state_true = mactive_state.copy()
state_true_succ_addr = find_state_succ_cmovxx(proj, base, state_true, False, real_blocks, real_block_addr, path)
state_false = mactive_state.copy()
state_false_succ_addr = find_state_succ_cmovxx(proj, base, state_false, True, real_blocks, real_block_addr, path)
if state_true_succ_addr is None or state_false_succ_addr is None:
print("cmovz/cmove error at:", hex(ins_offset))
print(f"cmovz/cmove successor error:{hex(real_block_addr)}-->{hex(state_true_succ_addr) if state_true_succ_addr is not None else state_true_succ_addr},"
f"{hex(state_false_succ_addr) if state_false_succ_addr is not None else state_false_succ_addr}")
return "erro"
#cmovne
print(f"cmovz/cmove branch relation:{hex(real_block_addr)}-->{hex(state_true_succ_addr)} zf = 1, {hex(state_false_succ_addr)} zf != 1")
#print(f"csel branch relation:{hex(real_block_addr)}-->{hex(state_true_succ_addr)},{hex(state_false_succ_addr)}")
return
msm2.step(num_inst=1)
# if the last block in the real-block set never found a successor, it's the return block; just return
return
msm.step(num_inst=1)
def angr_main(real_blocks,func_offset,file_path):
proj = angr.Project(file_path, auto_load_libs=False)
base = 0
func_addr = base + func_offset
init_state = proj.factory.blank_state(addr=func_addr)
init_state.options.add(angr.options.CALLLESS)
path = {addr: [] for addr in real_blocks}
ret_addr = real_blocks[len(real_blocks) - 1]
first_block = proj.factory.block(func_addr)
first_block_insns = first_block.capstone.insns
# last instruction of the main prologue
first_block_last_ins = first_block_insns[len(first_block_insns) - 1]
print(hex(first_block_last_ins.address))
for real_block_addr in tqdm(real_blocks):
if ret_addr == real_block_addr:
continue
state = init_state.copy()
print("searching:",hex(real_block_addr))
def jump_to_address(state):
#print(state.regs.pc)
state.regs.pc = base + real_block_addr - 6
print("jumping to:", hex(base + real_block_addr - 6))
proj.unhook(0x400675)
print(hex(real_block_addr),hex(func_offset))
if real_block_addr != func_offset:
print("prologue done")
proj.hook(0x400675, jump_to_address, first_block_last_ins.size)
ret = find_block_succ(proj, base, func_offset, state, real_block_addr, real_blocks, path)
if ret == "erro":
return
hex_dict = {
hex(key): [hex(value) for value in values]
for key, values in path.items()
}
for i in hex_dict.keys():
print(f"{i}: {hex_dict[i]}")
print(hex_dict)
return hex_dict
all_real_blocks: list[int] =[4195872, 4198689, 4198808, 4198878, 4198991, 4199006, 4199076, 4199158, 4199173, 4199206, 4199276, 4199346, 4199375, 4199407, 4199477, 4199553, 4199568, 4199598, 4199634, 4199656, 4199671, 4199686, 4199713, 4199783, 4199862, 4199877, 4199892, 4199917, 4199932, 4200002, 4200081, 4200096, 4200166, 4200249, 4200264, 4200293, 4200363, 4200444, 4200459, 4200496, 4200521, 4200545, 4200615, 4200698, 4200713, 4200742, 4200768, 4200800, 4200829, 4200859, 4200929, 4201021, 4201036, 4201075, 4201101, 4201171, 4201253, 4201268, 4201294, 4201309, 4201333, 4201403, 4201485, 4201500, 4201515, 4201542, 4201577, 4201647, 4201731, 4201746, 4201773, 4201797, 4201812, 4201882, 4201984, 4201999, 4202029, 4202099, 4202169, 4202198, 4202234, 4202249, 4202285, 4202300, 4202336, 4202391, 4202406, 4202421, 4202445, 4202469, 4202484, 4202508, 4202523, 4202547, 4202573, 4202610, 4202646, 4202675, 4202690, 4202324]
angr_main(all_real_blocks, 0x400620, "polyre")A few key points.
def jump_to_address(state):
#print(state.regs.pc)
state.regs.pc = base + real_block_addr - 6
print("jumping to:", hex(base + real_block_addr - 6))
proj.unhook(0x400675)We hook the last instruction of the main dispatcher, run until we jump straight to the block we want, then unhook — otherwise when execution loops back it jumps there again and spins forever. As for the -6: in theory it shouldn’t need an offset, but in practice without it angr mis-decodes the instruction and capstone produces garbage. I found -6 by trial and error; if it doesn’t work for you, try something else and print the disassembly to compare (I left it commented out). The result:
0x400620: ['0x401121']
0x401121: ['0x401198']
0x401198: ['0x401f60', '0x4011de']
0x4011de: ['0x401f60', '0x40124f']
0x40124f: ['0x40125e']
0x40125e: ['0x401f97', '0x4012a4']
0x4012a4: ['0x401f97', '0x4012f6']
0x4012f6: ['0x401305']
0x401305: ['0x401326']
0x401326: ['0x401fa6', '0x40136c']
0x40136c: ['0x401fa6', '0x4013b2']
0x4013b2: ['0x4015d4', '0x4013cf']
0x4013cf: ['0x4013ef']
0x4013ef: ['0x401fb5', '0x401435']
0x401435: ['0x401fb5', '0x401481']
0x401481: ['0x401490']
0x401490: ['0x4014ae', '0x4014f7']
0x4014ae: ['0x4014d2']
0x4014d2: ['0x4014e8']
0x4014e8: ['0x4015d4']
0x4014f7: ['0x401506']
0x401506: ['0x401521']
0x401521: ['0x401fcd', '0x401567']
0x401567: ['0x401fcd', '0x4015b6']
0x4015b6: ['0x4015c5']
0x4015c5: ['0x40125e']
0x4015d4: ['0x4015ed']
0x4015ed: ['0x4015fc']
0x4015fc: ['0x401fe5', '0x401642']
0x401642: ['0x401fe5', '0x401691']
0x401691: ['0x4016a0']
0x4016a0: ['0x401ff4', '0x4016e6']
0x4016e6: ['0x401ff4', '0x401739']
0x401739: ['0x401748']
0x401748: ['0x401d54', '0x401765']
0x401765: ['0x40200c', '0x4017ab']
0x4017ab: ['0x40200c', '0x4017fc']
0x4017fc: ['0x40180b']
0x40180b: ['0x401830']
0x401830: ['0x401849']
0x401849: ['0x401861']
0x401861: ['0x40201b', '0x4018a7']
0x4018a7: ['0x40201b', '0x4018fa']
0x4018fa: ['0x401909']
0x401909: ['0x401c2b', '0x401926']
0x401926: ['0x401940']
0x401940: ['0x401960']
0x401960: ['0x401a73', '0x40197d']
0x40197d: ['0x40199b']
0x40199b: ['0x402033', '0x4019e1']
0x4019e1: ['0x402033', '0x401a3d']
0x401a3d: ['0x401a4c']
0x401a4c: ['0x401b4e']
0x401a73: ['0x401a8d']
0x401a8d: ['0x40204d', '0x401ad3']
0x401ad3: ['0x40204d', '0x401b25']
0x401b25: ['0x401b34']
0x401b34: ['0x401b4e']
0x401b4e: ['0x401b5d']
0x401b5d: ['0x401b75']
0x401b75: ['0x402072', '0x401bbb']
0x401bbb: ['0x402072', '0x401c0d']
0x401c0d: ['0x401c1c']
0x401c1c: ['0x401849']
0x401c2b: ['0x401c46']
0x401c46: ['0x401c69']
0x401c69: ['0x402096', '0x401caf']
0x401caf: ['0x402096', '0x401d03']
0x401d03: ['0x401d12']
0x401d12: ['0x401d2d']
0x401d2d: ['0x401d45']
0x401d45: ['0x4015fc']
0x401d54: ['0x4020b3', '0x401d9a']
0x401d9a: ['0x4020b3', '0x401e00']
0x401e00: ['0x401e0f']
0x401e0f: ['0x401e2d']
0x401e2d: ['0x4020c2', '0x401e73']
0x401e73: ['0x4020c2', '0x401eb9']
0x401eb9: ['0x401ed6', '0x401f09']
0x401ed6: ['0x401efa']
0x401efa: ['0x401f3c']
0x401f09: ['0x401f2d']
0x401f2d: ['0x401f3c']
0x401f3c: ['0x401f54']
0x401f60: ['0x4011de']
0x401f97: ['0x4012a4']
0x401fa6: ['0x40136c']
0x401fb5: ['0x401435']
0x401fcd: ['0x401567']
0x401fe5: ['0x401642']
0x401ff4: ['0x4016e6']
0x40200c: ['0x4017ab']
0x40201b: ['0x4018a7']
0x402033: ['0x4019e1']
0x40204d: ['0x401ad3']
0x402072: ['0x401bbb']
0x402096: ['0x401caf']
0x4020b3: ['0x401d9a']
0x4020c2: ['0x401e73']
0x401f54: []Patching
- If a block has a single successor, patch its last instruction to:
jmp <successor>- If it has two successors — left is
ZF=1, right isZF=0, starting at thecmovxx— patch to:
jz <left address>
jmp <right address>Fill the rest with NOPs, then NOP the blocks that never got used.
from collections import deque
import ida_funcs
import idaapi
import idautils
import idc
import keystone
def patch_ins_to_nop(ins):
size = idc.get_item_size(ins)
for i in range(size):
idc.patch_byte(ins + i,0x90)
def patch_bytes(addr, data):
for i, b in enumerate(data):
idc.patch_byte(addr + i, b)
def fill_nop(start_ea, end_ea):
# [FIX 1] should be end - start, otherwise negative
size = end_ea - start_ea
if size > 0:
# [FIX 2] use patch_bytes for bulk write
patch_bytes(start_ea, b'\x90' * size)
def get_block_by_address(ea):
func = idaapi.get_func(ea)
blocks = idaapi.FlowChart(func)
for block in blocks:
if block.start_ea <= ea < block.end_ea:
return block
return None
def generate_jmp_code(src, dst):
# E9 xx xx xx xx
offset = dst - (src + 5)
return b'\xE9' + offset.to_bytes(4, 'little', signed=True)
def generate_jz_code(src, dst):
# 0F 84 xx xx xx xx
offset = dst - (src + 6)
return b'\x0F\x84' + offset.to_bytes(4, 'little', signed=True)
def patch_branch(patch_dict):
for ea in patch_dict:
values = patch_dict[ea]
if len(values) == 0:# if there are zero successors, it's basically a return block — no patch needed, skip
continue
block = get_block_by_address(int(ea, 16))
start_ea = block.start_ea
end_ea = block.end_ea
last_ins_ea = idc.prev_head(end_ea)# block.end_ea is the address after the block's last instruction, so step back one
if len(values) == 2:
for ins in idautils.Heads(start_ea,end_ea):
if idc.print_insn_mnem(ins).startswith("cmov"):
print("find cmov")
jz_code = generate_jz_code(ins, int(values[0],16))
jmp_code = generate_jmp_code(ins + len(jz_code), int(values[1],16))
# [FIX 2] actually write to memory!
patch_bytes(ins, jz_code)
patch_bytes(ins + len(jz_code), jmp_code)
# 3. fill with NOP
nop_start = ins + len(jz_code) + len(jmp_code)
fill_nop(nop_start, end_ea)
if len(values) == 1:
mnem = idc.print_insn_mnem(last_ins_ea)
if mnem.startswith("jmp"):
jmp_code = generate_jmp_code(last_ins_ea, int(values[0],16))
patch_bytes(last_ins_ea, jmp_code)
nop_start = last_ins_ea + len(jmp_code)
fill_nop(nop_start, end_ea)
def find_all_useless_block(func_ea,real_blocks):
blocks = idaapi.FlowChart(idaapi.get_func(func_ea))
local_real_blocks = real_blocks.copy()
useless_blocks = []
# local_real_blocks.extend(succ.start_ea for succ in cur_block.succs())
for block in blocks:
start_ea = block.start_ea
end_ea = block.end_ea
if start_ea not in local_real_blocks:
useless_blocks.append([start_ea,end_ea])
print("all useless blocks:",[b for b in useless_blocks])
return useless_blocks
def patch_useless_blocks(useless_blocks):
# print(useless_blocks)
for useless_block in useless_blocks:
print(f"Nop-ing useless block from {hex(useless_block[0])} to {hex(useless_block[1])}")
fill_nop(useless_block[0], useless_block[1])
print("useless blocks nop'd")
func_ea = 0x400620
all_real_blocks =[4195872, 4198689, 4198808, 4198878, 4198991, 4199006, 4199076, 4199158, 4199173, 4199206, 4199276, 4199346, 4199375, 4199407, 4199477, 4199553, 4199568, 4199598, 4199634, 4199656, 4199671, 4199686, 4199713, 4199783, 4199862, 4199877, 4199892, 4199917, 4199932, 4200002, 4200081, 4200096, 4200166, 4200249, 4200264, 4200293, 4200363, 4200444, 4200459, 4200496, 4200521, 4200545, 4200615, 4200698, 4200713, 4200742, 4200768, 4200800, 4200829, 4200859, 4200929, 4201021, 4201036, 4201075, 4201101, 4201171, 4201253, 4201268, 4201294, 4201309, 4201333, 4201403, 4201485, 4201500, 4201515, 4201542, 4201577, 4201647, 4201731, 4201746, 4201773, 4201797, 4201812, 4201882, 4201984, 4201999, 4202029, 4202099, 4202169, 4202198, 4202234, 4202249, 4202285, 4202300, 4202336, 4202391, 4202406, 4202421, 4202445, 4202469, 4202484, 4202508, 4202523, 4202547, 4202573, 4202610, 4202646, 4202675, 4202690, 4202324]
useless_blocks = find_all_useless_block(func_ea,all_real_blocks)
patch_branch({'0x400620': ['0x401121'], '0x401121': ['0x401198'], '0x401198': ['0x401f60', '0x4011de'], '0x4011de': ['0x401f60', '0x40124f'], '0x40124f': ['0x40125e'], '0x40125e': ['0x401f97', '0x4012a4'], '0x4012a4': ['0x401f97', '0x4012f6'], '0x4012f6': ['0x401305'], '0x401305': ['0x401326'], '0x401326': ['0x401fa6', '0x40136c'], '0x40136c': ['0x401fa6', '0x4013b2'], '0x4013b2': ['0x4015d4', '0x4013cf'], '0x4013cf': ['0x4013ef'], '0x4013ef': ['0x401fb5', '0x401435'], '0x401435': ['0x401fb5', '0x401481'], '0x401481': ['0x401490'], '0x401490': ['0x4014ae', '0x4014f7'], '0x4014ae': ['0x4014d2'], '0x4014d2': ['0x4014e8'], '0x4014e8': ['0x4015d4'], '0x4014f7': ['0x401506'], '0x401506': ['0x401521'], '0x401521': ['0x401fcd', '0x401567'], '0x401567': ['0x401fcd', '0x4015b6'], '0x4015b6': ['0x4015c5'], '0x4015c5': ['0x40125e'], '0x4015d4': ['0x4015ed'], '0x4015ed': ['0x4015fc'], '0x4015fc': ['0x401fe5', '0x401642'], '0x401642': ['0x401fe5', '0x401691'], '0x401691': ['0x4016a0'], '0x4016a0': ['0x401ff4', '0x4016e6'], '0x4016e6': ['0x401ff4', '0x401739'], '0x401739': ['0x401748'], '0x401748': ['0x401d54', '0x401765'], '0x401765': ['0x40200c', '0x4017ab'], '0x4017ab': ['0x40200c', '0x4017fc'], '0x4017fc': ['0x40180b'], '0x40180b': ['0x401830'], '0x401830': ['0x401849'], '0x401849': ['0x401861'], '0x401861': ['0x40201b', '0x4018a7'], '0x4018a7': ['0x40201b', '0x4018fa'], '0x4018fa': ['0x401909'], '0x401909': ['0x401c2b', '0x401926'], '0x401926': ['0x401940'], '0x401940': ['0x401960'], '0x401960': ['0x401a73', '0x40197d'], '0x40197d': ['0x40199b'], '0x40199b': ['0x402033', '0x4019e1'], '0x4019e1': ['0x402033', '0x401a3d'], '0x401a3d': ['0x401a4c'], '0x401a4c': ['0x401b4e'], '0x401a73': ['0x401a8d'], '0x401a8d': ['0x40204d', '0x401ad3'], '0x401ad3': ['0x40204d', '0x401b25'], '0x401b25': ['0x401b34'], '0x401b34': ['0x401b4e'], '0x401b4e': ['0x401b5d'], '0x401b5d': ['0x401b75'], '0x401b75': ['0x402072', '0x401bbb'], '0x401bbb': ['0x402072', '0x401c0d'], '0x401c0d': ['0x401c1c'], '0x401c1c': ['0x401849'], '0x401c2b': ['0x401c46'], '0x401c46': ['0x401c69'], '0x401c69': ['0x402096', '0x401caf'], '0x401caf': ['0x402096', '0x401d03'], '0x401d03': ['0x401d12'], '0x401d12': ['0x401d2d'], '0x401d2d': ['0x401d45'], '0x401d45': ['0x4015fc'], '0x401d54': ['0x4020b3', '0x401d9a'], '0x401d9a': ['0x4020b3', '0x401e00'], '0x401e00': ['0x401e0f'], '0x401e0f': ['0x401e2d'], '0x401e2d': ['0x4020c2', '0x401e73'], '0x401e73': ['0x4020c2', '0x401eb9'], '0x401eb9': ['0x401ed6', '0x401f09'], '0x401ed6': ['0x401efa'], '0x401efa': ['0x401f3c'], '0x401f09': ['0x401f2d'], '0x401f2d': ['0x401f3c'], '0x401f3c': ['0x401f54'], '0x401f60': ['0x4011de'], '0x401f97': ['0x4012a4'], '0x401fa6': ['0x40136c'], '0x401fb5': ['0x401435'], '0x401fcd': ['0x401567'], '0x401fe5': ['0x401642'], '0x401ff4': ['0x4016e6'], '0x40200c': ['0x4017ab'], '0x40201b': ['0x4018a7'], '0x402033': ['0x4019e1'], '0x40204d': ['0x401ad3'], '0x402072': ['0x401bbb'], '0x402096': ['0x401caf'], '0x4020b3': ['0x401d9a'], '0x4020c2': ['0x401e73'], '0x401f54': []})
patch_useless_blocks(useless_blocks)
ida_funcs.reanalyze_function(ida_funcs.get_func(func_ea))# refresh the function's control-flow graph
print("control-flow graph refreshed")One small ordering detail here: first find all the useless blocks (their start/end addresses), then patch the control flow, and only then NOP the useless blocks using that earlier list. If you patch the control flow first and then go looking for useless blocks, the ones you find will be wrong.
Here’s the recovered code:
__int64 __fastcall main(int a1, char **a2, char **a3)
{
signed __int64 v4; // [rsp+1E0h] [rbp-110h]
int n6; // [rsp+1E8h] [rbp-108h]
int n64; // [rsp+1ECh] [rbp-104h]
int n64_1; // [rsp+1ECh] [rbp-104h]
_BYTE s1[48]; // [rsp+1F0h] [rbp-100h] BYREF
_BYTE s[60]; // [rsp+220h] [rbp-D0h] BYREF
unsigned int v10; // [rsp+25Ch] [rbp-94h]
_BYTE *p_s; // [rsp+260h] [rbp-90h]
int n64_3; // [rsp+26Ch] [rbp-84h]
bool v13; // [rsp+272h] [rbp-7Eh]
unsigned __int8 n10; // [rsp+273h] [rbp-7Dh]
int n10_1; // [rsp+274h] [rbp-7Ch]
_BYTE *v16; // [rsp+278h] [rbp-78h]
int v17; // [rsp+284h] [rbp-6Ch]
int n6_1; // [rsp+288h] [rbp-68h]
bool v19; // [rsp+28Fh] [rbp-61h]
_BYTE *p_s_1; // [rsp+290h] [rbp-60h]
int n64_4; // [rsp+298h] [rbp-58h]
bool v22; // [rsp+29Fh] [rbp-51h]
signed __int64 v23; // [rsp+2A0h] [rbp-50h]
bool v24; // [rsp+2AFh] [rbp-41h]
__int64 v25; // [rsp+2B0h] [rbp-40h]
__int64 v26; // [rsp+2B8h] [rbp-38h]
signed __int64 v27; // [rsp+2C0h] [rbp-30h]
__int64 v28; // [rsp+2C8h] [rbp-28h]
int n64_2; // [rsp+2D0h] [rbp-20h]
int v30; // [rsp+2D4h] [rbp-1Ch]
signed __int64 *v31; // [rsp+2D8h] [rbp-18h]
int v32; // [rsp+2E0h] [rbp-10h]
int v33; // [rsp+2E4h] [rbp-Ch]
bool v34; // [rsp+2EBh] [rbp-5h]
v10 = 0;
memset(s, 0, 0x30u);
memset(s1, 0, sizeof(s1));
printf("Input:");
p_s = s;
__isoc99_scanf("%s", s);
for ( n64 = 0; ; ++n64 )
{
n64_3 = n64;
v13 = n64 < 64;
if ( n64 >= 64 )
break;
n10 = s[n64];
n10_1 = n10;
if ( n10 == 10 )
{
v16 = &s[n64];
*v16 = 0;
break;
}
v17 = n64 + 1;
}
for ( n6 = 0; ; ++n6 )
{
n6_1 = n6;
v19 = n6 < 6;
if ( n6 >= 6 )
break;
p_s_1 = s;
v4 = *(_QWORD *)&s[8 * n6];
for ( n64_1 = 0; ; ++n64_1 )
{
n64_4 = n64_1;
v22 = n64_1 < 64;
if ( n64_1 >= 64 )
break;
v23 = v4;
v24 = v4 < 0;
if ( v4 >= 0 )
{
v27 = v4;
v28 = 2 * v4;
v4 *= 2LL;
}
else
{
v25 = 2 * v4;
v26 = 2 * v4;
v4 = (2 * v4) ^ 0xB0004B7679FA26B3LL;
}
n64_2 = n64_1;
}
v30 = 8 * n6;
v31 = (signed __int64 *)&s1[8 * n6];
*v31 = v4;
v32 = n6 + 1;
}
v33 = memcmp(s1, &s2_, 0x30u);
v34 = v33 != 0;
if ( v33 )
puts("Wrong!");
else
puts("Correct!");
return v10;
}And the final exploit:
import struct
# Data from s2_
s2_bytes = [
0x96, 0x62, 0x53, 0x43, 0x6d, 0xf2, 0x8f, 0xbc,
0x16, 0xee, 0x30, 0x5, 0x78, 0x0, 0x1, 0x52,
0xec, 0x8, 0x5f, 0x93, 0xea, 0xb5, 0xc0, 0x4d,
0x50, 0xf4, 0x53, 0xd8, 0xaf, 0x90, 0x2b, 0x34,
0x81, 0x36, 0x2c, 0xaa, 0xbc, 0xe, 0x25, 0x8b,
0xe4, 0x8a, 0xc6, 0xa2, 0x81, 0x9f, 0x75, 0x55
]
POLY = 0xB0004B7679FA26B3
# Convert bytes to QWORDs (little endian)
qwords = []
for i in range(0, len(s2_bytes), 8):
chunk = s2_bytes[i:i+8]
val = struct.unpack('<Q', bytes(chunk))[0]
qwords.append(val)
start_vals = []
for val in qwords:
curr = val
for _ in range(64):
if curr & 1:
curr = (curr ^ POLY) >> 1
curr |= 0x8000000000000000
else:
curr = curr >> 1
start_vals.append(curr)
# Convert back to bytes
flag = b''
for val in start_vals:
flag += struct.pack('<Q', val)
print(flag)
try:
print("Flag:", flag.decode('utf-8'))
except:
pass
#flag{6ff29390-6c20-4c56-ba70-a95758e3d1f8}Colin Xiao — I take on contract reverse-engineering work, and writeups like this are how I show what I can do. If you need help with anything in this space — including but not limited to the areas I cover above — I’m available. No project is too big or too small — if it’s reverse engineering and I can do it, I’ll take it. DM me on X (@colinxiao_lab) or Telegram (@ColinXiao666).
Get in touch