(self, current_block, block_artifacts, cfg_node)
return new_concrete_successors
Symbolically executes from ancestor nodes (2-5 times removed) finding paths of execution through the given
        CFGNode.

        :param SimSuccessors current_block: SimSuccessor with address to attempt to navigate to.
        :param dict block_artifacts:  Container of IRSB data - specifically used for known persistant register values.
        :param CFGNode cfg_node:      Current node interested around.
        :returns:                     Double checked concrete successors.
        :rtype: List
Symbolically executes from ancestor nodes (2-5 times removed) finding paths of execution through the given
        CFGNode.
Symbolically executes from ancestor nodes ( 2 - 5 times removed ) finding paths of execution through the given CFGNode .
def _symbolically_back_traverse(self, current_block, block_artifacts, cfg_node):
        """
        Symbolically executes from ancestor nodes (2-5 times removed) finding paths of execution through the given
        CFGNode.

        :param SimSuccessors current_block: SimSuccessor with address to attempt to navigate to.
        :param dict block_artifacts:  Container of IRSB data - specifically used for known persistant register values.
        :param CFGNode cfg_node:      Current node interested around.
        :returns:                     Double checked concrete successors.
        :rtype: List
        """
        class register_protector:
            def __init__(self, reg_offset, info_collection):
                """
                Class to overwrite registers.

                :param int reg_offest:       Register offset to overwrite from.
                :param dict info_collection: New register offsets to use (in container).
                """
                self._reg_offset = reg_offset
                self._info_collection = info_collection

            def write_persistent_register(self, state_):
                """
                Writes over given registers from self._info_coollection (taken from block_artifacts)

                :param SimSuccessors state_: state to update registers for
                """
                if state_.inspect.address is None:
                    l.error('state.inspect.address is None. It will be fixed by Yan later.')
                    return

                if state_.registers.load(self._reg_offset).symbolic:
                    current_run = state_.inspect.address
                    if current_run in self._info_collection and \
                            not state_.solver.symbolic(self._info_collection[current_run][self._reg_offset]):
                        l.debug("Overwriting %s with %s", state_.registers.load(self._reg_offset),
                                self._info_collection[current_run][self._reg_offset])
                        state_.registers.store(
                            self._reg_offset,
                            self._info_collection[current_run][self._reg_offset]
                        )

        l.debug("Start back traversal from %s", current_block)

        # Create a partial CFG first
        temp_cfg = networkx.DiGraph(self.graph)
        # Reverse it
        temp_cfg.reverse(copy=False)

        path_length = 0
        concrete_exits = []
        if cfg_node not in temp_cfg.nodes():
            # TODO: Figure out why this is happening
            return concrete_exits

        keep_running = True
        while not concrete_exits and path_length < 5 and keep_running:
            path_length += 1
            queue = [cfg_node]
            avoid = set()
            for _ in range(path_length):
                new_queue = []
                for n in queue:
                    successors = list(temp_cfg.successors(n))
                    for suc in successors:
                        jk = temp_cfg.get_edge_data(n, suc)['jumpkind']
                        if jk != 'Ijk_Ret':
                            # We don't want to trace into libraries
                            predecessors = list(temp_cfg.predecessors(suc))
                            avoid |= {p.addr for p in predecessors if p is not n}
                            new_queue.append(suc)
                queue = new_queue

            if path_length <= 1:
                continue

            for n in queue:
                # Start symbolic exploration from each block
                state = self.project.factory.blank_state(addr=n.addr,
                                                         mode='symbolic',
                                                         add_options={
                                                                         o.DO_RET_EMULATION,
                                                                         o.CONSERVATIVE_READ_STRATEGY,
                                                                     } | o.resilience
                                                         )
                # Avoid concretization of any symbolic read address that is over a certain limit
                # TODO: test case is needed for this option

                # Set initial values of persistent regs
                if n.addr in block_artifacts:
                    for reg in state.arch.persistent_regs:
                        state.registers.store(reg, block_artifacts[n.addr][reg])
                for reg in state.arch.persistent_regs:
                    reg_protector = register_protector(reg, block_artifacts)
                    state.inspect.add_breakpoint('reg_write',
                                                 BP(
                                                     BP_AFTER,
                                                     reg_write_offset=state.arch.registers[reg][0],
                                                     action=reg_protector.write_persistent_register
                                                 )
                                                 )
                simgr = self.project.factory.simulation_manager(state)
                simgr.use_technique(LoopSeer(bound=10))
                simgr.use_technique(Explorer(find=current_block.addr, avoid=avoid))
                simgr.use_technique(LengthLimiter(path_length))
                simgr.run()

                if simgr.found:
                    simgr = self.project.factory.simulation_manager(simgr.one_found, save_unsat=True)
                    simgr.step()
                    if simgr.active or simgr.unsat:
                        keep_running = False
                        concrete_exits.extend(simgr.active)
                        concrete_exits.extend(simgr.unsat)
                if keep_running:
                    l.debug('Step back for one more run...')

        # Make sure these successors are actually concrete
        # We just use the ip, persistent registers, and jumpkind to initialize the original unsat state
        # TODO: It works for jumptables, but not for calls. We should also handle changes in sp
        new_concrete_successors = []
        for c in concrete_exits:
            unsat_state = current_block.unsat_successors[0].copy()
            unsat_state.history.jumpkind = c.history.jumpkind
            for reg in unsat_state.arch.persistent_regs + ['ip']:
                unsat_state.registers.store(reg, c.registers.load(reg))
            new_concrete_successors.append(unsat_state)

        return new_concrete_successors
def _symbolically_back_traverse ( self , current_block , block_artifacts , cfg_node ) : class register_protector : def __init__ ( self , reg_offset , info_collection ) : """
                Class to overwrite registers.

                :param int reg_offest:       Register offset to overwrite from.
                :param dict info_collection: New register offsets to use (in container).
                """ self . _reg_offset = reg_offset self . _info_collection = info_collection def write_persistent_register ( self , state_ ) : """
                Writes over given registers from self._info_coollection (taken from block_artifacts)

                :param SimSuccessors state_: state to update registers for
                """ if state_ . inspect . address is None : l . error ( 'state.inspect.address is None. It will be fixed by Yan later.' ) return if state_ . registers . load ( self . _reg_offset ) . symbolic : current_run = state_ . inspect . address if current_run in self . _info_collection and not state_ . solver . symbolic ( self . _info_collection [ current_run ] [ self . _reg_offset ] ) : l . debug ( "Overwriting %s with %s" , state_ . registers . load ( self . _reg_offset ) , self . _info_collection [ current_run ] [ self . _reg_offset ] ) state_ . registers . store ( self . _reg_offset , self . _info_collection [ current_run ] [ self . _reg_offset ] ) l . debug ( "Start back traversal from %s" , current_block ) # Create a partial CFG first temp_cfg = networkx . DiGraph ( self . graph ) # Reverse it temp_cfg . reverse ( copy = False ) path_length = 0 concrete_exits = [ ] if cfg_node not in temp_cfg . nodes ( ) : # TODO: Figure out why this is happening return concrete_exits keep_running = True while not concrete_exits and path_length < 5 and keep_running : path_length += 1 queue = [ cfg_node ] avoid = set ( ) for _ in range ( path_length ) : new_queue = [ ] for n in queue : successors = list ( temp_cfg . successors ( n ) ) for suc in successors : jk = temp_cfg . get_edge_data ( n , suc ) [ 'jumpkind' ] if jk != 'Ijk_Ret' : # We don't want to trace into libraries predecessors = list ( temp_cfg . predecessors ( suc ) ) avoid |= { p . addr for p in predecessors if p is not n } new_queue . append ( suc ) queue = new_queue if path_length <= 1 : continue for n in queue : # Start symbolic exploration from each block state = self . project . factory . blank_state ( addr = n . addr , mode = 'symbolic' , add_options = { o . DO_RET_EMULATION , o . CONSERVATIVE_READ_STRATEGY , } | o . resilience ) # Avoid concretization of any symbolic read address that is over a certain limit # TODO: test case is needed for this option # Set initial values of persistent regs if n . addr in block_artifacts : for reg in state . arch . persistent_regs : state . registers . store ( reg , block_artifacts [ n . addr ] [ reg ] ) for reg in state . arch . persistent_regs : reg_protector = register_protector ( reg , block_artifacts ) state . inspect . add_breakpoint ( 'reg_write' , BP ( BP_AFTER , reg_write_offset = state . arch . registers [ reg ] [ 0 ] , action = reg_protector . write_persistent_register ) ) simgr = self . project . factory . simulation_manager ( state ) simgr . use_technique ( LoopSeer ( bound = 10 ) ) simgr . use_technique ( Explorer ( find = current_block . addr , avoid = avoid ) ) simgr . use_technique ( LengthLimiter ( path_length ) ) simgr . run ( ) if simgr . found : simgr = self . project . factory . simulation_manager ( simgr . one_found , save_unsat = True ) simgr . step ( ) if simgr . active or simgr . unsat : keep_running = False concrete_exits . extend ( simgr . active ) concrete_exits . extend ( simgr . unsat ) if keep_running : l . debug ( 'Step back for one more run...' ) # Make sure these successors are actually concrete # We just use the ip, persistent registers, and jumpkind to initialize the original unsat state # TODO: It works for jumptables, but not for calls. We should also handle changes in sp new_concrete_successors = [ ] for c in concrete_exits : unsat_state = current_block . unsat_successors [ 0 ] . copy ( ) unsat_state . history . jumpkind = c . history . jumpkind for reg in unsat_state . arch . persistent_regs + [ 'ip' ] : unsat_state . registers . store ( reg , c . registers . load ( reg ) ) new_concrete_successors . append ( unsat_state ) return new_concrete_successors