Reversing OLLVM Control-Flow Flattening with IDA Hex-Rays Microcode

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.

Reversing a plain OLLVM control-flow-flattening (CFF) pass usually reduces to a small set of stages:

  • (a) find the dispatcher
  • (b) find the real (effective) blocks
  • (c) find the state variable
  • (d) find which block assigns which value to the state variable
  • (e) find what the state variable equals at each block’s entry
  • (f) build the block-to-block mapping
  • (g) patch the control flow

A lot of customized or hardened OLLVM CFF can’t be handled this simply, so this article only covers the base case. I relied on two earlier writeups for most of the engineering detail, but they were missing a complete, self-contained PoC — so that’s what I’m providing here, along with a small test sample.

Value-range analysis and its limits

IDA’s microcode exposes a value-range analysis feature that, in the base case, can cover stage (e). It has a serious shortcoming though: many non-standard CFFs are complex enough that value-range analysis alone won’t recover the mapping from state variable to effective block, so stage (e) can’t be fully implemented.

The D810 plugin goes further than IDA’s built-in value-range analysis. It uses a MicroCodeInterpreter to simulate the dispatcher’s logic, feeding in a chosen state-variable value and watching which effective block the dispatcher routes to. One researcher tried to extract MicroCodeInterpreter for standalone use, then found a large number of micro-instructions are unimplemented and dropped the effort — the gap is significant. But D810’s framework itself is powerful, and worth studying closely if you’re serious about de-flattening.

Relying on IDA microcode alone is essentially static analysis, and plenty of complex cases resist it. Angr’s symbolic execution is stronger than MicroCodeInterpreter; it often doesn’t even need you to explicitly identify the state variable — you just simulate execution and build the block-to-block mapping directly, which collapses stages (c) through (f) into one shot. The downside of Angr is that it’s heavy.

I’d previously worked with D810 and Angr but never with IDA’s built-in value-range analysis. The two prior writeups supply most of the engineering detail for the technique; the one thing missing was a complete PoC. This article fills that gap with a full PoC and a small test sample so the technique stays focused. The sample was carved out of an original target, so it can’t be loaded and debugged in a real environment — but it’s sufficient for testing the de-flattening inside IDA.

Test cases (verified on IDA 8.4.1 and 9.2; the foo, bar, and baz functions in the sample .so are the targets):

The full PoC

The script below installs a Hex-Rays optimizer block (optblock_t) that detects a CFF structure, recovers the state variable, builds both the block→next-state and state→block mappings (the latter via value-range analysis), and rewrites the edges to de-flatten the function:

import sys
import ida_hexrays

def set_vivl ( vivl, state_var ) :
    t   = state_var[0]
    if t == ida_hexrays.mop_r :
        vivl.set_reg( state_var[1], state_var[2] )
    elif t == ida_hexrays.mop_S :
        vivl.set_stkoff( state_var[1], state_var[2] )
    else :
        assert False

def get_dispatcher_id ( mba, threshold=2 ) :
    dispatcher_id   = None
    max_in_num      = -1
    for i in range( mba.qty ) :
        mblock  = mba.get_mblock( i )
        if mblock.type != ida_hexrays.BLT_2WAY :
            continue
        in_num  = mblock.npred()
        if in_num > max_in_num :
            max_in_num      = in_num
            dispatcher_id   = i
    if max_in_num < threshold :
        return None
    return dispatcher_id

def get_var_from_mop ( mop ) :
    if mop.t == ida_hexrays.mop_r :
        return ( ida_hexrays.mop_r, mop.r, mop.size )
    if mop.t == ida_hexrays.mop_S :
        return ( ida_hexrays.mop_S, mop.s.off, mop.size )
    if mop.t == ida_hexrays.mop_d :
        if mop.d.r.t == ida_hexrays.mop_n :
            return get_var_from_mop( mop.d.l )
    return None

def get_state_var ( mba, dispatcher_id ) :
    mblock  = mba.get_mblock( dispatcher_id )
    minsn   = mblock.tail
    if not minsn :
        return None
    if not (
        ida_hexrays.is_mcode_jcond( minsn.opcode ) and
        minsn.r.t == ida_hexrays.mop_n and
        minsn.d.t == ida_hexrays.mop_b
    ) :
        return None
    print( f"Block {mblock.serial} : {minsn.ea:#x} {minsn.dstr()}" )
    return get_var_from_mop( minsn.l )

def check_var ( mop, var ) :
    ret     = False
    t       = var[0]
    size    = var[2]
    while True :
        if mop.t != t or mop.size != size :
            break
        if mop.t == ida_hexrays.mop_r :
            if mop.r == var[1] :
                ret = True
        elif t == ida_hexrays.mop_S :
            if mop.s.off == var[1] :
                ret = True
        else :
            assert False
        break
    return ret

def get_block_id_2_next_state_dict ( mba, dispatcher_id, state_var, block_id_2_next_state_dict ) :
    dispatcher_mblock   = mba.get_mblock( dispatcher_id )
    for i in range( dispatcher_mblock.npred() ) :
        block_id    = dispatcher_mblock.pred( i )
        mblock      = mba.get_mblock( block_id )
        if mblock.type != ida_hexrays.BLT_1WAY :
            continue
        minsn       = mblock.tail
        while minsn :
            if (
                minsn.opcode == ida_hexrays.m_mov and
                minsn.l.t == ida_hexrays.mop_n and
                check_var( minsn.d, state_var )
            ) :
                block_id_2_next_state_dict[block_id]    = minsn.l.nnn.value
                break
            minsn   = minsn.prev

def get_block_id_2_next_state_dict_1 ( mba, dispatcher_id, state_var, block_id_2_next_state_dict ) :
    dispatcher_mblock   = mba.get_mblock( dispatcher_id )
    for i in range( dispatcher_mblock.npred() ) :
        block_id    = dispatcher_mblock.pred( i )
        mblock      = mba.get_mblock( block_id )
        if mblock.type != ida_hexrays.BLT_1WAY :
            continue
        minsn       = mblock.tail
        while minsn :
            if (
                minsn.opcode == ida_hexrays.m_mov and
                minsn.l.t != ida_hexrays.mop_n and
                check_var( minsn.d, state_var )
            ) :
                prev_state_var  = get_var_from_mop( minsn.l )
                assert prev_state_var is not None
                get_block_id_2_next_state_dict( mba, block_id, prev_state_var, block_id_2_next_state_dict )
                break
            minsn   = minsn.prev

def get_block_id_2_next_state_dict_2 ( mba, dispatcher_id, state_var, block_id_2_next_state_dict ) :
    dispatcher_mblock   = mba.get_mblock( dispatcher_id )
    minsn               = dispatcher_mblock.tail.prev
    while minsn :
        if (
            minsn.opcode == ida_hexrays.m_mov and
            minsn.l.t != ida_hexrays.mop_n and
            check_var( minsn.d, state_var )
        ) :
            prev_state_var  = get_var_from_mop( minsn.l )
            assert prev_state_var is not None
            get_block_id_2_next_state_dict( mba, dispatcher_id, prev_state_var, block_id_2_next_state_dict )
            break
        minsn   = minsn.prev

def get_state_2_block_id_dict ( mba, dispatcher_id, state_var, state_2_block_id_dict ) :
    dispatcher_mblock \
                    = mba.get_mblock( dispatcher_id )
    pred_block_ids  = [dispatcher_mblock.pred(i) for i in range(dispatcher_mblock.npred())]
    block_id_begin  = min( dispatcher_id+1, *pred_block_ids )
    block_id_end    = max( pred_block_ids ) + 1 + 1
    for i in range( block_id_begin, block_id_end ) :
        mblock  = mba.get_mblock( i )
        res     = ida_hexrays.valrng_t()
        vivl    = ida_hexrays.vivl_t()
        set_vivl( vivl, state_var )
        ok      = mblock.get_valranges( res, vivl, ida_hexrays.VR_AT_START )
        if ok :
            ok, v   = res.cvt_to_single_value()
            if ok and v not in state_2_block_id_dict :
                state_2_block_id_dict[v]    = mblock.serial

def add_new_goto ( mba, block_id, next_block_id ) :
    mblock          = mba.get_mblock( block_id )
    new_mop         = ida_hexrays.mop_t()
    new_mop.t       = ida_hexrays.mop_b
    new_mop.b       = next_block_id
    new_mop.size    = ida_hexrays.NOSIZE
    new_goto        = ida_hexrays.minsn_t( mblock.tail.ea )
    new_goto.opcode = ida_hexrays.m_goto
    new_goto.l      = new_mop
    mblock.insert_into_block( new_goto, mblock.tail )

def change_goto ( mba, block_id, next_block_id ) :
    mblock  = mba.get_mblock( block_id )
    assert mblock.type == ida_hexrays.BLT_1WAY

    if mblock.tail.opcode == ida_hexrays.m_goto :
        old_next_block_id   = mblock.tail.l.b
        mblock.tail.l.b     = next_block_id
    else :
        old_next_block_id   = mblock.succset[0]
        add_new_goto( mba, block_id, next_block_id )
    return old_next_block_id

def modify_edge ( mba, block_id, old_next_block_id, next_block_id ) :
    mblock          = mba.get_mblock( block_id )
    old_next_mblock = mba.get_mblock( old_next_block_id )
    next_mblock     = mba.get_mblock( next_block_id )
    mblock.succset._del( old_next_block_id )
    old_next_mblock.predset._del( block_id )
    mblock.succset.push_back( next_block_id )
    next_mblock.predset.push_back( block_id )

def unflatten ( mba, state_var, block_id_2_next_state_dict, state_2_block_id_dict ) :
    ret = False
    for block_id, next_state in sorted( block_id_2_next_state_dict.items() ) :
        if next_state in state_2_block_id_dict :
            next_block_id       = state_2_block_id_dict[next_state]
            print( f"Block {block_id:<2} -> Block {next_block_id}" )
            old_next_block_id   = change_goto( mba, block_id, next_block_id )
            modify_edge( mba, block_id, old_next_block_id, next_block_id )
            mba.mark_chains_dirty()
            mba.verify( True )
            ret                 = True
    return ret

def process_one_cff_structure ( mba, dispatcher_id=None, state_var=None ) :
    if dispatcher_id is None :
        print( "Attempting to automatically find dispatcher..." )
        dispatcher_id   = get_dispatcher_id( mba )
        if dispatcher_id is None :
            print( "Could not find a candidate for the dispatcher block." )
            return False
        print( f"Dispatcher candidate found: Block {dispatcher_id}" )
    else :
        print( f"Using manually specified dispatcher: Block {dispatcher_id}" )

    if state_var is None :
        print( "Attempting to automatically find state variable..." )
        state_var   = get_state_var( mba, dispatcher_id )
        if state_var is None :
            print( "Could not identify the state variable." )
            return False
        print( f"State variable candidate found: {state_var}" )
    else :
        print( f"Using manually specified state variable: {state_var}" )

    block_id_2_next_state_dict  = {}
    get_block_id_2_next_state_dict( mba, dispatcher_id, state_var, block_id_2_next_state_dict )
    if not block_id_2_next_state_dict :
        get_block_id_2_next_state_dict_1( mba, dispatcher_id, state_var, block_id_2_next_state_dict )
    if not block_id_2_next_state_dict :
        get_block_id_2_next_state_dict_2( mba, dispatcher_id, state_var, block_id_2_next_state_dict )
    if not block_id_2_next_state_dict :
        print( "Could not find block-to-next_state mappings. Aborting this pass." )
        return False
    for block_id, next_state in sorted( block_id_2_next_state_dict.items() ) :
        print( f"Block {block_id:<2} -> State {next_state:#x}" )

    state_2_block_id_dict       = {}
    get_state_2_block_id_dict( mba, dispatcher_id, state_var, state_2_block_id_dict )
    if not state_2_block_id_dict :
        print( "Could not find state-to-block_id mappings. Aborting this pass." )
        return False
    for state, block_id in sorted( state_2_block_id_dict.items() ) :
        print( f"State 0x{state:<8x} -> Block {block_id}" )

    return unflatten( mba, state_var, block_id_2_next_state_dict, state_2_block_id_dict )

def analyzer_main ( mba ) :
    func_ea     = mba.entry_ea
    changes     = 0
    print( f"Automatic analysis for function {func_ea:#x} mba.qty {mba.qty} maturity {mba.maturity} ..." )
    ok      = process_one_cff_structure( mba )
    if not ok :
        print( f"Automatic analysis for function {func_ea:#x} mba.qty {mba.qty} maturity {mba.maturity} did not find a CFF structure or failed." )
    else :
        print( f"Automatic analysis for function {func_ea:#x} mba.qty {mba.qty} maturity {mba.maturity} completed." )
        changes    += 1
    return changes

class CFF_Analyzer ( ida_hexrays.optblock_t ) :

    def __init__ ( self ) :
        super().__init__()
        self.analyzed_functions = set()

    def func ( self, *args ) :
        mblock          = args[0]
        mba             = mblock.mba
        if mba.maturity not in ( ida_hexrays.MMAT_GLBOPT1, ida_hexrays.MMAT_GLBOPT2 ) :
            if (
                mblock.serial == 1 and
                mba.maturity == ida_hexrays.MMAT_LOCOPT
            ) :
                self.analyzed_functions = {
                    key for key in self.analyzed_functions
                    if not ( key[0] == mba.entry_ea and key[1] in ( ida_hexrays.MMAT_GLBOPT1, ida_hexrays.MMAT_GLBOPT2 ) )
                }
            return 0

        func_ea         = mba.entry_ea
        analysis_key    = ( func_ea, mba.maturity, mba.qty )
        if analysis_key in self.analyzed_functions :
            return 0
        self.analyzed_functions.add( analysis_key )
        changes         = analyzer_main( mba )
        return changes

ANALYZER_INSTANCE_NAME  = "__cff_analyzer_instance__"

def install_handler () :
    if hasattr( sys, ANALYZER_INSTANCE_NAME ) :
        print( "Removing existing CFF Analyzer instance..." )
        remove_handler()
    print( "Installing new CFF Analyzer instance..." )
    instance    = CFF_Analyzer()
    instance.install()
    setattr( sys, ANALYZER_INSTANCE_NAME, instance )
    print( "CFF Analyzer installed successfully." )
    ida_hexrays.clear_cached_cfuncs()

def remove_handler () :
    if hasattr( sys, ANALYZER_INSTANCE_NAME ) :
        print( "Removing existing CFF Analyzer instance..." )
        instance    = getattr( sys, ANALYZER_INSTANCE_NAME )
        instance.remove()
        delattr( sys, ANALYZER_INSTANCE_NAME )
        print( "CFF Analyzer removed." )
        ida_hexrays.clear_cached_cfuncs()
    else :
        print( "No existing CFF Analyzer instance to remove." )

def main () :
    if ida_hexrays.init_hexrays_plugin() :
        install_handler()
    else :
        print( "Error: Hex-Rays is not available." )

if "__main__" == __name__ :
    main()

The state-to-block mapping in get_state_2_block_id_dict() is where value-range analysis comes in. It iterates candidate blocks, runs get_valranges at VR_AT_START against the state variable, and converts the result to a single value via cvt_to_single_value.

Scope and limitations

This code only handles plain OLLVM CFF. As a counterexample, libgeiri.so contains many functions far too complex for this approach. For those, refer to the xdefla project, which tackles non-standard OLLVM-fla de-obfuscation:

The takeaway: IDA’s built-in value-range analysis is enough for the base case, but the moment the dispatcher logic gets non-trivial you’ll want either D810’s interpreter or a proper symbolic engine. This PoC gives you a clean starting point to build on.

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