(self, cfgnode, sim_successors, current_function_addr)
return list(successing_addresses)
Try to resolve an indirect jump by slicing backwards
Try to resolve an indirect jump by slicing backwards
Try to resolve an indirect jump by slicing backwards
def _backward_slice_indirect(self, cfgnode, sim_successors, current_function_addr):
        """
        Try to resolve an indirect jump by slicing backwards
        """
        # TODO: make this a real indirect jump resolver under the new paradigm

        irsb = sim_successors.artifacts['irsb']  # shorthand

        l.debug("Resolving indirect jump at IRSB %s", irsb)

        # Let's slice backwards from the end of this exit
        next_tmp = irsb.next.tmp
        stmt_id = [i for i, s in enumerate(irsb.statements)
                   if isinstance(s, pyvex.IRStmt.WrTmp) and s.tmp == next_tmp][0]

        cdg = self.project.analyses.CDG(cfg=self, fail_fast=self._fail_fast)
        ddg = self.project.analyses.DDG(cfg=self, start=current_function_addr, call_depth=0, fail_fast=self._fail_fast)

        bc = self.project.analyses.BackwardSlice(self,
                                                 cdg,
                                                 ddg,
                                                 targets=[(cfgnode, stmt_id)],
                                                 same_function=True,
                                                 fail_fast=self._fail_fast)
        taint_graph = bc.taint_graph
        # Find the correct taint
        next_nodes = [cl for cl in taint_graph.nodes() if cl.block_addr == sim_successors.addr]

        if not next_nodes:
            l.error('The target exit is not included in the slice. Something is wrong')
            return []

        next_node = next_nodes[0]

        # Get the weakly-connected subgraph that contains `next_node`
        all_subgraphs = ( networkx.induced_subgraph(taint_graph, nodes) for nodes in networkx.weakly_connected_components(taint_graph))
        starts = set()
        for subgraph in all_subgraphs:
            if next_node in subgraph:
                # Make sure there is no symbolic read...
                # FIXME: This is an over-approximation. We should try to limit the starts more
                nodes = [n for n in subgraph.nodes() if subgraph.in_degree(n) == 0]
                for n in nodes:
                    starts.add(n.block_addr)

        # Execute the slice
        successing_addresses = set()
        annotated_cfg = bc.annotated_cfg()
        for start in starts:
            l.debug('Start symbolic execution at 0x%x on program slice.', start)
            # Get the state from our CFG
            node = self.get_any_node(start)
            if node is None:
                # Well, we have to live with an empty state
                base_state = self.project.factory.blank_state(addr=start)
            else:
                base_state = node.input_state.copy()
                base_state.set_mode('symbolic')
                base_state.ip = start

                # Clear all initial taints (register values, memory values, etc.)
                initial_nodes = [n for n in bc.taint_graph.nodes() if bc.taint_graph.in_degree(n) == 0]
                for cl in initial_nodes:
                    # Iterate in all actions of this node, and pick corresponding actions
                    cfg_nodes = self.get_all_nodes(cl.block_addr)
                    for n in cfg_nodes:
                        if not n.final_states:
                            continue
                        actions = [ac for ac in n.final_states[0].history.recent_actions
                                   # Normally it's enough to only use the first final state
                                   if ac.bbl_addr == cl.block_addr and
                                   ac.stmt_idx == cl.stmt_idx
                                   ]
                        for ac in actions:
                            if not hasattr(ac, 'action'):
                                continue
                            if ac.action == 'read':
                                if ac.type == 'mem':
                                    unconstrained_value = base_state.solver.Unconstrained('unconstrained',
                                                                                      ac.size.ast * 8)
                                    base_state.memory.store(ac.addr,
                                                            unconstrained_value,
                                                            endness=self.project.arch.memory_endness)
                                elif ac.type == 'reg':
                                    unconstrained_value = base_state.solver.Unconstrained('unconstrained',
                                                                                      ac.size.ast * 8)
                                    base_state.registers.store(ac.offset,
                                                               unconstrained_value,
                                                               endness=self.project.arch.register_endness)

                # Clear the constraints!
                base_state.release_plugin('solver')

            # For speed concerns, we are limiting the timeout for z3 solver to 5 seconds
            base_state.solver._solver.timeout = 5000

            sc = self.project.factory.simulation_manager(base_state)
            sc.use_technique(Slicecutor(annotated_cfg))
            sc.use_technique(LoopSeer(bound=1))
            sc.run()

            if sc.cut or sc.deadended:
                all_deadended_states = sc.cut + sc.deadended
                for s in all_deadended_states:
                    if s.addr == sim_successors.addr:
                        # We want to get its successors
                        succs = s.step()
                        for succ in succs.flat_successors:
                            successing_addresses.add(succ.addr)

            else:
                l.debug("Cannot determine the exit. You need some better ways to recover the exits :-(")

        l.debug('Resolution is done, and we have %d new successors.', len(successing_addresses))

        return list(successing_addresses)
def _backward_slice_indirect ( self , cfgnode , sim_successors , current_function_addr ) : # TODO: make this a real indirect jump resolver under the new paradigm irsb = sim_successors . artifacts [ 'irsb' ] # shorthand l . debug ( "Resolving indirect jump at IRSB %s" , irsb ) # Let's slice backwards from the end of this exit next_tmp = irsb . next . tmp stmt_id = [ i for i , s in enumerate ( irsb . statements ) if isinstance ( s , pyvex . IRStmt . WrTmp ) and s . tmp == next_tmp ] [ 0 ] cdg = self . project . analyses . CDG ( cfg = self , fail_fast = self . _fail_fast ) ddg = self . project . analyses . DDG ( cfg = self , start = current_function_addr , call_depth = 0 , fail_fast = self . _fail_fast ) bc = self . project . analyses . BackwardSlice ( self , cdg , ddg , targets = [ ( cfgnode , stmt_id ) ] , same_function = True , fail_fast = self . _fail_fast ) taint_graph = bc . taint_graph # Find the correct taint next_nodes = [ cl for cl in taint_graph . nodes ( ) if cl . block_addr == sim_successors . addr ] if not next_nodes : l . error ( 'The target exit is not included in the slice. Something is wrong' ) return [ ] next_node = next_nodes [ 0 ] # Get the weakly-connected subgraph that contains `next_node` all_subgraphs = ( networkx . induced_subgraph ( taint_graph , nodes ) for nodes in networkx . weakly_connected_components ( taint_graph ) ) starts = set ( ) for subgraph in all_subgraphs : if next_node in subgraph : # Make sure there is no symbolic read... # FIXME: This is an over-approximation. We should try to limit the starts more nodes = [ n for n in subgraph . nodes ( ) if subgraph . in_degree ( n ) == 0 ] for n in nodes : starts . add ( n . block_addr ) # Execute the slice successing_addresses = set ( ) annotated_cfg = bc . annotated_cfg ( ) for start in starts : l . debug ( 'Start symbolic execution at 0x%x on program slice.' , start ) # Get the state from our CFG node = self . get_any_node ( start ) if node is None : # Well, we have to live with an empty state base_state = self . project . factory . blank_state ( addr = start ) else : base_state = node . input_state . copy ( ) base_state . set_mode ( 'symbolic' ) base_state . ip = start # Clear all initial taints (register values, memory values, etc.) initial_nodes = [ n for n in bc . taint_graph . nodes ( ) if bc . taint_graph . in_degree ( n ) == 0 ] for cl in initial_nodes : # Iterate in all actions of this node, and pick corresponding actions cfg_nodes = self . get_all_nodes ( cl . block_addr ) for n in cfg_nodes : if not n . final_states : continue actions = [ ac for ac in n . final_states [ 0 ] . history . recent_actions # Normally it's enough to only use the first final state if ac . bbl_addr == cl . block_addr and ac . stmt_idx == cl . stmt_idx ] for ac in actions : if not hasattr ( ac , 'action' ) : continue if ac . action == 'read' : if ac . type == 'mem' : unconstrained_value = base_state . solver . Unconstrained ( 'unconstrained' , ac . size . ast * 8 ) base_state . memory . store ( ac . addr , unconstrained_value , endness = self . project . arch . memory_endness ) elif ac . type == 'reg' : unconstrained_value = base_state . solver . Unconstrained ( 'unconstrained' , ac . size . ast * 8 ) base_state . registers . store ( ac . offset , unconstrained_value , endness = self . project . arch . register_endness ) # Clear the constraints! base_state . release_plugin ( 'solver' ) # For speed concerns, we are limiting the timeout for z3 solver to 5 seconds base_state . solver . _solver . timeout = 5000 sc = self . project . factory . simulation_manager ( base_state ) sc . use_technique ( Slicecutor ( annotated_cfg ) ) sc . use_technique ( LoopSeer ( bound = 1 ) ) sc . run ( ) if sc . cut or sc . deadended : all_deadended_states = sc . cut + sc . deadended for s in all_deadended_states : if s . addr == sim_successors . addr : # We want to get its successors succs = s . step ( ) for succ in succs . flat_successors : successing_addresses . add ( succ . addr ) else : l . debug ( "Cannot determine the exit. You need some better ways to recover the exits :-(" ) l . debug ( 'Resolution is done, and we have %d new successors.' , len ( successing_addresses ) ) return list ( successing_addresses )