使用Cutter和Radare2对APT32恶意程序流程图进行反混淆处理

class GraphDeobfuscator:   def __init__(self, pipe):       """an initializationfunction for the class             Arguments:           pipe {r2pipe} -- an instance ofr2pipe or Cutter's wrapper       """       self.pipe = pipe

def clean_junk_blocks(self):       """Search a givenfunction for junk blocks, remove them and fix the flow.       """       # Get all the basic blocks of thefunction       blocks = self.pipe.cmdj("afbj @$F")       if not blocks:           print("[X] No blocks found. Is it afunction?")           return       modified = False       # Iterate over all the basic blocks ofthe function       for block in blocks:           # do something

def get_fail_block(self, block):       """Return the block towhich a block branches if the condition is fails             Arguments:           block {block_context} -- A JSONrepresentation of a block             Returns:           block_context -- The block to whichthe branch fails. If not exists, returns None       """       # Get the address of the"fail" branch       fail_addr = self.get_fail(block)       if not fail_addr:           return None       # Get a block context of the failaddress       fail_block = self.get_block(fail_addr)       return fail_block if fail_block elseNone   def is_successive_fail(self, block_A,block_B):       """Check if the endaddress of block_A is the start of block_B       Arguments:           block_A {block_context} -- A JSONobject to represent the first block           block_B {block_context} -- A JSONobject to represent the second block             Returns:           bool -- True if block_B comes immediatelyafter block_A, False otherwise       """      return ((block_A["addr"] +block_A["size"]) == block_B["addr"])

接下来，我们要判断候选垃圾代码段是否包含无效指令：

def contains_meaningful_instructions (self,block):       '''Check if a block contains meaningfulinstructions (references, calls, strings,...)             Arguments:           block {block_context} -- A JSONobject which represents a block             Returns:           bool -- True if the block containsmeaningful instructions, False otherwise       '''       # Get summary of block - strings, calls,references       summary = self.pipe.cmd("pdsb @{addr}".format(addr=block["addr"]))       return summary != ""

最后，枚举出所有对立的跳转条件：

jmp_pairs = [       ['jno', 'jo'],       ['jnp', 'jp'],       ['jb', 'jnb'],       ['jl', 'jnl'],       ['je', 'jne'],       ['jns', 'js'],       ['jnz', 'jz'],       ['jc', 'jnc'],       ['ja', 'jbe'],       ['jae', 'jb'],       ['je', 'jnz'],       ['jg', 'jle'],       ['jge', 'jl'],       ['jpe', 'jpo'],       ['jne', 'jz']]   def is_opposite_conditional(self, cond_A,cond_B):       """Check if two operandsare opposite conditional jump operands             Arguments:           cond_A {string} -- the conditionaljump operand of the first block           cond_B {string} -- the conditionaljump operand of the second block             Returns:           bool -- True if the operands areopposite, False otherwise       """       sorted_pair = sorted([cond_A, cond_B])       for pair in self.jmp_pairs:           if sorted_pair == pair:               return True       return False

def clean_junk_blocks(self):       """Search a givenfunction for junk blocks, remove them and fix the flow.       """       # Get all the basic blocks of thefunction       blocks = self.pipe.cmdj("afbj @$F")       if not blocks:           print("[X] No blocks found. Isit a function?")           return       modified = False       # Iterate over all the basic blocks ofthe function       for block in blocks:           fail_block =self.get_fail_block(block)           if not fail_block or \           not self.is_successive_fail(block,fail_block) or \          self.contains_meaningful_instructions(fail_block) or \           notself.is_opposite_conditional(self.get_last_mnem_of_block(block),self.get_last_mnem_of_block(fail_block)):               continue

if__name__ == "__main__":   graph_deobfuscator = GraphDeobfuscator(pipe)   graph_deobfuscator.clean_graph()

ifcutter_available:   # This part will be executed only if Cutteris available.   # This will create the cutter plugin and UIobjects for the plugin   classGraphDeobfuscatorCutter(cutter.CutterPlugin):       name = "APT32 GraphDeobfuscator"       description = "Graph Deobfuscatorfor APT32 Samples"       version = "1.0"       author = "Itay Cohen(@Megabeets_)"       def setupPlugin(self):           pass       def setupInterface(self, main):           pass      def create_cutter_plugin():       return GraphDeobfuscatorCutter()

ifcutter_available:   # This part will be executed only if Cutteris available. This will   # create the cutter plugin and UI objectsfor the plugin   classGraphDeobfuscatorCutter(cutter.CutterPlugin):       name = "APT32 GraphDeobfuscator"       description = "Graph Deobfuscatorfor APT32 Samples"       version = "1.0"       author = "Megabeets"       def setupPlugin(self):           pass       def setupInterface(self, main):           # Create a new action (menu item)           action = QAction("APT32 GraphDeobfuscator", main)           action.setCheckable(False)          # Connect the action to a function - cleaner.           # A click on this action willtrigger the function          action.triggered.connect(self.cleaner)           # Add the action to the"Windows -> Plugins" menu           pluginsMenu =main.getMenuByType(main.MenuType.Plugins)           pluginsMenu.addAction(action)       def cleaner(self):           graph_deobfuscator =GraphDeobfuscator(pipe)           graph_deobfuscator.clean_graph()           cutter.refresh()   def create_cutter_plugin():       return GraphDeobfuscatorCutter()

Be6d5973452248cb18949711645990b6a56e7442dc30cc48a607a2afe7d8ec668d74d544396b57e6faa4f8fdf96a1a5e30b196d56c15f7cf05767a406708a6b2

/*
* 提示：该行代码过长，系统自动注释不进行高亮。一键复制会移除系统注释 
* """A plugin for Cutter and Radare2 to deobfuscate APT32 flow graphsThisis a python plugin for Cutter that is compatible as an r2pipe script forradare2as well. The plugin will help reverse engineers to deobfuscate and removejunkblocks from APT32 (Ocean Lotus) samples."""__author__  = "Itay Cohen, aka @megabeets_"__company__= "Check Point Software Technologies Ltd"#Check if we're running from cuttertry:    import cutter    from PySide2.QtWidgets import QAction    pipe = cutter    cutter_available = True# Ifno, assume running from radare2except:    import r2pipe    pipe = r2pipe.open()    cutter_available = FalseclassGraphDeobfuscator:    # A list of pairs of opposite conditionaljumps    jmp_pairs = [        ['jno', 'jo'],        ['jnp', 'jp'],        ['jb', 'jnb'],        ['jl', 'jnl'],        ['je', 'jne'],        ['jns', 'js'],        ['jnz', 'jz'],        ['jc', 'jnc'],        ['ja', 'jbe'],        ['jae', 'jb'],        ['je', 'jnz'],        ['jg', 'jle'],        ['jge', 'jl'],        ['jpe', 'jpo'],       ['jne', 'jz']]    def __init__(self, pipe, verbose=False):        """an initializationfunction for the class                Arguments:            pipe {r2pipe} -- an instance ofr2pipe or Cutter's wrapper                Keyword Arguments:            verbose {bool} -- if True willprint logs to the screen (default: {False})        """        self.pipe = pipe        self.verbose = verbose    def is_successive_fail(self, block_A,block_B):        """Check if the endaddress of block_A is the start of block_B        Arguments:            block_A {block_context} -- A JSONobject to represent the first block            block_B {block_context} -- A JSONobject to represent the second block                Returns:            bool -- True if block_B comesimmediately after block_A, False otherwise        """        return ((block_A["addr"] +block_A["size"]) == block_B["addr"])    def is_opposite_conditional(self, cond_A,cond_B):        """Check if two operandsare opposite conditional jump operands                Arguments:            cond_A {string} -- the conditionaljump operand of the first block            cond_B {string} -- the conditionaljump operand of the second block        Returns:            bool -- True if the operands areopposite, False otherwise        """        sorted_pair = sorted([cond_A, cond_B])        for pair in self.jmp_pairs:            if sorted_pair == pair:                return True        return False    defcontains_meaningful_instructions (self, block):        '''Check if a block contains meaningfulinstructions (references, calls, strings,...)                Arguments:            block {block_context} -- A JSONobject which represents a block                Returns:            bool -- True if the block containsmeaningful instructions, False otherwise        '''        # Get summary of block - strings,calls, references        summary = self.pipe.cmd("pdsb @{addr}".format(addr=block["addr"]))        return summary != ""    def get_block_end(self, block):        """Get the address ofthe last instruction in a given block                Arguments:            block {block_context} -- A JSONobject which represents a block                Returns:            The address of the last instructionin the block        """        # save current seek        self.pipe.cmd("s{addr}".format(addr=block['addr']))        # This will return the address of ablock's last instruction        block_end = self.pipe.cmd("?v $@B:-1")        return block_end    def get_last_mnem_of_block(self, block):        """Get the mnemonic ofthe last instruction in a block                Arguments:            block {block_context} -- A JSONobject which represents a block                Returns:            string -- the mnemonic of the lastinstruction in the given block        """        inst_info = self.pipe.cmdj("aoj @{addr}".format(addr=self.get_block_end(block)))[0]        return inst_info["mnemonic"]    def get_jump(self, block):        """Get the address towhich a block jumps                Arguments:            block {block_context} -- A JSONobject which represents a block                Returns:            addr -- the address to which theblock jumps to. If such address doesn't exist, returns False         """        return block["jump"] if"jump" in block else None    def get_fail_addr(self, block):        """Get the address towhich a block fails                Arguments:            block {block_context} -- A JSONobject which represents a block                Returns:            addr -- the address to which theblock fail-branches to. If such address doesn't exist, returns False         """        return block["fail"] if"fail" in block else None    def get_block(self, addr):        """Get the block contextin a given address                Arguments:            addr {addr} -- An address in ablock                Returns:            block_context -- the block to whichthe address belongs        """        block = self.pipe.cmdj("abj. @{offset}".format(offset=addr))        return block[0] if block else None    def get_fail_block(self, block):        """Return the block towhich a block branches if the condition is fails                Arguments:            block {block_context} -- A JSONrepresentation of a block                Returns:            block_context -- The block to whichthe branch fails. If not exists, returns None        """        # Get the address of the"fail" branch        fail_addr = self.get_fail_addr(block)        if not fail_addr:            return None        # Get a block context of the failaddress        fail_block = self.get_block(fail_addr)        return fail_block if fail_block elseNone    def reanalize_function(self):        """Re-Analyze a functionat a given address                Arguments:            addr {addr} -- an address of afunction to be re-analyze        """        # Seek to the function's start        self.pipe.cmd("s $F")        # Undefine the function in this address        self.pipe.cmd("af- $")        # Define and analyze a function in thisaddress        self.pipe.cmd("afr @ $")           def overwrite_instruction(self, addr):        """Overwrite aconditional jump to an address, with a JMP to it                Arguments:            addr {addr} -- address of aninstruction to be overwritten        """        jump_destination =self.get_jump(self.pipe.cmdj("aoj @ {addr}".format(addr=addr))[0])        if (jump_destination):            self.pipe.cmd("wai jmp0x{dest:x} @ {addr}".format(dest=jump_destination, addr=addr))    def get_current_function(self):        """Return the startaddress of the current function        Return Value:            The address of the currentfunction. None if no function found.        """        function_start =int(self.pipe.cmd("?vi $FB"))        return function_start if function_start!= 0 else None    def clean_junk_blocks(self):        """Search a givenfunction for junk blocks, remove them and fix the flow.        """        # Get all the basic blocks of thefunction        blocks = self.pipe.cmdj("afbj @$F")        if not blocks:            print("[X] No blocks found. Isit a function?")            return        # Have we modified any instruction inthe function?        # If so, a reanalyze of the function isrequired        modified = False        # Iterate over all the basic blocks ofthe function        for block in blocks:            fail_block =self.get_fail_block(block)            # Make validation checks            if not fail_block or \            not self.is_successive_fail(block,fail_block) or \           self.contains_meaningful_instructions(fail_block) or \            notself.is_opposite_conditional(self.get_last_mnem_of_block(block),self.get_last_mnem_of_block(fail_block)):                continue            if self.verbose:                print ("Potential junk:0x{junk_block:x}(0x{fix_block:x})".format(junk_block=fail_block["addr"],fix_block=block["addr"]))           self.overwrite_instruction(self.get_block_end(block))            modified = True        if modified:            self.reanalize_function()            def clean_graph(self):        """the initial functionof the class. Responsible to enable cache and start the cleaning        """        # Enable cache writing mode. changeswill only take place in the session and        # will not override the binary        self.pipe.cmd("eio.cache=true")        self.clean_junk_blocks()        ifcutter_available:    # This part will be executed only if Cutteris available. This will    # create the cutter plugin and UI objectsfor the plugin    classGraphDeobfuscatorCutter(cutter.CutterPlugin):        name = "APT32 GraphDeobfuscator"        description = "Graph Deobfuscatorfor APT32 Samples"        version = "1.0"        author = "Itay Cohen(@Megabeets_)"        def setupPlugin(self):            pass        def setupInterface(self, main):            # Create a new action (menu item)            action = QAction("APT32 GraphDeobfuscator", main)            action.setCheckable(False)            # Connect the action to a function- cleaner.            # A click on this action willtrigger the function            action.triggered.connect(self.cleaner)            # Add the action to the"Windows -> Plugins" menu            pluginsMenu =main.getMenuByType(main.MenuType.Plugins)            pluginsMenu.addAction(action)        def cleaner(self):            graph_deobfuscator =GraphDeobfuscator(pipe)            graph_deobfuscator.clean_graph()            cutter.refresh()    def create_cutter_plugin():        return GraphDeobfuscatorCutter()if__name__ == "__main__":    graph_deobfuscator =GraphDeobfuscator(pipe)graph_deobfuscator.clean_graph()
*/

*本文作者：checkpoint，转载请注明来自FreeBuf.COM